如果你正在做基于大模型的智能体应用LangGraph 大概率已经在你的候选清单里出现了。这个框架把智能体的运行流程建模成一张状态图节点是业务动作边是流转条件一个可序列化的状态对象是节点之间的公共数据通道。结构很直观但真正用起来有一道绕不开的坎——当智能体需要执行真实任务时比如查库、改文件、调用外部API这些能力怎么干净地接进图里直接在节点里写一堆函数调用图就会越来越脏代码很快失控。Skills技能就是为这个场景设计的工程惯例。它的思路很朴素把一组可以被大模型调用、同时又能被图节点稳定执行的动作封装成带元信息的可复用单元。每一个技能都有清晰的输入输出接口、描述、验证规则智能体可以根据任务描述动态地选择技能图节点则负责把技能稳定地执行完。相当于给智能体装了一排“工具抽屉”每个抽屉上贴了标签大模型负责挑图负责拉开抽屉干活。这篇文章我想把 LangGraph 中集成 Skills 的完整思路拆开讲一遍。从底层状态图模型说起讲清楚为什么要做技能层、如何设计技能接口、如何把技能注册进图节点、如何让大模型自主选技能以及我在实际项目中踩过的坑和排查方法。适合两类人一是刚接触 LangGraph、想把工具调用做得更规范的开发者二是已经用 LangGraph 写过几个 Agent、但感觉代码越来越难维护、想重构出一套技能体系的人。1. LangGraph 与 Skills 的底层关系先读懂状态图再谈技能1.1 LangGraph 的核心模型LangGraph 本质上是一个基于图结构的状态机。你定义一张图节点是函数边是节点之间的流转规则整张图共享一个状态对象State。每次运行Agent 按图走一遍节点函数读取状态、执行逻辑、把结果写回状态然后根据状态中的某个字段决定走哪条边。这个模型最大的好处是可控。普通 ReAct 风格的 Agent 把“思考-行动-观察”循环写在一个大循环里行为对开发者来说是个黑盒而 LangGraph 把每个动作拆到独立节点里你可以在任意节点之间插入人工审核、条件判断、重试逻辑。你还可以随时把图状态打印出来看调试体验比纯循环式 Agent 好太多了。但图模型也有代价节点和边的数量一旦多起来图的复杂度会迅速失控。如果你的每个业务能力都直接写成一个节点比如“查订单”“查库存”“改价格”“发消息”图会变成一张巨大的意大利面。这也是为什么需要用 Skills 做一层抽象——把图节点的粒度从“业务动作”提升到“技能执行器”让图的拓扑保持稳定具体能力以技能为单位挂载上去。1.2 Skills 到底是个什么东西Skills 不是一个严格的学术概念更像是一种模块化惯例。一个 Skill 至少包含三样东西技能描述、输入输出接口、执行逻辑。描述是给大模型看的接口是给图节点校验用的执行逻辑是真正干活的部分。你可以把 Skill 理解成“带说明书的函数”。普通函数只有入参和返回值而 Skill 还额外带了一段自然语言说明告诉大模型“这个技能能解决什么问题、什么情况下该用我”。这种设计是 Agent 工具调用的标准套路但在 LangGraph 里技能被提升到了图级别技能不只是给大模型选的还能被图节点主动调度、编排、重试。举一个生活类比。你请了一个助理大模型助理手里有一本工具手册技能库每页写着一种工具的功能和适用场景。助理接到任务后翻手册决定用哪个工具、按什么顺序用。但真正把工具从仓库里拿出来、确认没坏、用完放回去的是你定好的流程LangGraph 图。如果工具坏了流程会决定是换一把还是让助理重新描述任务。1.3 为什么要在 LangGraph 里单独做一层技能抽象很多人一开始直接让大模型调普通函数也就是 Function Calling。在小型 Demo 里这完全够用任务多了以后问题就暴露了没有统一的生命周期管理。函数调用完就结束了没有超时、重试、日志的统一处理。没有输入输出的强制性校验。大模型可能传错参数类型函数直接抛异常图崩溃。无法编排。比如“先查库存再生成报价单最后发邮件”这个流程用函数调用只能靠大模型自己在一次对话里想办法很难拆成可控制的独立步骤。无法复用和共享。两个 Agent 都要用同一个查库存能力只能复制粘贴。Skills 层把这些问题统一解决掉。每个技能都有 schema 校验、幂等设计、可观测性并且可以跨 Agent 共享。技能的粒度可以自由控制小到“查询天气”大到“生成月度报表”。在 LangGraph 里集成技能之后图就变成了一个稳定的“执行骨架”业务逻辑的扩展不再需要改图只需要新增技能。2. 技能拆解与声明式设计好的技能不是代码堆出来的2.1 识别可落地的技能边界我见过不少项目技能拆得乱七八糟要么一个技能塞了太多功能要么一个简单动作拆成五六个技能大模型选起来晕头转向。技能拆分的判断标准其实很简单一个技能应该对应一个完整的、可验收的任务闭环。什么叫“可验收”就是执行完这个技能后你能明确地说“这件事做完了”。比如“生成一份销售周报”做完后输出一份完整的报告文件这是可验收的。但“计算销售额总和”就不是一个合格的技能它只是报告生成过程中的一个步骤应该作为内部逻辑藏在技能实现里而不是暴露给大模型。另一个判断标准是失败边界。如果技能执行到一半失败用户能明确知道是什么任务失败了而不是一个语义模糊的中间状态。选“上传文件到服务器”当技能失败时可以告诉用户“上传失败”但如果把“解析文件、提取关键字段、上报系统”拆成三个技能失败时用户看到的是“提取字段失败”反而难以理解整体目标进行到哪一步。2.2 用类配置定义技能接口技能接口的核心是对输入输出做形式化描述。这里我强烈建议用 JSON Schema 或类配置结构而不是散乱的参数列表。因为大模型对结构化的描述理解更准确且 schema 可以直接复用于校验逻辑。一个常规的技能定义长这样# 技能定义使用 dataclass 或者字典结构 SKILL_REVIEW { name: code_review, description: 对一段源代码进行静态审查返回问题清单和修改建议。适用于代码合并前的质量检查。, input_schema: { type: object, properties: { code: {type: string, description: 待审查的完整代码}, language: {type: string, enum: [python, javascript, java], description: 代码语言} }, required: [code] }, output_schema: { type: object, properties: { issues: {type: array, items: {type: string}}, suggestions: {type: array, items: {type: string}} } }, timeout_seconds: 30, retry_policy: {max_retries: 2, backoff_seconds: 1} }注意看几个细节。description 字段写的是“什么时候该用我”而不是“我能做什么”这决定了模型选错的概率。input_schema 里每个字段都带 description是为了让大模型填参数时更加准确。更关键的是 timeout_seconds 和 retry_policy这两个字段是 LangGraph 图节点执行时用的体现了技能和图框架的配合。2.3 技能粒度一组动作还是一条完整流程技能粒度没有标准答案取决于你的图结构。如果技能只做单个动作比如查天气、发消息那么 LangGraph 的图会更像“动态路由表”大模型每轮选一个技能执行这是最简单的形态。如果技能本身包含多个步骤比如“处理客户投诉”包含拉取工单、查询历史订单、生成回复草稿三个动作你可以把它作为一个组合技能封装起来。组合技能内部自己管理子步骤对外只暴露一个入口和出口。这样做的好处是图里只有一个节点坏处是内部流程对 LangGraph 变得不可见无法利用图的中间检查点能力。我个人的经验是优先让技能保持“单目标多步骤”的粒度即一个技能内部可以有多个操作步骤但必须围绕一个目标。真正需要跨技能协作的流程应该留在 LangGraph 的图层面编排而不是硬塞进一个技能里。这样兼顾了图的可控性和技能的复用性。2.4 把复杂技能拆成可组合的子步骤有些技能确实很大拆成一个整体太笨重拆成多个又太碎。这时候我建议在技能内部实现子步骤调度但不要让子步骤暴露给大模型。比如某个“数据迁移”技能内部包含连接源库、抽取数据、清洗转换、写入目标库、校验一致性五个步骤。对外它是一个技能对内它有一套自己的执行管线。你可以用 LangGraph 的子图来实现这种技能也可以用普通 Python 函数编排。用子图的好处是可以在子步骤之间添加断点、日志、人工确认坏处是实现成本高。小团队项目就别折腾了普通函数编排足够了。拆子步骤的时候有个原则子步骤之间用明确的数据契约连接。前一步的输出是后一步的输入每步都有独立的异常处理。这样即使某个子步骤失败也能定位到具体环节并触发重试。3. 实操在 LangGraph 中集成 Skills 的完整步骤3.1 先搭一个最小的 LangGraph 骨架在加技能之前先确保你的最小图能跑通。下面的代码定义了一个最简单的状态图只有一个节点执行后进入结束状态from typing import TypedDict from langgraph.graph import StateGraph, END class AgentState(TypedDict): messages: list current_skill: str skill_input: dict skill_output: dict def noop_node(state: AgentState) - dict: return {messages: state[messages] [节点已执行]} graph StateGraph(AgentState) graph.add_node(entry, noop_node) graph.set_entry_point(entry) graph.add_edge(entry, END) app graph.compile() result app.invoke({messages: [], current_skill: , skill_input: {}, skill_output: {}})这个代码有两层含义需要理解。一是 State 里增加了 current_skill、skill_input、skill_output 三个字段这是为技能预留的通道。二是整个图只有一个节点你可以在这个节点的位置换成真正的技能执行器。3.2 实现一个通用的技能执行器节点技能执行器是整个架构的核心节点。它的职责是从状态中读取技能名称去注册表找到技能定义校验输入参数执行技能把结果写回状态。import time import logging SKILL_REGISTRY { code_review: SKILL_REVIEW, # ... 更多技能 } def find_skill(name: str): skill_def SKILL_REGISTRY.get(name) if not skill_def: raise ValueError(f技能 {name} 未注册) return skill_def def execute_skill_node(state: AgentState) - dict: skill_name state.get(current_skill) skill_input state.get(skill_input, {}) if not skill_name: return {skill_output: {error: 未指定技能名称}} skill_def find_skill(skill_name) # 输入校验 errors validate_schema(skill_input, skill_def[input_schema]) if errors: return {skill_output: {error: f参数校验失败: {errors}}} # 执行技能带超时 start time.time() try: result run_skill_impl(skill_def, skill_input) except Exception as e: result {error: str(e)} elapsed time.time() - start logging.info(f技能 {skill_name} 执行耗时 {elapsed:.2f}s) return {skill_output: result}这里有三个容易被忽略的点。第一validate_schema 这一步很关键它避免了大模型乱传参数导致技能内部崩溃。第二run_skill_impl 根据技能定义里的 type 或 name 分发到具体的实现函数这个分发机制可以用装饰器实现。第三返回的字典里包含 error 字段这为后面图里做条件分支提供了信号。3.3 让大模型自主选择技能动态路由的实现前面那个图是静态的真正的智能体需要让大模型根据用户任务动态选择技能。这个场景的标准做法是在图里加一个“规划节点”由大模型分析当前状态决策下一步要执行哪个技能。def planner_node(state: AgentState) - dict: user_request state[messages][-1][content] if state[messages] else # 把所有技能的描述构建成提示 skill_descriptions \n.join([ f- {name}: {detail[description]} for name, detail in SKILL_REGISTRY.items() ]) prompt f根据用户请求,从以下技能中选择一个最合适的执行。 如果所有技能都不适合,回复 no_skill。 技能列表: {skill_descriptions} 用户请求: {user_request} 只输出技能名称,不要额外解释。 skill_name call_llm(prompt) # 你的 LLM 调用封装 return {current_skill: skill_name, skill_input: extract_input_from_request(user_request, skill_name)}在这个思路下整体图结构变成入口节点 - 规划节点 - 技能执行器 - 判断结果 - 回到规划节点或结束。这就是标准的 Agent 循环只是“动作空间”变成了技能注册表。有一个关键细节你可以用上技能描述中如果包含“适用于什么情况、不要用于什么情况”大模型选错的概率会显著降低。3.4 利用条件边处理技能执行结果技能执行不保证成功所以图里必须有分支逻辑。LangGraph 的条件边非常适合做这件事。判断逻辑通常看 skill_output 里是否有 error 字段。def after_skill_router(state: AgentState) - str: output state.get(skill_output, {}) if output.get(error): return retry return next graph.add_conditional_edges( skill_executor, after_skill_router, { retry: planner, next: END } )这个分支的价值在于技能失败后可以自动回到规划节点让大模型重新换一个技能或者调整参数。如果大模型连续失败几次你还需要一个计数器字段超过上限直接进入人工处理节点。这种控制流在纯函数调用式的 Agent 里很难优雅实现在图模型里却非常顺手。3.5 状态上下文与技能参数的传递技能执行过程中往往需要访问上下文信息比如用户ID、会话ID、历史消息、外部系统凭证。这些数据不能每次靠大模型传应该由执行器节点从状态中自动注入。我一般会在状态里放一个 context 字段里面是所有技能需要的基础上下文。execute_skill_node 在调用技能实现之前把 context 中的字段合并进技能输入def execute_skill_node(state: AgentState) - dict: skill_name state.get(current_skill) skill_input dict(state.get(skill_input, {})) context state.get(context, {}) # 注入上下文,但技能显式传入的参数优先 for k, v in context.items(): skill_input.setdefault(k, v) # ... 后续执行逻辑用 setdefault 而不是直接覆盖是为了避免上下文里的默认值覆盖掉大模型根据用户请求提取出来的参数。这个坑我踩过之前直接把 context 和 skill_input 做并集合并结果大模型提取到的“当前用户ID”老是被上下文里的默认值覆盖导致技能屡屡操作到错误对象。3.6 流式输出与进度回调LangGraph 本身支持流式输出对于耗时技能来说非常有用。比如数据清洗技能可能要跑几十秒如果用户看不到进度体验会很差。实现方式有两种。一种是用 LangGraph 原生的事件流机制async for event in app.astream_events(state, versionv2): if event[event] on_chain_stream: # 透传技能实现中自定的进度事件 chunk event.get(data, {}).get(chunk) if chunk and chunk.get(type) progress: yield f进度: {chunk[message]}另一种更简单的方式是在技能实现里通过生成器逐段返回结果图节点把生成器产生的每个块即时写入状态的子字段中。如果你用的是流式 LLM 界面这个方法对调试也友好得多。4. 常见问题与排查技巧实录4.1 技能重复执行的幂等性LangGraph 支持对节点进行重试图模型本身允许同一节点被多次执行。问题在于如果你的技能不是幂等的比如“发短信”“扣余额”“创建订单”重试一次就重复执行一次轻则系统报错重则产生脏数据。解决思路是让技能实现里带一个操作记录在执行前生成一个唯一的 execution_id用外部存储记录执行状态。如果重复执行时发现同样的 execution_id 已经成功完成直接返回上次结果。EXECUTION_CACHE {} def run_skill_impl(skill_def, skill_input): exec_id skill_input.get(execution_id) or str(uuid.uuid4()) if exec_id in EXECUTION_CACHE: return EXECUTION_CACHE[exec_id] # 真正的技能逻辑 result do_skill_work(skill_input) EXECUTION_CACHE[exec_id] result return result如果你用的是 Kafka 或 Redis 这类中间件也可以把 EXECUTION_CACHE 换成 Redis 存储天然支持分布式环境的幂等。4.2 技能响应超时与重试策略外部 API 调用是最容易超时的环节。在技能定义里加 timeout_seconds 字段只是第一步关键还在于执行节点里真的用超时控制包住调用。from concurrent.futures import ThreadPoolExecutor, TimeoutError def execute_with_timeout(skill_def, skill_input): io_pool ThreadPoolExecutor(max_workers1) future io_pool.submit(run_skill_impl, skill_def, skill_input) try: return future.result(timeoutskill_def.get(timeout_seconds, 30)) except TimeoutError: future.cancel() return {error: 技能执行超时}超时之后的处理也要想清楚。如果技能是幂等安全的直接重试如果不是超时只能把控制权交还给规划节点让大模型决定改参数还是换技能。不要盲目套用固定重试策略我在生产环境里就因为固定重试三次把外部系统的限流打爆过。4.3 状态超限与序列化失败LangGraph 要求状态是 “可序列化”的。技能执行过程中如果往状态里塞了 Python 对象、文件句柄、连接池对象invoke 的时候就会发现序列化报错非常隐蔽。我遇到过一个场景技能里加载了一个大模型 tokenizer 对象顺手放进了状态里图跑完保存检查点时直接报错。排查了很久才发现是状态里多了一个不可序列化的成员。后来我把状态字段时间设置得非常明确只允许 JSON 基础类型、列表、字典。任何复杂对象要么转成 JSON 字符串要么放进外部存储。另外状态体积也要控制。技能执行结果如果太大比如拉取全量数据状态会被撑爆。我会在技能输出到状态之前做摘要化处理只保留关键统计量完整数据放进文件存储状态里只存引用路径。4.4 并行技能的状态写冲突LangGraph 支持并行节点执行比如同时调用多个技能但如果多个技能都往同一个状态字段里写就会有写冲突。默认的 reducer 策略是直接覆盖后写的覆盖先写的结果可能丢失数据。解决办法是在状态字段定义时用自定义 reducer。最常见的是把同名消息做追加合并这个 LangGraph 官方已经有内置支持。对于技能结果字段我会让每个技能把结果写入独立的键class AgentState(TypedDict): messages: Annotated[list, operator.add] skill_results: dict # 技能名称作为 key并行节点里各自写入 skill_results[skill_a] 和 skill_results[skill_b]就避免了互相覆盖。如果你需要一个总结果字段做后续判断那就用一个专门的汇总节点在所有并行技能跑完之后执行合并逻辑。4.5 条件边太多导致图难以追踪图分支太多调试时会乱。我的经验是不要把所有判断逻辑都放在条件边里。Locate 判断的复杂度应该拆开简单的成功/失败分支放条件边复杂的路由决策放一个专门的路由节点里。比如一个图有十几个技能如果每个技能后面都接一串条件分支图面会非常难看。我会压缩成所有技能节点都汇聚到一个统一出口节点由出口节点根据技能执行结果统一决定下一步去向。这相当于用中心化的路由代替了放射状的条件边逻辑清晰维护也简单。常见问题典型表现排查方向推荐做法技能重复执行重试后生成了重复订单/重复扣款检查技能实现是否具备幂等保护增加 execution_id 去重外部接口超时图运行到技能节点卡住很久检查 timeout 参数是否真正生效用线程池包裹调用设置硬超时状态序列化失败invoke 阶段直接报 pickle/JSON 错误检查状态里是否有非序列化对象严格控制状态字段类型并行技能写冲突多个技能结果互相覆盖检查状态字段是否使用合并 reducer按技能名分键写入图分支爆炸添加技能时图变得难以维护观察条件边数量是否过多用统一出口节点做中心化路由5. 从单技能到技能编排给智能体一张技能地图5.1 技能的层次化组织技能数量几十个之后就不能平铺在一张注册表里了。大模型面对几十个候选技能时选择准确率会明显下降。我早期遇到的情况是一次规划请求里大模型经常选错技能尤其是描述相似的技能。解决思路是分层组织。按照领域把技能分组每组设置一个“领域路由技能”大模型先选领域再选具体技能。这个方式相当于两级检索大幅降低单层选择的压力。SKILL_GROUPS { 数据查询: { description: 用于查询各类业务数据, skills: [order_query, user_query, product_query] }, 数据分析: { description: 用于对已有数据进行处理和统计, skills: [sales_summary, trend_analysis] }, 内容生成: { description: 用于生成报告、邮件、文案等内容, skills: [report_generate, email_draft] } }规划节点先从组描述中选组再从组内选技能。这个方法实测下来技能选择准确率比平铺式高了很多。代价是多了一次 LLM 调用但对于准确性要求高的场景完全值得。5.2 让多个技能协作的编排模式有些任务无法靠单个技能完成需要多个技能按顺序配合。比如“根据最近的销售数据给重点客户写一封邮件”先调用数据查询技能拿销售数据再调用数据分析技能算增长率和重点客户最后调用内容生成技能写邮件。这种流程型任务有两种做法。第一种是图层面显式编排三个技能节点按顺序连起来状态依次传递。好处是流程清晰、每一步都可监控坏处是每增加一个流程就要改图结构。第二种是在规划节点中用“行动计划”指定一串技能技能执行器按顺序执行。这个模式下图本身保持通用具体流程由大模型动态规划。这更像真正的 Agent 行为但执行过程的可控性会差一些。我目前的建议是核心业务流程用第一种边缘的、低概率的流程用第二种。核心流程显式化能保证稳定性边缘流程交给 Agent 自由发挥能保留灵活性。5.3 人机协作技能失败时把控制权还给用户技能执行到不可恢复的错误时不要强制重试也不要让大模型无限思考。最合理的做法是进入一个人工处理节点把错误信息和上下文抛给用户让用户决定下一步。在 LangGraph 里实现这个很自然技能结果 error 字段为真且重试次数超限时条件边直接指向一个 “human_handoff” 节点。这个节点可以暂停图的执行等用户在界面上输入后再把用户的指令写回状态继续走图。这里有一个细节不要把原始错误堆栈直接展示给用户。我试过直接透传底层异常用户看到数据库连接串、内部 API 地址既体验差又不安全。应该先做错误脱敏把技术错误转成业务描述比如“订单查询服务暂时不可用请稍后重试”。写在最后的实操体会我在实际项目中集成这些技能模块时最深的体会是技能设计占了大半工作量代码实现反而简单。很多时候你以为难点在于大模型怎么选对技能真正调试之后发现问题出在技能描述含糊、输入输出定义不严谨、失败后没有兜底路径。另外一点很实在的建议技能注册表一定要做可视化。我在本地调试时写了一个简单的 HTML 页面把每个技能的描述、参数、最近执行时间、失败率都列出来肉眼能看到的技能分布比任何文档都直观。哪个技能经常被选错哪个技能失败率奇高看一眼就知道该优化哪块。最后再分享一个小技巧每新增一个技能前强制自己写出这个技能的三个边界条件——什么场景下不该用它、输入什么异常值会失败、失败后用户应该怎么处理。写不出这三个条件的技能说明你对它的理解还不够先别急着写代码。这个习惯帮我挡掉了很多后续的返工。 
长河君 · 2026/10/21 6:50:38 · 阅读 1.2k LangGraph集成Skills：智能体技能封装与状态图编排实践
如果你正在做基于大模型的智能体应用LangGraph 大概率已经在你的候选清单里出现了。这个框架把智能体的运行流程建模成一张状态图节点是业务动作边是流转条件一个可序列化的状态对象是节点之间的公共数据通道。结构很直观但真正用起来有一道绕不开的坎——当智能体需要执行真实任务时比如查库、改文件、调用外部API这些能力怎么干净地接进图里直接在节点里写一堆函数调用图就会越来越脏代码很快失控。Skills技能就是为这个场景设计的工程惯例。它的思路很朴素把一组可以被大模型调用、同时又能被图节点稳定执行的动作封装成带元信息的可复用单元。每一个技能都有清晰的输入输出接口、描述、验证规则智能体可以根据任务描述动态地选择技能图节点则负责把技能稳定地执行完。相当于给智能体装了一排“工具抽屉”每个抽屉上贴了标签大模型负责挑图负责拉开抽屉干活。这篇文章我想把 LangGraph 中集成 Skills 的完整思路拆开讲一遍。从底层状态图模型说起讲清楚为什么要做技能层、如何设计技能接口、如何把技能注册进图节点、如何让大模型自主选技能以及我在实际项目中踩过的坑和排查方法。适合两类人一是刚接触 LangGraph、想把工具调用做得更规范的开发者二是已经用 LangGraph 写过几个 Agent、但感觉代码越来越难维护、想重构出一套技能体系的人。1. LangGraph 与 Skills 的底层关系先读懂状态图再谈技能1.1 LangGraph 的核心模型LangGraph 本质上是一个基于图结构的状态机。你定义一张图节点是函数边是节点之间的流转规则整张图共享一个状态对象State。每次运行Agent 按图走一遍节点函数读取状态、执行逻辑、把结果写回状态然后根据状态中的某个字段决定走哪条边。这个模型最大的好处是可控。普通 ReAct 风格的 Agent 把“思考-行动-观察”循环写在一个大循环里行为对开发者来说是个黑盒而 LangGraph 把每个动作拆到独立节点里你可以在任意节点之间插入人工审核、条件判断、重试逻辑。你还可以随时把图状态打印出来看调试体验比纯循环式 Agent 好太多了。但图模型也有代价节点和边的数量一旦多起来图的复杂度会迅速失控。如果你的每个业务能力都直接写成一个节点比如“查订单”“查库存”“改价格”“发消息”图会变成一张巨大的意大利面。这也是为什么需要用 Skills 做一层抽象——把图节点的粒度从“业务动作”提升到“技能执行器”让图的拓扑保持稳定具体能力以技能为单位挂载上去。1.2 Skills 到底是个什么东西Skills 不是一个严格的学术概念更像是一种模块化惯例。一个 Skill 至少包含三样东西技能描述、输入输出接口、执行逻辑。描述是给大模型看的接口是给图节点校验用的执行逻辑是真正干活的部分。你可以把 Skill 理解成“带说明书的函数”。普通函数只有入参和返回值而 Skill 还额外带了一段自然语言说明告诉大模型“这个技能能解决什么问题、什么情况下该用我”。这种设计是 Agent 工具调用的标准套路但在 LangGraph 里技能被提升到了图级别技能不只是给大模型选的还能被图节点主动调度、编排、重试。举一个生活类比。你请了一个助理大模型助理手里有一本工具手册技能库每页写着一种工具的功能和适用场景。助理接到任务后翻手册决定用哪个工具、按什么顺序用。但真正把工具从仓库里拿出来、确认没坏、用完放回去的是你定好的流程LangGraph 图。如果工具坏了流程会决定是换一把还是让助理重新描述任务。1.3 为什么要在 LangGraph 里单独做一层技能抽象很多人一开始直接让大模型调普通函数也就是 Function Calling。在小型 Demo 里这完全够用任务多了以后问题就暴露了没有统一的生命周期管理。函数调用完就结束了没有超时、重试、日志的统一处理。没有输入输出的强制性校验。大模型可能传错参数类型函数直接抛异常图崩溃。无法编排。比如“先查库存再生成报价单最后发邮件”这个流程用函数调用只能靠大模型自己在一次对话里想办法很难拆成可控制的独立步骤。无法复用和共享。两个 Agent 都要用同一个查库存能力只能复制粘贴。Skills 层把这些问题统一解决掉。每个技能都有 schema 校验、幂等设计、可观测性并且可以跨 Agent 共享。技能的粒度可以自由控制小到“查询天气”大到“生成月度报表”。在 LangGraph 里集成技能之后图就变成了一个稳定的“执行骨架”业务逻辑的扩展不再需要改图只需要新增技能。2. 技能拆解与声明式设计好的技能不是代码堆出来的2.1 识别可落地的技能边界我见过不少项目技能拆得乱七八糟要么一个技能塞了太多功能要么一个简单动作拆成五六个技能大模型选起来晕头转向。技能拆分的判断标准其实很简单一个技能应该对应一个完整的、可验收的任务闭环。什么叫“可验收”就是执行完这个技能后你能明确地说“这件事做完了”。比如“生成一份销售周报”做完后输出一份完整的报告文件这是可验收的。但“计算销售额总和”就不是一个合格的技能它只是报告生成过程中的一个步骤应该作为内部逻辑藏在技能实现里而不是暴露给大模型。另一个判断标准是失败边界。如果技能执行到一半失败用户能明确知道是什么任务失败了而不是一个语义模糊的中间状态。选“上传文件到服务器”当技能失败时可以告诉用户“上传失败”但如果把“解析文件、提取关键字段、上报系统”拆成三个技能失败时用户看到的是“提取字段失败”反而难以理解整体目标进行到哪一步。2.2 用类配置定义技能接口技能接口的核心是对输入输出做形式化描述。这里我强烈建议用 JSON Schema 或类配置结构而不是散乱的参数列表。因为大模型对结构化的描述理解更准确且 schema 可以直接复用于校验逻辑。一个常规的技能定义长这样# 技能定义使用 dataclass 或者字典结构 SKILL_REVIEW { name: code_review, description: 对一段源代码进行静态审查返回问题清单和修改建议。适用于代码合并前的质量检查。, input_schema: { type: object, properties: { code: {type: string, description: 待审查的完整代码}, language: {type: string, enum: [python, javascript, java], description: 代码语言} }, required: [code] }, output_schema: { type: object, properties: { issues: {type: array, items: {type: string}}, suggestions: {type: array, items: {type: string}} } }, timeout_seconds: 30, retry_policy: {max_retries: 2, backoff_seconds: 1} }注意看几个细节。description 字段写的是“什么时候该用我”而不是“我能做什么”这决定了模型选错的概率。input_schema 里每个字段都带 description是为了让大模型填参数时更加准确。更关键的是 timeout_seconds 和 retry_policy这两个字段是 LangGraph 图节点执行时用的体现了技能和图框架的配合。2.3 技能粒度一组动作还是一条完整流程技能粒度没有标准答案取决于你的图结构。如果技能只做单个动作比如查天气、发消息那么 LangGraph 的图会更像“动态路由表”大模型每轮选一个技能执行这是最简单的形态。如果技能本身包含多个步骤比如“处理客户投诉”包含拉取工单、查询历史订单、生成回复草稿三个动作你可以把它作为一个组合技能封装起来。组合技能内部自己管理子步骤对外只暴露一个入口和出口。这样做的好处是图里只有一个节点坏处是内部流程对 LangGraph 变得不可见无法利用图的中间检查点能力。我个人的经验是优先让技能保持“单目标多步骤”的粒度即一个技能内部可以有多个操作步骤但必须围绕一个目标。真正需要跨技能协作的流程应该留在 LangGraph 的图层面编排而不是硬塞进一个技能里。这样兼顾了图的可控性和技能的复用性。2.4 把复杂技能拆成可组合的子步骤有些技能确实很大拆成一个整体太笨重拆成多个又太碎。这时候我建议在技能内部实现子步骤调度但不要让子步骤暴露给大模型。比如某个“数据迁移”技能内部包含连接源库、抽取数据、清洗转换、写入目标库、校验一致性五个步骤。对外它是一个技能对内它有一套自己的执行管线。你可以用 LangGraph 的子图来实现这种技能也可以用普通 Python 函数编排。用子图的好处是可以在子步骤之间添加断点、日志、人工确认坏处是实现成本高。小团队项目就别折腾了普通函数编排足够了。拆子步骤的时候有个原则子步骤之间用明确的数据契约连接。前一步的输出是后一步的输入每步都有独立的异常处理。这样即使某个子步骤失败也能定位到具体环节并触发重试。3. 实操在 LangGraph 中集成 Skills 的完整步骤3.1 先搭一个最小的 LangGraph 骨架在加技能之前先确保你的最小图能跑通。下面的代码定义了一个最简单的状态图只有一个节点执行后进入结束状态from typing import TypedDict from langgraph.graph import StateGraph, END class AgentState(TypedDict): messages: list current_skill: str skill_input: dict skill_output: dict def noop_node(state: AgentState) - dict: return {messages: state[messages] [节点已执行]} graph StateGraph(AgentState) graph.add_node(entry, noop_node) graph.set_entry_point(entry) graph.add_edge(entry, END) app graph.compile() result app.invoke({messages: [], current_skill: , skill_input: {}, skill_output: {}})这个代码有两层含义需要理解。一是 State 里增加了 current_skill、skill_input、skill_output 三个字段这是为技能预留的通道。二是整个图只有一个节点你可以在这个节点的位置换成真正的技能执行器。3.2 实现一个通用的技能执行器节点技能执行器是整个架构的核心节点。它的职责是从状态中读取技能名称去注册表找到技能定义校验输入参数执行技能把结果写回状态。import time import logging SKILL_REGISTRY { code_review: SKILL_REVIEW, # ... 更多技能 } def find_skill(name: str): skill_def SKILL_REGISTRY.get(name) if not skill_def: raise ValueError(f技能 {name} 未注册) return skill_def def execute_skill_node(state: AgentState) - dict: skill_name state.get(current_skill) skill_input state.get(skill_input, {}) if not skill_name: return {skill_output: {error: 未指定技能名称}} skill_def find_skill(skill_name) # 输入校验 errors validate_schema(skill_input, skill_def[input_schema]) if errors: return {skill_output: {error: f参数校验失败: {errors}}} # 执行技能带超时 start time.time() try: result run_skill_impl(skill_def, skill_input) except Exception as e: result {error: str(e)} elapsed time.time() - start logging.info(f技能 {skill_name} 执行耗时 {elapsed:.2f}s) return {skill_output: result}这里有三个容易被忽略的点。第一validate_schema 这一步很关键它避免了大模型乱传参数导致技能内部崩溃。第二run_skill_impl 根据技能定义里的 type 或 name 分发到具体的实现函数这个分发机制可以用装饰器实现。第三返回的字典里包含 error 字段这为后面图里做条件分支提供了信号。3.3 让大模型自主选择技能动态路由的实现前面那个图是静态的真正的智能体需要让大模型根据用户任务动态选择技能。这个场景的标准做法是在图里加一个“规划节点”由大模型分析当前状态决策下一步要执行哪个技能。def planner_node(state: AgentState) - dict: user_request state[messages][-1][content] if state[messages] else # 把所有技能的描述构建成提示 skill_descriptions \n.join([ f- {name}: {detail[description]} for name, detail in SKILL_REGISTRY.items() ]) prompt f根据用户请求,从以下技能中选择一个最合适的执行。 如果所有技能都不适合,回复 no_skill。 技能列表: {skill_descriptions} 用户请求: {user_request} 只输出技能名称,不要额外解释。 skill_name call_llm(prompt) # 你的 LLM 调用封装 return {current_skill: skill_name, skill_input: extract_input_from_request(user_request, skill_name)}在这个思路下整体图结构变成入口节点 - 规划节点 - 技能执行器 - 判断结果 - 回到规划节点或结束。这就是标准的 Agent 循环只是“动作空间”变成了技能注册表。有一个关键细节你可以用上技能描述中如果包含“适用于什么情况、不要用于什么情况”大模型选错的概率会显著降低。3.4 利用条件边处理技能执行结果技能执行不保证成功所以图里必须有分支逻辑。LangGraph 的条件边非常适合做这件事。判断逻辑通常看 skill_output 里是否有 error 字段。def after_skill_router(state: AgentState) - str: output state.get(skill_output, {}) if output.get(error): return retry return next graph.add_conditional_edges( skill_executor, after_skill_router, { retry: planner, next: END } )这个分支的价值在于技能失败后可以自动回到规划节点让大模型重新换一个技能或者调整参数。如果大模型连续失败几次你还需要一个计数器字段超过上限直接进入人工处理节点。这种控制流在纯函数调用式的 Agent 里很难优雅实现在图模型里却非常顺手。3.5 状态上下文与技能参数的传递技能执行过程中往往需要访问上下文信息比如用户ID、会话ID、历史消息、外部系统凭证。这些数据不能每次靠大模型传应该由执行器节点从状态中自动注入。我一般会在状态里放一个 context 字段里面是所有技能需要的基础上下文。execute_skill_node 在调用技能实现之前把 context 中的字段合并进技能输入def execute_skill_node(state: AgentState) - dict: skill_name state.get(current_skill) skill_input dict(state.get(skill_input, {})) context state.get(context, {}) # 注入上下文,但技能显式传入的参数优先 for k, v in context.items(): skill_input.setdefault(k, v) # ... 后续执行逻辑用 setdefault 而不是直接覆盖是为了避免上下文里的默认值覆盖掉大模型根据用户请求提取出来的参数。这个坑我踩过之前直接把 context 和 skill_input 做并集合并结果大模型提取到的“当前用户ID”老是被上下文里的默认值覆盖导致技能屡屡操作到错误对象。3.6 流式输出与进度回调LangGraph 本身支持流式输出对于耗时技能来说非常有用。比如数据清洗技能可能要跑几十秒如果用户看不到进度体验会很差。实现方式有两种。一种是用 LangGraph 原生的事件流机制async for event in app.astream_events(state, versionv2): if event[event] on_chain_stream: # 透传技能实现中自定的进度事件 chunk event.get(data, {}).get(chunk) if chunk and chunk.get(type) progress: yield f进度: {chunk[message]}另一种更简单的方式是在技能实现里通过生成器逐段返回结果图节点把生成器产生的每个块即时写入状态的子字段中。如果你用的是流式 LLM 界面这个方法对调试也友好得多。4. 常见问题与排查技巧实录4.1 技能重复执行的幂等性LangGraph 支持对节点进行重试图模型本身允许同一节点被多次执行。问题在于如果你的技能不是幂等的比如“发短信”“扣余额”“创建订单”重试一次就重复执行一次轻则系统报错重则产生脏数据。解决思路是让技能实现里带一个操作记录在执行前生成一个唯一的 execution_id用外部存储记录执行状态。如果重复执行时发现同样的 execution_id 已经成功完成直接返回上次结果。EXECUTION_CACHE {} def run_skill_impl(skill_def, skill_input): exec_id skill_input.get(execution_id) or str(uuid.uuid4()) if exec_id in EXECUTION_CACHE: return EXECUTION_CACHE[exec_id] # 真正的技能逻辑 result do_skill_work(skill_input) EXECUTION_CACHE[exec_id] result return result如果你用的是 Kafka 或 Redis 这类中间件也可以把 EXECUTION_CACHE 换成 Redis 存储天然支持分布式环境的幂等。4.2 技能响应超时与重试策略外部 API 调用是最容易超时的环节。在技能定义里加 timeout_seconds 字段只是第一步关键还在于执行节点里真的用超时控制包住调用。from concurrent.futures import ThreadPoolExecutor, TimeoutError def execute_with_timeout(skill_def, skill_input): io_pool ThreadPoolExecutor(max_workers1) future io_pool.submit(run_skill_impl, skill_def, skill_input) try: return future.result(timeoutskill_def.get(timeout_seconds, 30)) except TimeoutError: future.cancel() return {error: 技能执行超时}超时之后的处理也要想清楚。如果技能是幂等安全的直接重试如果不是超时只能把控制权交还给规划节点让大模型决定改参数还是换技能。不要盲目套用固定重试策略我在生产环境里就因为固定重试三次把外部系统的限流打爆过。4.3 状态超限与序列化失败LangGraph 要求状态是 “可序列化”的。技能执行过程中如果往状态里塞了 Python 对象、文件句柄、连接池对象invoke 的时候就会发现序列化报错非常隐蔽。我遇到过一个场景技能里加载了一个大模型 tokenizer 对象顺手放进了状态里图跑完保存检查点时直接报错。排查了很久才发现是状态里多了一个不可序列化的成员。后来我把状态字段时间设置得非常明确只允许 JSON 基础类型、列表、字典。任何复杂对象要么转成 JSON 字符串要么放进外部存储。另外状态体积也要控制。技能执行结果如果太大比如拉取全量数据状态会被撑爆。我会在技能输出到状态之前做摘要化处理只保留关键统计量完整数据放进文件存储状态里只存引用路径。4.4 并行技能的状态写冲突LangGraph 支持并行节点执行比如同时调用多个技能但如果多个技能都往同一个状态字段里写就会有写冲突。默认的 reducer 策略是直接覆盖后写的覆盖先写的结果可能丢失数据。解决办法是在状态字段定义时用自定义 reducer。最常见的是把同名消息做追加合并这个 LangGraph 官方已经有内置支持。对于技能结果字段我会让每个技能把结果写入独立的键class AgentState(TypedDict): messages: Annotated[list, operator.add] skill_results: dict # 技能名称作为 key并行节点里各自写入 skill_results[skill_a] 和 skill_results[skill_b]就避免了互相覆盖。如果你需要一个总结果字段做后续判断那就用一个专门的汇总节点在所有并行技能跑完之后执行合并逻辑。4.5 条件边太多导致图难以追踪图分支太多调试时会乱。我的经验是不要把所有判断逻辑都放在条件边里。Locate 判断的复杂度应该拆开简单的成功/失败分支放条件边复杂的路由决策放一个专门的路由节点里。比如一个图有十几个技能如果每个技能后面都接一串条件分支图面会非常难看。我会压缩成所有技能节点都汇聚到一个统一出口节点由出口节点根据技能执行结果统一决定下一步去向。这相当于用中心化的路由代替了放射状的条件边逻辑清晰维护也简单。常见问题典型表现排查方向推荐做法技能重复执行重试后生成了重复订单/重复扣款检查技能实现是否具备幂等保护增加 execution_id 去重外部接口超时图运行到技能节点卡住很久检查 timeout 参数是否真正生效用线程池包裹调用设置硬超时状态序列化失败invoke 阶段直接报 pickle/JSON 错误检查状态里是否有非序列化对象严格控制状态字段类型并行技能写冲突多个技能结果互相覆盖检查状态字段是否使用合并 reducer按技能名分键写入图分支爆炸添加技能时图变得难以维护观察条件边数量是否过多用统一出口节点做中心化路由5. 从单技能到技能编排给智能体一张技能地图5.1 技能的层次化组织技能数量几十个之后就不能平铺在一张注册表里了。大模型面对几十个候选技能时选择准确率会明显下降。我早期遇到的情况是一次规划请求里大模型经常选错技能尤其是描述相似的技能。解决思路是分层组织。按照领域把技能分组每组设置一个“领域路由技能”大模型先选领域再选具体技能。这个方式相当于两级检索大幅降低单层选择的压力。SKILL_GROUPS { 数据查询: { description: 用于查询各类业务数据, skills: [order_query, user_query, product_query] }, 数据分析: { description: 用于对已有数据进行处理和统计, skills: [sales_summary, trend_analysis] }, 内容生成: { description: 用于生成报告、邮件、文案等内容, skills: [report_generate, email_draft] } }规划节点先从组描述中选组再从组内选技能。这个方法实测下来技能选择准确率比平铺式高了很多。代价是多了一次 LLM 调用但对于准确性要求高的场景完全值得。5.2 让多个技能协作的编排模式有些任务无法靠单个技能完成需要多个技能按顺序配合。比如“根据最近的销售数据给重点客户写一封邮件”先调用数据查询技能拿销售数据再调用数据分析技能算增长率和重点客户最后调用内容生成技能写邮件。这种流程型任务有两种做法。第一种是图层面显式编排三个技能节点按顺序连起来状态依次传递。好处是流程清晰、每一步都可监控坏处是每增加一个流程就要改图结构。第二种是在规划节点中用“行动计划”指定一串技能技能执行器按顺序执行。这个模式下图本身保持通用具体流程由大模型动态规划。这更像真正的 Agent 行为但执行过程的可控性会差一些。我目前的建议是核心业务流程用第一种边缘的、低概率的流程用第二种。核心流程显式化能保证稳定性边缘流程交给 Agent 自由发挥能保留灵活性。5.3 人机协作技能失败时把控制权还给用户技能执行到不可恢复的错误时不要强制重试也不要让大模型无限思考。最合理的做法是进入一个人工处理节点把错误信息和上下文抛给用户让用户决定下一步。在 LangGraph 里实现这个很自然技能结果 error 字段为真且重试次数超限时条件边直接指向一个 “human_handoff” 节点。这个节点可以暂停图的执行等用户在界面上输入后再把用户的指令写回状态继续走图。这里有一个细节不要把原始错误堆栈直接展示给用户。我试过直接透传底层异常用户看到数据库连接串、内部 API 地址既体验差又不安全。应该先做错误脱敏把技术错误转成业务描述比如“订单查询服务暂时不可用请稍后重试”。写在最后的实操体会我在实际项目中集成这些技能模块时最深的体会是技能设计占了大半工作量代码实现反而简单。很多时候你以为难点在于大模型怎么选对技能真正调试之后发现问题出在技能描述含糊、输入输出定义不严谨、失败后没有兜底路径。另外一点很实在的建议技能注册表一定要做可视化。我在本地调试时写了一个简单的 HTML 页面把每个技能的描述、参数、最近执行时间、失败率都列出来肉眼能看到的技能分布比任何文档都直观。哪个技能经常被选错哪个技能失败率奇高看一眼就知道该优化哪块。最后再分享一个小技巧每新增一个技能前强制自己写出这个技能的三个边界条件——什么场景下不该用它、输入什么异常值会失败、失败后用户应该怎么处理。写不出这三个条件的技能说明你对它的理解还不够先别急着写代码。这个习惯帮我挡掉了很多后续的返工。