构建一个具有按需技能的 SQL 助手

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

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

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

NOTE

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

TIP

渐进式披露由 Anthropic 推广，是一种构建可扩展智能体技能系统的技术。这种方法采用三层架构（元数据 → 核心内容 → 详细资源），智能体仅在需要时加载信息。有关该技术的更多信息，请参阅 Equipping agents for the real world with Agent Skills。

工作原理

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

%%{init: {'theme':'base', 'themeVariables': { 'primaryColor':'#E5F4FF','primaryTextColor':'#030710','primaryBorderColor':'#006DDD','lineColor':'#40668D','secondaryColor':'#F6FFDB','tertiaryColor':'#FDF3FF','tertiaryBorderColor':'#7E65AE','tertiaryTextColor':'#504B5F'}}}%%
flowchart TD
    Start([💬 User: Write SQL query<br/>for high-value customers]) --> SystemPrompt[📋 Agent sees skill descriptions:<br/>• sales_analytics<br/>• inventory_management]

    SystemPrompt --> Decide{🤔 Need sales schema}

    Decide --> LoadSkill[🔧 load_skill<br/>'sales_analytics']

    LoadSkill --> Schema[📊 Schema loaded:<br/>customers, orders tables<br/>+ business logic]

    Schema --> WriteQuery[✍️ Agent writes SQL query<br/>using schema knowledge]

    WriteQuery --> Response([✅ Returns valid SQL<br/>following business rules])

    %% Styling for light and dark modes
    classDef startEnd fill:#F6FFDB,stroke:#6E8900,stroke-width:2px,color:#2E3900
    classDef process fill:#FDF3FF,stroke:#7E65AE,stroke-width:2px,color:#504B5F
    classDef decision fill:#E5F4FF,stroke:#006DDD,stroke-width:2px,color:#030710
    classDef enrichment fill:#EBD0F0,stroke:#885270,stroke-width:2px,color:#441E33

    class Start,Response startEnd
    class SystemPrompt,LoadSkill,WriteQuery process
    class Decide decision
    class Schema enrichment

为什么使用渐进式披露：

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

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

TIP

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

权衡：

延迟：按需加载技能需要额外的工具调用，这会为第一个需要每项技能的请求增加延迟
工作流控制：基本实现依赖提示词来引导技能使用 - 如果没有自定义逻辑，你无法强制执行诸如“总是先尝试技能 A 再尝试技能 B”这样的硬性约束

TIP

实现你自己的技能系统

在构建你自己的技能实现时（如本教程所做的那样），核心概念是渐进式披露 - 即按需加载信息。除此之外，你在实现上拥有完全的灵活性：

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

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

设置

安装

本教程需要 langchain 包：

pipuvconda

bash

pip install langchain

bash

uv add langchain

bash

conda install langchain -c conda-forge

更多细节请参阅我们的安装指南。

LangSmith

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

ShellPython

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()

选择 LLM

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

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

pipuv

bash

pip install -U "langchain[openai]"

bash

uv add "langchain[openai]"

init_chat_modelModel Class

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")

1. 定义技能

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

python

from typing import TypedDict

class Skill(TypedDict):  
    """A skill that can be progressively disclosed to the agent."""
    name: str  # Unique identifier for the skill
    description: str  # 1-2 sentence description to show in system prompt
    content: str  # Full skill content with detailed instructions

现在为 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;
""",
    },
]

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")
    """
    # Find and return the requested skill
    for skill in SKILLS:
        if skill["name"] == skill_name:
            return f"Loaded skill: {skill_name}\n\n{skill['content']}"

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

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

3. 构建技能中间件

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

NOTE

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

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."""

    # Register the load_skill tool as a class variable
    tools = [load_skill]  

    def __init__(self):
        """Initialize and generate the skills prompt from SKILLS."""
        # Build skills prompt from the SKILLS list
        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."""
        # Build the skills addendum
        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."
        )

        # Append to system message content blocks
        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 工具被注册为类变量，从而使智能体可以使用它。

NOTE

生产环境考量：为简单起见，本教程在 __init__ 中加载技能列表。在生产系统中，你可能希望在 before\_agent 钩子中加载技能，从而可以定期刷新以反映最新的更改（例如，当新增技能或修改现有技能时）。有关详细信息，请参阅 before_agent 钩子文档。

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

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

python

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

# Create the agent with skill support
agent = create_agent(
    model,
    system_prompt=(
        "You are a SQL query assistant that helps users "
        "write queries against business databases."
    ),
    middleware=[SkillMiddleware()],  
    checkpointer=InMemorySaver(),
)

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

5. 测试渐进式披露

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

python

from langchain_core.utils.uuid import uuid7

# Configuration for this conversation thread
thread_id = str(uuid7())
config = {"configurable": {"thread_id": thread_id}}

# Ask for a SQL query
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
)

# Print the conversation
for message in result["messages"]:
    if hasattr(message, 'pretty_print'):
        message.pretty_print()
    else:
        print(f"{message.type}: {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]]  # Track which skills have been loaded  #

更新 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
    """
    # Find and return the requested skill
    for skill in SKILLS:
        if skill["name"] == skill_name:
            skill_content = f"Loaded skill: {skill_name}\n\n{skill['content']}"

            # Update state to track loaded skill
            return Command(  
                update={  
                    "messages": [  
                        ToolMessage(  
                            content=skill_content,  
                            tool_call_id=runtime.tool_call_id,  
                        )  
                    ],  
                    "skills_loaded": [skill_name],  
                }  
            )  

    # Skill not found
    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,
                )
            ]
        }
    )

创建受限工具

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

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)
    """
    # Check if the required skill has been loaded
    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."
        )  

    # Validate and format the query
    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."
    )

更新中间件和智能体

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

python

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

    state_schema = CustomState  
    tools = [load_skill, write_sql_query]  

    # ... rest of the middleware implementation stays the same

创建智能体，并使用注册该受限工具的中间件：

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(),
)

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

完整示例

View complete runnable script

下面是一个完整的、可运行的实现，汇总了本教程中的所有部分：

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

# Define skill structure
class Skill(TypedDict):
    """A skill that can be progressively disclosed to the agent."""
    name: str
    description: str
    content: str

# Define skills with schemas and business logic
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;
""",
    },
]

# Create skill loading 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., "sales_analytics", "inventory_management")
    """
    # Find and return the requested skill
    for skill in SKILLS:
        if skill["name"] == skill_name:
            return f"Loaded skill: {skill_name}\n\n{skill['content']}"

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

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

    # Register the load_skill tool as a class variable
    tools = [load_skill]

    def __init__(self):
        """Initialize and generate the skills prompt from SKILLS."""
        # Build skills prompt from the SKILLS list
        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."""
        # Build the skills addendum
        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."
        )

        # Append to system message content blocks
        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)

# Initialize your chat model (replace with your model)
# Example: from langchain_anthropic import ChatAnthropic
# model = ChatAnthropic(model="claude-3-5-sonnet-20241022")
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-5.5")

# Create the agent with skill support
agent = create_agent(
    model,
    system_prompt=(
        "You are a SQL query assistant that helps users "
        "write queries against business databases."
    ),
    middleware=[SkillMiddleware()],
    checkpointer=InMemorySaver(),
)

# Example usage
if __name__ == "__main__":
    # Configuration for this conversation thread
    thread_id = str(uuid7())
    config = {"configurable": {"thread_id": thread_id}}

    # Ask for a SQL query
    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
    )

    # Print the conversation
    for message in result["messages"]:
        if hasattr(message, 'pretty_print'):
            message.pretty_print()
        else:
            print(f"{message.type}: {message.content}")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248

这个完整示例包含：

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

要运行它，你需要：

安装所需的包：pip install langchain langchain-openai langgraph
设置你的 API key（例如 export OPENAI_API_KEY=...）
把模型初始化替换为你偏好的 LLM 服务商

实现变体

查看实现方案与取舍

本教程把技能实现为通过工具调用加载的内存中 Python 字典。不过，要用技能实现渐进式披露，有几种方式：

存储后端：

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

技能发现（智能体如何得知存在哪些技能）：

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

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

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

规模考量（未经校准的心智模型 - 请针对你的系统进行优化）：

小型技能（< 1K tokens / ~750 个单词）：可直接放进系统提示词，并用提示词缓存来节省成本、加快响应
中型技能（1-10K tokens / ~750-7.5K 个单词）：适合按需加载，以避免上下文开销（本教程）
大型技能（> 10K tokens / ~7.5K 个单词，或 > 上下文窗口的 5-10%）：应使用分页、基于搜索的加载或分层探索等渐进式披露技术，避免占用过多上下文

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

渐进式披露与上下文工程

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

渐进式披露本质上是一种**上下文工程技术** - 你在管理智能体在何时能获取哪些信息。本教程聚焦于加载数据库 schema，但同样的原则也适用于其他类型的上下文。

与少样本提示结合

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

示例思路：

用户提问："找出 6 个月内没有下过单的客户"
智能体加载 sales_analytics schema（如本教程所示）
智能体还会加载 2-3 条相关的示例查询（通过语义搜索或基于标签的查找）：
- 用于查找不活跃客户的查询
- 带日期过滤的查询
- 联接客户表和订单表的查询
智能体结合 schema 知识和示例模式来编写查询

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

后续步骤

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

数据集

创建数据集

运行评估

评估技术

定义评估目标

打分方法

实验配置

多模态评估

分析实验结果

微调

教程

常见数据类型

评估器类型

UI

SDK

框架与集成

改进评估器

反馈

追踪设置

集成

手动埋点

配置与故障排除

项目与环境设置

高级追踪技术

数据与隐私

故障排除指南

查看与管理追踪

批量导出追踪数据

轨迹视图

数据类型参考

监控与告警

在线评估器

自动化

开发你的应用

设置依赖

持久化

能力

助手

运行

并发消息处理

如何构建

鉴权与访问控制

自定义鉴权教程

服务器定制

静态加密

替换内置后端

扩展 HTTP 服务器

请求头与日志

部署指南

参考

配置

参考

提示词

创建与管理提示词

连接模型

上下文中枢

教程

全栈 Web 应用

Agent Server API

控制平面 API

计费与用量

参考

按云服务商起步

用 Terraform 部署

AWS

GCP

Azure

安装实例

配置实例

连接外部服务

平台鉴权与访问控制

管理实例

自托管可观测性

混合部署

脚本

参考

组织