Trong bài viết trước, chúng ta đã khám phá cách mô phỏng mô hình sử dụng Skills của Deep Agent thông qua Deep Agents CLI. Ngày nay, LangChain đã hỗ trợ nguyên bản tính năng này, đơn giản hóa đáng kể quy trình phát triển. Bài viết này sẽ hướng dẫn bạn trải nghiệm sâu hơn về chức năng này, xây dựng một Trợ lý SQL thông minh hơn.

Khi xây dựng các AI Agent phức tạp, các nhà phát triển thường rơi vào tình thế tiến thoái lưỡng nan: có nên đưa tất cả ngữ cảnh (cấu trúc bảng cơ sở dữ liệu, tài liệu API, quy tắc nghiệp vụ) vào System Prompt một lần, dẫn đến tràn cửa sổ ngữ cảnh (Context Window) và phân tán sự chú ý của mô hình? Hay chọn tinh chỉnh thường xuyên (Fine-tuning) tốn kém?

Mô hình Skills (Skills Pattern) cung cấp một con đường trung gian thanh lịch. Nó đạt được hiệu quả sử dụng ngữ cảnh cao thông qua việc tải động kiến thức cần thiết. Sự hỗ trợ nguyên bản của LangChain cho mô hình này có nghĩa là chúng ta có thể dễ dàng xây dựng các Agent có khả năng "học theo yêu cầu".

Bài viết này sẽ kết hợp với tài liệu chính thức Build a SQL assistant with on-demand skills, hướng dẫn độc giả từ đầu, xây dựng một SQL Assistant hỗ trợ "tải kiến thức theo yêu cầu".

1. Khái niệm cốt lõi: Tại sao chọn mô hình Skills?

Hạn chế của SQL Agent truyền thống

Trong kiến trúc SQL Agent truyền thống, chúng ta thường cần cung cấp đầy đủ Database Schema trong System Prompt. Khi nghiệp vụ phát triển, khi số lượng bảng mở rộng lên hàng trăm, phương pháp này sẽ gây ra những vấn đề đáng kể:

Tiêu thụ Token lớn: Mỗi cuộc trò chuyện đều mang theo một lượng lớn cấu trúc bảng không liên quan, gây lãng phí tài nguyên.
Tăng rủi ro ảo giác: Quá nhiều thông tin gây nhiễu không liên quan sẽ làm giảm độ chính xác suy luận của mô hình.
Khó bảo trì: Kiến thức của tất cả các dòng nghiệp vụ được liên kết chặt chẽ, khó lặp lại độc lập.

Mô hình Skills: Giải pháp dựa trên tiết lộ tăng dần

Mô hình Skills dựa trên nguyên tắc tiết lộ tăng dần (Progressive Disclosure), xử lý quá trình thu thập kiến thức theo lớp:

Trạng thái ban đầu của Agent: Chỉ nắm vững những "kỹ năng" (Skills) nào và mô tả ngắn gọn của chúng (Description), giữ cho nó nhẹ nhàng.
Tải trong thời gian chạy: Khi đối mặt với một vấn đề cụ thể (ví dụ: "truy vấn hàng tồn kho"), Agent chủ động gọi công cụ (load_skill) để tải ngữ cảnh chi tiết của kỹ năng đó (Schema + Prompt).
Thực hiện nhiệm vụ: Dựa trên ngữ cảnh chính xác đã tải, thực hiện các nhiệm vụ cụ thể (ví dụ: viết và thực thi SQL).

Mô hình này hỗ trợ hiệu quả mở rộng vô hạn và phân tách nhóm, cho phép Agent thích ứng với các tình huống nghiệp vụ ngày càng phức tạp.

2. Thiết kế kiến trúc hệ thống

Trong dự án thực tế này, chúng ta sẽ xây dựng một SQL Assistant bao gồm hai Skills cốt lõi để trình diễn ứng dụng thực tế của mô hình này:

Sales Analytics (Phân tích bán hàng): Chịu trách nhiệm cho bảng sales_data, xử lý thống kê doanh thu, phân tích xu hướng đơn hàng, v.v.
Inventory Management (Quản lý kho): Chịu trách nhiệm cho bảng inventory_items, xử lý giám sát mức tồn kho, truy vấn vị trí, v.v.

3. Thiết lập môi trường phát triển

Dự án này sử dụng Python uv để quản lý phụ thuộc hiệu quả.

Cài đặt phụ thuộc cốt lõi

uv add langchain langchain-openai langgraph psycopg2-binary python-dotenv langchain-community

Cấu hình môi trường PostgreSQL

Khởi động một phiên bản Postgres cục bộ và tạo cơ sở dữ liệu agent_platform. Chúng tôi cung cấp tập lệnh setup_db.py để tự động khởi tạo cấu trúc bảng và dữ liệu thử nghiệm (xem mã nguồn ở cuối bài viết).

4. Giải thích chi tiết các bước triển khai cốt lõi### Bước 1: Định nghĩa kỹ năng lĩnh vực (The Knowledge)

Chúng ta sẽ định nghĩa kỹ năng dưới dạng cấu trúc từ điển, mô phỏng quá trình tải từ hệ thống tệp hoặc cơ sở dữ liệu. Xin lưu ý sự khác biệt giữa description (để Agent quyết định lựa chọn) và content (ngữ cảnh chi tiết thực tế được tải).

SKILLS = {"sales_analytics": {"description":"Hữu ích cho việc phân tích doanh thu bán hàng, xu hướng...","content":"""... Table Schema: sales_data ..."" },"inventory_management": {"description":"Hữu ích cho việc kiểm tra mức tồn kho...","content":"""... Table Schema: inventory_items ..."" }}

Bước 2: Triển khai các công cụ cốt lõi (The Capabilities)

Agent phụ thuộc vào hai công cụ quan trọng để hoàn thành nhiệm vụ:

load_skill(skill_name)**: Tải chi tiết của kỹ năng được chỉ định một cách động trong thời gian chạy. **
run_sql_query(query)**: Thực thi các câu lệnh SQL cụ thể. **

Bước 3: Sắp xếp logic Agent (The Brain)

Sử dụng LangGraph để xây dựng ReAct Agent. System Prompt đóng vai trò quan trọng ở đây, nó hướng dẫn Agent tuân thủ nghiêm ngặt quy trình làm việc tiêu chuẩn (SOP) Identify -> Load -> Query.

system_prompt ="""1. Identify the relevant skill.2. Use 'load_skill' to get schema.3. Write and execute SQL using 'run_sql_query'....Do not guess table names. Always load the skill first."""

5. Xác minh hiệu quả hoạt động

Bằng cách chạy test_agent.py, chúng tôi đã thử nghiệm các truy vấn trong hai lĩnh vực khác nhau là Sales và Inventory. Dưới đây là nhật ký đầu ra thực tế từ bảng điều khiển, hiển thị cách Agent tải kỹ năng một cách động dựa trên câu hỏi:

Testing Sales Query...Agent calling tools: [{'name': 'load_skill', 'args': {'skill_name': 'sales_analytics'}, 'id': 'call_f270d76b7ce4404cb5f61bf2', 'type': 'tool_call'}]Tool output:You are a Sales Analytics Expert.You have access to the 'sales_data' table.Table Schema:- id: integer...Agent calling tools: [{'name': 'run_sql_query', 'args': {'query': 'SELECT SUM(amount) as total_revenue FROM sales_data;'}, 'id': 'call_b4f3e686cc7f4f22b3bb9ea7', 'type': 'tool_call'}]Tool output: [(Decimal('730.50'),)]...Agent response: The total revenue is $730.50.Testing Inventory Query...Agent calling tools: [{'name': 'load_skill', 'args': {'skill_name': 'inventory_management'}, 'id': 'call_18c823b2d5064e95a0cfe2e3', 'type': 'tool_call'}]Tool output:You are an Inventory Management Expert.You have access to the 'inventory_items' table.Table Schema...Agent calling tools: [{'name': 'run_sql_query', 'args': {'query': "SELECT warehouse_location FROM inventory_items WHERE product_name = 'Laptop';"}, 'id': 'call_647ee3a444804bd98a045f00', 'type': 'tool_call'}]Tool output: [('Warehouse A',)]...Agent response: The Laptop is located in **Warehouse A**.## 6. Tham khảo mã nguồn hoàn chỉnh\n\nDưới đây là mã nguồn hoàn chỉnh của dự án, bao gồm script khởi tạo cơ sở dữ liệu và chương trình Agent chính.\n\n### 1. Khởi tạo cơ sở dữ liệu (setup_db.py)\n\n`importpsycopg2frompsycopg2.extensionsimportISOLATION_LEVEL_AUTOCOMMITimportosfromdotenvimportload_dotenvload_dotenv()# Vui lòng đảm bảo cấu hình thông tin kết nối cơ sở dữ liệu trong .envDB_HOST = os.getenv(### 2. Agent 主程序 (main.py)

`importosfromtypingimportAnnotated, Literal, TypedDict, Union, Dictfromdotenvimportload_dotenvfromlangchain_openaiimportChatOpenAIfromlangchain_core.toolsimporttoolfromlangchain_core.messagesimportSystemMessage, HumanMessage, AIMessage, ToolMessagefromlangchain_community.utilitiesimportSQLDatabasefromlangchain_community.agent_toolkitsimportSQLDatabaseToolkitfromlanggraph.graphimportStateGraph, START, END, MessagesStatefromlanggraph.prebuiltimportToolNode, tools_conditionload_dotenv()# --- Configuration ---BASE_URL = os.getenv("BASIC_MODEL_BASE_URL")API_KEY = os.getenv("BASIC_MODEL_API_KEY")MODEL_NAME = os.getenv("BASIC_MODEL_MODEL")DB_URI =f"postgresql://{os.getenv('DB_USER')}:{os.getenv('DB_PASSWORD')}@{os.getenv('DB_HOST')}:{os.getenv('DB_PORT')}/{os.getenv('DB_NAME')}"# --- Database Setup ---db = SQLDatabase.from_uri(DB_URI)# --- Skills Definition ---SKILLS: Dict[str, Dict[str, str]] = {"sales_analytics": {"description":"Useful for analyzing sales revenue, trends, and regional performance.","content":"""You are a Sales Analytics Expert.You have access to the 'sales_data' table.Table Schema:- id: integer (primary key)- transaction_date: date- product_id: varchar(50)- amount: decimal(10, 2)- region: varchar(50)Common queries:- Total revenue: SUM(amount)- Revenue by region: GROUP BY region- Sales trend: GROUP BY transaction_date""" },"inventory_management": {"description":"Useful for checking stock levels, product locations, and warehouse management.","content":"""You are an Inventory Management Expert.You have access to the 'inventory_item"Table Schema:

id: integer (primary key)
product_id: varchar(50)
product_name: varchar(100)
stock_count: integer
warehouse_location: varchar(50)

Common queries:

Check stock: WHERE product_name = '...'
Low stock: WHERE stock_count < threshold""" }}# --- Tools ---@tooldefload_skill(skill_name: str)-> str:""" Load the detailed prompt and schema for a specific skill. Available skills:
sales_analytics: For sales, revenue, and transaction analysis.
inventory_management: For stock, products, and warehouse queries. """ skill = SKILLS.get(skill_name) if not skill: return f"Error: Skill '{skill_name}' not found. Available skills: {list(SKILLS.keys())}" return skill["content"]

@tool def run_sql_query(query: str) -> str: """ Execute a SQL query against the database. Only use this tool AFTER loading the appropriate skill to understand the schema. """ try: return db.run(query) except Exception as e: return f"Error executing SQL: {e}"

@tool def list_tables() -> str: """List all available tables in the database.""" return str(db.get_usable_table_names())

tools = [load_skill, run_sql_query, list_tables]

--- Agent Setup ---

llm = ChatOpenAI( base_url=BASE_URL, api_key=API_KEY, model=MODEL_NAME, temperature=0 ) llm_with_tools = llm.bind_tools(tools)

--- Graph Definition ---

class AgentState(MessagesState):

We can add custom state if needed, but MessagesState is sufficient for simple chat

pass

def agent_node(state: AgentState): messages = state["messages"] response = llm_with_tools.invoke(messages) return {"messages": [response]}

workflow = StateGraph(AgentState) workflow.add_node("agent", agent_node)workflow.add_node("tools", ToolNode(tools))workflow.add_edge(START,"agent")workflow.add_conditional_edges("agent", tools_condition)workflow.add_edge("tools","agent")app = workflow.compile()# --- Main Execution ---if__name__ =="main": system_prompt ="""Bạn là một Trợ lý SQL hữu ích.Bạn có quyền truy cập vào các kỹ năng chuyên biệt chứa lược đồ cơ sở dữ liệu và kiến thức miền.Để trả lời câu hỏi của người dùng:1. Xác định kỹ năng liên quan (sales_analytics hoặc inventory_management).2. Sử dụng công cụ 'load_skill' để lấy lược đồ và hướng dẫn.3. Dựa trên kỹ năng đã tải, hãy viết và thực thi một truy vấn SQL bằng 'run_sql_query'.4. Trả lời câu hỏi của người dùng dựa trên kết quả truy vấn.Không đoán tên bảng. Luôn tải kỹ năng trước.""" print("SQL Assistant initialized. Type 'quit' to exit.") print("-"*50) messages = [SystemMessage(content=system_prompt)]# Pre-warm connection checktry: print(f"Connected to database:{DB_URI.split('@')[-1]}")exceptExceptionase: print(f"Database connection warning:{e}")whileTrue:try: user_input = input("User: ")ifuser_input.lower()in["quit","exit"]:break messages.append(HumanMessage(content=user_input))# Stream the execution print("Agent: ", end="", flush=True) final_response =Noneforeventinapp.stream({"messages": messages}, stream_mode="values"):# In 'values' mode, we get the full state. We just want to see the last message if it's new. last_message = event["messages"][-1]# Update our message history with the latest statepass# After stream finishes, the last state has the final answer final_state = app.invoke({"messages": messages}) last_msg = final_state["messages"][-1] if isinstance(last_msg, AIMessage): print(last_msg.content) messages = final_state["messages"]