使用按需技能构建 SQL 助手

本教程展示如何使用渐进式披露——一种上下文管理技术，智能体按需加载信息而非预先加载——来实现技能（基于提示的专门指令）。智能体通过工具调用加载技能，而不是动态更改系统提示，仅发现并加载每个任务所需的技能。 用例： 想象构建一个智能体，帮助在大型企业的不同业务垂直领域编写 SQL 查询。您的组织可能为每个垂直领域拥有独立的数据存储，或者一个包含数千张表的单一整体数据库。无论哪种情况，预先加载所有模式都会使上下文窗口不堪重负。渐进式披露通过仅在需要时加载相关模式来解决此问题。这种架构还允许不同的产品负责人和利益相关者独立贡献和维护其特定业务垂直领域的技能。 您将构建什么： 一个具有两个技能（销售分析和库存管理）的 SQL 查询助手。智能体在其系统提示中看到轻量级的技能描述，然后仅在与用户查询相关时，通过工具调用加载完整的数据库模式和业务逻辑。

有关具有查询执行、错误纠正和验证功能的完整 SQL 智能体示例，请参阅我们的 SQL 智能体教程。本教程重点介绍渐进式披露模式，该模式可应用于任何领域。

渐进式披露由 Anthropic 推广，是一种构建可扩展智能体技能系统的技术。该方法使用三级架构（元数据 → 核心内容 → 详细资源），智能体仅按需加载信息。有关此技术的更多信息，请参阅使用智能体技能为现实世界装备智能体。

工作原理

以下是用户请求 SQL 查询时的流程： 为什么使用渐进式披露：

减少上下文使用 - 仅加载任务所需的 2-3 个技能，而非所有可用技能
实现团队自主性 - 不同团队可以独立开发专门技能（类似于其他多智能体架构）
高效扩展 - 添加数十或数百个技能而不会使上下文不堪重负
简化对话历史 - 具有单一对话线程的单一智能体

什么是技能： 技能，如 Claude Code 所推广的，主要是基于提示的：针对特定业务任务的自包含的专门指令单元。在 Claude Code 中，技能以文件系统上的目录和文件形式暴露，通过文件操作发现。技能通过提示指导行为，并可以提供有关工具使用的信息，或包含供编码智能体执行的示例代码。

具有渐进式披露的技能可以被视为一种 RAG（检索增强生成）形式，其中每个技能都是一个检索单元——尽管不一定由嵌入或关键词搜索支持，而是由浏览内容的工具支持（如文件操作，或在本教程中，直接查找）。

权衡：

延迟：按需加载技能需要额外的工具调用，这会增加需要每个技能的第一个请求的延迟
工作流控制：基本实现依赖于提示来指导技能使用 - 您无法在没有自定义逻辑的情况下强制执行硬约束，例如“始终在技能 B 之前尝试技能 A”

实现您自己的技能系统在构建您自己的技能实现时（正如我们在本教程中所做的），核心概念是渐进式披露 - 按需加载信息。除此之外，您在实现方面拥有完全的灵活性：

存储：数据库、S3、内存数据结构或任何后端
发现：直接查找（本教程）、用于大型技能集合的 RAG、文件系统扫描或 API 调用
加载逻辑：自定义延迟特性，并添加逻辑以搜索技能内容或对相关性进行排序
副作用：定义技能加载时发生的情况，例如暴露与该技能关联的工具（在第 8 节中介绍）

这种灵活性使您可以针对性能、存储和工作流控制的特定要求进行优化。

设置

安装

本教程需要 langchain 包：

pip install langchain

有关更多详细信息，请参阅我们的安装指南。

LangSmith

设置 LangSmith 以检查智能体内部发生的情况。然后设置以下环境变量：

export LANGSMITH_TRACING="true"
export LANGSMITH_API_KEY="..."

选择 LLM

从 LangChain 的集成套件中选择一个聊天模型：

👉 阅读 OpenAI 聊天模型集成文档

pip install -U "langchain[openai]"

import os
from langchain.chat_models import init_chat_model

os.environ["OPENAI_API_KEY"] = "sk-..."

model = init_chat_model("gpt-5.4")

👉 阅读 Anthropic 聊天模型集成文档

pip install -U "langchain[anthropic]"

import os
from langchain.chat_models import init_chat_model

os.environ["ANTHROPIC_API_KEY"] = "sk-..."

model = init_chat_model("claude-sonnet-4-6")

👉 阅读 Azure 聊天模型集成文档

pip install -U "langchain[openai]"

import os
from langchain.chat_models import init_chat_model

os.environ["AZURE_OPENAI_API_KEY"] = "..."
os.environ["AZURE_OPENAI_ENDPOINT"] = "..."
os.environ["OPENAI_API_VERSION"] = "2025-03-01-preview"

model = init_chat_model(
    "azure_openai:gpt-5.4",
    azure_deployment=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"],
)

👉 阅读 Google GenAI 聊天模型集成文档

pip install -U "langchain[google-genai]"

import os
from langchain.chat_models import init_chat_model

os.environ["GOOGLE_API_KEY"] = "..."

model = init_chat_model("google_genai:gemini-2.5-flash-lite")

👉 阅读 AWS Bedrock 聊天模型集成文档

pip install -U "langchain[aws]"

from langchain.chat_models import init_chat_model

# 按照此处步骤配置您的凭证：
# https://docs.aws.amazon.com/bedrock/latest/userguide/getting-started.html

model = init_chat_model(
    "anthropic.claude-3-5-sonnet-20240620-v1:0",
    model_provider="bedrock_converse",
)

👉 阅读 HuggingFace 聊天模型集成文档

pip install -U "langchain[huggingface]"

import os
from langchain.chat_models import init_chat_model

os.environ["HUGGINGFACEHUB_API_TOKEN"] = "hf_..."

model = init_chat_model(
    "microsoft/Phi-3-mini-4k-instruct",
    model_provider="huggingface",
    temperature=0.7,
    max_tokens=1024,
)

👉 阅读 OpenRouter 聊天模型集成文档

pip install -U "langchain-openrouter"

import os
from langchain.chat_models import init_chat_model

os.environ["OPENROUTER_API_KEY"] = "sk-..."

model = init_chat_model(
    "auto",
    model_provider="openrouter",
)

1. 定义技能

首先，定义技能的结构。每个技能都有一个名称、一个简短描述（显示在系统提示中）和完整内容（按需加载）：

from typing import TypedDict

class Skill(TypedDict):
    """可以渐进式披露给智能体的技能。"""
    name: str  # 技能的唯一标识符
    description: str  # 显示在系统提示中的 1-2 句描述
    content: str  # 包含详细说明的完整技能内容

现在为 SQL 查询助手定义示例技能。这些技能设计为描述轻量级（预先显示给智能体）但内容详细（仅在需要时加载）：

查看完整的技能定义

SKILLS: list[Skill] = [
    {
        "name": "sales_analytics",
        "description": "用于销售数据分析的数据库模式和业务逻辑，包括客户、订单和收入。",
        "content": """# 销售分析模式

## 表

### customers
- customer_id (主键)
- name
- email
- signup_date
- status (active/inactive)
- customer_tier (bronze/silver/gold/platinum)

### orders
- order_id (主键)
- customer_id (外键 -> customers)
- order_date
- status (pending/completed/cancelled/refunded)
- total_amount
- sales_region (north/south/east/west)

### order_items
- item_id (主键)
- order_id (外键 -> orders)
- product_id
- quantity
- unit_price
- discount_percent

## 业务逻辑

**活跃客户**：status = 'active' AND signup_date <= CURRENT_DATE - INTERVAL '90 days'

**收入计算**：仅计算 status = 'completed' 的订单。使用 orders 表中的 total_amount，该字段已考虑折扣。

**客户生命周期价值 (CLV)**：客户所有已完成订单金额的总和。

**高价值订单**：total_amount > 1000 的订单

## 示例查询

-- 获取上个季度收入最高的前 10 名客户
SELECT
    c.customer_id,
    c.name,
    c.customer_tier,
    SUM(o.total_amount) as total_revenue
FROM customers c
JOIN orders o ON c.customer_id = o.customer_id
WHERE o.status = 'completed'
  AND o.order_date >= CURRENT_DATE - INTERVAL '3 months'
GROUP BY c.customer_id, c.name, c.customer_tier
ORDER BY total_revenue DESC
LIMIT 10;
""",
    },
    {
        "name": "inventory_management",
        "description": "用于库存跟踪的数据库模式和业务逻辑，包括产品、仓库和库存水平。",
        "content": """# 库存管理模式

## 表

### products
- product_id (主键)
- product_name
- sku
- category
- unit_cost
- reorder_point (重新订购前的最低库存水平)
- discontinued (布尔值)

### warehouses
- warehouse_id (主键)
- warehouse_name
- location
- capacity

### inventory
- inventory_id (主键)
- product_id (外键 -> products)
- warehouse_id (外键 -> warehouses)
- quantity_on_hand
- last_updated

### stock_movements
- movement_id (主键)
- product_id (外键 -> products)
- warehouse_id (外键 -> warehouses)
- movement_type (inbound/outbound/transfer/adjustment)
- quantity (入站为正，出站为负)
- movement_date
- reference_number

## 业务逻辑

**可用库存**：inventory 表中 quantity_on_hand > 0 的 quantity_on_hand

**需要重新订购的产品**：所有仓库中 quantity_on_hand 总和小于或等于产品 reorder_point 的产品

**仅限活跃产品**：除非专门分析已停产产品，否则排除 discontinued = true 的产品

**库存估值**：每个产品的 quantity_on_hand * unit_cost

## 示例查询

-- 查找所有仓库中低于重新订购点的产品
SELECT
    p.product_id,
    p.product_name,
    p.reorder_point,
    SUM(i.quantity_on_hand) as total_stock,
    p.unit_cost,
    (p.reorder_point - SUM(i.quantity_on_hand)) as units_to_reorder
FROM products p
JOIN inventory i ON p.product_id = i.product_id
WHERE p.discontinued = false
GROUP BY p.product_id, p.product_name, p.reorder_point, p.unit_cost
HAVING SUM(i.quantity_on_hand) <= p.reorder_point
ORDER BY units_to_reorder DESC;
""",
    },
]

2. 创建技能加载工具

创建一个工具，用于按需加载完整的技能内容：

from langchain.tools import tool

@tool
def load_skill(skill_name: str) -> str:
    """将技能的完整内容加载到智能体的上下文中。

    当您需要有关如何处理特定类型请求的详细信息时使用此工具。这将为您提供有关该技能领域的全面说明、策略和指南。

    Args:
        skill_name: 要加载的技能的名称（例如 "expense_reporting"、"travel_booking"）
    """
    # 查找并返回请求的技能
    for skill in SKILLS:
        if skill["name"] == skill_name:
            return f"已加载技能：{skill_name}\n\n{skill['content']}"

    # 未找到技能
    available = ", ".join(s["name"] for s in SKILLS)
    return f"未找到技能 '{skill_name}'。可用技能：{available}"

load_skill 工具将完整的技能内容作为字符串返回，该字符串成为 ToolMessage 对话的一部分。有关创建和使用工具的更多详细信息，请参阅工具指南。

3. 构建技能中间件

创建自定义中间件，将技能描述注入系统提示。此中间件使技能可被发现，而无需预先加载其完整内容。

本指南演示了如何创建自定义中间件。有关中间件概念和模式的全面指南，请参阅自定义中间件文档。

from langchain.agents.middleware import ModelRequest, ModelResponse, AgentMiddleware
from langchain.messages import SystemMessage
from typing import Callable

class SkillMiddleware(AgentMiddleware):
    """将技能描述注入系统提示的中间件。"""

    # 将 load_skill 工具注册为类变量
    tools = [load_skill]

    def __init__(self):
        """初始化并从 SKILLS 生成技能提示。"""
        # 从 SKILLS 列表构建技能提示
        skills_list = []
        for skill in SKILLS:
            skills_list.append(
                f"- **{skill['name']}**: {skill['description']}"
            )
        self.skills_prompt = "\n".join(skills_list)

    def wrap_model_call(
        self,
        request: ModelRequest,
        handler: Callable[[ModelRequest], ModelResponse],
    ) -> ModelResponse:
        """同步：将技能描述注入系统提示。"""
        # 构建技能附录
        skills_addendum = (
            f"\n\n## 可用技能\n\n{self.skills_prompt}\n\n"
            "当您需要有关处理特定类型请求的详细信息时，"
            "请使用 load_skill 工具。"
        )

        # 追加到系统消息内容块
        new_content = list(request.system_message.content_blocks) + [
            {"type": "text", "text": skills_addendum}
        ]
        new_system_message = SystemMessage(content=new_content)
        modified_request = request.override(system_message=new_system_message)
        return handler(modified_request)

中间件将技能描述追加到系统提示，使智能体能够感知可用技能，而无需加载其完整内容。load_skill 工具注册为类变量，使其可供智能体使用。

生产环境考虑：本教程为简单起见在 __init__ 中加载技能列表。在生产系统中，您可能希望在 before_agent 钩子中加载技能，以便定期刷新它们以反映最新更改（例如，添加新技能或修改现有技能时）。有关详细信息，请参阅 before_agent 钩子文档。

4. 创建支持技能的智能体

现在创建具有技能中间件和用于状态持久化的检查点的智能体：

from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver

# 创建支持技能的智能体
agent = create_agent(
    model,
    system_prompt=(
        "您是一个 SQL 查询助手，帮助用户"
        "针对业务数据库编写查询。"
    ),
    middleware=[SkillMiddleware()],
    checkpointer=InMemorySaver(),
)

智能体现在可以在其系统提示中访问技能描述，并可以在需要时调用 load_skill 来检索完整的技能内容。检查点维护跨轮次的对话历史。

5. 测试渐进式披露

使用需要特定技能知识的问题测试智能体：

from langchain_core.utils.uuid import uuid7

# 此对话线程的配置
thread_id = str(uuid7())
config = {"configurable": {"thread_id": thread_id}}

# 请求 SQL 查询
result = agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": (
                    "编写一个 SQL 查询，查找上个月所有"
                    "订单金额超过 1000 美元的客户"
                ),
            }
        ]
    },
    config
)

# 打印对话
for message in result["messages"]:
    if hasattr(message, 'pretty_print'):
        message.pretty_print()
    else:
        print(f"{message.type}: {message.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_skill

已加载技能：sales_analytics

# 销售分析模式

## 表

### customers
- customer_id (主键)
- name
- email
- signup_date
- status (active/inactive)
- customer_tier (bronze/silver/gold/platinum)

### orders
- order_id (主键)
- customer_id (外键 -> customers)
- order_date
- status (pending/completed/cancelled/refunded)
- total_amount
- sales_region (north/south/east/west)

[... 模式的其余部分 ...]

## 业务逻辑

**高价值订单**：`total_amount > 1000` 的订单
**收入计算**：仅计算 `status = 'completed'` 的订单

================================== Ai Message ==================================

这是一个 SQL 查询，用于查找上个月所有订单金额超过 1000 美元的客户：

\`\`\`sql
SELECT DISTINCT
    c.customer_id,
    c.name,
    c.email,
    c.customer_tier
FROM customers c
JOIN orders o ON c.customer_id = o.customer_id
WHERE o.total_amount > 1000
  AND o.status = 'completed'
  AND o.order_date >= CURRENT_DATE - INTERVAL '1 month'
ORDER BY c.customer_id;
\`\`\`

此查询：
- 将客户与其订单连接
- 使用 total_amount 字段筛选高价值订单（>$1000）
- 仅包含已完成的订单（根据业务逻辑）
- 限制为上个月的订单
- 返回不同的客户，以避免如果他们下了多个符合条件的订单时出现重复

智能体在其系统提示中看到了轻量级的技能描述，识别出该问题需要销售数据库知识，调用 load_skill("sales_analytics") 获取完整的模式和业务逻辑，然后使用该信息编写遵循数据库约定的正确查询。

6. 高级：使用自定义状态添加约束

可选：跟踪已加载的技能并强制执行工具约束

您可以添加约束，以确保某些工具仅在特定技能加载后才可用。这需要跟踪哪些技能已加载在自定义智能体状态中。

定义自定义状态

首先，扩展智能体状态以跟踪已加载的技能：

from langchain.agents.middleware import AgentState

class CustomState(AgentState):
    skills_loaded: NotRequired[list[str]]  # 跟踪已加载的技能  #

更新 load_skill 以修改状态

修改 load_skill 工具，以便在技能加载时更新状态：

from langgraph.types import Command  
from langchain.tools import tool, ToolRuntime
from langchain.messages import ToolMessage  

@tool
def load_skill(skill_name: str, runtime: ToolRuntime) -> Command:
    """将技能的完整内容加载到智能体的上下文中。

    当您需要有关如何处理特定类型请求的详细信息时使用此工具。这将为您提供有关该技能领域的全面说明、策略和指南。

    Args:
        skill_name: 要加载的技能的名称
    """
    # 查找并返回请求的技能
    for skill in SKILLS:
        if skill["name"] == skill_name:
            skill_content = f"已加载技能：{skill_name}\n\n{skill['content']}"

            # 更新状态以跟踪已加载的技能
            return Command(
                update={
                    "messages": [
                        ToolMessage(
                            content=skill_content,
                            tool_call_id=runtime.tool_call_id,
                        )
                    ],
                    "skills_loaded": [skill_name],
                }
            )

    # 未找到技能
    available = ", ".join(s["name"] for s in SKILLS)
    return Command(
        update={
            "messages": [
                ToolMessage(
                    content=f"未找到技能 '{skill_name}'。可用技能：{available}",
                    tool_call_id=runtime.tool_call_id,
                )
            ]
        }
    )

创建受约束的工具

创建一个仅在特定技能加载后才可用的工具：

@tool
def write_sql_query(
    query: str,
    vertical: str,
    runtime: ToolRuntime,
) -> str:
    """为特定业务垂直领域编写和验证 SQL 查询。

    此工具有助于格式化和验证 SQL 查询。您必须先加载相应的技能以了解数据库模式。

    Args:
        query: 要编写的 SQL 查询
        vertical: 业务垂直领域（sales_analytics 或 inventory_management）
    """
    # 检查是否已加载所需的技能
    skills_loaded = runtime.state.get("skills_loaded", [])

    if vertical not in skills_loaded:
        return (
            f"错误：您必须先加载 '{vertical}' 技能 "
            f"以了解数据库模式，然后才能编写查询。 "
            f"使用 load_skill('{vertical}') 加载模式。"
        )

    # 验证并格式化查询
    return (
        f"针对 {vertical} 的 SQL 查询：\n\n"
        f"```sql\n{query}\n```\n\n"
        f"✓ 查询已根据 {vertical} 模式验证\n"
        f"准备在数据库上执行。"
    )

更新中间件和智能体

更新中间件以使用自定义状态模式：

class SkillMiddleware(AgentMiddleware[CustomState]):
    """将技能描述注入系统提示的中间件。"""

    state_schema = CustomState  
    tools = [load_skill, write_sql_query]

    # ... 中间件实现的其余部分保持不变

创建具有注册受约束工具的中间件的智能体：

agent = create_agent(
    model,
    system_prompt=(
        "您是一个 SQL 查询助手，帮助用户"
        "针对业务数据库编写查询。"
    ),
    middleware=[SkillMiddleware()],
    checkpointer=InMemorySaver(),
)

现在，如果智能体在加载所需技能之前尝试使用 write_sql_query，它将收到一条错误消息，提示它先加载相应的技能（例如 sales_analytics 或 inventory_management）。这确保了智能体在尝试验证查询之前具有必要的模式知识。

完整示例

查看完整的可运行脚本

这是一个完整的、可运行的实现，结合了本教程中的所有部分：

from langchain_core.utils.uuid import uuid7
from typing import TypedDict, NotRequired
from langchain.tools import tool
from langchain.agents import create_agent
from langchain.agents.middleware import ModelRequest, ModelResponse, AgentMiddleware
from langchain.messages import SystemMessage
from langgraph.checkpoint.memory import InMemorySaver
from typing import Callable

# 定义技能结构
class Skill(TypedDict):
    """可以渐进式披露给智能体的技能。"""
    name: str
    description: str
    content: str

# 定义包含模式和业务逻辑的技能
SKILLS: list[Skill] = [
    {
        "name": "sales_analytics",
        "description": "用于销售数据分析的数据库模式和业务逻辑，包括客户、订单和收入。",
        "content": """# 销售分析模式

## 表

### customers
- customer_id (主键)
- name
- email
- signup_date
- status (active/inactive)
- customer_tier (bronze/silver/gold/platinum)

### orders
- order_id (主键)
- customer_id (外键 -> customers)
- order_date
- status (pending/completed/cancelled/refunded)
- total_amount
- sales_region (north/south/east/west)

### order_items
- item_id (主键)
- order_id (外键 -> orders)
- product_id
- quantity
- unit_price
- discount_percent

## 业务逻辑

**活跃客户**：status = 'active' AND signup_date <= CURRENT_DATE - INTERVAL '90 days'

**收入计算**：仅计算 status = 'completed' 的订单。使用 orders 表中的 total_amount，该字段已考虑折扣。

**客户生命周期价值 (CLV)**：客户所有已完成订单金额的总和。

**高价值订单**：total_amount > 1000 的订单

## 示例查询

-- 获取上个季度收入最高的前 10 名客户
SELECT
    c.customer_id,
    c.name,
    c.customer_tier,
    SUM(o.total_amount) as total_revenue
FROM customers c
JOIN orders o ON c.customer_id = o.customer_id
WHERE o.status = 'completed'
  AND o.order_date >= CURRENT_DATE - INTERVAL '3 months'
GROUP BY c.customer_id, c.name, c.customer_tier
ORDER BY total_revenue DESC
LIMIT 10;
""",
    },
    {
        "name": "inventory_management",
        "description": "用于库存跟踪的数据库模式和业务逻辑，包括产品、仓库和库存水平。",
        "content": """# 库存管理模式

## 表

### products
- product_id (主键)
- product_name
- sku
- category
- unit_cost
- reorder_point (重新订购前的最低库存水平)
- discontinued (布尔值)

### warehouses
- warehouse_id (主键)
- warehouse_name
- location
- capacity

### inventory
- inventory_id (主键)
- product_id (外键 -> products)
- warehouse_id (外键 -> warehouses)
- quantity_on_hand
- last_updated

### stock_movements
- movement_id (主键)
- product_id (外键 -> products)
- warehouse_id (外键 -> warehouses)
- movement_type (inbound/outbound/transfer/adjustment)
- quantity (入站为正，出站为负)
- movement_date
- reference_number

## 业务逻辑

**可用库存**：inventory 表中 quantity_on_hand > 0 的 quantity_on_hand

**需要重新订购的产品**：所有仓库中 quantity_on_hand 总和小于或等于产品 reorder_point 的产品

**仅限活跃产品**：除非专门分析已停产产品，否则排除 discontinued = true 的产品

**库存估值**：每个产品的 quantity_on_hand * unit_cost

## 示例查询

-- 查找所有仓库中低于重新订购点的产品
SELECT
    p.product_id,
    p.product_name,
    p.reorder_point,
    SUM(i.quantity_on_hand) as total_stock,
    p.unit_cost,
    (p.reorder_point - SUM(i.quantity_on_hand)) as units_to_reorder
FROM products p
JOIN inventory i ON p.product_id = i.product_id
WHERE p.discontinued = false
GROUP BY p.product_id, p.product_name, p.reorder_point, p.unit_cost
HAVING SUM(i.quantity_on_hand) <= p.reorder_point
ORDER BY units_to_reorder DESC;
""",
    },
]

# 创建技能加载工具
@tool
def load_skill(skill_name: str) -> str:
    """将技能的完整内容加载到智能体的上下文中。

    当您需要有关如何处理特定类型请求的详细信息时使用此工具。这将为您提供有关该技能领域的全面说明、策略和指南。

    Args:
        skill_name: 要加载的技能的名称（例如 "sales_analytics"、"inventory_management"）
    """
    # 查找并返回请求的技能
    for skill in SKILLS:
        if skill["name"] == skill_name:
            return f"已加载技能：{skill_name}\n\n{skill['content']}"

    # 未找到技能
    available = ", ".join(s["name"] for s in SKILLS)
    return f"未找到技能 '{skill_name}'。可用技能：{available}"

# 创建技能中间件
class SkillMiddleware(AgentMiddleware):
    """将技能描述注入系统提示的中间件。"""

    # 将 load_skill 工具注册为类变量
    tools = [load_skill]

    def __init__(self):
        """初始化并从 SKILLS 生成技能提示。"""
        # 从 SKILLS 列表构建技能提示
        skills_list = []
        for skill in SKILLS:
            skills_list.append(
                f"- **{skill['name']}**: {skill['description']}"
            )
        self.skills_prompt = "\n".join(skills_list)

    def wrap_model_call(
        self,
        request: ModelRequest,
        handler: Callable[[ModelRequest], ModelResponse],
    ) -> ModelResponse:
        """同步：将技能描述注入系统提示。"""
        # 构建技能附录
        skills_addendum = (
            f"\n\n## 可用技能\n\n{self.skills_prompt}\n\n"
            "当您需要有关处理特定类型请求的详细信息时，"
            "请使用 load_skill 工具。"
        )

        # 追加到系统消息内容块
        new_content = list(request.system_message.content_blocks) + [
            {"type": "text", "text": skills_addendum}
        ]
        new_system_message = SystemMessage(content=new_content)
        modified_request = request.override(system_message=new_system_message)
        return handler(modified_request)

# 初始化您的聊天模型（替换为您的模型）
# 示例：from langchain_anthropic import ChatAnthropic
# model = ChatAnthropic(model="claude-3-5-sonnet-20241022")
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-4")

# 创建支持技能的智能体
agent = create_agent(
    model,
    system_prompt=(
        "您是一个 SQL 查询助手，帮助用户"
        "针对业务数据库编写查询。"
    ),
    middleware=[SkillMiddleware()],
    checkpointer=InMemorySaver(),
)

# 示例用法
if __name__ == "__main__":
    # 此对话线程的配置
    thread_id = str(uuid7())
    config = {"configurable": {"thread_id": thread_id}}

    # 请求 SQL 查询
    result = agent.invoke(
        {
            "messages": [
                {
                    "role": "user",
                    "content": (
                        "编写一个 SQL 查询，查找上个月所有"
                        "订单金额超过 1000 美元的客户"
                    ),
                }
            ]
        },
        config
    )

    # 打印对话
    for message in result["messages"]:
        if hasattr(message, 'pretty_print'):
            message.pretty_print()
        else:
            print(f"{message.type}: {message.content}")

此完整示例包括：

包含完整数据库模式的技能定义
用于按需加载的 load_skill 工具
将技能描述注入系统提示的 SkillMiddleware
使用中间件和检查点创建智能体
展示智能体如何加载技能和编写 SQL 查询的示例用法

要运行此示例，您需要：

安装所需的包：pip install langchain langchain-openai langgraph
设置您的 API 密钥（例如 export OPENAI_API_KEY=...）
将模型初始化替换为您首选的 LLM 提供商

实现变体

查看实现选项和权衡

本教程将技能实现为通过工具调用加载的内存 Python 字典。但是，有几种方法可以使用技能实现渐进式披露：存储后端：

内存（本教程）：技能定义为 Python 数据结构，访问速度快，无 I/O 开销
文件系统（Claude Code 方法）：技能作为包含文件的目录，通过 read_file 等文件操作发现
远程存储：技能存储在 S3、数据库、Notion 或 API 中，按需获取

技能发现（智能体如何了解存在哪些技能）：

系统提示列表：系统提示中的技能描述（本教程中使用）
基于文件：通过扫描目录发现技能（Claude Code 方法）
基于注册表：查询技能注册表服务或 API 以获取可用技能
动态查找：通过工具调用列出可用技能

渐进式披露策略（如何加载技能内容）：

单次加载：在一次工具调用中加载整个技能内容（本教程中使用）
分页：对于大型技能，分多个页面/块加载技能内容
基于搜索：在特定技能内容中搜索相关部分（例如，使用 grep/read 操作处理技能文件）
分层：首先加载技能概述，然后深入到特定子部分

大小考虑（未校准的心理模型 - 根据您的系统进行优化）：

小型技能（< 1K 令牌 / ~750 个单词）：可以直接包含在系统提示中，并通过提示缓存进行缓存以节省成本并加快响应速度
中型技能（1-10K 令牌 / ~750-7.5K 个单词）：受益于按需加载以避免上下文开销（本教程）
大型技能（> 10K 令牌 / ~7.5K 个单词，或 > 上下文窗口的 5-10%）：应使用渐进式披露技术，如分页、基于搜索的加载或分层探索，以避免消耗过多上下文

选择取决于您的要求：内存最快，但需要重新部署才能更新技能，而基于文件或远程存储支持动态技能管理，无需更改代码。

渐进式披露与上下文工程

结合少样本提示和其他技术

渐进式披露从根本上说是一种**上下文工程技术** - 您正在管理哪些信息可供智能体使用以及何时使用。本教程重点介绍加载数据库模式，但相同的原则适用于其他类型的上下文。

结合少样本提示

对于 SQL 查询用例，您可以扩展渐进式披露以动态加载与用户查询匹配的少样本示例：示例方法：

用户询问：“查找 6 个月未下订单的客户”
智能体加载 sales_analytics 模式（如本教程所示）
智能体还加载 2-3 个相关的示例查询（通过语义搜索或基于标签的查找）：
- 查找不活跃客户的查询
- 带有基于日期过滤的查询
- 连接客户和订单表的查询
智能体使用模式知识和示例模式编写查询

渐进式披露（按需加载模式）和动态少样本提示（加载相关示例）的这种组合创建了一种强大的上下文工程模式，可以扩展到大型知识库，同时提供高质量、有依据的输出。

后续步骤

了解中间件以实现更动态的智能体行为
探索上下文工程技术以管理智能体上下文
探索交接模式以实现顺序工作流
阅读子智能体模式以实现并行任务路由
查看多智能体模式以了解其他专门智能体方法
使用 LangSmith 调试和监控技能加载

将这些文档连接到 Claude、VSCode 等，通过 MCP 获取实时答案。

在 GitHub 上编辑此页面或提交问题。

​工作原理

​设置

​安装

​LangSmith

​选择 LLM

​1. 定义技能

​2. 创建技能加载工具

​3. 构建技能中间件

​4. 创建支持技能的智能体

​5. 测试渐进式披露

​6. 高级：使用自定义状态添加约束

​定义自定义状态

​更新 load_skill 以修改状态

​创建受约束的工具

​更新中间件和智能体

​完整示例

​实现变体

​渐进式披露与上下文工程

​结合少样本提示

​后续步骤

工作原理

设置

安装

LangSmith

选择 LLM

1. 定义技能

2. 创建技能加载工具

3. 构建技能中间件

4. 创建支持技能的智能体

5. 测试渐进式披露

6. 高级：使用自定义状态添加约束

定义自定义状态

更新 load_skill 以修改状态

创建受约束的工具

更新中间件和智能体

完整示例

实现变体

渐进式披露与上下文工程

结合少样本提示

后续步骤