用户提问 → Agent扫描系统提示中的"Skill菜单" → LLM 内部推理决定是否需要某个 Skill 的详细内容 → 是:调用 load_skill 工具,Skill 内容作为 ToolMessage 注入对话历史 → Agent基于完整上下文作答;否:直接基于通用知识作答

python
from typing import TypedDictclassSkill(TypedDict):"""可按需渐进式披露给Agent的Skill"""    name: str# Skill唯一标识    description: str# 1-2 句简介,展示在系统提示中    content: str# 完整Skill内容(只在加载时注入上下文)

python
from langchain.tools import tool@tooldefload_skill(skill_name: str) -> str:"""按需加载Skill的完整内容"""for skill in SKILLS:if skill["name"] == skill_name:returnf"Loaded skill: {skill_name}\n\n{skill['content']}"    available = ", ".join(s["name"] for s in SKILLS)returnf"Skill '{skill_name}' not found. Available skills: {available}"

python
from langchain.agents.middleware import (    AgentMiddleware, ModelRequest, ModelResponse)from langchain.messages import SystemMessageclassSkillMiddleware(AgentMiddleware):"""将Skill描述注入系统提示的中间件"""    tools = [load_skill]   # 类变量注册工具defwrap_model_call(self, request, handler):        skills_addendum = (f"\n\n## Available Skills\n\n{self.skills_prompt}\n\n""Use the load_skill tool when you need detailed information ""about handling a specific type of request."        )        new_content = list(request.system_message.content_blocks) + [            {"type": "text", "text": skills_addendum}        ]        new_system = SystemMessage(content=new_content)return handler(request.override(system_message=new_system))

python
from typing import NotRequiredfrom langgraph.types import Commandfrom langchain.agents.middleware import AgentStateclassCustomState(AgentState):    skills_loaded: NotRequired[list[str]]   # 已加载的Skill名@tooldefload_skill(skill_name: str, runtime) -> Command:# 查找Skill逻辑同上return Command(update={"messages": [ToolMessage(content=...)],"skills_loaded": [skill_name],   # 累积到状态字段    })

bash
pip install -U "langchain[openai]"# 或 uv add / conda 等价命令export OPENAI_API_KEY="sk-..."

python
from langchain.chat_models import init_chat_modelmodel = init_chat_model("openai:gpt-4o")# 也支持: anthropic / azure_openai / google_genai / bedrock / openrouter / huggingface ...

python
SKILLS: list[Skill] = [    {"name": "sales_analytics","description": "销售分析的数据库 schema 与业务规则,涵盖 customers、orders、revenue。","content": """# Sales Analytics Schema## Tables### customers(关键字段)- customer_id (PRIMARY KEY)- customer_tier (bronze/silver/gold/platinum)### orders(关键字段)- order_id, customer_id (FK), order_date- status (pending/completed/cancelled/refunded)- total_amount, sales_region### order_items- item_id, order_id, product_id, quantity, unit_price## Business Logic- **Active customers**:status = 'active' AND signup_date <= CURRENT_DATE - INTERVAL '90 days'- **Revenue 仅算 completed 订单**- **High-value orders**:total_amount > 1000## Example QuerySELECT c.customer_id, c.name, SUM(o.total_amount) ...;""",    },    {"name": "inventory_management","description": "库存管理的数据库 schema 与业务规则,涵盖 products、warehouses、stock levels。","content": """# Inventory Management Schema## Tables### products(关键字段)- product_id, product_name, sku, category- reorder_point, discontinued### warehouses- warehouse_id, warehouse_name, location, capacity### inventory- inventory_id, product_id, warehouse_id, quantity_on_hand## Business Logic- **库存预警**:SUM(quantity_on_hand) <= reorder_point- **停售商品默认排除**(discontinued = false)## Example QuerySELECT p.product_id, p.product_name, SUM(i.quantity_on_hand) AS total_stock ...HAVING total_stock <= p.reorder_point;""",    },]

python
@tooldefload_skill(skill_name: str) -> str:"""按需加载Skill的完整内容..."""for skill in SKILLS:if skill["name"] == skill_name:returnf"Loaded skill: {skill_name}\n\n{skill['content']}"    available = ", ".join(s["name"] for s in SKILLS)returnf"Skill '{skill_name}' not found. Available skills: {available}"classSkillMiddleware(AgentMiddleware):    tools = [load_skill]def__init__(self):self.skills_prompt = "\n".join(f"- **{s['name']}**: {s['description']}"for s in SKILLS        )defwrap_model_call(self, request, handler):        addendum = (f"\n\n## Available Skills\n\n{self.skills_prompt}\n\n""Use the load_skill tool when you need detailed information ""about handling a specific type of request."        )        new_content = list(request.system_message.content_blocks) + [            {"type": "text", "text": addendum}        ]        new_system = SystemMessage(content=new_content)return handler(request.override(system_message=new_system))

python
from langchain.agents import create_agentfrom langgraph.checkpoint.memory import InMemorySaveragent = create_agent(    model,    system_prompt=("你是一名 SQL 查询助手,帮助用户针对业务数据库编写合规 SQL。"    ),    middleware=[SkillMiddleware()],    checkpointer=InMemorySaver(),)

python
from langchain_core.utils.uuid import uuid7thread_id = str(uuid7())   # 每次会话唯一config = {"configurable": {"thread_id": thread_id}}result = agent.invoke(    {"messages": [{"role": "user","content": "编写 SQL:查找上个月下过超过 1000 元订单的客户"    }]},    config,)for msg in result["messages"]:ifhasattr(msg, 'pretty_print'):        msg.pretty_print()else:print(f"{msg.type}: {msg.content}")

================================ Human Message =================================编写 SQL:查找上个月下过超过 1000 元订单的客户================================== Ai Message ==================================Tool Calls:  load_skill (call_abc123)    Call ID: call_abc123     Args:      skill_name: sales_analytics================================= Tool Message =================================Name: load_skillLoaded skill: sales_analytics# Sales Analytics Schema[... 完整 Schema 与业务规则被注入,详细见第 5.2 节 ...]## Business Logic- **Revenue 仅算 completed 订单**- **High-value orders**:total_amount > 1000================================== Ai Message ==================================以下是符合业务规则的 SQL:```sql-- 查找上个月所有单笔订单 > 1000 元的客户(去重)SELECT DISTINCT    c.customer_id,    c.name,    c.email,    c.customer_tierFROM customers cJOIN orders o ON c.customer_id = o.customer_idWHERE o.total_amount > 1000  AND o.status = 'completed'           -- 仅计完成订单(对应 Revenue 规则)  AND o.order_date >= CURRENT_DATE - INTERVAL '1 month'ORDER BY c.customer_id;

bash
export OPENAI_API_KEY="sk-..."python skills_sql_agent.py

python
"""Skill 模式 SQL 助手示例(production-ready 注解版)本示例演示多智能体系统中的"Skill(Skills)"模式 —— 渐进式披露:  1. Agent 系统提示中仅含 Skill 简介(~50 token);  2. 收到用户提问后,Agent 决定加载哪些 Skill(通过工具调用);  3. Skill 完整内容作为 ToolMessage 注入对话历史;  4. Agent 基于完整 Schema + 业务规则生成合规 SQL。应用场景对比:  - 路由器: 适合"查询多个数据源并并行汇总"  - Skill:   适合"知识库极大,需按业务域按需加载"  - 子Agent:  适合"任务可分解为多个独立子流程"  - 交接:    适合"流程图能画成状态机""""import osfrom typing import TypedDictfrom langchain.agents import create_agentfrom langchain.agents.middleware import AgentMiddlewarefrom langchain.chat_models import init_chat_modelfrom langchain.tools import toolfrom langgraph.checkpoint.memory import InMemorySaver# ======================== Skill 数据结构 ========================classSkill(TypedDict):"""单一Skill的完整定义:名称 + 简介 + 详细内容"""    name: str    description: str    content: str# ======================== 业务Skill定义 ========================SKILLS: list[Skill] = [    {"name": "sales_analytics","description": "销售分析:customers、orders、revenue 相关查询","content": """# Sales Analytics Schema### customers- customer_id (PK), name, email, customer_tier (bronze/silver/gold/platinum)- signup_date, region### orders- order_id (PK), customer_id (FK), order_date- status (pending/completed/cancelled/refunded), total_amount### order_items- item_id (PK), order_id (FK), product_id, quantity, unit_price## 业务规则- 活跃客户:status = 'active' AND signup_date <= CURRENT_DATE - INTERVAL '90 days'- 营收:仅 sum(total_amount) WHERE status = 'completed'- 高价值订单:total_amount > 1000""",    },    {"name": "inventory_management","description": "库存管理:products、warehouses、stock levels","content": """# Inventory Management Schema### products- product_id (PK), product_name, sku, category, reorder_point, discontinued### warehouses- warehouse_id (PK), warehouse_name, location, capacity### inventory- inventory_id (PK), product_id (FK), warehouse_id (FK), quantity_on_hand## 业务规则- 补货预警:SUM(quantity_on_hand) <= reorder_point- 默认排除 discontinued 商品""",    },]# ======================== load_skill 工具 ========================@tooldefload_skill(skill_name: str) -> str:"""按需加载Skill的完整内容(返回字符串注入 ToolMessage)"""for skill in SKILLS:if skill["name"] == skill_name:returnf"Loaded skill: {skill_name}\n\n{skill['content']}"    available = ", ".join(s["name"] for s in SKILLS)returnf"Skill '{skill_name}' not found. Available skills: {available}"# ======================== SkillMiddleware ========================classSkillMiddleware(AgentMiddleware):"""将Skill菜单拼接到系统消息末尾"""    tools = [load_skill]   # 类变量注册工具(被 LangChain 框架发现)def__init__(self):self.skills_prompt = "\n".join(f"- **{s['name']}**: {s['description']}"for s in SKILLS        )defwrap_model_call(self, request, handler):        addendum = (f"\n\n## Available Skills\n\n{self.skills_prompt}\n\n""Use load_skill when you need detailed information ""about a specific business area."        )        new_content = list(request.system_message.content_blocks) + [            {"type": "text", "text": addendum}        ]from langchain.messages import SystemMessage        new_system = SystemMessage(content=new_content)return handler(request.override(system_message=new_system))# ======================== Agent 创建 ========================os.environ.setdefault("OPENAI_API_KEY", "sk-...")   # 真实使用时替换为你的 Keymodel = init_chat_model("openai:gpt-4o")agent = create_agent(    model,    system_prompt="你是一名 SQL 助手,帮助用户编写业务合规的 SQL。",    middleware=[SkillMiddleware()],    checkpointer=InMemorySaver(),)# ======================== 调用测试 ========================if __name__ == "__main__":from langchain_core.utils.uuid import uuid7    config = {"configurable": {"thread_id": str(uuid7())}}    result = agent.invoke(        {"messages": [{"role": "user","content": "编写 SQL:查找上个月所有订单 > 1000 的客户",        }]},        config,    )for msg in result["messages"]:ifhasattr(msg, "pretty_print"):            msg.pretty_print()else:print(f"{msg.type}: {msg.content}")