LangChain 中文文档

外观

本教程演示如何使用渐进式披露（progressive disclosure）——一种上下文管理技术，智能体按需而非预先加载信息——来实现技能（基于提示词的专业指令）。智能体通过工具调用加载技能，而不是动态更改系统提示词，从而只发现并加载每个任务所需的技能。

使用场景： 设想构建一个智能体，帮助在大型企业的不同业务垂直领域编写 SQL 查询。你的组织可能为每个垂直领域设置独立的数据存储，或者使用包含数千张表的单一单体数据库。无论哪种方式，预先加载所有模式都会使上下文窗口不堪重负。渐进式披露通过仅在需要时加载相关的模式来解决这个问题。这种架构还使不同的产品负责人和利益相关者能够独立地为各自特定的业务垂直领域贡献和维护技能。

你将构建的内容： 一个带有两个技能（销售分析和库存管理）的 SQL 查询助手。智能体在系统提示词中看到轻量级的技能描述，然后仅在与用户查询相关时通过工具调用加载完整的数据库模式和业务逻辑。

INFO

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

TIP

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

工作原理 ​

以下是用户请求 SQL 查询时的流程：

为什么使用渐进式披露：

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

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

TIP

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

权衡：

  • 延迟：按需加载技能需要额外的工具调用，这会增加每个技能首次被请求时的延迟
  • 工作流控制：基本实现依赖提示词来引导技能使用——没有自定义逻辑，你就无法强制执行"始终先尝试技能 A 再尝试技能 B"这样的硬约束

TIP

实现你自己的技能系统

在构建你自己的技能实现时（正如我们在本教程中所做的），核心概念是渐进式披露——按需加载信息。除此之外，你在实现上有充分的灵活性：

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

这种灵活性让你可以针对性能、存储和工作流控制方面的具体需求进行优化。

设置 ​

安装 ​

本教程需要 langchain 包：

bash
pip install langchain
bash
uv add langchain
bash
conda install langchain -c conda-forge
bash
npm install langchain
bash
yarn add langchain
bash
pnpm add langchain

更多详情，请参阅我们的安装指南。

LangSmith ​

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

bash
export LANGSMITH_TRACING="true"
export LANGSMITH_API_KEY="..."
python
import getpass
import os

os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = getpass.getpass()
bash
export LANGSMITH_TRACING="true"
export LANGSMITH_API_KEY="..."
typescript
process.env.LANGSMITH_TRACING = "true";
process.env.LANGSMITH_API_KEY = "...";

选择 LLM ​

从 LangChain 的集成套件中选择一个对话模型：

OpenAI ​

    👉 Read the [OpenAI chat model integration docs](/oss/python/integrations/chat/openai)
bash
pip install -U "langchain[openai]"
python
import os
from langchain.chat_models import init_chat_model

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

model = init_chat_model("gpt-5.5")
python
import os
from langchain_openai import ChatOpenAI

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

model = ChatOpenAI(model="gpt-5.5")

Anthropic ​

    👉 Read the [Anthropic chat model integration docs](/oss/python/integrations/chat/anthropic)
bash
pip install -U "langchain[anthropic]"
python
import os
from langchain.chat_models import init_chat_model

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

model = init_chat_model("claude-sonnet-4-6")
python
import os
from langchain_anthropic import ChatAnthropic

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

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

Azure ​

    👉 Read the [Azure chat model integration docs](/oss/python/integrations/chat/azure_chat_openai)
bash
pip install -U "langchain[openai]"
python
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.5",
    azure_deployment=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"],
)
python
import os
from langchain_openai import AzureChatOpenAI

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

model = AzureChatOpenAI(
    model="gpt-5.5",
    azure_deployment=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"]
)

Google Gemini ​

    👉 Read the [Google GenAI chat model integration docs](/oss/python/integrations/chat/google_generative_ai)
bash
pip install -U "langchain[google-genai]"
python
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")
python
import os
from langchain_google_genai import ChatGoogleGenerativeAI

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

model = ChatGoogleGenerativeAI(model="gemini-2.5-flash-lite")

AWS Bedrock ​

    👉 Read the [AWS Bedrock chat model integration docs](/oss/python/integrations/chat/bedrock)
bash
pip install -U "langchain[aws]"
python
from langchain.chat_models import init_chat_model

# Follow the steps here to configure your credentials:
# https://docs.aws.amazon.com/bedrock/latest/userguide/getting-started.html

model = init_chat_model(
    "us.anthropic.claude-sonnet-4-6",
    model_provider="bedrock_converse",
)
python
from langchain_aws import ChatBedrock

model = ChatBedrock(model="us.anthropic.claude-sonnet-4-6")

HuggingFace ​

    👉 Read the [HuggingFace chat model integration docs](/oss/python/integrations/chat/huggingface)
bash
pip install -U "langchain[huggingface]"
python
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,
)
python
import os
from langchain_huggingface import ChatHuggingFace, HuggingFaceEndpoint

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

llm = HuggingFaceEndpoint(
    repo_id="microsoft/Phi-3-mini-4k-instruct",
    temperature=0.7,
    max_length=1024,
)
model = ChatHuggingFace(llm=llm)

OpenRouter ​

    👉 Read the [OpenRouter chat model integration docs](/oss/python/integrations/chat/openrouter)
bash
pip install -U "langchain-openrouter"
python
import os
from langchain.chat_models import init_chat_model

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

model = init_chat_model(
    "auto",
    model_provider="openrouter",
)
python
import os
from langchain_openrouter import ChatOpenRouter

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

model = ChatOpenRouter(model="auto")

OpenAI ​

    👉 Read the [OpenAI chat model integration docs](/oss/javascript/integrations/chat/openai)
bash
npm install @langchain/openai
bash
pnpm install @langchain/openai
bash
yarn add @langchain/openai
bash
bun add @langchain/openai
typescript
import { initChatModel } from "langchain";

process.env.OPENAI_API_KEY = "your-api-key";

const model = await initChatModel("gpt-5.5");
typescript
import { ChatOpenAI } from "@langchain/openai";

const model = new ChatOpenAI({
  model: "gpt-5.5",
  apiKey: "your-api-key"
});

Anthropic ​

    👉 Read the [Anthropic chat model integration docs](/oss/javascript/integrations/chat/anthropic)
bash
npm install @langchain/anthropic
bash
pnpm install @langchain/anthropic
bash
yarn add @langchain/anthropic
bash
pnpm add @langchain/anthropic
typescript
import { initChatModel } from "langchain";

process.env.ANTHROPIC_API_KEY = "your-api-key";

const model = await initChatModel("claude-sonnet-4-6");
typescript
import { ChatAnthropic } from "@langchain/anthropic";

const model = new ChatAnthropic({
  model: "claude-sonnet-4-6",
  apiKey: "your-api-key"
});

Azure ​

    👉 Read the [Azure chat model integration docs](/oss/javascript/integrations/chat/azure)
bash
npm install @langchain/azure
bash
pnpm install @langchain/azure
bash
yarn add @langchain/azure
bash
bun add @langchain/azure
typescript
import { initChatModel } from "langchain";

process.env.AZURE_OPENAI_API_KEY = "your-api-key";
process.env.AZURE_OPENAI_ENDPOINT = "your-endpoint";
process.env.OPENAI_API_VERSION = "your-api-version";

const model = await initChatModel("azure_openai:gpt-5.5");
typescript
import { AzureChatOpenAI } from "@langchain/openai";

const model = new AzureChatOpenAI({
  model: "gpt-5.5",
  azureOpenAIApiKey: "your-api-key",
  azureOpenAIApiEndpoint: "your-endpoint",
  azureOpenAIApiVersion: "your-api-version"
});

Google Gemini ​

    👉 Read the [Google GenAI chat model integration docs](/oss/javascript/integrations/chat/google_generative_ai)
bash
npm install @langchain/google-genai
bash
pnpm install @langchain/google-genai
bash
yarn add @langchain/google-genai
bash
bun add @langchain/google-genai
typescript
import { initChatModel } from "langchain";

process.env.GOOGLE_API_KEY = "your-api-key";

const model = await initChatModel("google-genai:gemini-2.5-flash-lite");
typescript
import { ChatGoogleGenerativeAI } from "@langchain/google-genai";

const model = new ChatGoogleGenerativeAI({
  model: "gemini-2.5-flash-lite",
  apiKey: "your-api-key"
});

Bedrock Converse ​

    👉 Read the [AWS Bedrock chat model integration docs](/oss/javascript/integrations/chat/bedrock_converse)
bash
npm install @langchain/aws
bash
pnpm install @langchain/aws
bash
yarn add @langchain/aws
bash
bun add @langchain/aws
typescript
import { initChatModel } from "langchain";

// Follow the steps here to configure your credentials:
// https://docs.aws.amazon.com/bedrock/latest/userguide/getting-started.html

const model = await initChatModel("bedrock:gpt-5.5");
typescript
import { ChatBedrockConverse } from "@langchain/aws";

// Follow the steps here to configure your credentials:
// https://docs.aws.amazon.com/bedrock/latest/userguide/getting-started.html

const model = new ChatBedrockConverse({
  model: "gpt-5.5",
  region: "us-east-2"
});

1. 定义技能 ​

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

python
from typing import TypedDict

class Skill(TypedDict):  
    """A skill that can be progressively disclosed to the agent."""
    name: str  # 技能的唯一标识符
    description: str  # 显示在系统提示词中的 1-2 句描述
    content: str  # 包含详细指令的完整技能内容
typescript
import { z } from "zod";

// 可渐进式披露给智能体的技能
const SkillSchema = z.object({  
  name: z.string(),  // 技能的唯一标识符
  description: z.string(),  // 显示在系统提示词中的 1-2 句描述
  content: z.string(),  // 包含详细指令的完整技能内容
});

type Skill = z.infer<typeof SkillSchema>;

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

查看完整的技能定义 ​

python
SKILLS: list[Skill] = [
    {
        "name": "sales_analytics",
        "description": "Database schema and business logic for sales data analysis including customers, orders, and revenue.",
        "content": """# Sales Analytics Schema

## Tables

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

### orders
- order_id (PRIMARY KEY)
- customer_id (FOREIGN KEY -> customers)
- order_date
- status (pending/completed/cancelled/refunded)
- total_amount
- sales_region (north/south/east/west)

### order_items
- item_id (PRIMARY KEY)
- order_id (FOREIGN KEY -> orders)
- product_id
- quantity
- unit_price
- discount_percent

## Business Logic

**Active customers**: status = 'active' AND signup_date <= CURRENT_DATE - INTERVAL '90 days'

**Revenue calculation**: Only count orders with status = 'completed'. Use total_amount from orders table, which already accounts for discounts.

**Customer lifetime value (CLV)**: Sum of all completed order amounts for a customer.

**High-value orders**: Orders with total_amount > 1000

## Example Query

-- Get top 10 customers by revenue in the last quarter
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": "Database schema and business logic for inventory tracking including products, warehouses, and stock levels.",
        "content": """# Inventory Management Schema

## Tables

### products
- product_id (PRIMARY KEY)
- product_name
- sku
- category
- unit_cost
- reorder_point (minimum stock level before reordering)
- discontinued (boolean)

### warehouses
- warehouse_id (PRIMARY KEY)
- warehouse_name
- location
- capacity

### inventory
- inventory_id (PRIMARY KEY)
- product_id (FOREIGN KEY -> products)
- warehouse_id (FOREIGN KEY -> warehouses)
- quantity_on_hand
- last_updated

### stock_movements
- movement_id (PRIMARY KEY)
- product_id (FOREIGN KEY -> products)
- warehouse_id (FOREIGN KEY -> warehouses)
- movement_type (inbound/outbound/transfer/adjustment)
- quantity (positive for inbound, negative for outbound)
- movement_date
- reference_number

## Business Logic

**Available stock**: quantity_on_hand from inventory table where quantity_on_hand > 0

**Products needing reorder**: Products where total quantity_on_hand across all warehouses is less than or equal to the product's reorder_point

**Active products only**: Exclude products where discontinued = true unless specifically analyzing discontinued items

**Stock valuation**: quantity_on_hand * unit_cost for each product

## Example Query

-- Find products below reorder point across all warehouses
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;
""",
    },
]
typescript
import { context } from "langchain";

const SKILLS: Skill[] = [
  {
    name: "sales_analytics",
    description:
      "Database schema and business logic for sales data analysis including customers, orders, and revenue.",
    content: context`
    # Sales Analytics Schema

    ## Tables

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

    ### orders
    - order_id (PRIMARY KEY)
    - customer_id (FOREIGN KEY -> customers)
    - order_date
    - status (pending/completed/cancelled/refunded)
    - total_amount
    - sales_region (north/south/east/west)

    ### order_items
    - item_id (PRIMARY KEY)
    - order_id (FOREIGN KEY -> orders)
    - product_id
    - quantity
    - unit_price
    - discount_percent

    ## Business Logic

    **Active customers**:
    status = 'active' AND signup_date <= CURRENT_DATE - INTERVAL '90 days'

    **Revenue calculation**:
    Only count orders with status = 'completed'.
    Use total_amount from orders table, which already accounts for discounts.

    **Customer lifetime value (CLV)**:
    Sum of all completed order amounts for a customer.

    **High-value orders**:
    Orders with total_amount > 1000

    ## Example Query

    -- Get top 10 customers by revenue in the last quarter
    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:
      "Database schema and business logic for inventory tracking including products, warehouses, and stock levels.",
    content: context`
    # Inventory Management Schema

    ## Tables

    ### products
    - product_id (PRIMARY KEY)
    - product_name
    - sku
    - category
    - unit_cost
    - reorder_point (minimum stock level before reordering)
    - discontinued (boolean)

    ### warehouses
    - warehouse_id (PRIMARY KEY)
    - warehouse_name
    - location
    - capacity

    ### inventory
    - inventory_id (PRIMARY KEY)
    - product_id (FOREIGN KEY -> products)
    - warehouse_id (FOREIGN KEY -> warehouses)
    - quantity_on_hand
    - last_updated

    ### stock_movements
    - movement_id (PRIMARY KEY)
    - product_id (FOREIGN KEY -> products)
    - warehouse_id (FOREIGN KEY -> warehouses)
    - movement_type (inbound/outbound/transfer/adjustment)
    - quantity (positive for inbound, negative for outbound)
    - movement_date
    - reference_number

    ## Business Logic

    **Available stock**:
    quantity_on_hand from inventory table where quantity_on_hand > 0

    **Products needing reorder**:
    Products where total quantity_on_hand across all warehouses is less
    than or equal to the product's reorder_point

    **Active products only**:
    Exclude products where discontinued = true unless specifically analyzing discontinued items

    **Stock valuation**:
    quantity_on_hand * unit_cost for each product

    ## Example Query

    -- Find products below reorder point across all warehouses
    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. 创建技能加载工具 ​

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

python
from langchain.tools import tool

@tool
def load_skill(skill_name: str) -> str:
    """Load the full content of a skill into the agent's context.

    Use this when you need detailed information about how to handle a specific
    type of request. This will provide you with comprehensive instructions,
    policies, and guidelines for the skill area.

    Args:
        skill_name: The name of the skill to load (e.g., "expense_reporting", "travel_booking")
    """
    # 查找并返回所请求的技能
    for skill in SKILLS:
        if skill["name"] == skill_name:
            return f"Loaded skill: {skill_name}\n\n{skill['content']}"

    # 未找到该技能
    available = ", ".join(s["name"] for s in SKILLS)
    return f"Skill '{skill_name}' not found. Available skills: {available}"
typescript
import { tool } from "langchain";
import { z } from "zod";

const loadSkill = tool(  
  async ({ skillName }) => {
    // 查找并返回所请求的技能
    const skill = SKILLS.find((s) => s.name === skillName);
    if (skill) {
      return `Loaded skill: ${skillName}\n\n${skill.content}`;  
    }

    // 未找到该技能
    const available = SKILLS.map((s) => s.name).join(", ");
    return `Skill '${skillName}' not found. Available skills: ${available}`;
  },
  {
    name: "load_skill",
    description: `Load the full content of a skill into the agent's context.

Use this when you need detailed information about how to handle a specific
type of request. This will provide you with comprehensive instructions,
policies, and guidelines for the skill area.`,
    schema: z.object({
      skillName: z.string().describe("The name of the skill to load"),
    }),
  }
);

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

3. 构建技能中间件 ​

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

INFO

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

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

class SkillMiddleware(AgentMiddleware):  
    """Middleware that injects skill descriptions into the system prompt."""

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

    def __init__(self):
        """Initialize and generate the skills prompt from 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:
        """Sync: Inject skill descriptions into system prompt."""
        # 构建技能补充内容
        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_message = SystemMessage(content=new_content)
        modified_request = request.override(system_message=new_system_message)
        return handler(modified_request)
typescript
import { createMiddleware } from "langchain";

// 根据 SKILLS 列表构建技能提示词
const skillsPrompt = SKILLS.map(
  (skill) => `- **${skill.name}**: ${skill.description}`
).join("\n");

const skillMiddleware = createMiddleware({  
  name: "skillMiddleware",
  tools: [loadSkill],  
  wrapModelCall: async (request, handler) => {
    // 构建技能补充内容
    const skillsAddendum =
      `\n\n## Available Skills\n\n${skillsPrompt}\n\n` +
      "Use the load_skill tool when you need detailed information " +
      "about handling a specific type of request.";  

    // 追加到系统提示词
    const newSystemPrompt = request.systemPrompt + skillsAddendum;

    return handler({
      ...request,
      systemPrompt: newSystemPrompt,
    });
  },
});

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

INFO

生产环境注意事项：本教程为简单起见在 __init__ 中加载技能列表。在生产系统中，你可能希望在 before_agent 钩子中加载技能，以便定期刷新以反映最新的更改（例如，当添加新技能或修改现有技能时）。详情请参阅 before_agent 钩子文档。

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

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

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

# 创建带技能支持的智能体
agent = create_agent(
    model,
    system_prompt=(
        "You are a SQL query assistant that helps users "
        "write queries against business databases."
    ),
    middleware=[SkillMiddleware()],  
    checkpointer=InMemorySaver(),
)
typescript
import { createAgent } from "langchain";
import { MemorySaver } from "@langchain/langgraph";

// 创建带技能支持的智能体
const agent = createAgent({
  model,
  systemPrompt:
    "You are a SQL query assistant that helps users " +
    "write queries against business databases.",
  middleware: [skillMiddleware],  
  checkpointer: new MemorySaver(),
});

智能体现在可以在系统提示词中访问技能描述，并且可以在需要时调用 load_skill 获取完整的技能内容。检查点在多个轮次之间维护对话历史。

5. 测试渐进式披露 ​

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

python
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": (
                    "Write a SQL query to find all customers "
                    "who made orders over $1000 in the last month"
                ),
            }
        ]
    },
    config
)

# 打印对话
for message in result["messages"]:
    if hasattr(message, 'pretty_print'):
        message.pretty_print()
    else:
        print(f"{message.type}: {message.content}")
typescript
// 此对话线程的配置
const threadId = crypto.randomUUID();
const config = { configurable: { thread_id: threadId } };

// 请求编写 SQL 查询
const result = await agent.invoke(  
  {
    messages: [
      {
        role: "user",
        content:
          "Write a SQL query to find all customers " +
          "who made orders over $1000 in the last month",
      },
    ],
  },
  config
);

// 打印对话
for (const message of result.messages) {
  console.log(`${message._getType()}: ${message.content}`);
}

预期输出：

================================ Human Message =================================

Write a SQL query to find all customers who made orders over $1000 in the last month
================================== Ai Message ==================================
Tool Calls:
  load_skill (call_abc123)
 Call ID: call_abc123
  Args:
    skill_name: sales_analytics
================================= Tool Message =================================
Name: load_skill

Loaded skill: sales_analytics

# Sales Analytics Schema

## Tables

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

### orders
- order_id (PRIMARY KEY)
- customer_id (FOREIGN KEY -> customers)
- order_date
- status (pending/completed/cancelled/refunded)
- total_amount
- sales_region (north/south/east/west)

[... rest of schema ...]

## Business Logic

**High-value orders**: Orders with `total_amount > 1000`
**Revenue calculation**: Only count orders with `status = 'completed'`

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

Here's a SQL query to find all customers who made orders over $1000 in the last month:

\`\`\`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;
\`\`\`

This query:
- Joins customers with their orders
- Filters for high-value orders (>$1000) using the total_amount field
- Only includes completed orders (as per the business logic)
- Restricts to orders from the last month
- Returns distinct customers to avoid duplicates if they made multiple qualifying orders

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

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

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

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

定义自定义状态 ​

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

python
from langchain.agents.middleware import AgentState

class CustomState(AgentState):  
    skills_loaded: NotRequired[list[str]]  # 记录已加载的技能  #
typescript
import { StateSchema } from "@langchain/langgraph";
import { z } from "zod";

const CustomState = new StateSchema({
  skillsLoaded: z.array(z.string()).optional(),  // 记录已加载的技能
});

更新 load_skill 以修改状态 ​

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

python
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:  
    """Load the full content of a skill into the agent's context.

    Use this when you need detailed information about how to handle a specific
    type of request. This will provide you with comprehensive instructions,
    policies, and guidelines for the skill area.

    Args:
        skill_name: The name of the skill to load
    """
    # 查找并返回所请求的技能
    for skill in SKILLS:
        if skill["name"] == skill_name:
            skill_content = f"Loaded skill: {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 '{skill_name}' not found. Available skills: {available}",
                    tool_call_id=runtime.tool_call_id,
                )
            ]
        }
    )
typescript
import { tool, ToolMessage, type ToolRuntime } from "langchain";
import { Command } from "@langchain/langgraph";  
import { z } from "zod";

const loadSkill = tool(  
  async ({ skillName }, runtime: ToolRuntime<typeof CustomState.State>) => {
    // 查找并返回所请求的技能
    const skill = SKILLS.find((s) => s.name === skillName);

    if (skill) {
      const skillContent = `Loaded skill: ${skillName}\n\n${skill.content}`;

      // 更新状态以记录已加载的技能
      return new Command({  
        update: {  
          messages: [  
            new ToolMessage({  
              content: skillContent,  
              tool_call_id: runtime.toolCallId,  
            }),  
          ],  
          skillsLoaded: [skillName],  
        },  
      });  
    }

    // 未找到该技能
    const available = SKILLS.map((s) => s.name).join(", ");
    return new Command({
      update: {
        messages: [
          new ToolMessage({
            content: `Skill '${skillName}' not found. Available skills: ${available}`,
            tool_call_id: runtime.toolCallId,
          }),
        ],
      },
    });
  },
  {
    name: "load_skill",
    description: `Load the full content of a skill into the agent's context.`,
    schema: z.object({
      skillName: z.string().describe("The name of the skill to load"),
    }),
  }
);

创建受约束的工具 ​

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

python
@tool
def write_sql_query(  
    query: str,
    vertical: str,
    runtime: ToolRuntime,
) -> str:
    """Write and validate a SQL query for a specific business vertical.

    This tool helps format and validate SQL queries. You must load the
    appropriate skill first to understand the database schema.

    Args:
        query: The SQL query to write
        vertical: The business vertical (sales_analytics or inventory_management)
    """
    # 检查所需的技能是否已加载
    skills_loaded = runtime.state.get("skills_loaded", [])  

    if vertical not in skills_loaded:  
        return (  
            f"Error: You must load the '{vertical}' skill first "
            f"to understand the database schema before writing queries. "
            f"Use load_skill('{vertical}') to load the schema."
        )  

    # 验证并格式化查询
    return (
        f"SQL Query for {vertical}:\n\n"
        f"```sql\n{query}\n```\n\n"
        f"✓ Query validated against {vertical} schema\n"
        f"Ready to execute against the database."
    )
typescript
const writeSqlQuery = tool(  
  async ({ query, vertical }, runtime: ToolRuntime<typeof CustomState.State>) => {
    // 检查所需的技能是否已加载
    const skillsLoaded = runtime.state.skillsLoaded ?? [];  

    if (!skillsLoaded.includes(vertical)) {  
      return (  
        `Error: You must load the '${vertical}' skill first ` +
        `to understand the database schema before writing queries. ` +
        `Use load_skill('${vertical}') to load the schema.`
      );  
    }

    // 验证并格式化查询
    return (
      `SQL Query for ${vertical}:\n\n` +
      `\`\`\`sql\n${query}\n\`\`\`\n\n` +
      `✓ Query validated against ${vertical} schema\n` +
      `Ready to execute against the database.`
    );
  },
  {
    name: "write_sql_query",
    description: `Write and validate a SQL query for a specific business vertical.

This tool helps format and validate SQL queries. You must load the
appropriate skill first to understand the database schema.`,
    schema: z.object({
      query: z.string().describe("The SQL query to write"),
      vertical: z.string().describe("The business vertical (sales_analytics or inventory_management)"),
    }),
  }
);

更新中间件和智能体 ​

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

python
class SkillMiddleware(AgentMiddleware[CustomState]):  
    """Middleware that injects skill descriptions into the system prompt."""

    state_schema = CustomState  
    tools = [load_skill, write_sql_query]  

    # ... 中间件的其余实现保持不变
typescript
const skillMiddleware = createMiddleware({  
  name: "skillMiddleware",
  stateSchema: CustomState,  
  tools: [loadSkill, writeSqlQuery],  
  // ... 中间件的其余实现保持不变
});

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

python
agent = create_agent(
    model,
    system_prompt=(
        "You are a SQL query assistant that helps users "
        "write queries against business databases."
    ),
    middleware=[SkillMiddleware()],  
    checkpointer=InMemorySaver(),
)
typescript
const agent = createAgent({
  model,
  systemPrompt:
    "You are a SQL query assistant that helps users " +
    "write queries against business databases.",
  middleware: [skillMiddleware],  
  checkpointer: new MemorySaver(),
});

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

完整示例 ​

查看完整可运行脚本 ​

以下是结合了本教程所有部分的完整可运行实现：

python
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):
    """A skill that can be progressively disclosed to the agent."""
    name: str
    description: str
    content: str

# 定义带模式和业务逻辑的技能
SKILLS: list[Skill] = [
    {
        "name": "sales_analytics",
        "description": "Database schema and business logic for sales data analysis including customers, orders, and revenue.",
        "content": """# Sales Analytics Schema

## Tables

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

### orders
- order_id (PRIMARY KEY)
- customer_id (FOREIGN KEY -> customers)
- order_date
- status (pending/completed/cancelled/refunded)
- total_amount
- sales_region (north/south/east/west)

### order_items
- item_id (PRIMARY KEY)
- order_id (FOREIGN KEY -> orders)
- product_id
- quantity
- unit_price
- discount_percent

## Business Logic

**Active customers**: status = 'active' AND signup_date <= CURRENT_DATE - INTERVAL '90 days'

**Revenue calculation**: Only count orders with status = 'completed'. Use total_amount from orders table, which already accounts for discounts.

**Customer lifetime value (CLV)**: Sum of all completed order amounts for a customer.

**High-value orders**: Orders with total_amount > 1000

## Example Query

-- Get top 10 customers by revenue in the last quarter
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": "Database schema and business logic for inventory tracking including products, warehouses, and stock levels.",
        "content": """# Inventory Management Schema

## Tables

### products
- product_id (PRIMARY KEY)
- product_name
- sku
- category
- unit_cost
- reorder_point (minimum stock level before reordering)
- discontinued (boolean)

### warehouses
- warehouse_id (PRIMARY KEY)
- warehouse_name
- location
- capacity

### inventory
- inventory_id (PRIMARY KEY)
- product_id (FOREIGN KEY -> products)
- warehouse_id (FOREIGN KEY -> warehouses)
- quantity_on_hand
- last_updated

### stock_movements
- movement_id (PRIMARY KEY)
- product_id (FOREIGN KEY -> products)
- warehouse_id (FOREIGN KEY -> warehouses)
- movement_type (inbound/outbound/transfer/adjustment)
- quantity (positive for inbound, negative for outbound)
- movement_date
- reference_number

## Business Logic

**Available stock**: quantity_on_hand from inventory table where quantity_on_hand > 0

**Products needing reorder**: Products where total quantity_on_hand across all warehouses is less than or equal to the product's reorder_point

**Active products only**: Exclude products where discontinued = true unless specifically analyzing discontinued items

**Stock valuation**: quantity_on_hand * unit_cost for each product

## Example Query

-- Find products below reorder point across all warehouses
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:
    """Load the full content of a skill into the agent's context.

    Use this when you need detailed information about how to handle a specific
    type of request. This will provide you with comprehensive instructions,
    policies, and guidelines for the skill area.

    Args:
        skill_name: The name of the skill to load (e.g., "sales_analytics", "inventory_management")
    """
    # 查找并返回所请求的技能
    for skill in SKILLS:
        if skill["name"] == skill_name:
            return f"Loaded skill: {skill_name}\n\n{skill['content']}"

    # 未找到该技能
    available = ", ".join(s["name"] for s in SKILLS)
    return f"Skill '{skill_name}' not found. Available skills: {available}"

# 创建技能中间件
class SkillMiddleware(AgentMiddleware):
    """Middleware that injects skill descriptions into the system prompt."""

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

    def __init__(self):
        """Initialize and generate the skills prompt from 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:
        """Sync: Inject skill descriptions into system prompt."""
        # 构建技能补充内容
        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_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-5.5")

# 创建带技能支持的智能体
agent = create_agent(
    model,
    system_prompt=(
        "You are a SQL query assistant that helps users "
        "write queries against business databases."
    ),
    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": (
                        "Write a SQL query to find all customers "
                        "who made orders over $1000 in the last month"
                    ),
                }
            ]
        },
        config
    )

    # 打印对话
    for message in result["messages"]:
        if hasattr(message, 'pretty_print'):
            message.pretty_print()
        else:
            print(f"{message.type}: {message.content}")
typescript
import {
  tool,
  createAgent,
  createMiddleware,
  ToolMessage,
  context,
  type ToolRuntime,
} from "langchain";
import { MemorySaver, Command } from "@langchain/langgraph";
import { ChatOpenAI } from "@langchain/openai";
import { z } from "zod";

// 可渐进式披露给智能体的技能
const SkillSchema = z.object({
  name: z.string(), // 技能的唯一标识符
  description: z.string(), // 显示在系统提示词中的 1-2 句描述
  content: z.string(), // 包含详细指令的完整技能内容
});

type Skill = z.infer<typeof SkillSchema>;

const SKILLS: Skill[] = [
  {
    name: "sales_analytics",
    description:
      "Database schema and business logic for sales data analysis including customers, orders, and revenue.",
    content: context`
    # Sales Analytics Schema

    ## Tables

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

    ### orders
    - order_id (PRIMARY KEY)
    - customer_id (FOREIGN KEY -> customers)
    - order_date
    - status (pending/completed/cancelled/refunded)
    - total_amount
    - sales_region (north/south/east/west)

    ### order_items
    - item_id (PRIMARY KEY)
    - order_id (FOREIGN KEY -> orders)
    - product_id
    - quantity
    - unit_price
    - discount_percent

    ## Business Logic

    **Active customers**: status = 'active' AND signup_date <= CURRENT_DATE - INTERVAL '90 days'

    **Revenue calculation**:
    Only count orders with status = 'completed'. Use total_amount from orders table,
    which already accounts for discounts.

    **Customer lifetime value (CLV)**:
    Sum of all completed order amounts for a customer.

    **High-value orders**:
    Orders with total_amount > 1000

    ## Example Query
    -- Get top 10 customers by revenue in the last quarter
    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:
      "Database schema and business logic for inventory tracking including products, warehouses, and stock levels.",
    content: context`
    # Inventory Management Schema

    ## Tables

    ### products
    - product_id (PRIMARY KEY)
    - product_name
    - sku
    - category
    - unit_cost
    - reorder_point (minimum stock level before reordering)
    - discontinued (boolean)

    ### warehouses
    - warehouse_id (PRIMARY KEY)
    - warehouse_name
    - location
    - capacity

    ### inventory
    - inventory_id (PRIMARY KEY)
    - product_id (FOREIGN KEY -> products)
    - warehouse_id (FOREIGN KEY -> warehouses)
    - quantity_on_hand
    - last_updated

    ### stock_movements
    - movement_id (PRIMARY KEY)
    - product_id (FOREIGN KEY -> products)
    - warehouse_id (FOREIGN KEY -> warehouses)
    - movement_type (inbound/outbound/transfer/adjustment)
    - quantity (positive for inbound, negative for outbound)
    - movement_date
    - reference_number

    ## Business Logic

    **Available stock**:
    quantity_on_hand from inventory table where quantity_on_hand > 0

    **Products needing reorder**:
    Products where total quantity_on_hand across all warehouses is
    less than or equal to the product's reorder_point

    **Active products only**:
    Exclude products where discontinued = true unless specifically
    analyzing discontinued items

    **Stock valuation**:
    quantity_on_hand * unit_cost for each product

    ## Example Query

    -- Find products below reorder point across all warehouses
    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;`,
  },
];

// const loadSkill = tool(
//   async ({ skillName }) => {
//     // 查找并返回所请求的技能
//     const skill = SKILLS.find((s) => s.name === skillName);
//     if (skill) {
//       return `Loaded skill: ${skillName}\n\n${skill.content}`;
//     }

//     // 未找到该技能
//     const available = SKILLS.map((s) => s.name).join(", ");
//     return `Skill '${skillName}' not found. Available skills: ${available}`;
//   },
//   {
//     name: "load_skill",
//     description: `Load the full content of a skill into the agent's context.

// Use this when you need detailed information about how to handle a specific
// type of request. This will provide you with comprehensive instructions,
// policies, and guidelines for the skill area.`,
//     schema: z.object({
//       skillName: z.string().describe("The name of the skill to load"),
//     }),
//   }
// );

// 根据 SKILLS 列表构建技能提示词
const skillsPrompt = SKILLS.map(
  (skill) => `- **${skill.name}**: ${skill.description}`
).join("\n");

const skillMiddleware = createMiddleware({
  name: "skillMiddleware",
  tools: [loadSkill],
  wrapModelCall: async (request, handler) => {
    // 构建技能补充内容
    const skillsAddendum =
      `\n\n## Available Skills\n\n${skillsPrompt}\n\n` +
      "Use the load_skill tool when you need detailed information " +
      "about handling a specific type of request.";

    // 追加到系统提示词
    const newSystemPrompt = request.systemPrompt + skillsAddendum;

    return handler({
      ...request,
      systemPrompt: newSystemPrompt,
    });
  },
});

const model = new ChatOpenAI({
  model: "gpt-5.4-mini",
  temperature: 0,
});

// 创建带技能支持的智能体
const agent = createAgent({
  model,
  systemPrompt:
    "You are a SQL query assistant that helps users " +
    "write queries against business databases.",
  middleware: [skillMiddleware],
  checkpointer: new MemorySaver(),
});

// 此对话线程的配置
const threadId = crypto.randomUUID();
const config = { configurable: { thread_id: threadId } };

// 请求编写 SQL 查询
const result = await agent.invoke(
  {
    messages: [
      {
        role: "user",
        content:
          "Write a SQL query to find all customers " +
          "who made orders over $1000 in the last month",
      },
    ],
  },
  config
);

// 打印对话
for (const message of result.messages) {
  console.log(`${message.type}: ${message.content}`);
}

这个完整示例包括：

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

要运行此示例，你需要：

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

实现变体 ​

查看实现选项和权衡 ​

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

存储后端：

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

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

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

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

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

大小考虑（未经校准的心智模型——针对你的系统进行优化）：

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

选择取决于你的需求：内存最快，但技能更新时需要重新部署，而基于文件或远程存储则无需更改代码即可实现动态技能管理。

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

与少样本提示及其他技术结合 ​

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

与少样本提示结合 ​

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

示例方法：

  1. 用户问："查找 6 个月内未下单的客户"
  2. 智能体加载 sales_analytics 模式（如本教程所示）
  3. 智能体还加载 2-3 个相关的示例查询（通过语义搜索或基于标签的查找）：
    • 用于查找非活跃客户的查询
    • 带日期过滤的查询
    • 连接 customers 和 orders 表的查询
  4. 智能体结合模式知识和示例模式编写查询

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

后续步骤 ​