★ FEATURED ARTICLE 去年冬天我手上有个任务型 Agent 的小项目需求一句话概括给它一句模糊的目标让它自己拆步骤、自己找工具、自己判断做没做完。第一版我用最朴素的写法——一个大 prompt 里塞进所有工具说明让模型边想边调循环到它自己说完成为止。结果跑了一晚上早上打开日志看到的是一个在原地打转的模型同一份计划被它换了三种措辞又执行了一遍最后信心十足地宣布任务成功而输出文件里只有三行占位文本。那次之后我重写了两遍最终落到LangGraph上把规划和验证从 prompt 里的软约束变成了图里的硬节点。这篇文章讲的就是这套东西一个具备自我路径规划能力和结果验证闭环的Agent怎么从状态定义、节点职责、条件边路由一路搭到熔断、观测和评测。适合已经写过几个 ReAct 式 Demo、发现一旦任务超过三步就开始失控的同学也适合完全没接触过状态图编排、但想搞清楚Agent 为什么需要图的人。我会把每一步为什么这么设计讲清楚也会把踩过的坑和排查链路原样端出来方便你直接对照自己的日志。1. 把自我路径规划从玄学拆成三件可工程化的事很多人第一次听到自我规划脑子里浮现的是模型在思考框里写一大段首先、其次、然后。这不叫规划这叫自然语言散文。真正的规划要能在代码里被检查、被修改、被重放所以我一般把它拆成三个可验证的动作缺一个这套系统就会退化成随机漫步。1.1 分解把目标切成有完成判据的步骤分解的关键不在于切几刀而在于每一步都带一个完成判据done_when。没有判据的步骤执行完之后没人知道算不算完成验证节点就只能靠模型感觉而模型的感觉在长任务里极不稳定。我第一版最大的错误就是把步骤写成搜集资料整理结论这种动词短语。这类描述无法判定。后来我把每一步强制成结构体用 Pydantic 约束字段模型必须在done_when里写一句可被第三方检查的话比如输出中至少包含三条带来源链接的事实。from pydantic import BaseModel, Field class Step(BaseModel): id: int intent: str Field(description这一步要达成什么) tool: str | None Field(defaultNone, description需要的工具名无需工具则为 None) args_hint: dict Field(default_factorydict) depends_on: list[int] Field(default_factorylist, description前置步骤 id) done_when: str Field(description可被检查的完成判据禁止写尽量充分) class Plan(BaseModel): steps: list[Step] rationale: str Field(description为什么这样拆不超过 80 字)depends_on这个字段是后面一切的基础。它让计划从一个列表变成一张有拓扑序的图执行器才知道哪些步骤可以并行、哪些必须等、某个步骤失败时下游哪些要作废。很多人省掉它然后在执行阶段硬编码顺序一旦需要插入新步骤就全乱。rationale限长也是个经验不限制的话模型会在里面写小作文白白吃掉上下文预算而这段文字真正的作用只是让它在生成计划时想一下并不需要留在状态里传给下游。1.2 编排决定顺序、并行与依赖分解完之后要决定的事情很具体哪些步骤能同时做哪些必须串行哪个步骤是信息瓶颈应该尽量前置。这里有个反直觉的结论——先做能证伪的步骤。举个例子任务要求分析某个数据集并给出三个结论如果你的计划里第一步是下载数据、最后一步才是校验数据完整性那很可能你花了八成预算做完分析才发现数据字段缺失。把确认输入可用放到最前面哪怕它看起来不像个正经步骤也能省下大量返工。我现在的默认编排习惯是探路步骤确认输入、确认工具可用→ 主干步骤按依赖拓扑序→ 收敛步骤汇总与自检。这个顺序写进规划节点的系统提示里比让模型自由发挥稳定得多。1.3 动态插入与删除计划必须能被改自我规划里的自我真正体现在这一步。执行到一半发现某步的产出推翻了前面的假设系统要能把剩余计划作废重排而不是硬着头皮走完。工程上我把它处理成一次计划替换而不是补丁式修改规划节点每次被调用都输出一份完整的Plan路由层拿新计划和旧计划做差异比对把已完成的步骤标记保留、未完成的丢弃。补丁式修改模型输出删除第 3 步、在第 2 步后插入一步听起来优雅实际调试时你会被索引错位折磨到怀疑人生。1.4 验证环节不是打补丁而是图的第二个入口大部分教程把验证写成执行完之后的最后一步execute → verify → end。这个结构能跑但它浪费了验证最大的价值验证的结论应该能反向驱动规划而不只是给个通过或不通过。所以验证节点必须有两个出边通过则走向汇总不通过则回到规划节点并把失败详情带回去。它在图里的位置不是一个终点而是一个岔路口。后面第 5 节讲熔断本质上就是在给这个岔路口装刹车。2. 为什么状态图比链式编排更适合这种循环我最初用的是链式编排的思路一串函数依次调用谁需要重试就在内部写个 while。三个节点以内还行到了规划—执行—验证—重规划—再执行这种带环的结构链式写法会迅速变成一团互相调用的意大利面。2.1 链式编排在回退时的结构性缺陷链式编排的根本问题是控制流藏在函数体里。你在执行函数内部判断要不要重试在验证函数内部判断要不要回退这些跳转对调用方完全不可见。等到线上出问题你想知道这次任务一共重规划了几次、每次是被什么触发的只能翻散落在各处的日志。更麻烦的是状态传递。链式写法通常靠函数返回值往下传一旦出现回退回退到的那一步需要的是很久以前的状态你就得手动保存快照或者重算。这段代码会随着需求增长而腐烂通常三周后就没人敢改了。2.2 状态、节点、边三个概念落到本项目上的映射LangGraph 的心智模型很简单一个共享的状态State若干个只读状态、返回增量的节点Node以及决定下一步去哪个节点的边Edge包括条件边。落到我们这个项目上映射关系是这样图元素本项目中的对应物职责边界State目标、计划、游标、观察记录、验证结论、预算计数唯一事实来源节点不得私自持有跨轮状态plan 节点大模型按 schema 输出Plan只产出计划不执行、不判断对错execute 节点取出游标指向的步骤调用工具只执行一步返回观察结果verify 节点三层校验输出结构化 verdict只判断不修改计划条件边根据 verdict 与预算决定去向无业务逻辑纯路由这个划分有一条铁律判断和执行必须分属不同节点。让执行节点自己说我做完了等于让考生自己判卷。分开之后验证节点拿到的是执行节点留下的原始观察可以独立复核。2.3 用条件边把重规划变成一等公民条件边是整个设计里我最喜欢的部分因为它把下一步去哪从一个隐含的控制流变成了一张显式的路由表。你会被迫在图构建阶段就把所有分支列全漏掉哪条编译时就能看出来。from langgraph.graph import StateGraph, START, END def route_after_verify(state: AgentState) - str: v state[verdict] if v[status] pass: return finalize if state[budget][replans] state[budget][max_replans]: return give_up # 预算耗尽走降级路径 if v[status] blocked: return ask_human # 缺外部信息挂起等人 return plan # 默认回到规划 g StateGraph(AgentState) g.add_node(plan, plan_node) g.add_node(execute, execute_node) g.add_node(verify, verify_node) g.add_node(finalize, finalize_node) g.add_node(give_up, give_up_node) g.add_edge(START, plan) g.add_edge(plan, execute) g.add_conditional_edges(execute, route_after_execute, {verify: verify, execute: execute}) g.add_conditional_edges(verify, route_after_verify, {plan: plan, finalize: finalize, give_up: give_up, ask_human: ask_human}) g.add_edge(finalize, END) g.add_edge(give_up, END)注意execute后面那个条件边它允许在执行器内部连续走多步游标没走完就继续而不是每执行一步就回去验证一次。验证是昂贵的操作逐步骤验证会让成本和延迟翻好几倍。我的做法是按依赖层验证——一个拓扑层里的步骤全执行完再统一验证一次。还有个小细节give_up一定要有而且它必须产出一个诚实的部分结果而不是抛异常了事。用户拿到我完成了三件事剩下两件因为某原因没做成这是中间产出体验远好于一个红叉。3. 状态设计与节点职责整篇文章里最容易被低估的部分如果你只从这篇里带走一样东西我希望是这一节。我见过太多 Agent 项目模型能力没问题工具也没问题就是状态设计得像个杂物间跑到第五轮就开始行为诡异。八成的原因都能在 State 里找到。3.1 State 里该放什么绝对不该放什么State 的定位是跨轮次必须共享的最小事实集。判断标准很简单这个东西下一轮还要不要用要就进 State不要就是节点内部的局部变量。该放的目标原文、当前计划、执行游标、已产生的观察结构化、最近一次验证结论、预算计数器、重规划次数。不该放的完整的模型对话历史放消息列表里别塞 State、工具的原始响应体只留摘要和关键字段、临时拼装的 prompt 字符串、任何可以在节点里一次算出来的派生值。把完整对话历史塞进 State 是最常见的错误。有人觉得反正模型都要看结果每轮状态快照都是几万 token检查点存储迅速膨胀恢复时加载一次要好几秒。正确的做法是只保留一份压缩后的事实摘要原始消息走单独的存储。3.2 Reducer 的选择覆盖、追加还是合并LangGraph 的 State 字段默认是覆盖语义。要用追加得显式给 reducer。这个选择直接决定了你会不会遇到观察记录莫名其妙变多的问题。import operator from typing import Annotated, TypedDict class AgentState(TypedDict): goal: str plan: dict # 覆盖每次规划都是新计划 cursor: int # 覆盖 observations: Annotated[list[dict], operator.add] # 追加 verdict: dict | None # 覆盖只关心最近一次 budget: dict # 覆盖但节点内要整体重建 plan_fingerprints: Annotated[list[str], operator.add] # 追加用于查重observations用追加是对的因为历史观察不该丢。但这里埋着一个坑如果某个节点被重试它的观察会被追加两次。后面第 6 节我会讲怎么用步骤 id 去重。budget用覆盖有个陷阱——节点返回{budget: {replans: n1}}会把整个字典替换掉max_replans直接丢了。要么整体重建字典要么给它也写个合并 reducer。我见过至少三个项目栽在这上面症状是预算限制永远不生效。def merge_budget(left: dict, right: dict) - dict: return {**left, **right} # State 里改成 budget: Annotated[dict, merge_budget]3.3 规划节点的输出契约必须硬约束规划节点我坚持用结构化输出而不是解析自然语言。LangGraph 本身不管你用什么方式但我的做法是把Plan的 JSON schema 直接喂给模型的结构化输出接口解析失败就重试一次两次失败直接走give_up。硬约束带来的好处是下游全部可以放心地用字段访问不需要写防御性的if steps in plan。而且 schema 里的description本身就是最好的提示词——你在字段说明里写清楚禁止写尽量充分比在系统提示里喊十遍都管用。规划节点的提示里我固定放三块内容目标原文、当前可用工具清单含每个工具的一句话说明和参数 schema、上一轮验证给出的失败详情如果是重规划。不放执行历史全文只放失败详情。原因很简单历史全文会让模型倾向于修补上一版计划而不是重新思考只给它失败原因它更容易跳出原来的框架。3.4 执行节点把一步做成一次可复现的调用执行节点的职责边界要收得很紧根据游标取一步解析参数调工具把结果规整成统一格式写进观察记录游标推进。就这些。参数解析是这里最容易出事的地方。模型给的args_hint经常是大概意思比如工具要求date是 ISO 格式它给你下周三。我的处理是让执行节点带一个轻量的参数修复步骤先用 schema 校验不通过就把校验错误原样返回给模型让它重写一次参数仍然不通过就标记这一步失败并附上原因。def execute_node(state: AgentState) - dict: step state[plan][steps][state[cursor]] try: args validate_args(step.tool, step.args_hint) except ValidationError as e: return { observations: [{ step_id: step.id, ok: False, error_type: bad_args, detail: str(e) }], cursor: state[cursor], # 游标不动让验证层决定是否重规划 } result call_tool(step.tool, args) return { observations: [{ step_id: step.id, ok: True, summary: summarize(result), raw_ref: store(result), }], cursor: state[cursor] 1, }注意raw_ref原始结果存外部State 里只留引用和摘要。这个习惯在长任务里能救命尤其是某一步返回了几百行 JSON 的时候。摘要怎么生成我的经验是让工具侧直接产出摘要字段比事后让模型总结更便宜也更准。4. 验证器怎么写才不是走过场验证节点是这套架构里最容易被做成摆设的地方。我见过的典型写法是把执行结果丢给模型问一句这个结果是否满足要求模型说是的满足流程结束。这种验证的通过率接近 100%等于没有。要让验证真正起作用核心思路是分层和找反例。4.1 三层校验结构、事实、目标我把校验拆成三层从便宜到贵前一层不过就不进下一层。这样大部分明显的问题在几十毫秒内就被拦下来了。层级检查内容实现方式成本L1 结构层字段完整、类型正确、数值范围、必填项Pydantic / JSON Schema极低纯代码L2 事实层断言是否有来源、引用是否可回读、数值是否自洽代码 工具回读低到中L3 目标层是否达成原始目标、每条 done_when 是否满足大模型 结构化 rubric中到高L2 是我最看重也最容易被跳过的一层。它的做法是对结果里的每一条关键断言要求带一个来源标识工具返回的哪一段、文件的哪一行然后代码去回读这个来源确认内容对得上。对不上就判事实层失败。这一层能拦下大量模型把数字记错了引用了不存在的段落的问题而这些恰恰是 LLM judge 最容易放过的东西。L3 用模型判但判的方式要改造。我不让它直接回答通过还是不通过而是给它一份 rubric逐条列出done_when要求它对每一条输出三态结论satisfied/violated/insufficient_evidence。三态比二态有用得多因为证据不足是个独立的失败模式它应该触发的是补做而不是重规划。4.2 拿模型当裁判的三个坑第一个坑是评分通胀。你让它打 1 到 10 分它几乎永远给 7 到 9 分。解决办法是换离散档位只允许pass / partial / fail三选一并且在提示里明确partial 表示部分判据未满足必须列出未满足项。第二个坑是自我偏袒。同一个模型既做执行又做裁判它会倾向于认可自己刚生成的内容。缓解办法有三个验证提示里不给它看这是你生成的这类上下文温度调到 0以及最有效的——在给结论之前强制它先写这段内容可能错在哪里列出至少两条质疑再下判断。这个先找反例的强制步骤实测能把误判率压下去一大截。第三个坑是位置偏好。如果你做的是 A/B 对比式评估模型会偏向先出现的那个。解法很简单两个顺序各评一次结论不一致就判insufficient_evidence。def verify_goal(goal: str, result: str, rubric: list[str]) - dict: prompt f目标{goal} 待检结果{result} 按下面的判据逐条判断每条给出 satisfied / violated / insufficient_evidence {chr(10).join(f- {r} for r in rubric)} 下结论前先列出这段结果中至少两条可能存在的问题或不足。 输出 JSON{{concerns: [...], items: [{{criterion:..., status:...}}], status: pass|partial|fail}} return call_llm_json(prompt, temperature0)4.3 失败反馈怎么回灌给规划器验证不通过之后最忌讳的是把整段 verdict 原样塞回规划节点。里面有大量已经满足的判据会稀释掉真正的问题。我的做法是构造一个精简的失败包只包含三样未满足的判据原文、支撑该判断的证据片段、以及一个可选的方向建议。方向建议也限长一句话不给模型发挥空间。def build_feedback(verdict: dict) - dict: return { failed: [ {criterion: it[criterion], evidence: it.get(evidence, )} for it in verdict[items] if it[status] ! satisfied ], hint: verdict.get(suggestion, )[:120], }这个精简包会被拼进规划节点的提示。实测下来比给全文的情况下重规划成功率高出不少而且规划输出的 token 数也明显更少——因为它接收的噪音少了。5. 循环控制怎么让 Agent 知道该停带环的图最大的风险不是跑得慢是它永远不停。LangGraph 有recursion_limit这个兜底但那玩意儿是防止你把进程跑死的最后一道闸触发它意味着这次任务已经彻底失败了你不该指望它来做正常的收敛控制。5.1 三重熔断步数、重规划次数、收敛判据我一般设三道互相独立的闸任何一道触发都走降级路径。第一道是单次任务步数上限。这个按任务的正常路径长度乘 2 到 3 来定比如正常 5 步能完成就设 15。超过说明它在原地打转。第二道是重规划次数上限。这个比步数更重要因为它直接对应成本。我的经验值是 2 到 3 次。超过 3 次还解决不了的问题绝大多数不是再想一次就能想到而是信息缺失或者工具能力不够继续重规划只是烧钱。第三道是收敛判据连续两次验证的失败判据集合如果没有缩小直接停。实现上就是比对两次 verdict 里failed的 criterion 集合如果新集合不是旧集合的真子集排除掉换了别的问题的情况判定为不收敛。def not_converging(prev: set[str], cur: set[str]) - bool: if not prev: return False return len(cur - prev) 0 and len(cur) len(prev) def route_after_verify(state: AgentState) - str: b state[budget] v state[verdict] if v[status] pass: return finalize if b[replans] b[max_replans] or b[steps] b[max_steps]: return give_up if not_converging(b.get(last_failed, set()), current_failed(v)): return give_up return plan三道闸的触发原因要写进最终输出。用户看到任务未完成连续两次重规划后失败判据没有减少比看到一个空结果要好得多至少他知道该往哪个方向补信息。5.2 计划指纹与重复检测熔断之外还有一类更难缠的情况它在两个方案之间来回横跳。方案 A 验证不过重规划出方案 BB 也不过又规划回 A如此往复。因为每次的失败判据确实变小了一点点收敛判据拦不住它。对付这个用计划指纹。把计划里每一步的intent和tool做归一化去空白、统一小写、去掉标点后排序拼起来算个哈希。每次规划产出新计划就查一下历史指纹列表命中就说明它在绕圈。import hashlib def fingerprint(plan: dict) - str: parts sorted( f{norm(s[intent])}|{s.get(tool) or } for s in plan[steps] ) return hashlib.sha1(\n.join(parts).encode()).hexdigest()[:16]注意这里的norm和sorted两个操作不能省。我第一版直接对原始 JSON 序列化取哈希结果模型每次把步骤顺序换一下、措辞改个同义词哈希就变了查重形同虚设。归一化加排序之后才算真正抓住了是不是同一个计划。命中重复指纹时的处理要分情况如果当前预算还充裕我倾向让规划节点带着上一版计划已被证伪必须换一种拆解思路的强提示再试一次如果预算不多了直接降级。5.3 人工介入点与断点恢复有些动作天然需要人点头——删除数据、发外部请求、覆盖已有文件。这类节点前面挂一个中断点图跑到那儿会停下来把当前完整状态交出去等人补一个决定再继续。这个能力依赖检查点。给图编译时挂上 checkpointer每次状态变更都会落一份快照用thread_id区分不同任务。好处不只是人工介入调试时也极香线上某次任务跑歪了你拿到thread_id就能把整条状态演变历史回放出来看每一步的计划、观察、验证结论分别是什么。代价是存储。前面说过别把原始响应塞进 State很大程度就是为了让快照足够小。我一般把快照里超过一定长度的字段替换成外部引用恢复时再按引用取回。6. 实测中踩到的坑与排查链路以下四个问题都是我真金白银撞过的排查过程我尽量按当时的顺序写方便你对照自己的现象。6.1 无限重规划从日志定位到指纹冲突现象某类任务的重规划次数稳定打满上限最后总是走降级路径。日志里看着每次计划都不太一样但人工一看就知道是同一个思路换了说法。排查第一步是打时间线。我把每次规划的指纹、步数、失败判据数量打在同一行很快看出规律指纹每次都不同但失败判据集合几乎一样。这说明它确实在换措辞没换思路。第二步是去看规划节点的输入。发现问题了——我把上一版的完整计划也塞进了提示里本意是让它参考实际效果是让它照着改。模型看到旧计划几乎必然会在旧结构上做微调。修复很直接把旧计划从提示里拿掉只保留失败详情并在系统提示里加一句不要复用之前的拆解方式如果上一版失败原因是 X考虑换一类方法。改完重规划的成功率肉眼可见地上升。顺带把指纹的归一化补上了现在同一思路换措辞会被识别出来直接触发必须换思路的强提示。6.2 上下文爆炸导致的失忆现象任务跑到第八、九步之后模型开始忘记最早的目标约束或者把前面已经确认过的事实重新推翻。排查时我先量了一下每轮送给规划节点的 token 数发现随着observations增长输入长度线性上升到第九步已经接近上下文窗口的七成。模型在这么长的输入里对开头部分的注意力确实会衰减。修复分三步。第一观察记录不再全量注入改成结构化摘要 最近三条完整观察 早期观察的一句话要点。要点由执行节点在写入时就生成好不额外调模型。第二目标原文单独放在提示的最开头和最末尾各出现一次——这个土办法对长上下文的注意力分配意外有效。第三给观察记录的摘要字段加长度上限超了就截断并标记。改完之后单轮输入 token 稳定在一个区间内不再随步数增长。任务成功率在长任务上提升最明显。6.3 验证器和规划器互相放水现象验证通过率异常高但抽样人工检查发现不少结果其实不达标。根因有两个。一是验证提示里我写了请判断以下结果是否满足目标这种开放式问法会让模型倾向于找能满足的解释。二是规划和验证用的是同一个模型、同一个温度风格高度一致验证器很难对自己人开火。修复验证改成逐条 rubric 三态判断强制先列质疑温度固定 0规划节点温度设 0.3 保留一点发散性。另外加了一条硬规则——L2 事实层只要有任何一条断言的来源回读失败无论 L3 怎么判整体直接判 fail。这条规则拦下的问题比 L3 多得多。6.4 状态被就地修改引发的诡异 bug现象同一个步骤的观察在列表里出现了两次导致验证时把重复内容当成多个来源 corroborate误判为事实成立。根因很隐蔽执行节点里我对传进来的 list 直接做了append而 LangGraph 的追加 reducer 又会把节点返回的增量合并进去等于写了两遍。这类 bug 的排查方法是在节点入口打印id(state[observations])看不同轮次是不是同一个对象。如果节点在就地修改共享状态这个 id 会一直不变。修复是养成习惯——节点内永远不修改传入的状态只构造新对象返回。另外在合并观察时按step_id去重作为第二道保险。def dedup_observations(obs: list[dict]) - list[dict]: seen, out set(), [] for o in obs: key (o.get(step_id), o.get(ok)) if key in seen: continue seen.add(key) out.append(o) return out7. 可观测性与评测怎么证明它真的变好了改完之后最危险的事情是感觉变好了。Agent 系统的行为方差很大凭几次手动测试得出的结论基本不可信。必须有一套自己的度量。7.1 每个节点埋什么指标我的原则是每个节点至少埋三样耗时、token 消耗、以及一个能反映该节点质量的业务指标。节点通用指标业务指标plan耗时、输入输出 token计划步数、格式校验失败率、指纹重复率execute耗时、工具调用次数参数校验失败率、单步重试率verify耗时、judge 调用次数各层失败占比、partial 占比整图总耗时、总 token重规划次数、熔断原因分布、最终状态分布熔断原因分布这个指标特别值得盯。如果max_replans触发占比持续上升说明任务难度分布变了或者某个工具的稳定性下降了。如果not_converging占比高通常是规划提示需要调整了。所有指标都要能按thread_id串起来这样任何一个失败任务都能一键拉出完整链路。7.2 构建一个小而刁的评测集我不追求评测集大追求它刁。我的做法是分三档各准备十几条第一档单步可达用来验证基础功能没坏第二档需要两到三步且有明确依赖用来验证编排正确第三档第一次计划必然失败必须靠重规划才能完成用来验证整个闭环。第三档最有价值也最难构造——我的办法是故意在工具描述里留一个容易误解的点或者在输入里埋一个需要先探路才能发现的问题。指标上我只看四个任务成功率、平均步数、平均 token、重规划率。前两个一起看防止用疯狂重试换成功率后两个一起看防止用暴力堆上下文换效果。跑评测时固定随机种子、固定温度每组跑三遍取中位数。这个流程听起来繁琐但它是我唯一敢用来判断这次改动是不是真的有用的依据。7.3 几个调出来的经验值这些数字没有普适性但可以作为起点max_replans设 2 到 3max_steps按正常路径的 2.5 倍设规划温度 0.2 到 0.3验证温度 0观察记录保留最近 3 条完整 全量摘要失败反馈包控制在 300 token 以内L3 rubric 的判据条数别超过 6 条多了模型会开始合并同类项判断质量下降。最后再分享一个排查小技巧给每个节点加一个可选的环境变量开关打开后把该节点的输入输出完整落盘到本地文件。线上出问题时不用重新跑直接拿thread_id找回放把某个节点的输入单独拎出来喂给模型几分钟就能定位是提示问题、状态问题还是模型能力问题。这个开关在我手上救过至少三次深夜的紧急排查。
阅读完成 · 觉得有帮助？
