> ## Documentation Index
> Fetch the complete documentation index at: https://langchain-zh.cn/llms.txt
> Use this file to discover all available pages before exploring further.
# 使用按需技能构建 SQL 助手
本教程展示了如何使用**渐进式披露**（progressive disclosure）——一种上下文管理技术，其中代理按需加载信息而非预先加载——来实现**技能**（基于特定提示的专用指令）。代理通过工具调用来加载技能，而不是动态更改系统提示词，仅发现并加载每个任务所需的技能。
**用例：** 想象构建一个代理来帮助大型企业中不同业务领域的 SQL 查询编写。您的组织可能为每个领域拥有独立的数据存储，或者拥有一个包含数千张表的单一单体数据库。无论哪种方式，预先加载所有模式都会使上下文窗口不堪重负。渐进式披露通过在需要时仅加载相关模式来解决此问题。这种架构还允许不同的产品负责人和利益相关者独立贡献和维护其特定业务领域的技能。
**您将构建什么：** 一个具有两个技能（销售分析和库存管理）的 SQL 查询助手。代理在系统提示词中查看轻量级的技能描述，然后通过工具调用仅在与其用户查询相关时才加载完整的数据库模式和业务逻辑。
有关带有查询执行、错误纠正和验证的 SQL 代理的完整示例，请参阅我们的 [SQL 代理教程](/oss/python/langchain/sql-agent)。本教程侧重于可应用于任何领域的渐进式披露模式。
渐进式披露由 Anthropic 推广，作为构建可扩展代理技能系统的技术。这种方法使用三级架构（元数据 → 核心内容 → 详细资源），其中代理仅在需要时加载信息。有关此技术的更多信息，请参阅 [为现实世界配备代理技能](https://www.anthropic.com/engineering/equipping-agents-for-the-real-world-with-agent-skills)。
## 工作原理
以下是用户请求 SQL 查询时的流程：
```mermaid theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
%%{init: {'theme':'base', 'themeVariables': { 'primaryColor':'#4CAF50','primaryTextColor':'#fff','primaryBorderColor':'#2E7D32','lineColor':'#666','secondaryColor':'#FF9800','tertiaryColor':'#2196F3','tertiaryBorderColor':'#1565C0','tertiaryTextColor':'#fff'}}}%%
flowchart TD
Start([💬 User: Write SQL query
for high-value customers]) --> SystemPrompt[📋 Agent sees skill descriptions:
• sales_analytics
• inventory_management]
SystemPrompt --> Decide{🤔 Need sales schema}
Decide --> LoadSkill[🔧 load_skill
'sales_analytics']
LoadSkill --> Schema[📊 Schema loaded:
customers, orders tables
+ business logic]
Schema --> WriteQuery[✍️ Agent writes SQL query
using schema knowledge]
WriteQuery --> Response([✅ Returns valid SQL
following business rules])
%% Styling for light and dark modes
classDef startEnd fill:#DCFCE7,stroke:#16A34A,stroke-width:2px,color:#14532D
classDef process fill:#FEF3C7,stroke:#F59E0B,stroke-width:2px,color:#78350F
classDef decision fill:#DBEAFE,stroke:#2563EB,stroke-width:2px,color:#1E3A8A
classDef enrichment fill:#F3E8FF,stroke:#9333EA,stroke-width:2px,color:#581C87
class Start,Response startEnd
class SystemPrompt,LoadSkill,WriteQuery process
class Decide decision
class Schema enrichment
```
**为什么需要渐进式披露：**
* **减少上下文使用** - 仅加载任务所需的 2-3 个技能，而不是所有可用技能
* **实现团队自主权** - 不同团队可以独立开发专用技能（类似于其他多代理架构）
* **高效扩展** - 添加数十或数百个技能而不会使上下文不堪重负
* **简化对话历史** - 单个代理，一个对话线程
**什么是技能：** 技能（如 Claude Code 所推广的）主要是基于提示的：针对特定业务任务的自包含专用指令单元。在 Claude Code 中，技能作为文件系统中的目录暴露，通过文件操作发现。技能通过提示词指导行为，并提供有关工具使用的信息，或包含供编码代理执行的示例代码。
具有渐进式披露的技能可以被视为一种 [RAG（检索增强生成）](/oss/python/langchain/rag)，其中每个技能都是一个检索单元——尽管不一定由嵌入或关键词搜索支持，而是由浏览内容的工具支持（如文件操作，或在本文中，直接查找）。
**权衡：**
* **延迟**：按需加载技能需要额外的工具调用，这会增加首次需要每个技能的请求的延迟
* **工作流控制**：基本实现依赖于提示词来引导技能使用——如果没有自定义逻辑，您无法强制执行硬性约束，如“始终先尝试技能 A 再尝试技能 B"
**实现您自己的技能系统**
在构建您自己的技能实现（正如我们在本教程中所做的那样）时，核心概念是渐进式披露——按需加载信息。除此之外，您在实现方面拥有完全灵活性：
* **存储**：数据库、S3、内存数据结构或任何后端
* **发现**：直接查找（本教程）、用于大型技能集合的 RAG、文件系统扫描或 API 调用
* **加载逻辑**：自定义延迟特性并添加逻辑以搜索技能内容或排名相关性
* **副作用**：定义技能加载时发生的情况，例如暴露与该技能关联的工具（在第 8 节中涵盖）
这种灵活性使您能够针对性能、存储和工作流控制方面的特定需求进行优化。
## 设置
### 安装
本教程需要 `langchain` 包：
```bash pip theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
pip install langchain
```
```bash uv theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
uv add langchain
```
```bash conda theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
conda install langchain -c conda-forge
```
更多详细信息，请参阅我们的 [安装指南](/oss/python/langchain/install)。
### LangSmith
设置 [LangSmith](https://smith.langchain.com) 以检查代理内部正在发生的事情。然后设置以下环境变量：
```bash bash theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
export LANGSMITH_TRACING="true"
export LANGSMITH_API_KEY="..."
```
```python python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
import getpass
import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = getpass.getpass()
```
### 选择 LLM
从 LangChain 的集成套件中选择聊天模型：
👉 阅读 [OpenAI 聊天模型集成文档](/oss/python/integrations/chat/openai/)
```shell theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
pip install -U "langchain[openai]"
```
```python init_chat_model theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
import os
from langchain.chat_models import init_chat_model
os.environ["OPENAI_API_KEY"] = "sk-..."
model = init_chat_model("gpt-5.2")
```
```python Model Class theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
import os
from langchain_openai import ChatOpenAI
os.environ["OPENAI_API_KEY"] = "sk-..."
model = ChatOpenAI(model="gpt-5.2")
```
👉 阅读 [Anthropic 聊天模型集成文档](/oss/python/integrations/chat/anthropic/)
```shell theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
pip install -U "langchain[anthropic]"
```
```python init_chat_model theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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 Model Class theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
import os
from langchain_anthropic import ChatAnthropic
os.environ["ANTHROPIC_API_KEY"] = "sk-..."
model = ChatAnthropic(model="claude-sonnet-4-6")
```
👉 阅读 [Azure 聊天模型集成文档](/oss/python/integrations/chat/azure_chat_openai/)
```shell theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
pip install -U "langchain[openai]"
```
```python init_chat_model theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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.2",
azure_deployment=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"],
)
```
```python Model Class theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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.2",
azure_deployment=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"]
)
```
👉 阅读 [Google GenAI 聊天模型集成文档](/oss/python/integrations/chat/google_generative_ai/)
```shell theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
pip install -U "langchain[google-genai]"
```
```python init_chat_model theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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 Model Class theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
import os
from langchain_google_genai import ChatGoogleGenerativeAI
os.environ["GOOGLE_API_KEY"] = "..."
model = ChatGoogleGenerativeAI(model="gemini-2.5-flash-lite")
```
👉 阅读 [AWS Bedrock 聊天模型集成文档](/oss/python/integrations/chat/bedrock/)
```shell theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
pip install -U "langchain[aws]"
```
```python init_chat_model theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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",
)
```
```python Model Class theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from langchain_aws import ChatBedrock
model = ChatBedrock(model="anthropic.claude-3-5-sonnet-20240620-v1:0")
```
👉 阅读 [HuggingFace 聊天模型集成文档](/oss/python/integrations/chat/huggingface/)
```shell theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
pip install -U "langchain[huggingface]"
```
```python init_chat_model theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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 Model Class theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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)
```
## 1. 定义技能
首先，定义技能的结构。每个技能都有一个名称、简要描述（显示在系统提示词中）和完整内容（按需加载）：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from typing import TypedDict
class Skill(TypedDict): # [!code highlight]
"""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 theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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 theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from langchain.tools import tool
@tool # [!code highlight]
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']}" # [!code highlight]
# Skill not found
available = ", ".join(s["name"] for s in SKILLS)
return f"Skill '{skill_name}' not found. Available skills: {available}"
```
`load_skill` 工具返回完整的技能内容作为字符串，该字符串成为对话的一部分作为 ToolMessage。有关创建和使用工具的更多详细信息，请参阅 [工具指南](/oss/python/langchain/tools)。
## 3. 构建技能中间件
创建自定义中间件，将技能描述注入到系统提示词中。此中间件使技能可被发现，而无需预先加载其完整内容。
本指南演示了创建自定义中间件。有关中间件概念和模式的全面指南，请参阅 [自定义中间件文档](/oss/python/langchain/middleware/custom)。
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from langchain.agents.middleware import ModelRequest, ModelResponse, AgentMiddleware
from langchain.messages import SystemMessage
from typing import Callable
class SkillMiddleware(AgentMiddleware): # [!code highlight]
"""Middleware that injects skill descriptions into the system prompt."""
# Register the load_skill tool as a class variable
tools = [load_skill] # [!code highlight]
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 = ( # [!code highlight]
f"\n\n## Available Skills\n\n{self.skills_prompt}\n\n" # [!code highlight]
"Use the load_skill tool when you need detailed information " # [!code highlight]
"about handling a specific type of request." # [!code highlight]
)
# 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` 工具注册为类变量，使其对代理可用。
**生产注意事项**：本教程为了简单起见在 `__init__` 中加载技能列表。在生产系统中，您可能希望在 `before_agent` 钩子中加载技能，以便定期刷新它们以反映最新更改（例如，当添加新技能或修改现有技能时）。有关详细信息，请参阅 [before\_agent 钩子文档](/oss/python/langchain/middleware/custom#node-style-hooks)。
## 4. 创建支持技能的代理
现在创建带有技能中间件和用于状态持久化的检查点器的代理：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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()], # [!code highlight]
checkpointer=InMemorySaver(),
)
```
代理现在可以在其系统提示词中访问技能描述，并在需要时调用 `load_skill` 以检索完整的技能内容。检查点器维护跨轮次的对话历史。
## 5. 测试渐进式披露
用需要特定技能知识的问题测试代理：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
import uuid
# Configuration for this conversation thread
thread_id = str(uuid.uuid4())
config = {"configurable": {"thread_id": thread_id}}
# Ask for a SQL query
result = agent.invoke( # [!code highlight]
{
"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 theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from langchain.agents.middleware import AgentState
class CustomState(AgentState): # [!code highlight]
skills_loaded: NotRequired[list[str]] # Track which skills have been loaded # [!code highlight]
```
### 更新 load\_skill 以修改状态
修改 `load_skill` 工具以在加载技能时更新状态：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from langgraph.types import Command # [!code highlight]
from langchain.tools import tool, ToolRuntime
from langchain.messages import ToolMessage # [!code highlight]
@tool
def load_skill(skill_name: str, runtime: ToolRuntime) -> Command: # [!code highlight]
"""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( # [!code highlight]
update={ # [!code highlight]
"messages": [ # [!code highlight]
ToolMessage( # [!code highlight]
content=skill_content, # [!code highlight]
tool_call_id=runtime.tool_call_id, # [!code highlight]
) # [!code highlight]
], # [!code highlight]
"skills_loaded": [skill_name], # [!code highlight]
} # [!code highlight]
) # [!code highlight]
# 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 theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
@tool
def write_sql_query( # [!code highlight]
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", []) # [!code highlight]
if vertical not in skills_loaded: # [!code highlight]
return ( # [!code highlight]
f"Error: You must load the '{vertical}' skill first " # [!code highlight]
f"to understand the database schema before writing queries. " # [!code highlight]
f"Use load_skill('{vertical}') to load the schema." # [!code highlight]
) # [!code highlight]
# 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."
)
````
### 更新中间件和代理
更新中间件以使用自定义状态模式：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
class SkillMiddleware(AgentMiddleware[CustomState]): # [!code highlight]
"""Middleware that injects skill descriptions into the system prompt."""
state_schema = CustomState # [!code highlight]
tools = [load_skill, write_sql_query] # [!code highlight]
# ... rest of the middleware implementation stays the same
```
使用注册受限工具的中间件创建代理：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
agent = create_agent(
model,
system_prompt=(
"You are a SQL query assistant that helps users "
"write queries against business databases."
),
middleware=[SkillMiddleware()], # [!code highlight]
checkpointer=InMemorySaver(),
)
```
现在，如果代理尝试在加载所需技能之前使用 `write_sql_query`，它将收到错误消息，提示它首先加载适当的技能（例如 `sales_analytics` 或 `inventory_management`）。这确保代理在尝试验证查询之前拥有必要的模式知识。
## 完整示例
以下是结合本教程中所有部分的完整可运行实现：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
import uuid
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-4")
# 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(uuid.uuid4())
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}")
```
此完整示例包括：
* 带有完整数据库模式的技能定义
* 用于按需加载的 `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/读取操作技能文件）
* **分层**：首先加载技能概述，然后钻取到特定子部分
**大小考虑**（未校准的心理模型 - 根据您的系统进行优化）：
* **小型技能**（\< 1K tokens / \~750 words）：可以直接包含在系统提示词中，并使用提示词缓存进行缓存以节省成本并加快响应速度
* **中型技能**（1-10K tokens / \~750-7.5K words）：受益于按需加载以避免上下文开销（本教程）
* **大型技能**（> 10K tokens / \~7.5K words，或 > 5-10% 的上下文窗口）：应使用分页、基于搜索的加载或分层探索等渐进式披露技术，以避免消耗过多的上下文
选择取决于您的需求：内存中是最快的，但需要重新部署才能更新技能，而基于文件或远程存储 enables 动态技能管理而无需代码更改。
## 渐进式披露与上下文工程
渐进式披露本质上是一种 **[上下文工程](/oss/python/langchain/context-engineering) 技术** —— 您正在管理代理何时可获得哪些信息。本教程侧重于加载数据库模式，但相同的原则适用于其他类型的上下文。
### 结合少样本提示
对于 SQL 查询用例，您可以扩展渐进式披露以动态加载与用户查询匹配的 **少样本示例**：
**示例方法：**
1. 用户询问："查找 6 个月未下单的客户"
2. 代理加载 `sales_analytics` 模式（如本教程所示）
3. 代理还加载 2-3 个相关的示例查询（通过语义搜索或基于标签的查找）：
* 查找不活跃客户的查询
* 基于日期过滤的查询
* 连接客户和订单表的查询
4. 代理使用模式知识和示例模式编写查询
这种渐进式披露（按需加载模式）和动态少样本提示（加载相关示例）的结合创造了一种强大的上下文工程模式，可扩展到大型知识库，同时提供高质量、接地气的输出。
## 下一步
* 了解 [中间件](/oss/python/langchain/middleware) 以实现更动态的代理行为
* 探索 [上下文工程](/oss/python/langchain/context-engineering) 技术以管理代理上下文
* 探索 [交接模式](/oss/python/langchain/multi-agent/handoffs-customer-support) 以实现顺序工作流
* 阅读 [子代理模式](/oss/python/langchain/multi-agent/subagents-personal-assistant) 以实现并行任务路由
* 查看 [多代理模式](/oss/python/langchain/multi-agent) 以获取其他专用代理方法
* 使用 [LangSmith](https://smith.langchain.com) 调试和监控技能加载
***
[Edit this page on GitHub](https://github.com/langchain-ai/docs/edit/main/src/i18n\zh-CN\oss\langchain\multi-agent\skills-sql-assistant.mdx) or [file an issue](https://github.com/langchain-ai/docs/issues/new/choose).
[Connect these docs](/use-these-docs) to Claude, VSCode, and more via MCP for real-time answers.