技能:SQL 助手
概述
本教程展示了如何使用渐进式披露——一种按需加载信息而非一次性加载的上下文管理技术——来实现技能(专门化的基于提示词的指令)。代理通过工具调用加载技能,而不是动态更改系统提示,只为每个任务发现和加载所需的技能。
使用场景: 想象构建一个帮助编写跨大型企业不同业务垂直领域的 SQL 查询的代理。您的组织可能为每个垂直领域设有独立的数据存储,或者有一个包含数千张表的单一数据库。无论哪种情况,一次性加载所有模式会压垮上下文窗口。渐进式披露通过仅在需要时加载相关模式来解决这个问题。这种架构还使不同的产品负责人和利益相关者能够独立贡献和维护其特定业务垂直领域的技能。
您将构建的内容: 一个具有两个技能(销售分析和库存管理)的 SQL 查询助手。代理在其系统提示中看到轻量级的技能描述,然后仅在与用户查询相关时通过工具调用加载完整的数据库模式。
关于带查询执行、错误修正和验证的 SQL 代理完整示例,请参见我们的 SQL 代理教程。本教程专注于可应用于任何领域的渐进式披露模式。
渐进式披露由 Anthropic 推广,作为构建可扩展代理技能系统的一种技术。该方法使用三级架构(元数据 → 核心内容 → 详细资源),代理仅在需要时加载信息。更多信息请参见 Equipping agents for the real world with Agent Skills。
工作原理
以下是用户请求 SQL 查询时的流程:
flowchart TD
Start([💬 用户:为高价值客户<br/>编写 SQL 查询]) --> SystemPrompt[📋 代理看到技能描述:<br/>• sales_analytics<br/>• inventory_management]
SystemPrompt --> Decide{🤔 需要销售模式}
Decide --> LoadSkill[🔧 load_skill<br/>'sales_analytics']
LoadSkill --> Schema[📊 模式已加载:<br/>customers, orders 表<br/>+ 业务逻辑]
Schema --> WriteQuery[✍️ 代理使用模式知识<br/>编写 SQL 查询]
WriteQuery --> Response([✅ 返回遵循<br/>业务规则的有效 SQL])为什么使用渐进式披露:
- 减少上下文使用 — 只加载任务所需的 2-3 个技能,而不是所有可用技能
- 支持团队自治 — 不同团队可以独立开发专门化技能(类似于其他多代理架构)
- 高效扩展 — 添加数十或数百个技能而不会压垮上下文
- 简化对话历史 — 单个代理,一条对话线程
什么是技能: 技能,由 Claude Code 推广,主要是基于提示词的:自包含的专门化指令单元,用于特定的业务任务。在 Claude Code 中,技能以文件系统中的目录形式暴露,通过文件操作发现。技能通过提示词引导行为,可以提供关于工具使用方式的信息,或包含供编码代理执行的示例代码。
具有渐进式披露的技能可以被视为一种 RAG(检索增强生成) 形式,其中每个技能是一个检索单元——尽管不一定由嵌入或关键词搜索支持,而是通过浏览内容的工具(如文件操作或本教程中的直接查找)。
权衡:
- 延迟:按需加载技能需要额外的工具调用,这增加了需要每个技能的首次请求的延迟
- 工作流控制:基本实现依赖提示词来引导技能使用——没有自定义逻辑就无法强制执行硬约束,如"总是先尝试技能 A 再尝试技能 B"
实现您自己的技能系统
在构建自己的技能实现时(如本教程所示),核心概念是渐进式披露——按需加载信息。除此之外,您有完全的实现灵活性:
- 存储:数据库、S3、内存数据结构或任何后端
- 发现:直接查找(本教程)、大规模技能集合的 RAG、文件系统扫描或 API 调用
- 加载逻辑:自定义延迟特征,添加搜索技能内容或排序相关性逻辑
- 副作用:定义技能加载时发生什么,如暴露与该技能关联的工具
这种灵活性使您能够针对性能、存储和工作流控制方面的特定需求进行优化。
设置
安装
本教程需要 langchain 包:
pip install langchain更多详情请参阅我们的安装指南。
LangSmith
设置 LangSmith:
import getpass
import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = getpass.getpass()选择模型
选择一个聊天模型:
from langchain.chat_models import init_chat_model
# OpenAI
model = init_chat_model("openai:gpt-5.4")
# Anthropic
model = init_chat_model("anthropic:claude-sonnet-4-6")
# Google Gemini
model = init_chat_model("google_genai:gemini-2.5-flash-lite")1. 定义技能
首先定义技能的结构。每个技能有一个名称、一个简要描述(在系统提示中显示)和完整内容(按需加载):
from typing import TypedDict
class Skill(TypedDict):
"""一个可以渐进式披露给代理的技能。"""
name: str # 技能的唯一标识符
description: str # 在系统提示中显示的 1-2 句描述
content: str # 包含详细说明的完整技能内容现在定义 SQL 查询助手的示例技能。技能设计为描述轻量级(预先展示给代理)但内容详细(仅在需要时加载):
SKILLS: list[Skill] = [
{
"name": "sales_analytics",
"description": "用于销售数据分析的数据库模式和业务逻辑,包括客户、订单和收入。",
"content": """# 销售分析模式
## 表
### customers
- customer_id (主键)
- name
- email
- signup_date
- status (active/inactive)
- customer_tier (bronze/silver/gold/platinum)
### orders
- order_id (主键)
- customer_id (外键 -> customers)
- order_date
- status (pending/completed/cancelled/refunded)
- total_amount
- sales_region (north/south/east/west)
### order_items
- item_id (主键)
- order_id (外键 -> orders)
- product_id
- quantity
- unit_price
- discount_percent
## 业务逻辑
**活跃客户**:status = 'active' AND signup_date <= CURRENT_DATE - INTERVAL '90 days'
**收入计算**:只计入 status = 'completed' 的订单。使用 orders 表中的 total_amount,该字段已考虑折扣。
**客户生命周期价值 (CLV)**:客户所有已完成订单金额的总和。
**高价值订单**:total_amount > 1000 的订单
## 示例查询
-- 获取上季度收入最高的前 10 名客户
SELECT
c.customer_id,
c.name,
c.customer_tier,
SUM(o.total_amount) as total_revenue
FROM customers c
JOIN orders o ON c.customer_id = o.customer_id
WHERE o.status = 'completed'
AND o.order_date >= CURRENT_DATE - INTERVAL '3 months'
GROUP BY c.customer_id, c.name, c.customer_tier
ORDER BY total_revenue DESC
LIMIT 10;
""",
},
{
"name": "inventory_management",
"description": "用于库存跟踪的数据库模式和业务逻辑,包括产品、仓库和库存水平。",
"content": """# 库存管理模式
## 表
### products
- product_id (主键)
- product_name
- sku
- category
- unit_cost
- reorder_point (补货前的最低库存水平)
- discontinued (布尔值)
### warehouses
- warehouse_id (主键)
- warehouse_name
- location
- capacity
### inventory
- inventory_id (主键)
- product_id (外键 -> products)
- warehouse_id (外键 -> warehouses)
- quantity_on_hand
- last_updated
### stock_movements
- movement_id (主键)
- product_id (外键 -> products)
- warehouse_id (外键 -> warehouses)
- movement_type (inbound/outbound/transfer/adjustment)
- quantity (入库为正,出库为负)
- movement_date
- reference_number
## 业务逻辑
**可用库存**:来自 inventory 表的 quantity_on_hand,其中 quantity_on_hand > 0
**需要补货的产品**:所有仓库的 quantity_on_hand 总和小于或等于产品 reorder_point 的产品
**仅活跃产品**:除非专门分析已停产产品,否则排除 discontinued = true 的产品
**库存估值**:每个产品的 quantity_on_hand * unit_cost
## 示例查询
-- 查找所有仓库中低于补货点的产品
SELECT
p.product_id,
p.product_name,
p.reorder_point,
SUM(i.quantity_on_hand) as total_stock,
p.unit_cost,
(p.reorder_point - SUM(i.quantity_on_hand)) as units_to_reorder
FROM products p
JOIN inventory i ON p.product_id = i.product_id
WHERE p.discontinued = false
GROUP BY p.product_id, p.product_name, p.reorder_point, p.unit_cost
HAVING SUM(i.quantity_on_hand) <= p.reorder_point
ORDER BY units_to_reorder DESC;
""",
},
]2. 创建技能加载工具
创建一个按需加载完整技能内容的工具:
from langchain.tools import tool
@tool
def load_skill(skill_name: str) -> str:
"""加载指定技能的完整内容,包括数据库模式、业务逻辑和示例查询。
可用技能:
- sales_analytics:销售数据模式(customers、orders、order_items 表)和业务逻辑
- inventory_management:库存模式(products、warehouses、inventory、stock_movements 表)和业务逻辑
将完整的技能内容注入到对话中,以便为编写 SQL 查询提供所需的模式信息。
"""
for skill in SKILLS:
if skill["name"] == skill_name:
return skill["content"]
return f"错误:未找到技能 '{skill_name}'"这个工具作为按需加载机制——代理只有在处理需要该特定模式信息的查询时才会调用它。
3. 创建代理
现在创建一个了解可用技能并知道如何加载它们的代理:
from langchain.agents import create_agent
AVAILABLE_SKILLS_DESCRIPTION = "\n".join(
f"- {s['name']}: {s['description']}" for s in SKILLS
)
SYSTEM_PROMPT = f"""你是一个 SQL 查询助手。你帮助用户为不同的业务领域编写 SQL 查询。
可用技能:
{AVAILABLE_SKILLS_DESCRIPTION}
工作流程:
1. 分析用户请求,确定需要哪个知识领域
2. 使用 load_skill 工具加载相关技能的内容(模式 + 业务逻辑)
3. 使用加载的模式信息编写准确的 SQL 查询
4. 如果用户请求涉及多个领域,加载多个技能
注意:一次加载多个技能可能会增加延迟。首先加载最相关的技能。
"""
sql_agent = create_agent(
model,
tools=[load_skill],
system_prompt=SYSTEM_PROMPT,
)4. 测试技能加载
现在测试代理如何按需发现和加载技能:
query = "编写一个 SQL 查询,找出过去 90 天内最活跃的客户"
for step in sql_agent.stream(
{"messages": [{"role": "user", "content": query}]}
):
for update in step.values():
for message in update.get("messages", []):
message.pretty_print()代理的推理过程:
- 识别需求:用户需要"活跃客户"——这在
sales_analytics技能中定义 - 加载技能:调用
load_skill("sales_analytics")获取完整的模式和业务逻辑 - 使用模式:观察到
customers表有signup_date和status字段,业务逻辑定义"活跃客户"为status = 'active'且注册超过 90 天 - 编写查询:根据加载的模式信息生成准确的 SQL:
SELECT
customer_id,
name,
email,
signup_date,
customer_tier
FROM customers
WHERE status = 'active'
AND signup_date <= CURRENT_DATE - INTERVAL '90 days'
ORDER BY signup_date DESC;5. 多技能场景
当查询跨越多个领域时,代理可以按顺序加载多个技能:
query = "列出当前库存低于补货点的产品,以及我们最活跃客户的销售区域分布"
for step in sql_agent.stream(
{"messages": [{"role": "user", "content": query}]}
):
for update in step.values():
for message in update.get("messages", []):
message.pretty_print()代理会先后调用 load_skill("inventory_management") 和 load_skill("sales_analytics"),分别加载两个领域的模式和业务逻辑,然后根据两组信息生成 SQL 查询。
6. 高级:技能加载时暴露工具
在某些实现中,加载技能还可以暴露与该技能关联的新工具。这在技能涉及动态工具注册时很有用:
# 技能到工具的映射
SKILL_TOOLS = {
"sales_analytics": [],
"inventory_management": [],
}
@tool
def get_warehouse_capacity(warehouse_id: str) -> str:
"""获取特定仓库的当前容量使用情况。"""
return f"仓库 {warehouse_id} 当前容量利用率:75%"
@tool
def get_top_products(n: int = 5) -> str:
"""获取销量最高的前 N 个产品。"""
return f"前 {n} 个产品:产品A(1200件)、产品B(980件)、产品C(750件)"
# 在更高级的设置中,load_skill 可以返回新工具的定义
# 代理可以使用这些工具进行后续操作概念回顾
在本教程中,您学到了:
- 渐进式披露 — 按需加载信息以管理上下文窗口的使用
- 技能系统 — 自包含的专门化指令单元,用于特定的业务领域
load_skill工具 — 代理在需要时调用的按需加载机制- 轻量描述 + 详细内容 — 描述始终可见,完整内容仅在需要时加载
- 单代理、多技能 — 一个代理实例,根据查询动态加载不同的专业知识
这种方法可以扩展到成百上千个技能,而不会压垮上下文。每个技能由独立的团队维护,使得知识管理去中心化且可扩展。
下一步
- 探索子代理模式,了解监督者协调多个专业代理的方式
- 学习路由器模式,构建并行查询多个知识源的系统
- 查看完整功能的 SQL 代理教程,包括查询执行和错误修正
- 深入了解 Anthropic 的 Agent Skills 文章