[LangChain中的Multi-Agent模式-04]Skill轻量化智能体构建：避免上下文污染的专业化路径

在技​​能模式（Skills）中，专门化的能力被打包成可调用的技能，以增强Agent的行为。技能主要是由提示驱动的专业化功能，Agent可以按需调用这些功能。关键Skills的详细说明，请参阅Anthropic的官方文档“[Agent Skills]（https://platform.claude.com/docs/en/agents-and-tools/agent-skills/overview）”。Skills模式具有如下的核心特征：

• 提示驱动的专业化：技能主要由特定提示定义；
• 渐进式披露：根据上下文或用户需求按序加载可用Skill。在此基础上更进一步，还包括根据Skill动态注册工具集；
• 团队分布：不同团队可以独立开发和维护技能；
• 轻量级构建：技能比完整的Sub-Agent更简单；
• 引用感知：技能可以引用脚本、模板和其他资源。虽然每个技能只有一个提示，但该提示可以引用其他资源的位置，并提供Agent何时应使用这些资源的信息。

当需要一个具备多种专业技能的单一Agent，且无需对技能之间施加特定约束，或者不同团队需要独立开发相应能力时，可以使用Skills模式。常见的例子包括编码助手（针对不同语言或任务的技能）、知识库（针对不同领域的技能）和创意助手（针对不同格式的技能）。我们已经尝试了三种模式(Sub-Agent、Router和Handoffs)来开发我们的差旅助手Agent，现在我们继续将它改写成Skills模式。虽然“Deep Agents”提供了针对Skills的原生支持，但是为了让大家对Skills的实现原理有更清晰的认知，我决定使用一个更笨的解决方案来改写我们：使用自定义工具加载Skill。

1. 定义加载Skill的工具

一个Skill由元数据和主体内容组成。以名称和描述为核心的元数据会全程绑定到Agent上，所以我们要保证命名准确，描述精炼，以免占据过多的上下文窗口。主体内容相当于一份用于指导Agent工作的说明书，一般采用Markdown格式编写。为此我们定义了如下这个Skill类，并创建了用于购买机票和酒店预订的Skill，并将其保存到全局字典all_skills中（Key为Skill的名称）。

classSkill(TypedDict):name:strdescription:strcontent:strbuy_airplane_ticket_skill="""\ ## 基本流程 - 确定当前是否注册了`buy_airplane_ticket`工具，如果没有注册该工具，拒绝执行并回复用户：抱歉，我无法购买机票，因为相关工具未注册。 - 调用`buy_airplane_ticket`工具购买机票，最后将工具返回的预订信息整理后返回给用户。 ## 注意事项 - 购买机票是唯一的任务 - 调用`buy_airplane_ticket`工具是购买机票唯一的方式 - 只需要根据`buy_airplane_ticket`工具的Schema来分析购买机票的信息是否充分 - 可以完全自主决定航司、舱位等级和航班等信息，不需用户确认 """book_hotel_skill="""\ ## 基本流程 - 确定当前是否注册了`book_hotel`工具，如果没有注册该工具,拒绝执行并回复用户：抱歉，我无法预订酒店，因为相关工具未注册。 - 调用`book_hotel`工具预订酒店，最后将工具返回的预订信息整理后返回给用户。 ## 注意事项 - 预订酒店是唯一的任务 - 调用`book_hotel`工具是预订酒店唯一的方式 - 只需要根据`book_hotel`工具的Schema来分析预订酒店的信息是否充分 - 可以完全自主决定酒店、价位和房型等信息，不需用户确认 """all_skills=[{"name":"buy-airplane-ticket","description":"购买机票，只有在用户明确提出要求购买机票时才会使用","content":buy_airplane_ticket_skill},{"name":"book-hotel","description":"预订酒店，只有在用户明确提出要求预订酒店时才会使用","content":book_hotel_skill},]

为了跟踪当前加载的Skill，我们在状态类型State中添加了loaded_skills字段，并利用自定义的reducer函数添加加载的Skill名称。在推理过程中，LLM会根据推理任务和预先加载的Skill元数据决定所需的Skill，并作针对性的加载。如下这个load_skill函数就是我们为它准备的Skill加载工具。如果成功加载，返回的Command会通过修改loaded_skills通道将建在Skill名称添加到状态中。

classState(AgentState):loaded_skills:Annotated[set[str],lambdax,y:x.union(y)]@tooldefload_skill(name:str,runtime:ToolRuntime)->Command|str:"""根据指定的技能名称加载技能详细内容"""forskillinall_skills:ifskill["name"]==name:returnCommand(update={"messages":[ToolMessage(content=skill["content"],tool_call_id=runtime.tool_call_id)],"loaded_skills":{name},})returnf"抱歉，未找到名为{name}的技能。"

2. 定义Skill中间件

如下定义的SkillMiddleware中间件旨在完成两项任务：

• 将所有可用Skill的元数据给格式化成员系统提示词的一部分；
• 根据当前加载的Skill提供对应的工具集。

这两项工作都是利用awrap_model_call方法针对模型调用的拦截实现的。

asyncdefmain():client=MultiServerMCPClient(connections={"server":{"transport":"stdio","command":"python","args":["server.py"]}})tools={tool.name:toolfortoolinawaitclient.get_tools(server_name="server")}skill_based_tools:dict[str,list[BaseTool]]={"buy-airplane-ticket":[tools["buy_airplane_ticket"]],"book-hotel":[tools["book_hotel"]],}classSkillMiddleware(AgentMiddleware):tools=[load_skill]def__init__(self):skills=(f" - **{skill['name']}**:{skill['description']}"forskillinall_skills)self.skills_prompt=f"""\ 你拥有如下可用Skill：{'\n'.join(skills)}可以指定Skill名称调用`load_skill`工具来获取指定技能的详细信息。 """asyncdefawrap_model_call(self,request:ModelRequest,handler:Callable[[ModelRequest],Awaitable[ModelResponse]],)->ModelResponse|AIMessage|ExtendedModelResponse:system_message=request.system_messageifsystem_messageisNone:system_message=SystemMessage(content=self.skills_prompt)else:cotents=system_message.content_blocks+[{"type":"text","text":self.skills_prompt}]system_message=SystemMessage(content_blocks=cotents)tools=[load_skill]loaded_skills=request.state.get("loaded_skills",set())forskillinloaded_skills:tools.extend(skill_based_tools.get(skill,[]))returnawaithandler(request.override(system_message=system_message,tools=tools))# type: ignore

3. 创建和测试Agent

如果采用Skills模式，我们主要工作会放在Skill文档的撰写上面，程序员俨然变成一个文字工作者。所以我们直接可以根据提供的工具（load_skill和由MCP服务器提供的buy_airplane_ticket和book_hotel工具）、系统提示词和SkillMiddleware创建我们所需的Agent。最后我们依然采用与前面一样的测试用例（供了四种输入来模拟四种情况：同时包含酒店预订和机票购买需求、只包含酒店预订或者机票购买需求以及不涉及这两种需求）对构建的Agent进行测试：

asyncdefmain():...agent=create_agent(model=ChatOpenAI(model="gpt-5.2-chat"),state_schema=State,tools=[load_skill,*tools.values()],system_prompt="你是一个差旅助理，请严格按照注册的技能要求来分析和执行用户的请求。如果用户没有明确预订酒店或者购买机票的需求，直接回复：意图不明，无法执行",middleware=[SkillMiddleware()])inputs=["我有明后两天(从上海)去成都的出差，请帮我预订酒店（明天入住，后天离开）和机票（往返，明天出发，后天返回）。","我有明后两天(从上海)去成都的出差，请帮我预订酒店（明天入住，后天离开）。","我有明后两天(从上海)去成都的出差，请帮我预订机票（往返，明天出发，后天返回）。","随便说说！"]formessageininputs:reuslt=awaitagent.ainvoke(input={"messages":[HumanMessage(content=message)]})# type: ignoreprint(f"用户输入：{message}")print(f"系统回复：{reuslt['messages'][-1].text}\n\n")

如下所示的是四段输出：

用户输入：我有明后两天(从上海)去成都的出差，请帮我预订酒店（明天入住，后天离开）和机票（往返，明天出发，后天返回）。 系统回复：好的，已为您完成全部预订，安排如下： ✈️ **机票预订** - **去程**：上海 → 成都 出发日期：2026-04-18 航班号：MU5401 - **返程**：成都 → 上海 出发日期：2026-04-19 航班号：MU5402 🏨 **酒店预订** - 城市：成都 - 酒店：如家酒店 - 入住时间：2026-04-18 - 退房时间：2026-04-19 如需调整航班时间、酒店档次或增加其他出行安排，请随时告诉我。祝您出差顺利！

用户输入：我有明后两天(从上海)去成都的出差，请帮我预订酒店（明天入住，后天离开）。 系统回复：✅ 酒店预订成功！ 已为您安排好本次成都出差的住宿，具体信息如下： - **酒店城市**：成都 - **酒店名称**：如家酒店 - **入住时间**：2026 年 4 月 18 日（明天）12:00 - **退房时间**：2026 年 4 月 19 日（后天）12:00 祝您出差顺利！如果后续还需要我帮您预订机票或调整行程，请随时告诉我。

用户输入：我有明后两天(从上海)去成都的出差，请帮我预订机票（往返，明天出发，后天返回）。 系统回复：✅ 机票已为您成功预订，行程如下： ### ✈️ 去程 - **航线**：上海 → 成都 - **出发日期**：明天（2026-04-18） - **航班号**：MU5401 - **起飞时间**：12:00 ### ✈️ 返程 - **航线**：成都 → 上海 - **出发日期**：后天（2026-04-19） - **航班号**：MU5402 - **起飞时间**：12:00 如需 **预订酒店、调整航班时间或座位等级**，请随时告诉我，祝您出差顺利！

用户输入：随便说说！ 系统回复：意图不明，无法执行

针对第一个请求，Agent内部的调用链如下所示。

4. 完整的实现

如下提供完整的实现：

fromtypingimportAnnotated,Awaitable,Callable,TypedDictfromlangchain.agentsimportcreate_agent,AgentStatefromlangchain_core.messagesimportAIMessagefromlangchain_openaiimportChatOpenAIfromlangchain_mcp_adapters.clientimportMultiServerMCPClientfromlangchain.agents.middlewareimportAgentMiddleware,ExtendedModelResponse,ModelResponse,wrap_model_call,ModelRequestfromlangchain.toolsimportToolRuntime,tool,BaseToolfromlangchain_core.messagesimportSystemMessage,HumanMessage,ToolMessagefromlanggraph.typesimportCommandimportasynciofromdotenvimportload_dotenv load_dotenv()classSkill(TypedDict):name:strdescription:strcontent:strbuy_airplane_ticket_skill="""\ ## 基本流程 - 确定当前是否注册了`buy_airplane_ticket`工具，如果没有注册该工具,拒绝执行并回复用户：抱歉，我无法购买机票，因为相关工具未注册。 - 调用`buy_airplane_ticket`工具购买机票，最后将工具返回的预订信息整理后返回给用户。 ## 注意事项 - 购买机票是唯一的任务 - 调用`buy_airplane_ticket`工具是购买机票唯一的方式 - 只需要根据`buy_airplane_ticket`工具的Schema来分析购买机票的信息是否充分 - 可以完全自主决定航司、舱位等级和航班等信息，不需用户确认 """book_hotel_skill="""\ ## 基本流程 - 确定当前是否注册了`book_hotel`工具，如果没有注册该工具,拒绝执行并回复用户：抱歉，我无法预订酒店，因为相关工具未注册。 - 调用`book_hotel`工具预订酒店，最后将工具返回的预订信息整理后返回给用户。 ## 注意事项 - 预订酒店是唯一的任务 - 调用`book_hotel`工具是预订酒店唯一的方式 - 只需要根据`book_hotel`工具的Schema来分析预订酒店的信息是否充分 - 可以完全自主决定酒店、价位和房型等信息，不需用户确认 """all_skills=[{"name":"buy-airplane-ticket","description":"购买机票，只有在用户明确提出要求购买机票时才会使用","content":buy_airplane_ticket_skill},{"name":"book-hotel","description":"预订酒店,只有在用户明确提出要求预订酒店时才会使用","content":book_hotel_skill},]classState(AgentState):loaded_skills:Annotated[set[str],lambdax,y:x.union(y)]@tooldefload_skill(name:str,runtime:ToolRuntime)->Command|str:"""根据指定的技能名称加载技能详细内容"""forskillinall_skills:ifskill["name"]==name:returnCommand(update={"messages":[ToolMessage(content=skill["content"],tool_call_id=runtime.tool_call_id)],"loaded_skills":{name},})returnf"抱歉，未找到名为{name}的技能。"asyncdefmain():client=MultiServerMCPClient(connections={"server":{"transport":"stdio","command":"python","args":["server.py"]}})tools={tool.name:toolfortoolinawaitclient.get_tools(server_name="server")}skill_based_tools:dict[str,list[BaseTool]]={"buy-airplane-ticket":[tools["buy_airplane_ticket"]],"book-hotel":[tools["book_hotel"]],}classSkillMiddleware(AgentMiddleware):tools=[load_skill]def__init__(self):skills=(f" - **{skill['name']}**:{skill['description']}"forskillinall_skills)self.skills_prompt=f"""\ 你拥有如下可用技能：{'\n'.join(skills)}可以指定技能名称调用`load_skill`工具来获取指定技能的详细信息。 """asyncdefawrap_model_call(self,request:ModelRequest,handler:Callable[[ModelRequest],Awaitable[ModelResponse]],)->ModelResponse|AIMessage|ExtendedModelResponse:system_message=request.system_messageifsystem_messageisNone:system_message=SystemMessage(content=self.skills_prompt)else:cotents=system_message.content_blocks+[{"type":"text","text":self.skills_prompt}]system_message=SystemMessage(content_blocks=cotents)tools=[load_skill]loaded_skills=request.state.get("loaded_skills",set())forskillinloaded_skills:tools.extend(skill_based_tools.get(skill,[]))returnawaithandler(request.override(system_message=system_message,tools=tools))# type: ignoreagent=create_agent(model=ChatOpenAI(model="gpt-5.2-chat"),state_schema=State,tools=[load_skill,*tools.values()],system_prompt="你是一个差旅助理，请严格按照注册的技能要求来分析和执行用户的请求。如果用户没有明确预订酒店或者购买机票的需求，直接回复：意图不明，无法执行",middleware=[SkillMiddleware()])inputs=["我有明后两天(从上海)去成都的出差，请帮我预订酒店（明天入住，后天离开）和机票（往返，明天出发，后天返回）。","我有明后两天(从上海)去成都的出差，请帮我预订酒店（明天入住，后天离开）。","我有明后两天(从上海)去成都的出差，请帮我预订机票（往返，明天出发，后天返回）。","随便说说！"]formessageininputs:reuslt=awaitagent.ainvoke(input={"messages":[HumanMessage(content=message)]})# type: ignoreprint(f"用户输入：{message}")print(f"系统回复：{reuslt['messages'][-1].text}\n\n")asyncio.run(main())