1. 从“单打独斗”到“团队协作”为什么我们需要LangGraph中的Skills如果你和我一样从LangChain一路摸索到LangGraph可能会经历一个相似的认知转变早期我们痴迷于如何让一个LLM大语言模型通过Prompt工程和工具调用完成一个又一个独立的任务比如查天气、写邮件、分析数据。这就像训练一个“全能超人”希望它无所不能。但很快你就会发现这种模式的瓶颈非常明显——随着任务复杂度的提升Prompt会变得无比臃肿逻辑分支多到难以维护一旦某个环节出错整个链条就可能崩溃。这时LangGraph的出现带来了新的范式图计算。它不再将智能体视为一个“黑盒”而是将其工作流拆解成一个个相互连接的“节点”State Nodes通过“边”Edges来控制流转逻辑。这解决了复杂流程编排的问题让智能体有了清晰的“思维步骤”。但问题又来了每个节点里的“干活”能力从哪来难道我们要在每个节点的函数里都写上一大堆重复的、与业务逻辑耦合的代码吗比如一个节点需要调用搜索引擎另一个节点需要访问数据库再一个节点需要生成图表。如果都硬编码进去这个图很快就会变得难以阅读、难以复用、难以测试。这就是Skills登场的核心场景。在我的实践中我把Skills理解为智能体工作流中的“标准化技能包”或“可插拔的功能模块”。它不是LangGraph官方定义的一个类而是一种被社区广泛采纳的最佳实践架构思想。其核心价值在于解耦与复用将具体的功能实现如调用API、执行计算、访问资源封装成独立的、定义良好的函数或类然后像乐高积木一样按需装配到LangGraph的各个节点中。一个设计良好的Skill应该只关心“如何完成某件具体的事”而不需要知道它会被用在哪个工作流、哪个节点以及前后发生了什么。这直接解决了智能体开发中的几个核心痛点代码重复、维护困难、能力边界模糊以及团队协作时的模块化分工。2. Skills的本质超越“工具”的模块化设计很多人会把Skills和LangChain中的Tools概念混淆。确实它们有相似之处都是可调用的功能单元。但在我看来Skills是Tools在LangGraph图计算范式下的进化与泛化其内涵和外延都更广。Tools工具通常特指那些能够被LLM通过特定格式如OpenAI的Function Calling识别、描述并自主决定调用的函数。它的核心交互对象是LLM设计重点是“如何让LLM理解并使用我”。一个Tool必须包含name、description、args_schema等元数据。Skills技能则跳出了“必须被LLM直接调用”的约束。它是一个更上层的抽象核心是功能模块化。一个Skill可以是一个Tool这是最常见的形式将一个封装好的功能暴露给LLM调用。是一个纯函数在图的某个节点中由开发者的逻辑决定何时调用无需LLM参与决策。例如一个专门用于数据清洗的clean_data()函数。是一个类Class封装更复杂的状态和行为。例如一个DatabaseConnector类内部管理连接池提供query、insert等方法可以被多个节点共享。是一组相关功能的集合比如一个DataVizSkill包里面包含了generate_bar_chart、plot_timeline、export_to_html等多个方法。为了更直观地对比我整理了以下表格特性维度LangChain ToolsLangGraph Skills (实践概念)核心目的让LLM能够识别并调用外部功能实现工作流中功能模块的解耦与复用调用方主要由LLM Agent自主决策调用可由LLM、图节点逻辑、或其他Skill调用设计重点规范的描述name, description以供LLM理解清晰的接口、独立的功能、可测试性依赖关系通常依赖LLM的调用格式仅依赖输入/输出约定与LLM解耦复用范围主要在Agent内部跨多个智能体、多个工作流、多个项目典型场景“请帮我搜索XXX” - LLM调用搜索Tool在“报告生成”工作流中节点A调用“数据获取”Skill节点B调用“图表生成”Skill所以当我们说“在LangGraph中集成Skills”时我们实际在做的是为我们的智能体工作流设计和构建一个专属于当前业务领域的、高内聚低耦合的“技能库”。这个技能库中的每一个技能都像螺丝刀、扳手一样功能明确随时可取用从而让我们能更专注于用LangGraph这张“蓝图”来设计和组装复杂的智能体行为而不是反复制造“工具”。3. 实战构建与集成Skills的完整流程理论讲完了我们直接上手。假设我们要构建一个“市场分析报告生成”智能体。它的工作流Graph可能包括获取最新行业数据、进行竞品对比分析、生成总结文案、制作可视化图表。我们将为这个工作流配套开发相应的Skills。3.1 第一步技能规划与设计在写代码之前先进行设计。根据工作流我们初步规划以下SkillsDataFetcherSkill负责从内部数据库或外部API如聚合数据平台获取原始市场数据。CompetitorAnalysisSkill接收数据运用一些分析逻辑如计算市场份额变化、关键词热度对比输出结构化分析结果。ReportWritingSkill基于分析结果调用LLM生成格式优美的中文报告段落。ChartGeneratorSkill调用如Matplotlib或Plotly的封装将数据转化为图表图片并保存。设计原则单一职责每个Skill只做一件事并把它做好。明确接口定义清晰的输入参数和返回类型推荐使用Pydantic模型。无状态性理想情况下Skill本身不维护内部状态除非是连接池这类资源管理。状态应由LangGraph的State来管理。错误处理Skill内部应妥善处理异常并抛出具有明确意义的错误类型方便上层节点捕获和决策。3.2 第二步实现基础Skill以DataFetcherSkill为例我们不直接写进节点而是先创建独立的技能模块。# skills/data_fetcher.py import aiohttp from pydantic import BaseModel, Field from typing import List, Dict, Any import asyncio class MarketDataQuery(BaseModel): 获取市场数据的查询参数模型 industry: str Field(description行业领域如新能源汽车、智能手机) timeframe: str Field(description时间范围如2024-Q1, last_30_days) metrics: List[str] Field(default_factorylist, description需要获取的指标如market_share, growth_rate, search_volume) class MarketDataResponse(BaseModel): 市场数据响应模型 success: bool data: Dict[str, Any] None error_message: str None class DataFetcherSkill: 数据获取技能 def __init__(self, api_base_url: str, api_key: str None): self.api_base_url api_base_url self.api_key api_key # 可以初始化一些会话资源如aiohttp.ClientSession self._session None async def _get_session(self): if self._session is None: self._session aiohttp.ClientSession() return self._session async def fetch_market_data(self, query: MarketDataQuery) - MarketDataResponse: 核心技能方法获取市场数据 # 1. 构建请求参数这里只是一个示例实际会更复杂 params { industry: query.industry, timeframe: query.timeframe, metrics: ,.join(query.metrics) } headers {} if self.api_key: headers[Authorization] fBearer {self.api_key} try: session await self._get_session() async with session.get( f{self.api_base_url}/market_data, paramsparams, headersheaders, timeoutaiohttp.ClientTimeout(total30) ) as response: if response.status 200: json_data await response.json() return MarketDataResponse(successTrue, datajson_data) else: error_text await response.text() return MarketDataResponse( successFalse, error_messagefAPI请求失败状态码{response.status}, 详情{error_text[:200]} ) except asyncio.TimeoutError: return MarketDataResponse(successFalse, error_message请求外部API超时) except Exception as e: return MarketDataResponse(successFalse, error_messagef未知错误{str(e)}) # 注意实际生产环境需要更精细的异常分类和处理 async def close(self): 清理资源 if self._session: await self._session.close() # 可以同时将其包装成一个标准的LangChain Tool供LLM直接调用 from langchain.tools import tool tool(args_schemaMarketDataQuery) async def fetch_market_data_tool(industry: str, timeframe: str, metrics: List[str] None): 获取指定行业和时间的市场数据。 skill DataFetcherSkill(api_base_urlhttps://api.example.com) # 实例化实际应从配置或上下文获取 query MarketDataQuery(industryindustry, timeframetimeframe, metricsmetrics or []) result await skill.fetch_market_data(query) if not result.success: return f获取数据失败{result.error_message} return f数据获取成功。摘要{str(result.data)[:500]}... # 返回给LLM的摘要这个实现展示了几个关键点使用Pydantic定义接口MarketDataQuery和MarketDataResponse明确了技能的“契约”便于使用方理解和验证。技能类封装DataFetcherSkill类封装了所有细节API地址、认证、请求逻辑、错误处理。它易于独立测试。可选的Tool包装通过tool装饰器我们可以轻松地将这个技能的核心方法暴露给LLM作为一个Tool。这体现了Skill的灵活性。3.3 第三步在LangGraph节点中集成Skill现在我们来到LangGraph的工作流定义中。关键是如何将这些技能实例优雅地注入到需要它们的节点函数里。我强烈推荐使用依赖注入的模式而不是在节点函数内部硬编码创建Skill实例。# graph.py from typing import Annotated, TypedDict from langgraph.graph import StateGraph, END import operator from skills.data_fetcher import DataFetcherSkill, MarketDataQuery from skills.competitor_analysis import CompetitorAnalysisSkill # ... 导入其他技能 # 1. 定义Graph的状态结构 class AgentState(TypedDict): 智能体的工作状态 industry: str timeframe: str required_metrics: list raw_market_data: dict None # 由fetch_node填充 analysis_result: dict None # 由analyze_node填充 report_text: str None # 由report_node填充 chart_paths: list None # 由chart_node填充 errors: list None # 用于收集各环节错误 # 2. 创建技能实例通常从配置或工厂类中获取 # 在实际项目中这些实例的创建和管理可以通过像fastapi.Depends或injector这样的依赖注入容器来优化。 data_fetcher DataFetcherSkill(api_base_urlos.getenv(MARKET_API_URL), api_keyos.getenv(API_KEY)) analyst CompetitorAnalysisSkill() # ... 初始化其他技能 # 3. 定义节点函数并将技能实例作为“上下文”或参数传入 async def fetch_data_node(state: AgentState): 节点获取市场数据 print(f[Fetch Node] 正在获取 {state[industry]} 行业的数据...) query MarketDataQuery( industrystate[industry], timeframestate[timeframe], metricsstate[required_metrics] ) # 调用Skill response await data_fetcher.fetch_market_data(query) new_state {raw_market_data: None, errors: state.get(errors, [])} if response.success: new_state[raw_market_data] response.data print([Fetch Node] 数据获取成功。) else: error_msg f数据获取失败{response.error_message} print(f[Fetch Node] {error_msg}) new_state[errors] state.get(errors, []) [error_msg] return new_state async def analyze_data_node(state: AgentState): 节点分析竞品数据 if state.get(raw_market_data) is None: error 分析节点上游数据为空无法进行分析。 return {errors: state.get(errors, []) [error]} print([Analysis Node] 开始竞品分析...) # 调用另一个Skill analysis analyst.analyze(state[raw_market_data]) return {analysis_result: analysis} # 4. 构建图 builder StateGraph(AgentState) builder.add_node(fetch, fetch_data_node) builder.add_node(analyze, analyze_data_node) # ... 添加 report_node, chart_node 等 # 5. 设置边条件边或固定边 builder.set_entry_point(fetch) builder.add_edge(fetch, analyze) # 假设获取数据后总是进行分析 # ... 设置更多边 graph builder.compile()注意上面的代码中技能实例data_fetcher,analyst是在全局作用域创建的。对于简单应用可行但对于复杂应用或需要动态配置的场景更好的模式是使用闭包或类来将技能绑定到图上。例如可以定义一个GraphBuilder类在__init__中初始化所有技能然后节点方法定义为实例方法这样它们就能通过self访问这些技能实例。或者使用更高级的依赖注入框架。3.4 第四步进阶集成模式——将Skill作为Tool提供给LLM节点在很多工作流中我们有一个专门的节点是“LLM思考与决策”节点它需要根据当前状态决定下一步做什么并可能调用各种Tools。这时我们需要将Skills包装的Tools提供给这个节点。from langchain_openai import ChatOpenAI from langgraph.prebuilt import ToolExecutor, ToolInvocation from langgraph.graph import StateGraph, MessagesState from typing import TypedDict, Annotated, Sequence from langchain_core.messages import BaseMessage, HumanMessage, ToolMessage import operator # 假设我们已经有了包装好的Tools列表 tools [fetch_market_data_tool, another_skill_tool, ...] # 这些tool来自我们的Skills tool_executor ToolExecutor(tools) llm ChatOpenAI(modelgpt-4-turbo-preview) llm_with_tools llm.bind_tools(tools) # 关键将tools绑定给LLM它才能知道有哪些工具可用 class AgentState(TypedDict): messages: Annotated[Sequence[BaseMessage], operator.add] # 消息序列 def llm_agent_node(state: AgentState): LLM决策节点思考并可能调用工具 print([LLM Node] 正在思考...) last_message state[messages][-1] # 1. 调用LLM它会根据对话历史和绑定的tools决定是回复还是调用工具 response llm_with_tools.invoke(state[messages]) # 2. 将LLM的响应追加到消息历史 new_messages [response] # 3. 检查LLM是否想调用工具 if response.tool_calls: print(f[LLM Node] 决定调用工具{[tc[name] for tc in response.tool_calls]}) # 对于每个工具调用执行它 for tool_call in response.tool_calls: # 使用ToolExecutor执行工具调用 result tool_executor.invoke(tool_call) # 将工具执行结果作为ToolMessage追加 new_messages.append(ToolMessage(contentstr(result), tool_call_idtool_call[id])) # 工具调用后通常需要让LLM再次思考所以这个节点之后应该再次连接到自己或一个处理节点 return {messages: new_messages} # 如果没有工具调用LLM已经给出了最终回答可以流向END return {messages: new_messages} # 构建图 builder StateGraph(AgentState) builder.add_node(agent, llm_agent_node) builder.set_entry_point(agent) # 这里需要更复杂的边逻辑来处理“是否调用了工具”的条件判断通常使用langgraph.graph.StateGraph.add_conditional_edges # 这是一个简化示例实际需要条件边来循环调用。这种模式下Skills通过Tool接口成为了LLM可以自主调用的“手”和“脚”极大地扩展了智能体的能力边界。而整个工作流的编排和控制仍然由LangGraph的图结构牢牢掌握。4. 架构思考如何设计可维护、可扩展的Skill体系当Skills数量增多后管理它们就成了一个挑战。以下是我从几个项目中总结出的架构经验1. 技能分类与分层基础技能与具体业务无关的通用能力如HTTPClientSkill、FileIOSkill、CalculationsSkill。这些可以作为所有项目的底层依赖。领域技能与特定业务领域相关如MarketDataSkill、CustomerServiceSkill、CodeReviewSkill。它们建立在基础技能之上。组合技能由多个更细粒度技能组合而成完成一个更复杂的子任务。例如GenerateQuarterlyReportSkill内部可能依次调用FetchDataSkill、AnalyzeSkill、WriteSummarySkill。2. 统一的技能注册与管理中心创建一个SkillRegistry技能注册表单例或类。所有技能在应用启动时向注册表注册自己并提供一个唯一的名称和版本。节点或LLM可以通过名称从注册表中获取技能实例。这带来了以下好处集中配置技能的初始化参数如API密钥、端点可以在注册中心统一管理。动态加载可以根据配置或环境动态启用或禁用某些技能。依赖注入注册中心可以处理技能之间的依赖关系例如ChartGeneratorSkill依赖于DataCacheSkill。# skill_registry.py (简化示例) class SkillRegistry: _instance None _skills {} def __new__(cls): if cls._instance is None: cls._instance super().__new__(cls) return cls._instance def register(self, name: str, skill_factory): 注册一个技能工厂函数 self._skills[name] skill_factory def get(self, name: str, **kwargs): 获取一个技能实例可以传入初始化参数 if name not in self._skills: raise KeyError(fSkill {name} not registered.) return self._skills[name](**kwargs) # 在应用初始化时注册技能 registry SkillRegistry() registry.register(data_fetcher, lambda: DataFetcherSkill(api_base_urlos.getenv(API_URL))) registry.register(competitor_analyst, lambda: CompetitorAnalysisSkill()) # 在节点中使用 data_fetcher registry.get(data_fetcher)3. 技能版本化与兼容性当技能逻辑更新时比如API接口变了为了避免破坏已有的工作流可以考虑引入版本号。节点在调用技能时指定所需版本注册中心返回对应版本的实例。这在大规模协作中尤为重要。4. 技能的可观测性为每个技能添加详细的日志记录、执行时间监控和错误统计。这不仅能帮助调试还能让你了解智能体工作流中各个技能的“健康度”和性能瓶颈。可以在技能基类或通过装饰器模式统一实现。5. 避坑指南Skills开发与集成中的常见陷阱在实际项目中踩过不少坑这里分享几个最典型的陷阱一技能与状态State的过度耦合症状Skill函数直接读取或修改LangGraph的整个State对象。 后果技能变得不可复用且测试困难。State结构的任何变动都会导致技能崩溃。 正确做法Skill只应通过明确的输入参数接收它需要的数据并通过返回值输出结果。状态管理是LangGraph节点的职责。节点作为“协调者”从State中提取数据传给Skill再将Skill的结果写回State。陷阱二忽视异步Async与同步Sync的混用症状在异步的LangGraph节点中调用了一个执行阻塞I/O操作的同步Skill函数。 后果整个事件循环被阻塞严重降低智能体的并发性能在高负载下可能导致系统瘫痪。 正确做法统一使用异步async/await。如果Skill内部涉及网络请求、文件读写、数据库查询等I/O操作务必使用异步库如aiohttp,aiomysql,aiofiles。如果必须使用同步库请使用asyncio.to_thread将其放到线程池中执行避免阻塞主事件循环。陷阱三脆弱的错误处理症状Skill内部只是简单打印错误或返回None节点函数没有检查Skill的失败情况。 后果工作流在静默中失败状态数据出现不一致难以定位问题根源。 正确做法Skill应定义明确的错误返回类型如前文的MarketDataResponse包含success和error_message。节点函数必须检查Skill的执行结果并根据业务逻辑决定下一步是重试、记录错误并继续、还是跳转到专门的“错误处理节点”一个健壮的工作流必须有错误处理路径。陷阱四技能粒度过粗或过细症状一个MegaSkill做了十件事或者十个TinySkill每个只做一行简单的字符串处理。 后果过粗的技能难以复用和测试过细的技能导致节点代码变成繁琐的“胶水代码”管理成本激增。 正确做法遵循“单一职责”和“高内聚”原则。一个Skill应该对应一个有意义的、独立的业务功能单元。例如“发送邮件”是一个好的Skill“构建邮件标题”可能就太细了。如果发现多个节点总是以固定顺序调用一组细粒度技能那么就应该考虑将它们组合成一个新的、更粗粒度的“组合技能”。陷阱五硬编码的配置和依赖症状Skill的API地址、密钥等直接写在类定义中。 后果无法适应不同环境开发、测试、生产也无法安全地管理密钥。 正确做法通过__init__方法接收配置或从环境变量、配置中心读取。使用依赖注入模式来管理Skill所依赖的其他服务如数据库连接池、缓存客户端。在LangGraph的生态中集成Skills本质上是一场关于如何构建模块化、可维护AI应用的工程实践。它要求我们不仅关注单个智能体的“智力”更要关注整个系统架构的“整洁度”。当你把那些杂乱无章的功能点规整成一个个边界清晰、接口明确的Skills时你会发现构建复杂智能体工作流从此变得像搭积木一样清晰而愉快。你的代码库将从一个“脚本集合”进化成一个真正的“技能库”而这正是AI应用工程化道路上至关重要的一步。 LangGraph Skills架构：构建模块化AI智能体的工程实践
1. 从“单打独斗”到“团队协作”为什么我们需要LangGraph中的Skills如果你和我一样从LangChain一路摸索到LangGraph可能会经历一个相似的认知转变早期我们痴迷于如何让一个LLM大语言模型通过Prompt工程和工具调用完成一个又一个独立的任务比如查天气、写邮件、分析数据。这就像训练一个“全能超人”希望它无所不能。但很快你就会发现这种模式的瓶颈非常明显——随着任务复杂度的提升Prompt会变得无比臃肿逻辑分支多到难以维护一旦某个环节出错整个链条就可能崩溃。这时LangGraph的出现带来了新的范式图计算。它不再将智能体视为一个“黑盒”而是将其工作流拆解成一个个相互连接的“节点”State Nodes通过“边”Edges来控制流转逻辑。这解决了复杂流程编排的问题让智能体有了清晰的“思维步骤”。但问题又来了每个节点里的“干活”能力从哪来难道我们要在每个节点的函数里都写上一大堆重复的、与业务逻辑耦合的代码吗比如一个节点需要调用搜索引擎另一个节点需要访问数据库再一个节点需要生成图表。如果都硬编码进去这个图很快就会变得难以阅读、难以复用、难以测试。这就是Skills登场的核心场景。在我的实践中我把Skills理解为智能体工作流中的“标准化技能包”或“可插拔的功能模块”。它不是LangGraph官方定义的一个类而是一种被社区广泛采纳的最佳实践架构思想。其核心价值在于解耦与复用将具体的功能实现如调用API、执行计算、访问资源封装成独立的、定义良好的函数或类然后像乐高积木一样按需装配到LangGraph的各个节点中。一个设计良好的Skill应该只关心“如何完成某件具体的事”而不需要知道它会被用在哪个工作流、哪个节点以及前后发生了什么。这直接解决了智能体开发中的几个核心痛点代码重复、维护困难、能力边界模糊以及团队协作时的模块化分工。2. Skills的本质超越“工具”的模块化设计很多人会把Skills和LangChain中的Tools概念混淆。确实它们有相似之处都是可调用的功能单元。但在我看来Skills是Tools在LangGraph图计算范式下的进化与泛化其内涵和外延都更广。Tools工具通常特指那些能够被LLM通过特定格式如OpenAI的Function Calling识别、描述并自主决定调用的函数。它的核心交互对象是LLM设计重点是“如何让LLM理解并使用我”。一个Tool必须包含name、description、args_schema等元数据。Skills技能则跳出了“必须被LLM直接调用”的约束。它是一个更上层的抽象核心是功能模块化。一个Skill可以是一个Tool这是最常见的形式将一个封装好的功能暴露给LLM调用。是一个纯函数在图的某个节点中由开发者的逻辑决定何时调用无需LLM参与决策。例如一个专门用于数据清洗的clean_data()函数。是一个类Class封装更复杂的状态和行为。例如一个DatabaseConnector类内部管理连接池提供query、insert等方法可以被多个节点共享。是一组相关功能的集合比如一个DataVizSkill包里面包含了generate_bar_chart、plot_timeline、export_to_html等多个方法。为了更直观地对比我整理了以下表格特性维度LangChain ToolsLangGraph Skills (实践概念)核心目的让LLM能够识别并调用外部功能实现工作流中功能模块的解耦与复用调用方主要由LLM Agent自主决策调用可由LLM、图节点逻辑、或其他Skill调用设计重点规范的描述name, description以供LLM理解清晰的接口、独立的功能、可测试性依赖关系通常依赖LLM的调用格式仅依赖输入/输出约定与LLM解耦复用范围主要在Agent内部跨多个智能体、多个工作流、多个项目典型场景“请帮我搜索XXX” - LLM调用搜索Tool在“报告生成”工作流中节点A调用“数据获取”Skill节点B调用“图表生成”Skill所以当我们说“在LangGraph中集成Skills”时我们实际在做的是为我们的智能体工作流设计和构建一个专属于当前业务领域的、高内聚低耦合的“技能库”。这个技能库中的每一个技能都像螺丝刀、扳手一样功能明确随时可取用从而让我们能更专注于用LangGraph这张“蓝图”来设计和组装复杂的智能体行为而不是反复制造“工具”。3. 实战构建与集成Skills的完整流程理论讲完了我们直接上手。假设我们要构建一个“市场分析报告生成”智能体。它的工作流Graph可能包括获取最新行业数据、进行竞品对比分析、生成总结文案、制作可视化图表。我们将为这个工作流配套开发相应的Skills。3.1 第一步技能规划与设计在写代码之前先进行设计。根据工作流我们初步规划以下SkillsDataFetcherSkill负责从内部数据库或外部API如聚合数据平台获取原始市场数据。CompetitorAnalysisSkill接收数据运用一些分析逻辑如计算市场份额变化、关键词热度对比输出结构化分析结果。ReportWritingSkill基于分析结果调用LLM生成格式优美的中文报告段落。ChartGeneratorSkill调用如Matplotlib或Plotly的封装将数据转化为图表图片并保存。设计原则单一职责每个Skill只做一件事并把它做好。明确接口定义清晰的输入参数和返回类型推荐使用Pydantic模型。无状态性理想情况下Skill本身不维护内部状态除非是连接池这类资源管理。状态应由LangGraph的State来管理。错误处理Skill内部应妥善处理异常并抛出具有明确意义的错误类型方便上层节点捕获和决策。3.2 第二步实现基础Skill以DataFetcherSkill为例我们不直接写进节点而是先创建独立的技能模块。# skills/data_fetcher.py import aiohttp from pydantic import BaseModel, Field from typing import List, Dict, Any import asyncio class MarketDataQuery(BaseModel): 获取市场数据的查询参数模型 industry: str Field(description行业领域如新能源汽车、智能手机) timeframe: str Field(description时间范围如2024-Q1, last_30_days) metrics: List[str] Field(default_factorylist, description需要获取的指标如market_share, growth_rate, search_volume) class MarketDataResponse(BaseModel): 市场数据响应模型 success: bool data: Dict[str, Any] None error_message: str None class DataFetcherSkill: 数据获取技能 def __init__(self, api_base_url: str, api_key: str None): self.api_base_url api_base_url self.api_key api_key # 可以初始化一些会话资源如aiohttp.ClientSession self._session None async def _get_session(self): if self._session is None: self._session aiohttp.ClientSession() return self._session async def fetch_market_data(self, query: MarketDataQuery) - MarketDataResponse: 核心技能方法获取市场数据 # 1. 构建请求参数这里只是一个示例实际会更复杂 params { industry: query.industry, timeframe: query.timeframe, metrics: ,.join(query.metrics) } headers {} if self.api_key: headers[Authorization] fBearer {self.api_key} try: session await self._get_session() async with session.get( f{self.api_base_url}/market_data, paramsparams, headersheaders, timeoutaiohttp.ClientTimeout(total30) ) as response: if response.status 200: json_data await response.json() return MarketDataResponse(successTrue, datajson_data) else: error_text await response.text() return MarketDataResponse( successFalse, error_messagefAPI请求失败状态码{response.status}, 详情{error_text[:200]} ) except asyncio.TimeoutError: return MarketDataResponse(successFalse, error_message请求外部API超时) except Exception as e: return MarketDataResponse(successFalse, error_messagef未知错误{str(e)}) # 注意实际生产环境需要更精细的异常分类和处理 async def close(self): 清理资源 if self._session: await self._session.close() # 可以同时将其包装成一个标准的LangChain Tool供LLM直接调用 from langchain.tools import tool tool(args_schemaMarketDataQuery) async def fetch_market_data_tool(industry: str, timeframe: str, metrics: List[str] None): 获取指定行业和时间的市场数据。 skill DataFetcherSkill(api_base_urlhttps://api.example.com) # 实例化实际应从配置或上下文获取 query MarketDataQuery(industryindustry, timeframetimeframe, metricsmetrics or []) result await skill.fetch_market_data(query) if not result.success: return f获取数据失败{result.error_message} return f数据获取成功。摘要{str(result.data)[:500]}... # 返回给LLM的摘要这个实现展示了几个关键点使用Pydantic定义接口MarketDataQuery和MarketDataResponse明确了技能的“契约”便于使用方理解和验证。技能类封装DataFetcherSkill类封装了所有细节API地址、认证、请求逻辑、错误处理。它易于独立测试。可选的Tool包装通过tool装饰器我们可以轻松地将这个技能的核心方法暴露给LLM作为一个Tool。这体现了Skill的灵活性。3.3 第三步在LangGraph节点中集成Skill现在我们来到LangGraph的工作流定义中。关键是如何将这些技能实例优雅地注入到需要它们的节点函数里。我强烈推荐使用依赖注入的模式而不是在节点函数内部硬编码创建Skill实例。# graph.py from typing import Annotated, TypedDict from langgraph.graph import StateGraph, END import operator from skills.data_fetcher import DataFetcherSkill, MarketDataQuery from skills.competitor_analysis import CompetitorAnalysisSkill # ... 导入其他技能 # 1. 定义Graph的状态结构 class AgentState(TypedDict): 智能体的工作状态 industry: str timeframe: str required_metrics: list raw_market_data: dict None # 由fetch_node填充 analysis_result: dict None # 由analyze_node填充 report_text: str None # 由report_node填充 chart_paths: list None # 由chart_node填充 errors: list None # 用于收集各环节错误 # 2. 创建技能实例通常从配置或工厂类中获取 # 在实际项目中这些实例的创建和管理可以通过像fastapi.Depends或injector这样的依赖注入容器来优化。 data_fetcher DataFetcherSkill(api_base_urlos.getenv(MARKET_API_URL), api_keyos.getenv(API_KEY)) analyst CompetitorAnalysisSkill() # ... 初始化其他技能 # 3. 定义节点函数并将技能实例作为“上下文”或参数传入 async def fetch_data_node(state: AgentState): 节点获取市场数据 print(f[Fetch Node] 正在获取 {state[industry]} 行业的数据...) query MarketDataQuery( industrystate[industry], timeframestate[timeframe], metricsstate[required_metrics] ) # 调用Skill response await data_fetcher.fetch_market_data(query) new_state {raw_market_data: None, errors: state.get(errors, [])} if response.success: new_state[raw_market_data] response.data print([Fetch Node] 数据获取成功。) else: error_msg f数据获取失败{response.error_message} print(f[Fetch Node] {error_msg}) new_state[errors] state.get(errors, []) [error_msg] return new_state async def analyze_data_node(state: AgentState): 节点分析竞品数据 if state.get(raw_market_data) is None: error 分析节点上游数据为空无法进行分析。 return {errors: state.get(errors, []) [error]} print([Analysis Node] 开始竞品分析...) # 调用另一个Skill analysis analyst.analyze(state[raw_market_data]) return {analysis_result: analysis} # 4. 构建图 builder StateGraph(AgentState) builder.add_node(fetch, fetch_data_node) builder.add_node(analyze, analyze_data_node) # ... 添加 report_node, chart_node 等 # 5. 设置边条件边或固定边 builder.set_entry_point(fetch) builder.add_edge(fetch, analyze) # 假设获取数据后总是进行分析 # ... 设置更多边 graph builder.compile()注意上面的代码中技能实例data_fetcher,analyst是在全局作用域创建的。对于简单应用可行但对于复杂应用或需要动态配置的场景更好的模式是使用闭包或类来将技能绑定到图上。例如可以定义一个GraphBuilder类在__init__中初始化所有技能然后节点方法定义为实例方法这样它们就能通过self访问这些技能实例。或者使用更高级的依赖注入框架。3.4 第四步进阶集成模式——将Skill作为Tool提供给LLM节点在很多工作流中我们有一个专门的节点是“LLM思考与决策”节点它需要根据当前状态决定下一步做什么并可能调用各种Tools。这时我们需要将Skills包装的Tools提供给这个节点。from langchain_openai import ChatOpenAI from langgraph.prebuilt import ToolExecutor, ToolInvocation from langgraph.graph import StateGraph, MessagesState from typing import TypedDict, Annotated, Sequence from langchain_core.messages import BaseMessage, HumanMessage, ToolMessage import operator # 假设我们已经有了包装好的Tools列表 tools [fetch_market_data_tool, another_skill_tool, ...] # 这些tool来自我们的Skills tool_executor ToolExecutor(tools) llm ChatOpenAI(modelgpt-4-turbo-preview) llm_with_tools llm.bind_tools(tools) # 关键将tools绑定给LLM它才能知道有哪些工具可用 class AgentState(TypedDict): messages: Annotated[Sequence[BaseMessage], operator.add] # 消息序列 def llm_agent_node(state: AgentState): LLM决策节点思考并可能调用工具 print([LLM Node] 正在思考...) last_message state[messages][-1] # 1. 调用LLM它会根据对话历史和绑定的tools决定是回复还是调用工具 response llm_with_tools.invoke(state[messages]) # 2. 将LLM的响应追加到消息历史 new_messages [response] # 3. 检查LLM是否想调用工具 if response.tool_calls: print(f[LLM Node] 决定调用工具{[tc[name] for tc in response.tool_calls]}) # 对于每个工具调用执行它 for tool_call in response.tool_calls: # 使用ToolExecutor执行工具调用 result tool_executor.invoke(tool_call) # 将工具执行结果作为ToolMessage追加 new_messages.append(ToolMessage(contentstr(result), tool_call_idtool_call[id])) # 工具调用后通常需要让LLM再次思考所以这个节点之后应该再次连接到自己或一个处理节点 return {messages: new_messages} # 如果没有工具调用LLM已经给出了最终回答可以流向END return {messages: new_messages} # 构建图 builder StateGraph(AgentState) builder.add_node(agent, llm_agent_node) builder.set_entry_point(agent) # 这里需要更复杂的边逻辑来处理“是否调用了工具”的条件判断通常使用langgraph.graph.StateGraph.add_conditional_edges # 这是一个简化示例实际需要条件边来循环调用。这种模式下Skills通过Tool接口成为了LLM可以自主调用的“手”和“脚”极大地扩展了智能体的能力边界。而整个工作流的编排和控制仍然由LangGraph的图结构牢牢掌握。4. 架构思考如何设计可维护、可扩展的Skill体系当Skills数量增多后管理它们就成了一个挑战。以下是我从几个项目中总结出的架构经验1. 技能分类与分层基础技能与具体业务无关的通用能力如HTTPClientSkill、FileIOSkill、CalculationsSkill。这些可以作为所有项目的底层依赖。领域技能与特定业务领域相关如MarketDataSkill、CustomerServiceSkill、CodeReviewSkill。它们建立在基础技能之上。组合技能由多个更细粒度技能组合而成完成一个更复杂的子任务。例如GenerateQuarterlyReportSkill内部可能依次调用FetchDataSkill、AnalyzeSkill、WriteSummarySkill。2. 统一的技能注册与管理中心创建一个SkillRegistry技能注册表单例或类。所有技能在应用启动时向注册表注册自己并提供一个唯一的名称和版本。节点或LLM可以通过名称从注册表中获取技能实例。这带来了以下好处集中配置技能的初始化参数如API密钥、端点可以在注册中心统一管理。动态加载可以根据配置或环境动态启用或禁用某些技能。依赖注入注册中心可以处理技能之间的依赖关系例如ChartGeneratorSkill依赖于DataCacheSkill。# skill_registry.py (简化示例) class SkillRegistry: _instance None _skills {} def __new__(cls): if cls._instance is None: cls._instance super().__new__(cls) return cls._instance def register(self, name: str, skill_factory): 注册一个技能工厂函数 self._skills[name] skill_factory def get(self, name: str, **kwargs): 获取一个技能实例可以传入初始化参数 if name not in self._skills: raise KeyError(fSkill {name} not registered.) return self._skills[name](**kwargs) # 在应用初始化时注册技能 registry SkillRegistry() registry.register(data_fetcher, lambda: DataFetcherSkill(api_base_urlos.getenv(API_URL))) registry.register(competitor_analyst, lambda: CompetitorAnalysisSkill()) # 在节点中使用 data_fetcher registry.get(data_fetcher)3. 技能版本化与兼容性当技能逻辑更新时比如API接口变了为了避免破坏已有的工作流可以考虑引入版本号。节点在调用技能时指定所需版本注册中心返回对应版本的实例。这在大规模协作中尤为重要。4. 技能的可观测性为每个技能添加详细的日志记录、执行时间监控和错误统计。这不仅能帮助调试还能让你了解智能体工作流中各个技能的“健康度”和性能瓶颈。可以在技能基类或通过装饰器模式统一实现。5. 避坑指南Skills开发与集成中的常见陷阱在实际项目中踩过不少坑这里分享几个最典型的陷阱一技能与状态State的过度耦合症状Skill函数直接读取或修改LangGraph的整个State对象。 后果技能变得不可复用且测试困难。State结构的任何变动都会导致技能崩溃。 正确做法Skill只应通过明确的输入参数接收它需要的数据并通过返回值输出结果。状态管理是LangGraph节点的职责。节点作为“协调者”从State中提取数据传给Skill再将Skill的结果写回State。陷阱二忽视异步Async与同步Sync的混用症状在异步的LangGraph节点中调用了一个执行阻塞I/O操作的同步Skill函数。 后果整个事件循环被阻塞严重降低智能体的并发性能在高负载下可能导致系统瘫痪。 正确做法统一使用异步async/await。如果Skill内部涉及网络请求、文件读写、数据库查询等I/O操作务必使用异步库如aiohttp,aiomysql,aiofiles。如果必须使用同步库请使用asyncio.to_thread将其放到线程池中执行避免阻塞主事件循环。陷阱三脆弱的错误处理症状Skill内部只是简单打印错误或返回None节点函数没有检查Skill的失败情况。 后果工作流在静默中失败状态数据出现不一致难以定位问题根源。 正确做法Skill应定义明确的错误返回类型如前文的MarketDataResponse包含success和error_message。节点函数必须检查Skill的执行结果并根据业务逻辑决定下一步是重试、记录错误并继续、还是跳转到专门的“错误处理节点”一个健壮的工作流必须有错误处理路径。陷阱四技能粒度过粗或过细症状一个MegaSkill做了十件事或者十个TinySkill每个只做一行简单的字符串处理。 后果过粗的技能难以复用和测试过细的技能导致节点代码变成繁琐的“胶水代码”管理成本激增。 正确做法遵循“单一职责”和“高内聚”原则。一个Skill应该对应一个有意义的、独立的业务功能单元。例如“发送邮件”是一个好的Skill“构建邮件标题”可能就太细了。如果发现多个节点总是以固定顺序调用一组细粒度技能那么就应该考虑将它们组合成一个新的、更粗粒度的“组合技能”。陷阱五硬编码的配置和依赖症状Skill的API地址、密钥等直接写在类定义中。 后果无法适应不同环境开发、测试、生产也无法安全地管理密钥。 正确做法通过__init__方法接收配置或从环境变量、配置中心读取。使用依赖注入模式来管理Skill所依赖的其他服务如数据库连接池、缓存客户端。在LangGraph的生态中集成Skills本质上是一场关于如何构建模块化、可维护AI应用的工程实践。它要求我们不仅关注单个智能体的“智力”更要关注整个系统架构的“整洁度”。当你把那些杂乱无章的功能点规整成一个个边界清晰、接口明确的Skills时你会发现构建复杂智能体工作流从此变得像搭积木一样清晰而愉快。你的代码库将从一个“脚本集合”进化成一个真正的“技能库”而这正是AI应用工程化道路上至关重要的一步。