11 实战：用 Skills 搭一个 SQL 助手（按需加载「查表手册」）

1
from typing import TypedDict
2

3
class Skill(TypedDict):
4
    """一个可以按需披露给 agent 的技能。"""
5
    name: str         # 技能唯一标识，load_skill 靠它点名
6
    description: str  # 1-2 句简介，写进系统提示词给模型看
7
    content: str      # 技能全文（表结构 + 业务规则），按需加载

1
SKILLS: list[Skill] = [
2
    {
3
        "name": "sales_analytics",
4
        "description": "销售数据分析（客户、订单、营收）的表结构与业务逻辑。",
5
        "content": """# 销售分析表结构
6
### customers
7
- customer_id (PRIMARY KEY)
8
# ...（略：name / signup_date / status / customer_tier）
9
### orders
10
- order_id / customer_id / order_date
11
- status (pending/completed/cancelled/refunded)
12
- total_amount / sales_region
13
# ...（略：order_items 表、业务规则、示例 SQL）
14
""",
15
    },
16
    {"name": "inventory_management", "description": "库存管理…（略）", "content": "# ...（略）"},
17
]

1
## 表结构 Tables
2
customers:   customer_id / name / email / signup_date / status / customer_tier
3
orders:      order_id / customer_id / order_date / status / total_amount / sales_region
4
order_items: item_id / order_id / product_id / quantity / unit_price / discount_percent
5

6
## 业务规则 Business Logic
7
活跃客户   = status = 'active' AND signup_date <= CURRENT_DATE - INTERVAL '90 days'
8
营收口径   = 只算 status = 'completed' 的订单；total_amount 已含折扣
9
客户生命周期价值 CLV = 该客户所有已完成订单金额之和
10
高价值订单 = total_amount > 1000
11

12
## 示例查询 Example Query（一条"上季度营收 Top 10 客户"的可照抄 SQL）

1
from langchain.tools import tool
2

3
@tool
4
def load_skill(skill_name: str) -> str:
5
    """把某个技能的完整内容加载进 agent 的上下文。
6

7
    当你需要处理某类请求的详细信息时使用。
8

9
    Args:
10
        skill_name: 要加载的技能名
11
    """
12
    for skill in SKILLS:
13
        if skill["name"] == skill_name:
14
            return f"Loaded skill: {skill_name}\n\n{skill['content']}"
15

16
    # 没找到：把可选技能列出来，让模型自己纠正
17
    available = ", ".join(s["name"] for s in SKILLS)
18
    return f"Skill '{skill_name}' not found. Available skills: {available}"

1
from typing import Callable
2
from langchain.agents.middleware import ModelRequest, ModelResponse, AgentMiddleware
3
from langchain.messages import SystemMessage
4

5
class SkillMiddleware(AgentMiddleware):
6
    """把技能目录注入系统提示词的中间件。"""
7

8
    tools = [load_skill]  # 类变量注册工具，agent 就能看到 load_skill
9

10
    def __init__(self):
11
        # 由 SKILLS 生成"目录"文本：- **技能名**: 一句话简介
12
        self.skills_prompt = "\n".join(
13
            f"- **{s['name']}**: {s['description']}" for s in SKILLS
14
        )
15

16
    def wrap_model_call(
17
        self, request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse]
18
    ) -> ModelResponse:
19
        """同步路径：把技能目录追加到系统提示词。"""
20
        addendum = (
21
            f"\n\n## Available Skills\n\n{self.skills_prompt}\n\n"
22
            "当你需要某类请求的详细信息时，调用 load_skill 工具。"
23
        )
24
        # 注意是"追加内容块"，不是覆盖原来的系统提示
25
        new_content = list(request.system_message.content_blocks) + [
26
            {"type": "text", "text": addendum}
27
        ]
28
        modified = request.override(system_message=SystemMessage(content=new_content))
29
        return handler(modified)

1
from langchain.agents import create_agent
2
from langgraph.checkpoint.memory import InMemorySaver
3

4
agent = create_agent(
5
    model,
6
    system_prompt=(
7
        "你是一个 SQL 查询助手，帮用户针对业务数据库写查询。"
8
    ),
9
    middleware=[SkillMiddleware()],   # 关键：挂上技能中间件
10
    checkpointer=InMemorySaver(),     # 记住对话，多轮才有上下文
11
)

1
from langchain_core.utils.uuid import uuid7
2

3
config = {"configurable": {"thread_id": str(uuid7())}}
4
result = agent.invoke(
5
    {"messages": [{"role": "user", "content": (
6
        "写一条 SQL，找出上个月下单金额超过 $1000 的所有客户"
7
    )}]},
8
    config,
9
)
10

11
for message in result["messages"]:
12
    message.pretty_print()   # 逐条打印：Human / AI(工具调用) / Tool / AI(答案)

1
================= Ai Message =================
2
Tool Calls:
3
  load_skill (call_abc123)   Args: skill_name: sales_analytics
4
================= Tool Message =================
5
Loaded skill: sales_analytics
6
...（技能全文：表结构 + 业务规则）...
7
================= Ai Message =================
8
SELECT DISTINCT c.customer_id, c.name, c.email, c.customer_tier
9
FROM customers c JOIN orders o ON c.customer_id = o.customer_id
10
WHERE o.total_amount > 1000
11
  AND o.status = 'completed'                      -- 出自技能里的业务规则
12
  AND o.order_date >= CURRENT_DATE - INTERVAL '1 month'

1
from typing import NotRequired
2
from langchain.agents.middleware import AgentState
3

4
class CustomState(AgentState):
5
    skills_loaded: NotRequired[list[str]]   # 记录已加载的技能名

1
from langgraph.types import Command
2
from langchain.tools import tool, ToolRuntime
3
from langchain.messages import ToolMessage
4

5
@tool
6
def load_skill(skill_name: str, runtime: ToolRuntime) -> Command:
7
    """把某个技能的完整内容加载进 agent 的上下文。"""
8
    for skill in SKILLS:
9
        if skill["name"] == skill_name:
10
            return Command(update={
11
                "messages": [ToolMessage(
12
                    content=f"Loaded skill: {skill_name}\n\n{skill['content']}",
13
                    tool_call_id=runtime.tool_call_id,
14
                )],
15
                "skills_loaded": [skill_name],   # 记一笔账
16
            })
17
    # 没找到同样要回一条 ToolMessage，否则工具调用没有结果
18
    return Command(update={"messages": [ToolMessage(
19
        content=f"Skill '{skill_name}' not found.",
20
        tool_call_id=runtime.tool_call_id)]})

1
@tool
2
def write_sql_query(query: str, vertical: str, runtime: ToolRuntime) -> str:
3
    """为某个业务线写入并校验 SQL 查询。必须先加载对应技能。"""
4
    skills_loaded = runtime.state.get("skills_loaded", [])
5
    if vertical not in skills_loaded:
6
        return (
7
            f"Error: 你必须先加载 '{vertical}' 技能，"
8
            f"才能了解表结构再写查询。请调用 load_skill('{vertical}')。"
9
        )
10
    return f"SQL Query for {vertical}:\n\n```sql\n{query}\n```\n\n已通过校验"

1
class SkillMiddleware(AgentMiddleware[CustomState]):
2
    state_schema = CustomState                  # 声明扩展后的状态
3
    tools = [load_skill, write_sql_query]       # 暴露两个工具
4
    # ...其余实现不变

1
import re
2
ALLOWED_TABLES = {"customers", "orders", "order_items"}   # 表白名单
3

4
def guard(query: str) -> str | None:
5
    """返回错误文案 = 拦截；返回 None = 放行。"""
6
    if re.search(r"\b(insert|update|delete|drop|alter|truncate)\b", query, re.I):
7
        return "Error: 只允许只读查询（SELECT）。"
8
    used = set(re.findall(r"\b(?:from|join)\s+([A-Za-z_]\w*)", query, re.I))
9
    if used - ALLOWED_TABLES:
10
        return f"Error: 不允许访问这些表：{sorted(used - ALLOWED_TABLES)}"
11
    return None