news 2026/10/21 18:52:30

LangGraph技能化改造:从提示词膨胀到结构化Skills的Agent架构实践

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
LangGraph技能化改造:从提示词膨胀到结构化Skills的Agent架构实践

上个月我把一个几乎跑不动的 Agent 项目整个推倒重做了。起因很现实:system prompt 里塞了十几个工具的调用说明,模型经常在"查订单"和"查物流"之间犯迷糊,A 需求改一个字段,B 流程跟着出幺蛾子。重构的核心动作只有一个——在 LangGraph 里把所有能力收编成结构化的 Skills,让模型通过技能注册表去选择能力,而不是靠一段越来越长的提示词去猜。

这篇文章把这段经历整理出来,围绕三个问题展开:Skill 到底解决了什么、LangGraph 里怎么给 Skill 安排位置、上线之后怎么排错和维护。从没接触过 LangGraph 的读者也能读,我会把每个"为什么这么做"都讲清楚;已经在折腾 Agent 的老手可以直接跳到第 5、6 节,那里有我踩了几次才想明白的东西。

1. 单块式 Agent 的失控现场:Skills 到底在解决什么问题

1.1 提示词膨胀不是数量问题,而是耦合问题

很多人把 Agent 能力不足归因于"工具太少",于是疯狂往提示词里加工具,但问题恰恰相反:工具越多、说明越杂,模型越无从下手。一个三千多词的 system prompt,里面混着工具定义、业务规则、输出格式、few-shot 示例,它们彼此之间没有任何边界。比如你让模型"先查天气再安排行程",模型可能直接在第一步就把行程安排一起生成了,因为"安排行程"的说明就在同一个上下文里触手可及。

提示词膨胀背后的本质是耦合:能力边界不清、触发条件互斥、更新互相干扰。加一个新工具,所有既有描述都要跟着重写一遍,回归测试等于全量回归。这种时候问题已经不是提示词工程能解决的了,得靠架构层面的拆分。我见过太多团队在 5000 词的提示词上打补丁,最后连自己都不敢改,因为改一句就可能让另一个功能悄悄失效。

1.2 技能(Skill)与普通工具的本质区别

普通工具是最小原子动作,输入输出都很薄,比如"执行搜索""发送消息"。而 Skill 是一整套可以独立交付的能力,内部可能用了多个工具、多轮模型调用、甚至一段确定性代码,但对外只暴露一个清晰的能力描述和输入输出契约。

打个比方:工具是单个零件,Skill 是"一台插上电就能工作的设备"。在 Agent 架构里,Skill 的价值体现在三个方面:

  • 边界清晰:模型只需要知道什么场景下用它,不需要理解内部实现;
  • 可测试:每个 Skill 都能脱离整个 Agent 单独跑单测;
  • 可组合:路由层可以把多个 Skill 拼成更复杂的流程,而不需要改 Skill 本身。
对比维度普通工具Skill
粒度原子操作完整流程
外部可见内容函数签名技能说明 + 输入输出契约
内部独立性通常依赖调用方上下文完全自包含
适合场景简单查询、写操作多步骤、多工具组合、需要专门提示词的场景

1.3 有些东西不值得做成 Skill

要克制。一个只在单处出现、逻辑少于十行、输入输出不确定的 helper,留在 node 里比做成 Skill 更划算。Skill 是有代价的:要写模型能看懂的描述,描述写得不好模型不会用;要维护 schema,字段一变全链路跟着变;还要在图里多一跳节点,增加调用链延迟。

我给自己定了三条准入规则,三个都是肯定答案才动手:它会被两个以上场景复用吗?它内部的逻辑足够独立、能离开整个图单独解释吗?它的输入输出稳定吗?过度抽象害死人,把一个只调用一次的函数包装成 Skill,只会让图和目录结构都变臃肿。

2. 用 LangGraph 的 State 和 Edge 给 Skill 腾位置

2.1 State 里专门开一条"技能通道"

LangGraph 的所有节点共享一份 State,数据怎么更新,决定了你会不会踩坑。如果技能节点直接往全局字段里写业务数据,多个技能在同一条链路上执行时,就会出现"上一手写的东西被下一手覆盖"的模糊地带。

我的做法是:在 State 里固定两个字段skill_requests和skill_results。skill_requests是路由层给技能下发的指令列表,skill_results是 dict,key 是技能名,value 是该技能返回的结构化结果。所有技能只往自己名字对应的 key 里写,读也只读自己的 key。这样整个图的数据流是单向的:路由层决定谁被调用,技能层只负责产出,汇总层再统一消费结果。

from typing import TypedDict, Any class AgentState(TypedDict): task: str # 用户原始任务 skill_requests: list[dict] # 待执行的技能请求 skill_results: dict[str, Any] # 技能名 -> 结构化结果 messages: list[dict] # 对话记录,用于回放和调试

我把这称为"技能通道"。它的价值在于:不管以后往图里加多少个技能,数据流的骨架不需要改。新技能只是往skill_results里多占一个 key,仅此而已。

2.2 节点注册的三种层级

在 LangGraph 里集成 Skills,节点注册有三种做法,我按项目演进顺序讲:

第一层,直接在构建图时写graph.add_node("fetch_news", fetch_news_fn)。最直接,但技能一多,组装代码会膨胀成一大坨。

第二层,把所有技能放进一个注册表,用循环统一注册。这一步让"技能清单"变成了数据,而不是散落的代码:

SKILLS = { "fetch_news": { "module": "skills.fetch_news", "enabled": True }, "summarize": { "module": "skills.summarize", "enabled": True }, # ... } for name, meta in SKILLS.items(): if meta["enabled"]: node_fn = load_skill_executor(meta["module"]) graph.add_node(f"{name}_node", make_adapter(node_fn, name))

第三层,把注册表结合配置文件做动态装配。测试环境加载全部技能,生产环境只加载enabled: true的技能。关键点是把"技能清单"变成配置的一部分,而不是埋在代码里,后面做灰度、做开关才有的放矢。

2.3 条件边才是真正做调度的人

LangGraph 条件边的工作方式是:执行完一个节点后,调用一个函数,根据返回值跳到对应的节点(或 END)。我习惯把这层叫"调度器"。它做的事情很纯粹:查看当前 state 里的skill_requests,取出第一个待执行技能的名字,返回给条件边映射表。

为什么不能让技能自己决定"我做完去哪"?因为那样技能就必须知道整个图的拓扑,又耦合了。调度器统一收口之后,技能节点完全不关心"下一步是谁",只管把手头的事做完。

from langgraph.graph import StateGraph, START, END def route_after_skill(state: AgentState) -> str: if state["skill_requests"]: return state["skill_requests"][0]["name"] return "finish" graph.add_conditional_edges( "router", route_after_skill, { "fetch_news": "fetch_news_node", "summarize": "summarize_node", "risk_alert": "risk_alert_node", "finish": END, }, )

写完这段代码我才真正理解了 LangGraph 的设计哲学:它不替你做调度决策,它只给你一个干净的路由机制,把"什么时候跑什么能力"完全交给你的逻辑。

3. 技能工厂设计:一份 manifest、一个执行器、一张注册表

3.1 技能说明书(manifest):模型靠它理解世界

在 LangGraph 里集成 Skills,最不能省的是那份给模型看的说明书。我把它做成每个技能目录里的 manifest 文件,JSON 格式,包含 name、description、input_schema、output_schema。description 是给路由模型看的,写作要遵守"场景前置 + 行为描述 + 边界提示"的格式:

{ "name": "fetch_news", "version": "1.2.0", "description": "当用户需要最新行业动态、资讯列表时使用。本技能只负责检索资讯标题和来源,不做摘要、筛选和风险判断。如果用户需要对资讯进行总结,请改用 summarize 技能。", "input_schema": { "topic": { "type": "string", "required": true }, "max_items": { "type": "integer", "default": 5 } }, "output_schema": { "items": [ { "title": "string", "source": "string", "published": "string" } ] } }

千万不要只写"抓取资讯"四个字。我第一次就栽在这上面,模型拿到这种描述,完全不知道什么时候该用它。另外 input_schema 要严格一点:模型填参数时很爱省略字段,required 标不好,后面 executor 就要到处处理 None。

3.2 执行器(executor):纯函数化是底线

每个技能目录里有个executor.py,对外只暴露run(inputs: dict, ctx: dict) -> dict。ctx 里放只读的环境信息(比如 API key、base_url),但绝不把整个 LangGraph state 传进去。

为什么这么设计?纯函数化之后,技能单测可以在完全离线、不启动 LangGraph 的环境下跑,直接在 pytest 里构造一个 inputs 就能验证。这是技能库能规模化维护的根本。换模型、换路由都不影响技能内部逻辑,回归测试基本不用动。

def run(inputs: dict, ctx: dict) -> dict: topic = inputs["topic"] limit = inputs.get("max_items", 5) items = search_provider.fetch(topic, limit=limit) return { "items": [ {"title": item.title, "source": item.source, "published": item.published} for item in items ] }

3.3 注册表与启动时的动态装配

启动时扫描技能目录,读 manifest,验证 schema,把 executor 包成符合 LangGraph node 签名的函数,注册进图里。同时把所有 manifest 的 name + description 汇总成一个工具列表,绑定给路由模型。

这里有个细节:技能包成节点后,函数签名需要接收 state 并返回 state,而 executor 只处理自己的 inputs。中间必须加一层 adapter,在 adapter 里从 state 取参数、调 executor、把结果写回skill_results[skill_name]:

def make_adapter(executor_fn: callable, skill_name: str): def adapter(state: AgentState) -> AgentState: # 取出当前技能请求,执行完后移出队列 request = state["skill_requests"].pop(0) try: result = executor_fn(request["arguments"], {"env": ...}) state["skill_results"][skill_name] = result except Exception as exc: state["skill_results"][skill_name] = { "error": {"code": type(exc).__name__, "message": str(exc)} } return state return adapter

这个 adapter 是通用模板代码,所有技能共用同一套。反正我从这个函数开始写第一版技能体系,后面的技能几乎都是复制目录改 manifest。

4. 实战集成:给"行业动态速报"Agent 挂上三个技能

4.1 场景拆解:任务的三个子问题

我以模拟项目X为例,它是一个"行业动态速报"Agent。用户给一个主题,最终要输出一份简报,这个任务可以拆成三个子问题:抓取该主题的最新资讯、逐条生成摘要、识别其中的风险点并提示。

这三个子问题我直接做成三个技能:fetch_news、summarize、risk_alert。按 Skill 的标准,它们都独立可复用:fetch_news 可以被舆情监控复用,summarize 能给邮件简报用,risk_alert 也能单独接进别的流程。这个拆分是"会被反复复用"的典型,不是拍脑袋抽象出来的。

4.2 状态定义与三个技能节点

状态沿用第 2 节的 AgentState。三个技能用同一个 adapter 注册成节点。summarize 内部是模型调用,输入 items 列表,输出每条摘要;risk_alert 内部是规则 + 模型判断,输入原文和摘要,输出风险等级与理由。

graph = StateGraph(AgentState) graph.add_node("router", router_node) graph.add_node("fetch_news_node", make_adapter(fetch_news_executor.run, "fetch_news")) graph.add_node("summarize_node", make_adapter(summarize_executor.run, "summarize")) graph.add_node("risk_alert_node", make_adapter(risk_alert_executor.run, "risk_alert")) graph.add_edge(START, "router") graph.add_conditional_edges( "router", route_after_skill, { "fetch_news": "fetch_news_node", "summarize": "summarize_node", "risk_alert": "risk_alert_node", "finish": END, }, ) # 技能执行完,回到路由继续排队 for node_name in ["fetch_news_node", "summarize_node", "risk_alert_node"]: graph.add_edge(node_name, "router")

注意这里我把技能执行完都拉回 router,形成一个循环。调度的核心设计是:路由节点一次性产出多个技能请求,然后逐个执行。上一个技能结束不代表整个任务结束,而是回到路由看是否还有待办。

4.3 路由模型:先选技能,再填参数

路由节点做的事情是:把三个技能的 manifest 转成模型可识别的 tool schema,绑定给模型调用一次,要求返回 tool_calls,再把这些调用写入skill_requests。

def router_node(state: AgentState) -> AgentState: tool_defs = [manifest_to_tool_schema(m) for m in reg.all_manifests()] llm = create_llm().bind_tools(tool_defs) response = llm.invoke([ {"role": "user", "content": state["task"]} ]) state["skill_requests"] = [ {"name": call.name, "arguments": call.args} for call in response.tool_calls ] return state

这样设计的好处是:路由模型不需要知道每个技能内部怎么跑,它只需要做"选择题"——根据用户任务挑出需要的技能并填好参数。选择题比开放式生成稳定得多,这是整个架构最值钱的地方。

4.4 跑通后的观察:token 与稳定性

几个真实体感供参考。system prompt 从原来的四五千字压到六百字左右,路由模型只需吃一份技能描述列表,上下文明显变轻;代价是每次路由多一次模型调用,但换来的是工具选择稳定很多。三个技能串行执行时,比较头疼的是 summarize 阶段模型容易忘掉 fetch_news 的原文内容,解决方式是把上一步结果合并进下一步的对话上下文,而不是只靠 state 传结构化数据。

诚实地说,技能化不是银弹。它解决了耦合问题,但没解决上下文管理问题。技能之间的数据怎么转换成模型可读的上下文,仍然需要针对每一条链路单独调。这一块没有通用解法,只能靠经验和观测数据。

5. 排错实录:技能不触发、状态串写、异常冒泡

5.1 症状一:技能被当成普通工具反复触发

现象:模型在一个任务里连续调用同一个技能两三次,或者明明该走风险识别,却反复去抓取资讯。我排查的第一件事是看路由模型的 tool_calls 日志,结果发现模型选了技能之后根本不看 description,纯粹按名字猜。

根因是 manifest 的 description 写得太粗糙。我一开始写的是"抓取资讯",没有任何场景边界。改成带有前置条件的描述后,误调用基本消失。后来我总结出一个通用模板:

当用户需要【场景】时使用本技能。本技能只负责【行为】,不负责【边界外的事】。如果遇到【互斥场景】,请改用其他技能。

这段模板里的三个空必须填实,否则 description 就是不合格的。模型不是看不懂,是你没给它足够的判断依据。

5.2 症状二:上一步的结果被下一步悄悄覆盖

现象:risk_alert 跑完,fetch_news 的结果变成空列表了。我一开始以为是被并发覆盖,回放 state 才发现,两个技能的 adapter 都往同一个 key 写数据——因为新技能在注册时名字写错了,覆盖到了已有技能的结果 key。

教训有三点:技能注册 key 必须全局唯一;adapter 只允许写自己注册时的 name,谁都不能碰别人的 key;汇总层永远先读请求列表再读结果列表,按请求顺序去取结果,而不是按固定 key 取。这个坑非常隐蔽,因为单技能测试看不到,只有多技能串行时才炸。

# 汇总层安全写法:按请求顺序取结果 def aggregate_results(state: AgentState) -> str: chain = [] for req in state["skill_requests_trace"]: result = state["skill_results"].get(req["name"]) if result and "error" not in result: chain.append((req["name"], result)) return format_report(chain)

注意我在 state 里加了一个skill_requests_trace,它只记录原始请求、不弹队列,专门用来给汇总层做顺序还原。弹队列的skill_requests是给调度用的,两者不冲突。

5.3 症状三:技能内部异常把整个图拖垮

现象:fetch_news 请求超时抛异常,整个图直接中断,用户等了个寂寞。第一次我天真地以为在 executor 里 try/except 就完事,后来发现异常被吞掉之后模型不知道发生了什么,开始瞎编内容,比报错还难查。

最佳实践是第 3 节 adapter 里的那层统一防护:异常转换成结构化 error 字段写回skill_results,然后让路由模型在汇总时看到 error 字段,由模型决定是重试还是换路径。注意不要无脑自动重试,要带着错误信息给模型判断,否则外部服务超时的时候,重试会成倍放大压力。

5.4 通用排查法:回放状态,而不是盯着日志猜

LangGraph 支持 checkpointer,把每一步的 state 存下来。我排错最常用的操作是:加载同一个线程 ID,按步回放,看skill_requests是什么时候变空的、skill_results是哪一步被覆盖的。这比翻一百行日志高效得多。

我的固定排查流程是:日志定位怀疑节点 -> 回放 state 确认数据 -> 用出问题的输入构造最小复现 -> 修 manifest 或 adapter -> 跑回归。这个流程走下来,平均一个技能问题半小时内能定位,比早期靠猜快了一个量级。

6. 生产化阶段的血泪建议:版本、成本和协作

6.1 Skill 目录也要做版本管理与灰度

技能是运行时代码,会持续更新。我一开始把 manifest 和代码混在一个仓里,改了个字段就全量上线,路由模型的表现波动很大。后来在 manifest 里加了 version 字段,并支持按环境加载技能清单:测试环境加载全部技能,生产环境只加载 enabled 的技能。

这里有一个真实教训:新版本尽量"先加后删"。如果新版本想改 description,直接替换会有过渡期,模型两边都不讨好。正确做法是先加一个 v2 技能,旧技能保留,观察一周调用成功率再下线 v1。技能下线前还要检查一下有没有其他流程引用了它的 schema,我吃过一次亏,删了技能之后汇总层直接报 KeyError。

6.2 给每个技能记账:token 成本的可观测化

技能化之后,每个技能的调用都是独立模型调用,成本散落在图里,不打点的话你对钱毫无概念。我在 state 里增加了 usage 聚合字段,每次模型调用后把 prompt_tokens 和 completion_tokens 累加到当前技能名下。

def record_usage(state: AgentState, skill_name: str, usage) -> None: if "usage" not in state: state["usage"] = {} entry = state["usage"].setdefault(skill_name, {"prompt_tokens": 0, "completion_tokens": 0}) entry["prompt_tokens"] += usage.prompt_tokens entry["completion_tokens"] += usage.completion_tokens

汇总层打印一份费用明细表之后,我很快就发现 summarize 技能占了六成成本。有了数据才能谈优化,不然只能拍脑袋换小模型。我建议每个技能上线时就带上这段打点逻辑,不然后面补会非常痛苦。

6.3 团队协作:让描述能力差的人写描述,是灾难

技能化的一个红利是后端同事不用懂 LangGraph 也能提交技能,但前提是 description 必须由最理解业务场景的人来写。我踩过的坑:让后端同事按函数注释的习惯写技能描述,结果模型把技能当摆设,因为描述全是技术词汇,完全没有"什么时候该用我"。

后来我定了一个硬性模板:描述里必须出现"当用户需要…时"和"本技能不负责…"这两类句式,缺一个都不允许合入。这个模板没几行字,但提交质量提升立竿见影。测试方面,每个新技能必须附带一个最小用例和一组期望结果,跑通才允许注册。

最后分享一个个人小习惯:每上线一个技能,我会在它的目录下放一个 examples/ 文件夹,把路由模型成功调用和失败误调用的完整日志都贴进去。过两周做回归测试时,这些例子就是现成的测试集。技能化这条路,难的不是写代码和搭图,而是学会用边界思维去切分能力——边界切得越干净,后面的路由、测试、协作就越省心。希望这篇把坑标出来的短文能让你少走我走过的弯路。

版权声明: 本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若内容造成侵权/违法违规/事实不符,请联系邮箱:809451989@qq.com进行投诉反馈,一经查实,立即删除!
网站建设 2026/10/21 18:51:18

GDAL VS2017库配置避坑指南:从目录结构到运行验证

简介:面向VS2017环境提供的GDAL预编译库,专供Windows平台C地理空间开发使用。GDAL为开源地理空间数据抽象库,支持超过250种栅格及矢量格式,涵盖读写、转换、投影等核心能力。压缩包共106个文件,以104个C头文件为主&…

作者头像 李华
网站建设 2026/10/21 18:50:37

台达PLC与三菱变频器Modbus RTU通讯:从接线到梯形图实战解析

1. 为什么是台达PLC去控三菱变频器先把结论放在前面:台达PLC和三菱变频器之间做Modbus通讯,真正麻烦的不是指令代码,而是地址映射和那根RS-485尾巴。只要把寄存器地址对准、把变频器的运行模式放对,DVP-ES2的COM2口加几条Modbus R…

作者头像 李华
网站建设 2026/10/21 18:49:02

C++继承实战指南:从代码复用到多态,避开对象切片与虚析构陷阱

前阵子我接手了一个叫“模拟项目X”的旧工程,里面有三个类,长得非常像:都有名称、编号、状态标记,也都实现了相同的保存和加载逻辑,区别只是各自多了一些字段。我一看这代码,第一反应是复制、粘贴、改几个类…

作者头像 李华
网站建设 2026/10/21 18:47:10

Java多态底层原理与面试要点:动态绑定、虚方法表到工程落地

1. 先抓核心:面试官问“多态”时,到底在问什么把这个问题拆开看,面试官手里其实握着一张“检查清单”:你知不知道多态的基本定义,能不能用一两句话说清楚?你说的是“重写”层面上的多态,还是“重…

作者头像 李华
网站建设 2026/10/21 18:47:09

数据库一体机的智能复兴:从Britton-Lee到Exadata与zData

1. 一个"生不逢时"的机器:Britton-Lee没能等来它的时代这几年我经常遇到一个有意思的问题:云计算都喊了这么多年,为什么金融、制造、医疗这些行业的用户,反而在关键系统上重新抱紧了数据库一体机?甚至有人告…

作者头像 李华
网站建设 2026/10/21 18:45:31

三级等保设备清单与落地部署实战:从采购到运维的完整指南

简介:这份PDF文档面向需要开展网络安全等级保护三级建设与测评的企业IT负责人、安全运维人员及合规咨询从业者,系统梳理了三级等保落地所需的设备与服务清单,帮助读者对照标准快速明确建设范围与配置要求。资源包共1个PDF文件,约2…

作者头像 李华