如果你正在做基于大模型的智能体应用，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, version="v2"): 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_workers=1) future = io_pool.submit(run_skill_impl, skill_def, skill_input) try: return future.result(timeout=skill_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 判断的复杂度应该拆开：简单的成功/失败分支放条件边，复杂的路由决策放一个专门的路由节点里。

比如一个图有十几个技能，如果每个技能后面都接一串条件分支，图面会非常难看。我会压缩成：所有技能节点都汇聚到一个统一出口节点，由出口节点根据技能执行结果统一决定下一步去向。这相当于用中心化的路由代替了放射状的条件边，逻辑清晰，维护也简单。

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 页面，把每个技能的描述、参数、最近执行时间、失败率都列出来，肉眼能看到的技能分布比任何文档都直观。哪个技能经常被选错，哪个技能失败率奇高，看一眼就知道该优化哪块。

最后再分享一个小技巧：每新增一个技能前，强制自己写出这个技能的三个边界条件——什么场景下不该用它、输入什么异常值会失败、失败后用户应该怎么处理。写不出这三个条件的技能，说明你对它的理解还不够，先别急着写代码。这个习惯帮我挡掉了很多后续的返工。