1. 从“工具调用”到“技能编排”为什么我们需要Agent Skills如果你最近在折腾LangChain或者AI Agent开发大概率会频繁听到一个词Skills。它听起来比单纯的“工具”Tools要酷一些但很多人可能只是把它当作一个花哨的同义词来用。在我过去一年多的Agent项目实战中我逐渐意识到从“工具”思维切换到“技能”思维是构建一个真正智能、可靠且易于维护的Agent系统的关键分水岭。简单来说一个“工具”就像一把螺丝刀它有一个明确的接口拧螺丝你告诉Agent“用这把螺丝刀”它就去拧。而一个“技能”则更像“组装一台电脑”它内部可能调用了螺丝刀、硅脂涂抹器、防静电手环等多种工具并且遵循一套逻辑先装CPU再涂硅脂然后装散热器。Skills的本质是对一个或多个底层工具Tool的封装、编排与逻辑增强使其能够完成一个更高级、更符合人类语义的任务单元。为什么这个区别如此重要因为当我们直接让大模型LLM去调用零散的工具时经常会遇到几个头疼的问题工具选择困难LLM面对十几个功能相近的工具时容易选错、复杂流程割裂一个用户查询需要连续调用多个工具LLM的上下文可能丢失中间状态、错误处理缺失工具调用失败后LLM往往不知道如何优雅地重试或降级。而Skills正是为了解决这些问题而生。它允许我们将领域知识、业务流程和异常处理逻辑固化下来让LLM从一个“微观操作工”升级为“宏观调度员”直接调用封装好的高级技能从而大幅提升Agent的可靠性、执行效率和可维护性。2. LangChain中Skills的三种实现范式与核心API在LangChain的生态里实现一个Skill并没有唯一的“标准答案”这取决于你的场景复杂度。根据我的经验可以归纳为三种主流的实现范式它们各有优劣适用于不同的阶段。2.1 范式一基于Tool类的直接封装入门级这是最直接的方式适合将单个功能封装成技能。LangChain的BaseTool类是所有工具的基类创建一个Skill本质上就是继承它并实现_run方法。from langchain.tools import BaseTool from typing import Optional, Type from pydantic import BaseModel, Field class WeatherQueryInput(BaseModel): 查询天气的输入参数 city_name: str Field(description城市名称例如北京、上海) class GetWeatherSkill(BaseTool): name get_weather description 根据城市名称查询实时天气情况。 args_schema: Type[BaseModel] WeatherQueryInput def _run(self, city_name: str) - str: # 这里是技能的核心逻辑 # 1. 可能调用一个第三方天气API # 2. 对API返回的原始数据进行清洗和格式化 # 3. 加入业务逻辑比如根据温度给出穿衣建议 try: # 模拟API调用 temperature 22 condition 晴 advice 天气舒适建议穿单衣。 return f{city_name}的天气是{condition}气温{temperature}摄氏度。{advice} except Exception as e: # 技能内部的错误处理 return f查询{city_name}天气时出错{str(e)}请检查城市名称或稍后重试。 async def _arun(self, city_name: str): 异步版本 # 实现异步调用逻辑 pass # 使用技能 weather_skill GetWeatherSkill()核心要点与避坑args_schema是灵魂务必使用Pydantic模型明确定义输入参数。这不仅是类型检查更重要的是为LLM提供了清晰、结构化的“使用说明书”。LLM会根据这个schema来生成调用你的技能所需的参数。description要精准描述必须清晰说明技能做什么以及何时使用。例如“查询天气”不如“根据中文城市名查询实时温度、天气状况并给出简要穿衣建议”来得有效。错误处理内置化在_run内部处理好异常并返回对用户友好的错误信息而不是抛出异常让Agent不知所措。2.2 范式二基于ToolRunnable的序列化技能进阶级当你的技能需要按顺序执行多个步骤或者需要在步骤间传递和加工数据时简单的_run方法就显得力不从心了。这时可以利用LangChain的Runnable协议来构建一个技能流水线。from langchain.tools import BaseTool from langchain.schema.runnable import RunnablePassthrough, RunnableLambda from langchain.schema import StrOutputParser from typing import Dict class DataAnalysisSkill(BaseTool): name analyze_sales_data description 分析上传的销售数据CSV文件生成月度总结报告。 # 注意这里我们可能不定义固定的args_schema因为输入可能是一个文件路径或数据字典。 def _run(self, file_path: str) - str: # 我们不直接把逻辑写在这里而是调用一个构建好的执行链 chain self._build_analysis_chain() return chain.invoke({file_path: file_path}) def _build_analysis_chain(self): 构建一个可运行的技能执行链 # 步骤1加载数据 load_data RunnableLambda(lambda x: self._load_csv(x[file_path])) # 步骤2清洗数据例如处理空值 clean_data RunnableLambda(lambda df: self._clean_data(df)) # 步骤3核心分析计算月度销售额、Top产品等 run_analysis RunnableLambda(lambda df: self._perform_analysis(df)) # 步骤4格式化报告 format_report RunnableLambda(lambda result: self._format_to_markdown(result)) # 组装链 chain ( RunnablePassthrough.assign(dataload_data) # 传递原始输入并添加data字段 | RunnablePassthrough.assign(cleaned_dataclean_data) # 传递并添加cleaned_data | RunnablePassthrough.assign(analysis_resultrun_analysis) | format_report | StrOutputParser() ) return chain def _load_csv(self, file_path): # 模拟函数 return fLoaded DataFrame from {file_path} def _clean_data(self, df): # 模拟函数 return fCleaned {df} def _perform_analysis(self, df): # 模拟函数 return {monthly_sales: 100000, top_product: Product_A} def _format_to_markdown(self, result): return f# 销售分析报告\n- 月度总销售额{result[monthly_sales]}\n- 畅销产品{result[top_product]} # 使用方式不变但内部是强大的流水线 analysis_skill DataAnalysisSkill()为什么选择这种范式可观测性与调试Runnable链的每个步骤都是独立的你可以轻松插入日志、监控点或者单独测试某个步骤。灵活组合你可以复用其他Runnable组件如其他Tool、LLM调用、条件判断等像搭积木一样构建复杂技能。与LangGraph无缝集成Runnable是LangGraph节点的天然组成部分当你需要将技能嵌入到有状态、可循环的Agent工作流时这种范式迁移成本最低。2.3 范式三基于自定义Agent的宏观技能专家级对于一些极其复杂、决策路径长的任务例如“为我规划一个为期三天的北京旅游行程”将其作为一个单独的Skill来开发可能更合适。这个Skill本身内部就运行着一个微型的、专门的Agent。from langchain.agents import AgentExecutor, create_react_agent from langchain.tools import Tool from langchain_community.chat_models import ChatOpenAI from langchain.prompts import PromptTemplate class TravelPlanningSkill(BaseTool): name plan_travel_itinerary description 为一个城市规划多日旅游行程会综合考虑景点、餐饮、交通和预算。 def __init__(self): super().__init__() # 1. 为这个技能专门初始化一个LLM self.llm ChatOpenAI(modelgpt-4, temperature0) # 2. 定义这个微型Agent所需的子工具 self.sub_tools [ Tool(namesearch_attractions, funcself._search_attractions, description搜索某个城市的景点信息), Tool(namecheck_restaurant, funcself._check_restaurant, description查询餐厅评价和人均消费), Tool(nameestimate_transit, funcself._estimate_transit, description估算两点间的交通时间和方式), ] # 3. 创建专属的Agent提示词 prompt PromptTemplate.from_template( 你是一个专业的旅行规划师。请为用户规划在{city}为期{days}天的行程。 用户要求{user_request} 请使用工具来获取必要信息并生成一份详细到小时、包含景点、餐饮、交通和预估费用的行程单。 思考过程{agent_scratchpad} ) # 4. 创建并初始化一个Agent执行器 self.agent create_react_agent(llmself.llm, toolsself.sub_tools, promptprompt) self.agent_executor AgentExecutor(agentself.agent, toolsself.sub_tools, verboseTrue, handle_parsing_errorsTrue) def _run(self, city: str, days: int, user_request: str ) - str: # 运行这个微型Agent inputs {city: city, days: days, user_request: user_request} result self.agent_executor.invoke(inputs) return result[output] # 子工具的实现模拟 def _search_attractions(self, query): return f景点信息{query} def _check_restaurant(self, query): return f餐厅信息{query} def _estimate_transit(self, query): return f交通信息{query}适用场景与挑战场景任务本身具有高度的探索性和不确定性需要多次调用不同工具并基于中间结果进行决策。将这部分复杂性封装在一个Skill内部对外提供简洁的接口。挑战技能内部的Agent也会消耗Token增加延迟和成本。需要精心设计提示词和工具集防止内部Agent陷入死循环或做出低效决策。3. 技能Skills与智能体Agent的协同四种集成模式解析创建了Skills之后如何让主Agent智能地调用它们这里我总结了四种常见的集成模式它们对应着不同的控制粒度与灵活性需求。3.1 模式一直接注入工具列表最常用这是最经典的模式。将所有的Skills实例化后直接作为tools参数列表提供给create_react_agent或类似的Agent创建函数。from langchain.agents import create_react_agent, AgentExecutor from langchain.prompts import PromptTemplate from langchain_openai import ChatOpenAI # 假设我们已经有了几个技能实例 weather_skill GetWeatherSkill() analysis_skill DataAnalysisSkill() travel_skill TravelPlanningSkill() all_skills [weather_skill, analysis_skill, travel_skill] llm ChatOpenAI(modelgpt-4, temperature0) prompt PromptTemplate.from_template(...你的提示词...) agent create_react_agent(llmllm, toolsall_skills, promptprompt) agent_executor AgentExecutor(agentagent, toolsall_skills, verboseTrue) # 现在Agent可以根据用户问题自主选择调用哪个技能 result agent_executor.invoke({input: 帮我看看北京明天天气怎么样})经验之谈技能描述description是导航图Agent完全依赖每个Skill的name和description来决定调用谁。因此描述的独特性至关重要。避免出现“处理数据”、“获取信息”这种模糊描述。数量控制通常一个Agent同时管理的Skills数量不宜超过10-15个。过多会导致LLM选择困难性能下降。可以考虑按功能域拆分出多个专门的Agent。3.2 模式二基于路由Router的技能分发当技能数量众多或功能有重叠时可以引入一个“路由Agent”作为调度层。用户请求先到达路由Agent由它决定将任务分发给哪个专用的“技能Agent”。用户 - 路由Agent - (判断) - 天气Skill Agent - 执行 - 结果返回给用户 - (判断) - 旅行Skill Agent - 执行 - (判断) - 数据分析Skill Agent - 执行在LangChain中这可以通过MultiRouteChain或利用LLMRouterChain配合ConversationChain来实现。更现代、更强大的方式是使用LangGraph来显式地构建这种有状态的工作流。在LangGraph中你可以定义一个Router节点根据LLM的判断将状态State指向不同的技能处理分支。优势架构清晰职责分离每个技能Agent可以有自己的优化提示词和工具集易于维护和扩展。3.3 模式三技能的动态加载与注册在某些场景下我们可能不希望一次性加载所有技能而是根据上下文、用户身份或会话阶段动态地添加或移除技能。这需要维护一个中央的技能注册表。class SkillRegistry: def __init__(self): self._skills {} def register(self, skill: BaseTool): if skill.name in self._skills: raise ValueError(fSkill {skill.name} already registered.) self._skills[skill.name] skill def get_skill(self, name: str) - Optional[BaseTool]: return self._skills.get(name) def get_skills_for_context(self, user_context: Dict) - List[BaseTool]: # 基于用户上下文如角色、权限、当前任务过滤技能 available_skills [] for skill in self._skills.values(): if self._is_skill_relevant(skill, user_context): available_skills.append(skill) return available_skills def _is_skill_relevant(self, skill, context): # 实现你的业务逻辑例如检查用户权限标签是否匹配技能所需标签 return True # 使用 registry SkillRegistry() registry.register(weather_skill) registry.register(analysis_skill) # 当新会话开始时根据用户信息获取可用技能 user_ctx {role: analyst, project: sales} available_tools registry.get_skills_for_context(user_ctx) # 然后用 available_tools 初始化Agent这种模式在构建企业级、多租户的Agent平台时非常有用。3.4 模式四技能作为LangGraph中的功能节点这是目前构建复杂、稳定Agent系统最推荐的方式。在LangGraph中每个Skill可以被建模为一个独立的节点Node而Agent或LLM则作为调度中心。from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated import operator # 1. 定义状态结构 class AgentState(TypedDict): messages: Annotated[list, operator.add] # 对话历史 next: str # 下一步该执行哪个节点 skill_result: str # 技能执行结果 # 2. 定义技能节点函数 def call_weather_skill(state: AgentState): # 从state中解析出参数例如从最新一条用户消息中提取城市 last_message state[“messages”][-1] city extract_city(last_message.content) # 假设的解析函数 result weather_skill.run(city) return {skill_result: result, next: process_result} def call_travel_skill(state: AgentState): # ...类似逻辑 pass # 3. 定义路由节点LLM决策 def router(state: AgentState): llm_decision llm.invoke(f根据对话历史决定下一步{state[messages]}。可选动作weather, travel, end。) if weather in llm_decision.content.lower(): return weather_node elif travel in llm_decision.content.lower(): return travel_node else: return end # 4. 构建图 workflow StateGraph(AgentState) workflow.add_node(router, router) workflow.add_node(weather_node, call_weather_skill) workflow.add_node(travel_node, call_travel_skill) workflow.add_node(process_result, process_result_node) # 处理技能结果并生成回复的节点 workflow.set_entry_point(router) workflow.add_conditional_edges(router, router) # 根据router的输出决定下一个节点 workflow.add_edge(weather_node, process_result) workflow.add_edge(travel_node, process_result) workflow.add_edge(process_result, END) app workflow.compile()这种模式的优势是革命性的显式控制流整个Agent的决策和执行流程一目了然不再是LLM内部的“黑箱”。持久化状态State可以持久化轻松实现长对话、复杂任务的中断与恢复。错误处理与重试可以轻松地在图中添加错误处理节点当技能调用失败时路由到重试或降级逻辑。并行与异步可以设计让多个技能节点并行执行极大提升效率。4. 实战避坑指南提升Skill可靠性的五个关键细节在真实项目中让Skills稳定工作远比跑通一个Demo复杂。下面是我从多个踩坑项目中总结出的核心经验。4.1 细节一为技能设计“防御性”的输入输出SchemaPydantic Schema不仅是类型声明更是与LLM沟通的契约。一个健壮的Schema能极大减少调用错误。反面教材class PoorInput(BaseModel): query: str问题query太模糊LLM可能传入“北京明天天气”也可能传入“帮我查下天气”。技能内部需要做复杂的字符串解析极易出错。最佳实践class RobustWeatherInput(BaseModel): location: str Field(description具体的城市或区县名称例如北京市海淀区) date: Optional[str] Field(default今天, description查询日期格式为‘今天’、‘明天’或‘YYYY-MM-DD’) need_detail: bool Field(defaultFalse, description是否需要湿度、风速等详细信息) validator(date) def validate_date(cls, v): # 添加自定义验证逻辑确保日期格式有效或转换为标准格式 if v not in [今天, 明天]: try: datetime.strptime(v, %Y-%m-%d) except ValueError: raise ValueError(日期格式错误请使用‘今天’、‘明天’或‘YYYY-MM-DD’格式) return v这样做的好处字段语义清晰LLM能准确理解每个参数要填什么。提供默认值降低LLM的思考负担。内置验证在参数传入技能逻辑前就拦截非法值返回清晰的错误信息给LLM让它能重新调整输入。4.2 细节二实现技能的“优雅降级”与重试机制网络调用、第三方API不稳定是常态。技能内部必须有容错设计。def _run_with_retry(self, api_func, *args, max_retries2, **kwargs): 带重试的调用封装 last_exception None for attempt in range(max_retries 1): try: return api_func(*args, **kwargs) except (TimeoutError, ConnectionError) as e: last_exception e if attempt max_retries: time.sleep(1 * (attempt 1)) # 指数退避 continue # 所有重试都失败执行降级逻辑 return self._fallback_logic(*args, **kwargs) def _fallback_logic(self, city): 降级逻辑返回缓存数据、估算数据或友好的错误提示 # 例如从本地缓存文件读取最近一次的成功数据 # 或者返回一个基于历史数据的估算值并明确告知用户“当前服务不稳定以下是预估数据” return f暂时无法获取{city}的实时天气。根据历史数据此时节平均气温约为15-25度。在LangGraph中你甚至可以将这个重试和降级逻辑建模为图中的一个独立节点形成调用技能 - 失败 - 重试节点/降级节点的清晰流程。4.3 细节三为技能添加可观测性Observability技能上线后你需要知道它被调用的频率、成功率、耗时。简单的打印日志是不够的。import time from functools import wraps import statsd # 或使用OpenTelemetry def observe_skill(func): 一个简单的技能执行观测装饰器 wraps(func) def wrapper(self, *args, **kwargs): skill_name self.name start_time time.time() try: result func(self, *args, **kwargs) duration (time.time() - start_time) * 1000 # 毫秒 # 发送指标到监控系统 statsd_client.timing(fskill.{skill_name}.duration, duration) statsd_client.incr(fskill.{skill_name}.success) return result except Exception as e: statsd_client.incr(fskill.{skill_name}.error) # 可以记录详细的错误信息和参数注意脱敏 logger.error(fSkill {skill_name} failed with args {args}, error: {e}) raise # 或者返回降级结果 return wrapper # 在技能的_run方法上使用装饰器 class MySkill(BaseTool): observe_skill def _run(self, ...): ...这样你就能在Grafana等看板上清晰地看到每个技能的健康状况快速定位瓶颈。4.4 细节四管理技能间的依赖与冲突当多个技能需要共享资源如数据库连接、API客户端或存在执行顺序依赖时需要妥善管理。共享依赖注入不要在技能内部硬编码创建资源客户端。应该在初始化技能时通过构造函数注入。class DBAnalysisSkill(BaseTool): def __init__(self, db_client): super().__init__() self.db_client db_client # 从外部传入共享的数据库客户端 ...技能执行副作用如果一个技能会修改某个共享状态如向一个公共任务队列添加作业必须在描述中清晰说明并考虑在LangGraph中用状态管理来协调避免并发冲突。4.5 细节五技能的版本化与热更新对于线上服务技能的迭代更新是常态。你需要考虑版本标识在技能类中添加version属性。向后兼容更新输入输出Schema时尽量保证旧版调用方式仍能工作一段时间。热更新策略通过技能注册表Skill Registry动态替换技能实例而无需重启整个Agent服务。新的用户会话将使用新技能而进行中的会话可能继续使用旧技能实例直至结束。5. 从Skills到智能工作流LangGraph的降维打击最后我想强调当你熟练掌握了Skills的构建后LangGraph是你将能力提升到下一个层次的必然选择。它让你从“思考如何让LLM调用工具”转变为“设计一个智能的工作流”。在LangGraph的范式下Skill就是一个Node每个技能被封装成一个纯净的函数节点只负责计算。LLM也是一个Node专门负责决策路由和生成自然语言回复。状态State是共享内存所有节点通过修改和读取State来协作。边Edges是控制流清晰地定义了工作流的走向包括条件分支、循环和并行。以前用LangChain Agent难以实现的复杂逻辑比如“先执行A技能如果结果满足条件X则执行B否则执行C最后汇总结果生成报告”在LangGraph中可以通过构图直观地实现。这大大降低了心智负担也让整个系统的可调试性和可维护性得到了质的飞跃。我个人的实践路径是先从封装好单个Tool开始然后用Runnable链组合成复杂Skill最后将这些Skill作为节点用LangGraph编织成强大的智能工作流。这个过程中对Skill的良好抽象和设计是最终构建出稳定、高效Agent系统的基石。
