Multi-agent [ 4 ]

模式 3: Skills
接下来我们来看第 3 种架构模式,它叫做Skills,翻译过来就是技能。
什么是Skills架构 & 何时使用
对Skills最简单的理解就是:一份高度完善、专门用来处理单一专项业务能力的提示词,这份专属提示词就叫做Skill。
在Skills架构中,我们将特定的能力封装为可调用的 “Skills”,用以增强智能体的行为。这些Skills 主要是由提示词驱动的专项能力,智能体可按需调用。

举个例子,一套智能系统内部往往需要具备多种不同业务能力,针对每一类独立能力,我们都可以单独设计专属的提示词,每一套专属提示词就是一个Skill:
-
skill a:负责处理专项能力 1; -
skill b:负责处理专项能力 2; -
skill c:负责处理专项能力 3;
所有这类专项能力,底层全部依靠提示词来驱动,这就是Skill的核心概念。
一份标准的Skill提示词需要完整定义对应业务的全部细节:
-
明确业务需求、完整业务流程、分步操作步骤;
-
说明代码编写规范、开发约束要求;
-
标注关联数据库结构、数据表、字段规则;
-
以及该专项任务涉及的所有配套信息。
这份完整、专门用于解决某一类专项任务的提示词,就是Skill。提示词写得越详尽、完善,智能体处理对应专项任务的准确性、专业度就越强。
而我们在 LangChain 框架中学习Skills模式,核心学习目标不是单纯编写提示词,而是如何高效调用、按需使用这些封装好的Skills提示词。
一个Agent智能体可以绑定多个Skill,但我们之前学过:原生Agent只能调用工具,无法直接读取多份提示词。这里的解决方案是:给Agent绑定唯一一个专用加载工具。
这个加载工具内部可以读取、管理全部的Skill资源。当Agent触发工具调用时,就能根据需求挑选对应Skill,工具与各类Skill之间用虚线表示关联关系,这也是下面Skills基础示例代码的核心设计思路。
整体实现逻辑:
-
Agent绑定一个专门用来加载Skill的工具; -
工具内部内置全部可用
Skills,比如 SQL 查询编写专家、法律文档审阅专家两套完整Skill提示词; -
每一份
Skill本质都是完整提示词,支持两种存储方式:本地文件、数据库; -
这个工具的核心作用:根据需求加载指定的
Skill提示词。
这类专项提示词内容量大、篇幅完整,不适合长期存放在内存中维护,最优方案是存放在磁盘文件或数据库里,需要时再通过工具读取加载,这就是 LangChain 中使用Skills的标准姿势。
解决了工具加载的基础逻辑后,还有一个关键问题:加载Skills时,是一次性把所有技能全部载入,还是需要哪个再加载哪个?这里分为两种方案。
1. 全量加载
一次性读取全部Skill,存入工具,再依靠属性匹配筛选对应技能。 这种设计和我们之前学的 RAG 检索增强生成有相似之处,都相当于内置知识库,由Agent通过工具检索对应内容。 但二者有明显区别:标准 RAG 可以直接基于用户问题query做语义检索匹配;而本模式下的Agent无法直接通过用户问题检索,只能依靠工具入参传入的名称,做精准匹配。
2. 渐进式披露(按需加载,推荐方案)
核心逻辑:用到哪个专项能力,就只加载对应的那一份Skill。
-
用户需求对应能力 2,就仅加载
skill b; -
用户需求对应能力 1,就仅加载
skill a。
按需加载的底层配置设计
每一个Skill都需要提前配置 3 项固定属性:
-
name:技能唯一名称,作为工具调用的入参标识; -
description:简短描述,100 字以内,简单说明该技能的业务用途; -
content:技能完整正文,完整提示词内容。若文件 / 数据库存储:该字段填资源存储路径;内存存储:该字段直接存放完整提示词文本。
所有Skill统一整理成一份配置清单,后续按需加载流程如下:
-
用户输入问题,需求对应某一个专项能力;
-
Agent识别需求,触发load_skill加载工具; -
Agent依靠各个Skill简短的description描述做语义匹配,判断当前需要调用哪一个Skill; -
匹配完成后,把目标
Skill的name传入工具作为参数; -
工具根据
name读取对应content完整提示词; -
读取到的完整
content会动态注入、替换智能体的系统提示词;
按需加载设计的核心优势
完整content包含数据表结构、代码规范、业务约束等全套信息,注入系统提示词后,大模型处理对应专项任务时逻辑严谨、结果精准,不会偏离业务要求。
这里解释为什么必须拆分description和content:description只是百字以内简短介绍,体量小,全部提前注入系统提示词不会占用过多上下文;而content是完整文档,篇幅巨大,不可能一次性全部放入上下文。大模型依靠轻量化的description就能自主判断需要调用哪一个技能,再按需加载庞大的完整正文,完美控制上下文长度,这也是渐进式披露设计的核心目的。
Skills架构五大关键特征
-
提示词驱动的专项化:所有
Skills的能力,全部依靠专属提示词实现定义; -
渐进式披露(上下文管理技术):轻量化描述常驻上下文,完整技能内容按需加载;
-
团队分布式开发:不同业务团队可以独立开发、维护各自领域的
Skill,互不干扰,适合多人协作开发; -
轻量级组合:对比完整独立子智能体,
Skill实现更轻量化;简单专项任务无需开发完整子 Agent,仅靠提示词就能实现; -
资源引用感知:单个
Skill内部可以引用外部脚本、模板、数据库等各类配套资源。
那我们应该何时选用Skills架构模式?
适用场景:需要单一智能体承载多类专项能力,技能之间不需要强制约束执行顺序,同时支持多团队独立开发业务能力。
典型落地案例:
-
编程助手:同一个 Agent 作为通用编程助手,针对 Java、C++、Python 等不同编程语言,分别设计独立
Skill,各语言代码生成规则、校验逻辑存在差异,拆分技能更易维护; -
领域知识库助手:同一个对话 Agent,针对法律、财务、运维等不同行业领域,拆分专属
Skill提示词;
简单总结:当你只需要一个统一交互入口(单一 Agent),但内部要区分多套独立业务处理规则时,优先使用Skills模式。
Skills 时间案例 - 构建一个按需加载 Skills 的 SQL 助手
基础实现示例
以下代码展示了 Skills 模式的基本概念:将一个 load_skill 函数作为工具提供给智能体。
from langchain.tools import tool
from langchain.agents import create_agent
@tool
def load_skill(skill_name: str) -> str:
"""加载一个专项 Skills 提示词。
可用 Skills:
- write_sql: SQL查询编写专家
- review_legal_doc: 法律文档审阅专家
返回 Skills 的提示词和上下文。
"""
# 实际应用中,从文件或数据库加载 Skills 内容
# 例如:skills = {"write_sql": "你是一个SQL专家...", "review_legal_doc": "你是一个法律专家..."}
# 这里仅为演示框架,实际需实现具体加载逻辑
pass
# 创建 Agent,配置使用的模型、工具列表和系统提示词
agent = create_agent(
model="gpt-5.4", # 指定使用的大语言模型
tools=[load_skill], # 注册工具列表,这里只注册了加载 Skill 的工具
system_prompt=( # 系统级提示词,定义 Agent 的角色和行为
"你是一个乐于助人的助手。"
"你可以使用两个 Skills : write_sql 和 review_legal_doc。"
"请使用 load_skill 工具来获取它们。"
),
)
说明:此示例仅用于阐述核心概念,并未实现完整的 Skills 加载逻辑。
架构与工作流
我们将展示如何使用渐进式披露技术(一种上下文管理技术)来实现 Skills 系统。当用户问一个问题时,智能体会先判断是否需要 “专业 Skills”。如果需要,就调用工具去加载该技能的详细说明书,然后基于说明书完成任务;如果不需要,就直接回答。核心就是一个“用到时才加载,而不是一开始全塞进去”的思路。
接下来,我们将构建一个 SQL 查询助手,它包含两个 Skills(销售分析和库存管理)。智能体在系统提示词中仅看到轻量级的 Skills 描述,只有当用户查询涉及特定领域时,才会通过工具调用来加载完整的数据库模式和业务逻辑。

为了更好的理解,使用比喻的方式逐步拆解工作流:
| 步骤 | 发生了什么 | 角色比喻 |
|---|---|---|
| 1. 用户提问 | 用户问:“查一下上个月谁买了超过 1000 美元的东西。给出SQL” |
你走进一家大公司的总台。 |
| 2. 智能体初判 | 智能体看到系统提示词里有一张 “Skill菜单”(只含名字和一句话描述),它意识到:“这问题得找销售数据部门。” |
总台看了部门索引卡,知道销售数据归 “销售分析部” 管。 |
| 3. 调用工具加载技能 | 智能体主动调用 load_skill("sales_analytics") 工具。 |
总台按下内线电话:“请销售分析部把他们的详细工作手册送过来。” |
| 4. 获取完整知识 | 工具返回 sales_analytics 技能的完整内容,包含所有表结构、字段名、业务规则(例如 “收入只算已完成的订单”)。这些信息现在进入了智能体的 “工作记忆”。 |
一本厚厚的部门手册被送到了总台,手册里有该部门的所有数据库图纸和办事规则。 |
| 5. 执行任务 | 智能体根据刚加载的详细知识,写出正确的SQL查询。 |
总台对照着手册里的图纸和规则,准确填写好了查询申请单。 |
为什么要绕这个弯(为什么使用渐进式披露)?
对比两种做法,就明白了:
笨办法(全量加载):一启动就把所有部门(销售、库存、人事、财务…)的厚手册全堆在总台桌上。结果是桌子堆不下(上下文窗口爆了),总台也找得眼花缭乱(模型注意力分散、成本剧增)。
聪明办法(渐进式披露):总台桌上只放一张部门索引卡(轻量级描述)。哪个手册用到了,再打电话去取(工具调用)。这样桌子永远清爽,总台效率最高。
优点与代价:
优点
| 优点 | 解释 |
|---|---|
| 省脑子(省 Token) | 一次只用加载 2-3 个相关技能,不浪费上下文。 |
| 部门独立 | 销售部改了自己手册,不影响总台和其他部门。 |
| 无限扩张 | 你可以有 100 个技能,总台桌上永远只有索引卡。 |
| 一问一答 | 始终是一个总台在和你对话,历史不乱。 |
代价
| 代价 | 解释 |
|---|---|
| 延迟 | 每个新问题如果需要新技能,都得先调用一次工具,比直接知道答案慢一点点。 |
| 流程控制弱 | 基础版【目前】只能靠提示词劝智能体 “先看手册再办事”,没法硬性规定 “必须先看 A 手册才能看 B 手册”。(不过高级版【下文】里可以用自定义状态实现硬约束。) |
概念解释:带渐进式披露的 Skills 可被视为一种 RAG 形式,其中每个 Skills 是一个检索单元,但检索方式不限于向量或关键词搜索,而是通过工具调用进行内容浏览。
第一步:定义 Skills
为了方便调试,直接将Skill存入内存变量;实际生产环境建议存入本地文件或数据库。 我们通过TypedDict定义Skill标准结构,包含name、description、content三个必填字段,示例中定义两套业务技能:
sales_analytics销售分析技能
description:用于销售数据分析的数据库模式与业务逻辑,包含客户、订单和收入;
content完整内容:
-
数据表定义:客户表、订单表、订单项表的主键、外键、字段枚举;
-
业务逻辑约束:高价值订单判定标准、收入统计规则(仅统计已完成订单); 当用户需要生成销售相关 SQL 时,大模型会读取这套表结构和业务规则,写出完全贴合业务的查询语句。
inventory_management库存管理技能
description:用于库存跟踪的数据库模式与业务逻辑,包含产品、仓库和库存水平;
content完整内容:
-
数据表定义:产品表、仓库表、库存表完整结构;
-
业务逻辑约束:可用库存判定、补货触发条件;
两套技能结构完全统一,仅业务内容区分,后续可以无限新增扩展。
from typing import TypedDict
class Skill(TypedDict):
"""一个可渐进披露给智能体的 Skills。"""
name: str # Skills 的唯一标识符
description: str # 在系统提示词中展示的简短描述
content: str # 包含详细指令的完整 Skills 内容
# 定义完整的 Skills 列表
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 (外键), order_date
- status (pending/completed/cancelled/refunded), total_amount
### order_items (订单项表)
- item_id (主键), order_id (外键), product_id, quantity, unit_price
## 业务逻辑
**高价值订单**: total_amount > 1000
**收入计算**: 仅统计 status = 'completed' 的订单。
"""
},
{
"name": "inventory_management",
"description": "用于库存跟踪的数据库模式与业务逻辑,包含产品、仓库和库存水平。",
"content": """# 库存管理模式
## 数据表
### products (产品表)
- product_id (主键), product_name, sku, category, unit_cost, reorder_point
### warehouses (仓库表)
- warehouse_id (主键), warehouse_name, location, capacity
### inventory (库存表)
- inventory_id (主键), product_id (外键), warehouse_id (外键), quantity_on_hand
## 业务逻辑
**可用库存**: quantity_on_hand > 0
**需补货产品**: 各仓库 quantity_on_hand 总和 <= reorder_point
"""
},
]
注意:此处的 content 字段为简化版,完整内容请参考源文档。
第二步:创建 Skills 加载工具
创建一个名为 load_skill 的工具,智能体将调用它来获取 Skills 的完整内容。
接收入参skill_name,根据名称匹配并返回对应完整content:
-
工具入参
skill_name由 Agent 自主传入,Agent 依靠提前注入的description判断需要加载哪套技能; -
遍历预定义的全部
SKILLS列表,匹配name后拼接完整技能内容返回; -
匹配失败时,返回提示并展示系统内全部可用技能名称;
-
工具描述写清使用场景:当需要专项业务完整规则时调用,为大模型提供领域完整指令、业务策略。
from langchain.tools import tool
@tool
def load_skill(skill_name: str) -> str:
"""将 Skills 的完整内容加载到智能体的上下文中。
当你需要处理特定类型请求的详细信息时使用此工具。
它将为你提供该 Skills 领域的全面指令、策略和指南。
Args:
skill_name: 要加载的 Skills 名称(例如 "sales_analytics")
"""
# 遍历 SKILLS 列表,查找与请求名称匹配的 Skill
for skill in SKILLS:
if skill["name"] == skill_name:
# 找到后返回 Skill 的名称和完整内容
return f"已加载 Skills : {skill_name}\n\n{skill['content']}"
# 如果未找到匹配的 Skill,则返回错误信息并列出所有可用的 Skill 名称
available = ", ".join(s["name"] for s in SKILLS)
return f"Skills '{skill_name}' 未找到。可用 Skills : {available}"
第三步:构建 Skills 中间件
中间件采用类继承模式实现,核心作用:在模型调用前,把全部Skill轻量化描述追加到系统提示词,让大模型能识别所有可用技能。
将load_skill工具注册为中间件类变量,Agent 可以直接调用;
初始化逻辑:遍历所有Skill,把name和description整理成 Markdown 格式文本; Markdown 分级格式更容易被大模型识别、区分信息,写提示词时推荐统一使用;
重写wrap_model_call拦截方法:
-
构建技能附录提示词,列出所有可用技能名称与简介,同时告知模型需要详情时调用
load_skill工具; -
读取原有系统提示词,把技能附录追加到原有提示词尾部,生成全新系统消息;
-
替换请求内的系统提示词,再放行请求发送给大模型;
这里区分两层提示词注入逻辑:
-
中间件仅注入轻量化
description,用于让模型自主选择技能; -
工具调用后,才会加载完整
content业务正文,真正用于业务生成。
创建一个自定义中间件 SkillMiddleware,它的作用是在不加载完整内容的前提下,将可用 Skills 的描述注入到系统提示词中,使 Skills 可被发现。
from langchain.agents.middleware import ModelRequest, ModelResponse, AgentMiddleware
from langchain.messages import SystemMessage
from typing import Callable
class SkillMiddleware(AgentMiddleware):
"""将 Skills 描述注入系统提示词的中间件。"""
# 将 load_skill 工具注册为类变量,使其对智能体可用
tools = [load_skill]
def __init__(self):
"""初始化时根据 SKILLS 列表生成 Skills 描述提示词。"""
skills_list = []
for skill in SKILLS:
# 遍历每个 Skill,格式化为"名称: 描述"的形式
skills_list.append(
f"- **{skill['name']}**: {skill['description']}"
)
# 将所有 Skill 描述用换行符拼接,存入实例变量供后续使用
self.skills_prompt = "\n".join(skills_list)
def wrap_model_call(
self,
request: ModelRequest,
handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
"""同步方法:在模型调用前,将 Skills 描述注入系统提示词。"""
# 构建要追加到系统提示词末尾的 Skills 附录
skills_addendum = (
f"\n\n## 可用 Skills \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)
说明:wrap_model_call 方法拦截发送给模型前的请求,修改系统提示词,从而实现渐进式披露的第一步。
第四步:创建支持 Skills 的智能体
创建 Agent 时传入四大核心配置:
-
大模型对象;
-
基础系统提示词:通用基础描述,示例为 “SQL 查询助手,编写业务数据库查询”;
-
绑定自定义
SkillMiddleware中间件; -
内存检查点器
InMemorySaver,持久化对话上下文;
工具无需手动传入 Agent,中间件已经提前注册工具,Agent 可自动识别调用。
现在,我们使用中间件和检查点器来创建智能体。
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
# 创建带有 Skills 支持的智能体
agent = create_agent(
model, # 指定使用的语言模型(需提前定义)
system_prompt=(
"你是一个SQL查询助手,帮助用户编写针对业务数据库的查询。"
),
middleware=[SkillMiddleware()], # 注册 SkillMiddleware 中间件,使 Skills 描述注入系统提示词
checkpointer=InMemorySaver(), # 使用内存检查点保存器,用于持久化对话状态(支持对话历史回溯)
)
第五步:测试渐进式披露
最后,我们用一个需要销售分析知识的问题来测试智能体。
配置对话线程 ID,区分多轮对话上下文;
用户测试提问:写一个 SQL 查询,找出上个月订单金额超过 1000 美元的所有客户;
执行agent.invoke发起对话,循环打印全部对话消息,观测执行链路;
# 为此对话线程配置ID,用于区分不同的对话会话
config = {"configurable": {"thread_id": "thread_123"}}
# 提出一个需要特定 Skills 的问题
result = agent.invoke(
{
"messages": [
{
"role": "user",
"content": (
"写一个SQL查询,找出上个月订单金额超过1000美元的所有客户。"
),
}
]
},
config # 传入配置,指定使用 thread_123 这个对话线程
)
# 打印对话过程中的所有消息
for message in result["messages"]:
# 如果消息对象有 pretty_print 方法,使用格式化打印(美观输出)
if hasattr(message, 'pretty_print'):
message.pretty_print()
else:
# 否则直接打印消息类型和内容
print(f"{message.type}: {message.content}")
预期执行流程拆解
-
Agent 读取中间件注入的技能简介,识别该销售统计需求对应
sales_analytics技能; -
Agent 主动调用
load_skill工具,入参传入skill_name="sales_analytics"; -
工具返回完整销售数据库表结构、订单收入业务规则;
-
完整技能内容注入上下文后,大模型参照数据表、业务约束生成合规 SQL;
测试结果解读
================================ Human Message =================================
写一个SQL查询,找出上个月订单金额超过1000美元的所有客户。
================================== Ai Message ==================================
好的,我需要先加载销售分析相关的技能,了解数据库模式和业务逻辑。
Tool Calls:
load_skill (call_00_nAG99t1y3jrvCVuOuFSf0198)
Call ID: call_00_nAG99t1y3jrvCVuOuFSf0198
Args:
skill_name: sales_analytics
================================= Tool Message =================================
Name: load_skill
已加载 Skills : sales_analytics
# 销售分析模式
## 数据表
### customers (客户表)
- customer_id (主键), name, email, signup_date
- status (active/inactive), customer_tier (bronze/silver/gold/platinum)
### orders (订单表)
- order_id (主键), customer_id (外键), order_date
- status (pending/completed/cancelled/refunded), total_amount
### order_items (订单项表)
- item_id (主键), order_id (外键), product_id, quantity, unit_price
## 业务逻辑
**高价值订单**: total_amount > 1000
**收入计算**: 仅统计 status = 'completed' 的订单。
================================== Ai Message ==================================
好的,我已经了解了数据库模式。现在来编写查询。
根据业务逻辑:
- 上个月的订单(使用 `order_date` 字段)
- 订单金额超过1000美元(`total_amount > 1000`)
- 只统计已完成订单(`status = 'completed'`)
以下是SQL查询:
```sql
SELECT
c.customer_id,
c.name,
c.email,
o.order_id,
o.total_amount,
o.order_date
FROM customers c
JOIN orders o ON c.customer_id = o.customer_id
WHERE o.status = 'completed'
AND o.total_amount > 1000
AND o.order_date >= DATE_TRUNC('month', CURRENT_DATE - INTERVAL '1 month')
AND o.order_date < DATE_TRUNC('month', CURRENT_DATE)
ORDER BY o.total_amount DESC;
```
### 查询说明:
1. **`DATE_TRUNC('month', CURRENT_DATE - INTERVAL '1 month')`** — 上个月的第一天(例如:如果现在是4月,则返回3月1日)
2. **`DATE_TRUNC('month', CURRENT_DATE)`** — 本月的第一天(作为上个月的结束边界)
3. **`o.status = 'completed'`** — 只统计已完成的订单(符合业务规则)
4. **`o.total_amount > 1000`** — 订单金额超过1000美元
5. **`JOIN customers c`** — 关联客户表获取客户详细信息
### 如果只需要客户列表(不重复):
```sql
SELECT DISTINCT
c.customer_id,
c.name,
c.email
FROM customers c
JOIN orders o ON c.customer_id = o.customer_id
WHERE o.status = 'completed'
AND o.total_amount > 1000
AND o.order_date >= DATE_TRUNC('month', CURRENT_DATE - INTERVAL '1 month')
AND o.order_date < DATE_TRUNC('month', CURRENT_DATE)
ORDER BY c.name;
```
这样就能找出上个月所有下过金额超过1000美元已完成订单的客户了。
日志中可以观测到工具调用参数正确传入销售技能名称,工具返回完整的数据库 Schema 与业务规则; 模型生成 SQL 时会严格遵循技能内的约束:仅统计completed已完成订单,关联客户表、订单表匹配字段,筛选金额大于 1000 的客户,表关联、字段、业务条件全部准确,证明完整Skill提示词成功生效。
高级主题与扩展
添加约束:自定义状态与受限工具
接下来我们来讲,如何给Skills模式新增受限工具,先讲清楚这个机制的作用。
回顾之前整套Skills运行流程:我们给Agent绑定了单一专用工具,这个工具的作用是按需加载各类Skill。但原有代码没有强制约束逻辑,系统不会强制要求必须先加载对应Skill,才能执行后续问答、生成业务结果,模型有可能跳过加载步骤直接作答,输出不符合业务规则的内容。 如果我们需要强约束机制:必须先加载对应Skill,才能开展后续问答、业务操作,就需要新增一套受限工具做强制校验,下面梳理完整实现思路。
可以通过自定义状态来跟踪已加载的 Skills,并创建仅当特定 Skills 加载后才可用的工具(例如 write_sql_query)。这通过扩展 AgentState 并修改 load_skill 工具以返回 Command 对象来更新状态实现。
简要步骤:
1. 自定义状态,记录已加载的技能
首先我们要自定义一套Agent全局状态,状态内部维护一个列表字段,专门记录当前对话线程中已经加载过的全部Skill名称。 所有工具、中间件都能读取、修改这份全局状态,以此判断目标技能是否完成加载。
定义包含
skills_loaded字段的CustomState。
from langchain.agents.middleware import AgentState
from typing import NotRequired
class CustomState(AgentState):
# 自定义状态类,扩展 AgentState
# 用于记录当前对话中哪些 Skill 已经被加载过
# 避免重复加载相同的 Skill,提高效率
skills_loaded: NotRequired[list[str]] # 记录已加载的 Skill 名称列表
2. 新增校验用受限工具
我们新增一个专门做校验的受限工具,核心逻辑:
调用该工具时会传入目标业务对应的Skill名称参数;
工具读取全局自定义状态,遍历已加载技能列表,判断目标Skill是否存在;
两种分支处理:
-
状态中已存在该
Skill:校验通过,允许继续执行、生成业务结果; -
状态中没有该
Skill:直接返回固定错误提示,告知模型:编写业务查询前必须先调用load_skill加载对应技能,读取数据库结构信息。
同时我们需要在基础系统提示词里补充约束规则:在生成 SQL、执行业务查询之前,必须先调用这个校验工具,确认对应Skill已加载完成,校验通过后才能生成最终查询语句。依靠提示词引导模型遵守这套强制校验流程。
创建受限工具(如
write_sql_query),在其逻辑中检查runtime.state.get("skills_loaded")是否包含所需Skills,若不包含则返回错误提示。
@tool
def write_sql_query(
query: str,
vertical: str,
runtime: ToolRuntime,
) -> str:
"""编写并验证针对特定业务领域的 SQL 查询。"""
# 从运行时状态中获取已加载的 Skill 列表,默认为空列表
skills_loaded = runtime.state.get("skills_loaded", [])
# 检查所需业务领域的 Skill 是否已加载
# 如果未加载,则无法了解数据库结构,无法生成正确的查询
if vertical not in skills_loaded:
return (
f"错误: 在编写查询之前,必须先加载 '{vertical}' skill以了解数据库结构。"
f"请使用 load_skill('{vertical}') 加载结构信息。"
)
# 验证通过,格式化并返回查询结果
# 此处可进一步扩展:实际验证 SQL 语法、检查表名/字段名是否存在于 Skill 定义的 schema 中
return (
f"{vertical} 的 SQL 查询: \n"
f"```sql\n{query}\n```\n"
f"✓ 查询已根据 {vertical} 结构进行验证\n"
f"✓ 准备就绪,可对数据库执行。"
)
3. 改造原有load_skill加载工具,同步更新全局状态
原来的load_skill工具只负责返回Skill完整内容,现在需要改造,增加状态更新逻辑:
每成功加载一个Skill,就把当前skill_name写入全局状态的skills_loaded列表,实时记录已加载技能;
如果传入的skill_name不存在、匹配失败,不再简单返回字符串,而是封装成标准ToolMessage工具消息返回。 ToolMessage会完整存入对话上下文,大模型读取消息后会识别 “目标技能不存在”,读取系统提示内全部可用技能清单,重新发起正确的加载调用,完成重试流程。
修改
load_skill工具,在返回Skills内容的同时,通过Command(update={"skills_loaded": [skill_name]})更新状态。
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:
"""加载指定的 Skill 并更新 Agent 状态。"""
# 遍历预定义的 SKILLS 列表,查找匹配的 Skill
for skill in SKILLS:
if skill["name"] == skill_name:
# 找到目标 Skill,构建返回内容
skill_content = f"已加载 skill: {skill_name}\n\n{skill['content']}"
# 返回 Command 对象,同时更新消息和状态
# update 中的字段会合并到 Agent 状态中
return Command(
update={
# 向消息列表中添加一条工具回复消息
"messages": [
ToolMessage(
content=skill_content,
tool_call_id=runtime.tool_call_id, # 关联到本次工具调用
)
],
# 记录已加载的 Skill 名称,供后续逻辑判断是否重复加载
"skills_loaded": [skill_name],
}
)
# 未找到匹配的 Skill,返回错误信息
available = ", ".join(s["name"] for s in SKILLS)
return Command(
update={
"messages": [
ToolMessage(
content=f"'{skill_name}' 未找到。可用skill: {available}",
tool_call_id=runtime.tool_call_id,
)
]
}
)
4. 修改自定义中间件,注册新工具、注入自定义状态
中间件有两处配套修改:
-
在中间件的
tools列表中,把新增的校验受限工具注册进去,让Agent识别、可以正常调用; -
将我们自定义的状态类声明为中间件的状态约束,把自定义全局状态注入整个智能体运行链路,保证所有工具都能读写这份加载记录。
更新
SkillMiddleware以使用CustomState并注册新工具。
class SkillMiddleware(AgentMiddleware[CustomState]):
"""将技能描述注入系统提示的中间件。"""
# 指定使用自定义状态模式,用于跟踪已加载的技能
state_schema = CustomState
# 注册该中间件提供的工具列表
# load_skill: 加载指定技能的内容
# write_sql_query: 编写并验证 SQL 查询
tools = [load_skill, write_sql_query]
# ... 中间件的其余实现保持不变
此模式确保了 Agent 在获得必要的知识后才能执行特定操作。
整套改造完成后,系统会形成强约束:模型想要生成业务答案、SQL 查询,必须先调用校验工具;校验工具会读取全局加载状态做判断,没加载对应技能就抛出错误引导模型先加载,只有成功加载并写入状态后,才能正常执行业务生成。
实现变体与扩展模式
讲完受限工具的高级约束,再介绍Skills模式其他通用扩展能力,适配更复杂的业务场景。
1. 动态工具注册
动态工具注册:加载 Skills 时,不仅提供上下文,还可动态注册新工具。
我们现有的中间件仅做了系统提示词追加、注入技能描述,还可以拓展动态工具注册能力:加载某一类Skill时,除了把业务上下文注入上下文,还能动态新增、注册配套专用工具。 举个例子:加载database_admin数据库管理技能后,自动注册backup数据库备份、restore数据恢复这类配套专属工具,技能和专属工具绑定,按需加载、按需注册。
2. 层级 Skills(树形子技能)
层级 Skills:Skills 可以定义子 Skills,形成树状结构。
上面示例只定义了 2 个顶层Skill,如果业务场景复杂、技能数量庞大,可以设计树形层级结构,拆分父Skill与子Skill: 比如顶层data_science数据分析父技能,加载完成后,再按需加载下属子技能pandas_expert表格处理、visualization可视化绘图,分层级逐步加载,避免一次性涌入大量技能描述占用上下文。
3. Skills 存储与自动发现机制
(1)存储方案
大量Skill资源有多种持久化存储选择:内存、本地文件系统、对象存储 S3、业务数据库,按需选用。
(2)自动发现方案,四种主流实现:
系统提示词列表:也是本教程使用的方案。把所有Skill简短description描述放入系统提示词,依靠大模型语义分析描述,自主判断需要调用哪个技能;描述文本 token 体量很小,不会挤占上下文。
文件系统扫描:程序自动扫描指定目录,读取本地存储的全部技能文件,自动识别、注册可用Skill。
RAG 向量检索:把所有Skill的描述、业务摘要存入向量知识库,用户输入问题后,通过语义检索匹配关联Skill,不用依靠大模型自主判断。
API 调用:通过外部接口远程拉取、查询可用技能列表,适合多服务分布式场景。
4. 分规格 Skills 加载策略
根据Skill完整文本的 token 体量,选择适配的加载方案:
-
小型 Skills(< 1K tokens):内容很短,可直接全部写入系统提示词,搭配缓存复用;
-
中型 Skills(1-10K tokens):本篇所采用的渐进式按需加载方案,用到再加载;
-
大型 Skills(> 10K tokens):内容体量极大,不能一次性全量载入,采用分页读取、分段搜索、层级拆分的加载方式。
与其他技术的结合
渐进式披露作为一种上下文工程技术,可以与少样本提示结合。
渐进式披露本身是一套上下文工程优化技术,可以和主流提示词技术结合使用,举典型落地案例: 将渐进式加载与少样本提示结合: 当Agent加载sales_analytics销售分析完整业务模式时,除了数据表、业务规则,同时通过语义检索拉取多条真实业务查询示例(少样本案例),一并注入上下文。完整数据表 Schema + 真实业务查询示例同时提供给大模型,模型既能遵循数据库字段、业务约束,又能参考标准示例的书写格式,最终生成精准、规范、贴合业务需求的 SQL 查询语句。
所有拓展策略、约束机制的核心目标只有一个:让大模型输出的业务结果更加准确规范,规避业务逻辑、语法、规则类错误。到这里,整套Skills架构模式从基础概念、完整代码、高级约束、多维度拓展全部讲解完毕。
更多推荐

所有评论(0)