LangChain多智能体之Skills

本文langchain版本为1.2.9

在 技能（Skills）架构 中，特定能力被封装为可调用的“技能”，用于增强 智能体（agent） 的行为。这些技能本质上是由专用提示词（prompt）驱动的专项能力，智能体可根据需要随时调用。

如需内置技能支持，请参阅 深度智能体（Deep Agents）。

💡 这一模式在概念上与 Agent Skills 和 Jeremy Howard 提出的 llms.txt 高度一致——后者通过工具调用（tool calling）实现文档的渐进式披露（progressive disclosure）。而技能模式则将这种“渐进式披露”应用于专用提示词和领域知识，而不仅限于文档页面。

核心特征

• 提示词驱动的专项能力：技能主要由专用提示词定义
• 渐进式披露：技能根据上下文或用户需求动态启用
• 团队协作友好：不同团队可独立开发和维护各自的技能
• 轻量级组合：技能比完整的子智能体（sub-agent）更简洁
• 资源引用感知：技能可引用脚本、模板及其他外部资源

适用场景

当您希望构建一个具备多种专项能力的单一 智能体，且满足以下任一条件时，推荐使用技能模式：

• 不需要在技能之间施加强制性约束
• 多个团队需独立开发各自的能力模块

典型应用场景包括：

• 编程助手：为不同语言或任务提供专属技能（如 Python 调试、SQL 生成）
• 知识库系统：按领域划分技能（如法律、医疗、金融）
• 创意助手：针对不同输出格式定制技能（如写诗、写邮件、生成剧本）

基础实现示例

from langchain.tools import tool
from langchain.agents import create_agent

@tool
def load_skill(skill_name: str) –> str:
"""加载一个专用技能的提示词。

可用技能包括：
– write_sql：SQL 查询编写专家
– review_legal_doc：法律文档审阅员

返回该技能对应的提示词及上下文信息。
"""
# 从文件或数据库中加载技能内容
...

agent = create_agent(
model="gpt-4.1",
tools=[load_skill],
system_prompt=(
"你是一个乐于助人的助手。"
"你拥有两项技能：write_sql 和 review_legal_doc。"
"请通过 load_skill 工具来调用它们。"
),
)

模式的扩展方式

在自定义实现中，您可以通过以下方式对基础技能模式进行增强：

• 动态工具注册：将渐进式披露与状态管理结合，在加载技能的同时动态注册新的 工具（tools）。 例如，加载 “database_admin” 技能时，不仅能注入专用上下文，还能自动注册数据库相关的工具（如备份、恢复、迁移）。这利用了多智能体架构中通用的“工具+状态”机制——通过工具修改状态，从而动态改变智能体的能力。
• 分层技能结构：技能可定义其他子技能，形成树状层级结构，实现嵌套式专项能力。 例如，加载 “data_science” 技能后，可进一步提供 “pandas_expert”、“visualization” 和 “statistical_analysis” 等子技能。每个子技能均可按需独立加载，实现对领域知识的细粒度渐进披露。这种分层组织方式有助于管理庞大的知识体系，将能力按逻辑分组，便于发现和按需加载。
• 资源引用感知：尽管每个技能仅包含一个提示词，但该提示词可指向其他资产（如文件、API、模板）的位置，并说明何时应使用这些资源。

当相关资产变得必要时，智能体会意识到它们的存在，并按需将其读入内存以完成任务。这也遵循了渐进式披露原则，有效控制上下文窗口中的信息量。

构建一个支持按需加载技能的 SQL 助手

本教程将演示如何使用 渐进式披露（progressive disclosure） ——一种上下文管理技术，让智能体仅在需要时才加载信息，而非一次性全部载入——来实现 技能（skills）（即基于专用提示词的专项指令）。智能体通过工具调用（tool calls）来加载技能，而不是动态修改系统提示词（system prompt），从而只为当前任务发现并加载所需的技能。

📝 若想了解一个功能更完整的 SQL 智能体（包含查询执行、错误修正和验证），请参阅官网的 SQL 智能体教程。本教程聚焦于 渐进式披露模式，该模式可应用于任何领域。

💡 渐进式披露由 Anthropic 推广，作为构建可扩展智能体技能系统的一种技术。该方法采用三层架构（元数据 → 核心内容 → 详细资源），智能体仅在需要时才加载信息。更多详情，请参阅 《为现实世界赋能智能体：Agent Skills》。

工作原理

当用户请求生成 SQL 查询时，流程如下：

为何采用渐进式披露：

• 减少上下文占用：仅加载任务所需的 2–3 项技能，而非全部可用技能
• 支持团队自治：不同团队可独立开发专属技能（与其他多智能体架构类似）
• 高效扩展：可轻松添加数十甚至上百项技能，而不会压垮上下文
• 简化对话历史：单个智能体维护单一对话线程

什么是技能（Skills）：

技能概念由 Claude Code 推广，本质上是基于提示词的：即针对特定业务任务的自包含专用指令单元。在 Claude Code 中，技能以文件系统中的目录形式存在，通过文件操作被发现。技能通过提示词引导行为，并可提供工具使用说明，或包含供编码智能体执行的示例代码。

💡 带有渐进式披露的技能可视为一种 RAG（检索增强生成） 形式，其中每项技能都是一个检索单元——尽管不一定依赖嵌入（embeddings）或关键词搜索，而是通过浏览内容的工具（如文件操作，或本教程中的直接查找）来实现。

权衡取舍：

• 延迟增加：按需加载技能需要额外的工具调用，会增加首次请求的响应延迟
• 工作流控制受限：基础实现依赖提示词引导技能使用——若无自定义逻辑，无法强制施加硬性约束（例如“必须先尝试技能 A，再尝试技能 B”）

💡 构建您自己的技能系统 在自行实现技能系统时（如本教程所示），核心理念是渐进式披露——按需加载信息。在此基础上，您拥有完全的实现灵活性：

• 存储方式：数据库、S3、内存数据结构，或任意后端
• 发现机制：直接查找（本教程）、大规模技能集的 RAG、文件系统扫描，或 API 调用
• 加载逻辑：自定义延迟特性，或添加搜索/相关性排序逻辑
• 副作用：定义技能加载时的行为，例如暴露与该技能关联的工具（见第 8 节） 这种灵活性让您可根据性能、存储和工作流控制的具体需求进行优化。

环境准备

安装

本教程需要 langchain 包：

pip install langchain

或用uv

uv add langchain

或用conda

conda install langchain -c conda-forge

更多详情，请参阅官网的 安装指南。

LangSmith

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

import getpass
import os

os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = getpass.getpass()

选择 LLM

从 LangChain 支持的模型中选择一个聊天模型：

# pip install -U "langchain[openai]"
import os
from langchain_openai import ChatOpenAI

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

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

1. 定义技能

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

from typing import TypedDict

class Skill(TypedDict): # [!code highlight]
"""可向智能体渐进式披露的技能。"""
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/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（产品ID）
– quantity（数量）
– unit_price（单价）
– discount_percent（折扣百分比）

## 业务逻辑

**活跃客户**：status = 'active' 且 signup_date <= 当前日期 – 90天

**收入计算**：仅统计 status = 'completed' 的订单。orders 表中的 total_amount 字段已包含折扣后的金额。

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

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

## 示例查询

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

## 表结构

### products（产品表）
– product_id（主键）
– product_name（产品名称）
– sku（库存单位编码）
– category（类别）
– unit_cost（单位成本）
– reorder_point（再订货点：库存低于此值时需补货）
– discontinued（是否停产：布尔值）

### warehouses（仓库表）
– warehouse_id（主键）
– warehouse_name（仓库名称）
– location（位置）
– capacity（容量）

### inventory（库存表）
– inventory_id（主键）
– product_id（外键 → products）
– warehouse_id（外键 → warehouses）
– quantity_on_hand（当前库存数量）
– last_updated（最后更新时间）

### stock_movements（库存变动记录表）
– movement_id（主键）
– product_id（外键 → products）
– warehouse_id（外键 → warehouses）
– movement_type（变动类型：inbound/入库、outbound/出库、transfer/调拨、adjustment/调整）
– quantity（数量：入库为正，出库为负）
– movement_date（变动日期）
– reference_number（参考编号）

## 业务逻辑

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

**需要补货的产品**：所有仓库中该产品的总库存量 ≤ 该产品的 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 # [!code highlight]
def load_skill(skill_name: str) –> str:
"""将技能的完整内容加载到智能体上下文中。

当您需要详细了解如何处理某类请求时，请使用此工具。
它将为您提供该技能领域的完整指令、策略和指南。

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

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

load_skill 工具以字符串形式返回完整技能内容，该内容会作为 ToolMessage 加入对话。有关创建和使用工具的更多详情，请参阅官网 工具指南。

3. 构建技能中间件

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

📝 本指南演示了如何创建自定义中间件。如需全面了解中间件的概念与模式，请参阅官网 自定义中间件文档。

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

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

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

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

def wrap_model_call(
self,
request: ModelRequest,
handler: Callable[[ModelRequest], ModelResponse],
) –> ModelResponse:
"""同步：将技能描述注入系统提示词。"""
# 构建技能附录
skills_addendum = ( # [!code highlight]
f"\\n\\n## 可用技能\\n\\n{self.skills_prompt}\\n\\n" # [!code highlight]
"当您需要详细了解如何处理某类请求时，请使用 load_skill 工具。" # [!code highlight]
)

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

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

📝 生产环境考量：本教程为简化起见，在 init 中加载技能列表。在生产系统中，您可能希望改在 before_agent 钩子中加载技能，以便定期刷新以反映最新变更（例如新增或修改技能）。详情请参阅 before_agent 钩子文档。

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

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

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

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

该智能体现在可在系统提示词中看到技能描述，并能在需要时调用 load_skill 获取完整技能内容。检查点用于跨轮次维护对话历史。

5. 测试渐进式披露

用一个需要技能专属知识的问题测试智能体：

import uuid

# 本次对话线程的配置
thread_id = str(uuid.uuid4())
config = {"configurable": {"thread_id": thread_id}}

# 请求生成 SQL 查询
result = agent.invoke( # [!code highlight]
{
"messages": [
{
"role": "user",
"content": (
"写一个 SQL 查询，找出上个月 "
"下单金额超过 $1000 的所有客户"
),
}
]
},
config
)

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

预期输出：

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

写一个 SQL 查询，找出上个月下单金额超过 $1000 的所有客户
================================== Ai Message ==================================
Tool Calls:
load_skill (call_abc123)
Call ID: call_abc123
Args:
skill_name: sales_analytics
================================= Tool Message =================================
Name: load_skill

已加载技能: sales_analytics

# 销售分析 Schema

## 表结构

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

[… 其余 Schema …]

## 业务逻辑

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

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

以下是查找上个月下单金额超过 $1000 的所有客户的 SQL 查询：

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

此查询：
– 关联 customers 与 orders 表
– 使用 total_amount 字段筛选高价值订单（>$1000）
– 仅包含已完成订单（根据业务逻辑）
– 限定为上个月的订单
– 返回去重的客户，避免同一客户多次符合条件时重复出现

智能体在系统提示词中看到轻量级技能描述，识别出问题需要销售数据库知识，于是调用 load_skill(“sales_analytics”) 获取完整 Schema 和业务逻辑，并据此编写出符合数据库规范的正确查询。

6. 进阶：通过自定义状态添加约束

定义自定义状态

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

from langchain.agents.middleware import AgentState

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

更新 load_skill 以修改状态

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

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]
"""将技能的完整内容加载到智能体上下文中。

当您需要详细了解如何处理某类请求时，请使用此工具。
它将为您提供该技能领域的完整指令、策略和指南。

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

# 更新状态以跟踪已加载技能
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]

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

创建受约束的工具

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

@tool
def write_sql_query( # [!code highlight]
query: str,
vertical: str,
runtime: ToolRuntime,
) –> str:
"""为特定业务线编写并验证 SQL 查询。

此工具用于格式化和验证 SQL 查询。您必须先加载
相应的技能，才能理解数据库 Schema。

参数:
query: 要编写的 SQL 查询
vertical: 业务线（sales_analytics 或 inventory_management）
"""
# 检查所需技能是否已加载
skills_loaded = runtime.state.get("skills_loaded", []) # [!code highlight]

if vertical not in skills_loaded: # [!code highlight]
return ( # [!code highlight]
f"错误：您必须先加载 '{vertical}' 技能 " # [!code highlight]
f"以理解数据库 Schema，然后才能编写查询。 " # [!code highlight]
f"请使用 load_skill('{vertical}') 加载 Schema。" # [!code highlight]
) # [!code highlight]

# 验证并格式化查询
return (
f"{vertical} 的 SQL 查询：\\n\\n"
f"```sql\\n{query}\\n```\\n\\n"
f"✓ 已根据 {vertical} Schema 验证查询\\n"
f"可执行于数据库。"
)

更新中间件和智能体

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

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

state_schema = CustomState # [!code highlight]
tools = [load_skill, write_sql_query] # [!code highlight]

# … 其余中间件实现保持不变

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

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

现在，如果智能体在加载所需技能前尝试使用 write_sql_query，将收到错误消息，提示其先加载相应技能（如 sales_analytics 或 inventory_management）。这确保了智能体在尝试验证查询前已掌握必要的 Schema 知识。

完整代码

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

# 定义技能结构
class Skill(TypedDict):
"""一个可以逐步向智能体披露的技能。"""
name: str
description: str
content: str

# 定义包含数据表结构和业务逻辑的技能
SKILLS: list[Skill] = [
{
"name": "sales_analytics",
"description": "用于销售数据分析的数据库结构和业务逻辑，包括客户、订单和收入。",
"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": "用于库存跟踪的数据库结构和业务逻辑，包括产品、仓库和库存水平。",
"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:
"""将指定技能的完整内容加载到智能体的上下文中。

当你需要关于如何处理某类请求的详细信息时，请使用此工具。
它会为你提供该技能领域的全面说明、策略和指南。

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

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

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

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

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

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

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

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

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

# 示例用法
if __name__ == "__main__":
# 为本次对话线程设置配置
thread_id = str(uuid.uuid4())
config = {"configurable": {"thread_id": thread_id}}

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

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

参考文献

https://docs.langchain.com/oss/python/langchain/multi-agent/skills

https://docs.langchain.com/oss/python/langchain/multi-agent/skills-sql-assistant