1. 从一句话到可靠交付中间隔着什么先说个真实场景。我接到的需求是做一个 AI 工作流系统用户输入一句话比如帮我把这份销售数据按区域汇总找出增长最快的三条线再生成周报系统要能自己理解意图规划步骤调用工具执行计算最后交付一份完整报告。听起来像是一个 ChatBot 加几个 Prompt 的事。但真正动手之后才发现从一句话到可靠交付中间隔着的不是模型能力而是工程能力。模型负责天马行空工程负责踩刹车、修轨道、确保结果稳定复现。这篇文章就围绕这个项目展开涉及意图识别、多 Agent 协作、任务规划、质量评审、LangGraph 状态恢复和接口幂等设计。核心关键词是意图识别、多 Agent、任务规划、LangGraph、幂等。适合谁来参考如果你在用 LangChain 或 LangGraph 搭 Agent 应用如果你被Agent 跑飞结果不稳定任务执行一半崩了要重新来这些问题折磨过这篇文章应该能帮上忙。我不会堆砌概念只讲我实测下来的方案、参数选择理由以及踩过的坑。2. 意图识别与任务规划别让模型自由发挥2.1 意图识别不是分类而是结构化理解很多人把意图识别做成一个分类任务——定义十几个意图类型让模型选一个。但实际业务里一句话往往包含多个意图而且意图之间还有依赖关系。比如对比本月和上月各区域销售找出下滑最严重的写个改进建议。这里至少有四个意图对比分析、筛选排序、归因分析、内容生成。如果只做一个单选题后面四个 Agent 的执行逻辑就完全没法展开。我的方案是第一层用大模型做结构化抽取输出一个包含意图标签、关键实体、约束条件的 JSON而不是单纯分类。具体实现上用 Pydantic 定义输出结构让模型严格按 schema 返回。from pydantic import BaseModel, Field from typing import List, Optional class IntentResult(BaseModel): primary_intent: str Field(description主意图必须是意图表中的一个) secondary_intents: List[str] Field(description从属意图列表可多个) entities: dict Field(description抽取的关键实体如时间范围、对象、指标) constraints: dict Field(description约束条件如排序方式、数量限制、输出格式) risk_flags: List[str] Field(description风险标记如无明确时间范围、指标缺失)模型输出后先过一层校验再进入下一步。实测下来这个设计有两点好处第一意图识别结果变成了结构化数据后续任务规划节点可以直接读取实体和约束不用重新解析原文减少误差传递。第二当模型识别不确定时它会主动暴露风险而不是猜一个答案蒙混过关。2.2 从意图到任务的拆解规划器是跑在轨道上的有了意图和实体下一步是把它们映射成具体任务序列。这一步我交给一个专门的规划 Agent但它的自由度是受限的。规划 Agent 的系统提示里有一个任务库每个任务有编号、输入参数、输出参数、前置依赖。它只能从任务库里选任务、确定顺序、填入参数不能自己发明任务。比如主意图是对比分析它就选load_data-aggregate_data-compare_periods-generate_insight这条链。这样做的原因是自由规划看起来很智能但在真实项目里你根本不知道模型会吐出什么步骤组合下游工具也没法保证被正确调用。把规划空间约束在一个封闭集合里既保留了灵活性又让后续的恢复、重试、幂等变得可管理。任务拆解的结果也是一个结构化对象包含任务列表、每个任务的输入来源前一个任务的哪个输出字段、执行模式同步还是并行。这一步的本质是把自然语言需求翻译成机器可执行的 DAG。2.3 规划校验宁可拒绝不要瞎跑任务规划完成后我会加一个校验节点做三件事依赖完整性检查每个任务的输入字段是否都有来源缺了直接拒绝。死循环检查虽然任务库不允许循环但规划器偶尔会输出重复任务要能识别。资源可行性检查比如并行任务数量是否超过限流阈值超了就改为串行。这个校验节点一开始并没有是后来加上的。原因是发生过一次事故规划器输出了一个任务依赖自己输出的配置Agent 在那个节点上反复重试了十多次直接打爆了上游数据服务的接口。所以我的建议是规划器输出之后必须有一个确定性代码来兜底不能完全信任模型输出。模型负责给出方案代码负责裁决方案是否合法。3. 多 Agent 协作各司其职但别各说各话3.1 角色划分Planner、Executor、Reviewer三分天下多 Agent 不是把一堆模型丢在一起聊天。我这里的 Agent 分三类职责边界非常清楚Agent 角色职责使用的工具输出Planner意图解析、任务规划、路线规划意图识别器、任务库结构化任务 DAGExecutor执行具体任务、调用工具、读取结果数据接口、代码执行器、检索器中间数据或结果片段Reviewer质量评审、结果验收、问题定位评审规则集、示例库通过 / 不通过 修改建议Executor 按角色还可以再拆成查询型、计算型、生成型但核心区别只在工具权限上。查询型只能读计算型只能跑代码生成型只能写文本权限隔离防止 Agent 越权操作数据。3.2 上下文传递只传必要信息不传聊天记录很多多 Agent 项目翻车翻在上下文管理上。每个 Agent 都拿到全部对话历史导致三个后果Token 消耗暴涨、无关信息干扰判断、敏感数据暴露面变大。我的做法是建立一个全局状态对象每个节点只读写自己需要的字段。比如数据加载节点只负责往状态里写data_summary字段评审节点只读final_report字段。节点之间不直接传递大段文本遇到大对象只传引用或摘要。这里补充说明一下实际项目里我用 LangGraph 来管理这个状态流。LangGraph 的状态图结构天然适合这种节点化、状态化的设计后面第 5 节我会重点讲它的恢复机制。3.3 Agent 跑飞了怎么办工具调用的硬约束多 Agent 最让人头疼的问题就是跑飞——模型不按既定步骤走自己编造工具参数或者拒绝调用工具直接编答案。我的方案是在 Executor 的工具调用层做硬校验。每个工具的参数都有 JSON SchemaAgent 在调用工具前代码先对参数做校验不符合 Schema 就拦截并返回错误信息。模型输出的 if 判断写错了工具层直接报参数类型错误Agent 就必须重新生成参数而不是将错就错。这一步是纯代码逻辑不依赖模型自律。实测下来工具调用失败率从最初的 16% 降到了 2% 左右。剩下的 2% 基本都是模型对参数理解确实有歧义需要规划器重新给出澄清参数。4. 质量评审让结果从能看变能用4.1 评审维度怎么定结果要用起来才算数评审节点最容易犯的错是把标准定成模型自评让生成结果的同一个模型给自己打分。这样评出来的结果基本都是 9 分以上一点参考价值都没有。我的评审 Agent 用的是独立模型而且是配置不同温度、不同提示词的独立模型。评审从五个维度打分结构完整性是否包含标题、摘要、正文、数据附录缺一项就不能通过数据一致性正文引用的数字是否和原始数据表一致抽查三五处逻辑连贯性结论是否由前面的分析推导而来有没有跳跃格式合规性是否符合交付模板比如 Markdown 结构、表格字段可执行性给出的建议是否具体到能直接落地还是正确的废话每个维度都有评分规则和失败示例。评审输出不是简单一个通过/不通过而是一份结构化报告包含每个维度的分数、问题列表、建议修改位置。4.2 评审失败后的迭代策略有限重试不能死循环刚开始时我把评审不通过的节点直接送回生成节点重新生成结果出现了一个经典问题生成节点重新生成的内容和上次几乎一样评审又不通过循环往复白白消耗大量 Token。后来我加了两条限制第一评审不通过时必须把哪里不通过、为什么、期望改成什么样这三条信息结构化回传给生成节点而不是只回传一句请改进。生成节点拿到具体的修改要求后才能有针对性地改。第二设置最大重试次数。我目前设为 2 次也就是说同一个生成节点最多被评审打回 2 次。第 3 次评审还不通过直接走人工兜底流程——把结果标记为需人工审核而不是继续烧钱让模型反复试。核心原则是AI 系统要承认自己能力的边界评审的重试可以暴露问题但不能因为反复重试把线上流程堵死。4.3 评审环节的成本控制评审 Agent 挂的是更强的模型成本比生成节点高。为了控成本我做了两个优化一是在评审之前先跑一个规则过滤器。用正则和代码检查格式合规性、数字一致性这些不需要模型判断的东西先拦下来。规则过滤器能发现的问题不让大模型重复判断能省下不少 Token。二是评审时只输入结果 评分标准 原始数据摘要不输入完整对话历史。评审需要的上下文其实很短完整对话历史里 99% 的信息对评审没有帮助。5. LangGraph 恢复机制与幂等设计可靠交付的底线5.1 为什么运行时恢复是刚需任何一个真实系统都会遇到进程崩溃、网络超时、数据库连接断开。Agent 应用更脆弱因为一次任务要跑十几个节点每个节点还可能调用外部服务任何一环断了整个任务就前功尽弃。没有恢复机制时任务失败只能从头开始。从头开始意味着又要重新调用数据接口、重新跑计算、重新生成内容。我遇到过一个真实案例上游数据服务在任务跑到第 8 个节点时超时了重试只能从节点 1 开始而节点 1 要扫描百万行数据整整跑了 40 分钟。这种体验没人能接受。LangGraph 的恢复机制解决的就是这个问题把每一步执行结果持久化崩溃后从最近的成功节点继续而不是从零开始。5.2 LangGraph 检查点持久化与恢复的实现LangGraph 自带检查点机制核心是checkpointer。我把检查点存储接入了 Redis原因是任务状态不仅要让应用进程自己在崩溃后能读到还需要让多个无状态服务实例共享状态。使用 LangGraph 的方式很简单在编译图的时候传入一个 Checkpointer 实例。每次节点执行完图的状态会自动被持久化。LangGraph 官方支持内存、SQLite、PostgreSQL 等多种存储我这里用 Redis 主要是考虑到部署环境的统一性。from langgraph.graph import StateGraph from langgraph.checkpoint.redis import RedisSaver checkpointer RedisSaver(hostlocalhost, port6379, db0) graph builder.compile(checkpointercheckpointer) config {configurable: {thread_id: task-20240512-001}}关键点在于thread_id。这个 ID 是任务的唯一标识LangGraph 靠它来关联同一个任务的历史状态。恢复执行时传入同样的thread_idLangGraph 会从持久化的状态里恢复中断的节点而不是重新构建一个新的状态。恢复后从哪里开始执行LangGraph 的机制是能恢复的节点直接返回缓存结果跳过不执行中断的节点重新执行下游节点继续往下走。这要求每个节点的执行结果必须是确定性的、可缓存的这正好引出了幂等设计这个话题。5.3 幂等设计让重复执行没有副作用幂等是什么简单说同一个操作执行一次和执行一百次对外部系统产生的结果是一致的。幂等是恢复机制的前置条件。如果节点在恢复后被重复执行而重复执行会产生副作用那系统就完了。5.3.1 接口幂等任务 ID 去重表外部接口调用是副作用的主要来源。我在所有 Agent 调用的外部接口上都加了幂等控制。实现方式很简单客户端在请求头里带一个Idempotency-Key服务端根据这个 key 判断请求是否已经处理过。处理过的直接返回缓存结果没处理过的执行并缓存。POST /api/v1/query-data Idempotency-Key: task-20240512-001-node-03-retry-02 X-Task-ID: task-20240512-001 X-Request-Hash: d41d8cd98f00b204e9800998ecf8427e服务端逻辑分三步先查去重表key 存在则直接返回上一次的响应体。不存在则执行正常逻辑结果写入去重表状态标记为 completed。若执行过程中出现异常写入去重表的记录标记为 failed允许下次重试。去重表我用 Redis 实现key 用 MD5 校验请求内容防止同一个 key 带上不同参数导致缓存错乱。过期时间设为 24 小时长任务也能覆盖。5.3.2 任务级幂等节点重复执行不产生脏数据接口幂等解决的是外部系统的重复调用问题。但任务内部还有一个坑Agent 节点执行时如果输入数据没变输出也应该保持一致。我之前遇到过的场景是生成报告节点被恢复机制重新执行了一次但这一次模型生成的报告内容跟第一次不一样——两篇报告都有价值但和后续的评审、结论都不一致导致整个任务的输出状态混乱。解决办法是在节点入口做状态级幂等判断每个节点的输出都带一个output_hash保存进状态。节点执行前先检查输入是否变化如果输入没变且已有输出直接返回缓存的输出结果。node_output_key f{task_id}:{node_id}:{input_hash} cached redis.get(node_output_key) if cached: return deserialize(cached)这个输入哈希 - 输出缓存的模式是保证恢复后整个工作流状态一致性的关键。5.4 幂等和恢复的配合它们是一套组合拳单独设计幂等或单独设计检查点恢复都解决不了问题。幂等的目的是保证重复执行没有副作用恢复的目标是崩溃后能继续执行。两者结合起来才能做到 at-most-once 语义 变成 effectively-once。实际执行时恢复机制从检查点恢复状态发现节点 3 已经成功执行过就跳过节点 3 直接执行节点 4。但中途可能遇到网络波动节点 6 其实已经执行了只是响应超时没有把结果写回状态此时恢复机制会让节点 6 再执行一次。如果节点 6 没有幂等保护数据就被写了两遍后面的数据汇总就全错了。用代码来表达,我的工作流节点统一封装了一层安全执行器逻辑是def safe_execute(node_name, execute_fn, task_id, input_data): # 1. 检查状态如果该节点已经有成功输出且输入数据未变化直接返回缓存 input_hash compute_hash(input_data) output_cache_key foutput:{task_id}:{node_name}:{input_hash} cached_output redis.get(output_cache_key) if cached_output: return deserialize(cached_output) # 2. 执行真实逻辑 result execute_fn(input_data) # 3. 写入输出缓存这一步保证重试时能拿到一致性结果 output_hash compute_hash(result) redis.set(output_cache_key, serialize(result), ex86400) # 4. 记录执行痕迹供评审和排障使用 redis.rpush(faudit:{task_id}, f{node_name}:{output_hash}) return result有了这个统一封装任何节点在恢复后都不会造成双重副作用整个工作流的状态就是可追溯、可验证的。5.5 关于 LangChain 与 LangGraph 的关系很多人会问 LangChain 和 LangGraph 到底什么关系。个人理解是这样的LangChain 是一套工具集合提供了大量 Tool 和链式调用的封装适合快速搭建串行流水线LangGraph 则是在其基础上提供了有向图的工作流编排核心差异就是状态管理、条件跳转和持久化检查点。标准 LangChain 的 LCELLangChain Expression Language适合固定链路的场景一旦链路分支复杂、需要人工干预、需要可暂停可恢复LCEL 就力不从心了。LangGraph 可以理解为带状态的图执行引擎它把节点、边、状态、检查点这四个概念作为一等公民天然适合我在这个项目里需要的多 Agent 编排和恢复机制。6. 常见问题与排查技巧实录6.1 意图识别被自由发挥改写现象用户输入帮我算一下每个月的增长率意图识别返回的主意图是generate_report实体里却出现了环比这类原文不存在的概念。这种自由发挥会把后续的执行带到完全错误的轨道上。排查思路打开意图识别节点的完整输出对比实体列表和原文。如果模型在抽取时补充了原文没有的信息多半是提示词里没有明确只能基于原文抽取不得推理补充的约束。解决办法在意图识别提示词里加了两条规则一是所有实体必须能在原文中找到对应词找不到就用原文原词二是如果原文没有时间范围不允许猜测本月或上月必须标记为 missing。另外我在解析阶段还会做一个简单的词汇对齐校验模型抽出的实体如果不在原文分词集合里就触发补充提醒。6.2 评审反复不通过的死循环现象生成节点生成报告评审节点不通过生成节点修改后再提交再被拒绝反复消耗大量 Token。排查思路查看评审节点的输出详情确认评审不通过的原因是否相同。如果两次都是同一个原因说明生成节点根本没有理解修改意见或者根本没有能力改到位。解决办法把评审意见从自然语言改为结构化问题列表每个问题带位置信息和建议修改方向。同时设置重试上限为 2 次超过直接转人工。另外还加了一个改良第一次评审不通过时生成节点可以用补丁模式只修改问题位置而不是重新生成全文这样既节省 Token又不容易破坏原本写得好的内容。6.3 恢复之后出现重复副作用现象原来已经执行成功的数据写入节点在恢复后又被执行了一次导致数据重复写入。排查思路先看检查点状态里该节点的执行标记再看输出缓存是否存在。如果状态里标记成功但输出缓存是空的说明该节点执行成功但没有持久化结果这是缓存写入时机的问题。解决办法把步骤拆成先缓存后返回节点执行成功的第一时间就写输出缓存然后再返回给调用方。这个顺序反过来就会造成状态和缓存不一致也是我踩过的最深的坑之一。另外所有写操作的工具调用必须走幂等接口用任务 ID 外加节点标识做去重键。6.4 任务一直重试恢复了还是失败现象任务执行到某个节点时崩溃恢复后还是在这个节点崩溃往复循环任务卡死。排查思路这类问题通常是死节点问题——某个外部服务不稳定或代码逻辑有 bug导致该节点每次都会失败。检查该节点在 Redis 里的执行记录如果失败次数反复增长且失败原因一致基本可以断定是死节点。解决办法加一个断路器逻辑。每个节点连续失败 3 次后自动熔断不再重试直接转入人工处理流程。这个策略看起来很基础但能避免系统在故障状态下继续浪费资源。同时为每个任务节点加上前置条件检查如果上游数据为空或格式不对提前终止而不是带着问题继续往下跑。6.5 幂等键应该选什么这个问题是实践中最高频的疑问。幂等键不是随便一个随机字符串它的选择直接决定了幂等的正确性。幂等键的三个层级层级作用范围推荐键说明任务级整个任务唯一的键业务单号 / 任务标识比如销售分析任务对应task-sales-Q1-20240512节点级某节点执行用任务 ID 节点名节点重试时复用参数级某次具体请求参数 MD5 摘要请求内容变化则视为新请求参数级幂等键有一个特别注意点请求中的时间戳字段会导致同样的业务请求生成不同的幂等键。所以我计算 MD5 时会剔除时间戳一类的非业务字段否则幂等形同虚设。7. 从一句话到可靠交付的完整链路复盘走完这一整套方案之后我最大的体会是AI 项目的大部分工作量不在模型怎么调而在模型外面那层工程骨架。意图识别是入口它的质量决定了后续所有环节的上限。多 Agent 是分工但分工的前提是明确的职责边界和硬约束。任务规划是调度需要把模型自由度约束在封闭集合内。质量评审是底线用独立视角给模型的输出踩一脚刹车。LangGraph 检查点是心脏让执行过程在故障时不至于断崖式归零。幂等设计是免疫系统让重复和重试不再造成二次伤害。这套链路跑通之后一个任务的交付时间从原来平均 30 分钟含人工介入压缩到 8 分钟失败率从 21% 降到 4% 左右。4% 的失败任务里绝大多数也已经能自动定位到具体的失败节点和原因真正需要人工从头干预的情况很少了。从一句话到可靠交付模型是很容易迭代的难的是把想说清楚要做的事和确实做到了的事这两者之间的差距用工程手段填平。如果你也在搭类似的 Agent 工作流我建议的顺序是先做幂等再做恢复最后才优化意图识别。前两者决定了系统的下限意图识别决定了上限。下限不稳上限再高都没意义。
从一句话到可靠交付：多Agent协作与LangGraph幂等设计实战
1. 从一句话到可靠交付中间隔着什么先说个真实场景。我接到的需求是做一个 AI 工作流系统用户输入一句话比如帮我把这份销售数据按区域汇总找出增长最快的三条线再生成周报系统要能自己理解意图规划步骤调用工具执行计算最后交付一份完整报告。听起来像是一个 ChatBot 加几个 Prompt 的事。但真正动手之后才发现从一句话到可靠交付中间隔着的不是模型能力而是工程能力。模型负责天马行空工程负责踩刹车、修轨道、确保结果稳定复现。这篇文章就围绕这个项目展开涉及意图识别、多 Agent 协作、任务规划、质量评审、LangGraph 状态恢复和接口幂等设计。核心关键词是意图识别、多 Agent、任务规划、LangGraph、幂等。适合谁来参考如果你在用 LangChain 或 LangGraph 搭 Agent 应用如果你被Agent 跑飞结果不稳定任务执行一半崩了要重新来这些问题折磨过这篇文章应该能帮上忙。我不会堆砌概念只讲我实测下来的方案、参数选择理由以及踩过的坑。2. 意图识别与任务规划别让模型自由发挥2.1 意图识别不是分类而是结构化理解很多人把意图识别做成一个分类任务——定义十几个意图类型让模型选一个。但实际业务里一句话往往包含多个意图而且意图之间还有依赖关系。比如对比本月和上月各区域销售找出下滑最严重的写个改进建议。这里至少有四个意图对比分析、筛选排序、归因分析、内容生成。如果只做一个单选题后面四个 Agent 的执行逻辑就完全没法展开。我的方案是第一层用大模型做结构化抽取输出一个包含意图标签、关键实体、约束条件的 JSON而不是单纯分类。具体实现上用 Pydantic 定义输出结构让模型严格按 schema 返回。from pydantic import BaseModel, Field from typing import List, Optional class IntentResult(BaseModel): primary_intent: str Field(description主意图必须是意图表中的一个) secondary_intents: List[str] Field(description从属意图列表可多个) entities: dict Field(description抽取的关键实体如时间范围、对象、指标) constraints: dict Field(description约束条件如排序方式、数量限制、输出格式) risk_flags: List[str] Field(description风险标记如无明确时间范围、指标缺失)模型输出后先过一层校验再进入下一步。实测下来这个设计有两点好处第一意图识别结果变成了结构化数据后续任务规划节点可以直接读取实体和约束不用重新解析原文减少误差传递。第二当模型识别不确定时它会主动暴露风险而不是猜一个答案蒙混过关。2.2 从意图到任务的拆解规划器是跑在轨道上的有了意图和实体下一步是把它们映射成具体任务序列。这一步我交给一个专门的规划 Agent但它的自由度是受限的。规划 Agent 的系统提示里有一个任务库每个任务有编号、输入参数、输出参数、前置依赖。它只能从任务库里选任务、确定顺序、填入参数不能自己发明任务。比如主意图是对比分析它就选load_data-aggregate_data-compare_periods-generate_insight这条链。这样做的原因是自由规划看起来很智能但在真实项目里你根本不知道模型会吐出什么步骤组合下游工具也没法保证被正确调用。把规划空间约束在一个封闭集合里既保留了灵活性又让后续的恢复、重试、幂等变得可管理。任务拆解的结果也是一个结构化对象包含任务列表、每个任务的输入来源前一个任务的哪个输出字段、执行模式同步还是并行。这一步的本质是把自然语言需求翻译成机器可执行的 DAG。2.3 规划校验宁可拒绝不要瞎跑任务规划完成后我会加一个校验节点做三件事依赖完整性检查每个任务的输入字段是否都有来源缺了直接拒绝。死循环检查虽然任务库不允许循环但规划器偶尔会输出重复任务要能识别。资源可行性检查比如并行任务数量是否超过限流阈值超了就改为串行。这个校验节点一开始并没有是后来加上的。原因是发生过一次事故规划器输出了一个任务依赖自己输出的配置Agent 在那个节点上反复重试了十多次直接打爆了上游数据服务的接口。所以我的建议是规划器输出之后必须有一个确定性代码来兜底不能完全信任模型输出。模型负责给出方案代码负责裁决方案是否合法。3. 多 Agent 协作各司其职但别各说各话3.1 角色划分Planner、Executor、Reviewer三分天下多 Agent 不是把一堆模型丢在一起聊天。我这里的 Agent 分三类职责边界非常清楚Agent 角色职责使用的工具输出Planner意图解析、任务规划、路线规划意图识别器、任务库结构化任务 DAGExecutor执行具体任务、调用工具、读取结果数据接口、代码执行器、检索器中间数据或结果片段Reviewer质量评审、结果验收、问题定位评审规则集、示例库通过 / 不通过 修改建议Executor 按角色还可以再拆成查询型、计算型、生成型但核心区别只在工具权限上。查询型只能读计算型只能跑代码生成型只能写文本权限隔离防止 Agent 越权操作数据。3.2 上下文传递只传必要信息不传聊天记录很多多 Agent 项目翻车翻在上下文管理上。每个 Agent 都拿到全部对话历史导致三个后果Token 消耗暴涨、无关信息干扰判断、敏感数据暴露面变大。我的做法是建立一个全局状态对象每个节点只读写自己需要的字段。比如数据加载节点只负责往状态里写data_summary字段评审节点只读final_report字段。节点之间不直接传递大段文本遇到大对象只传引用或摘要。这里补充说明一下实际项目里我用 LangGraph 来管理这个状态流。LangGraph 的状态图结构天然适合这种节点化、状态化的设计后面第 5 节我会重点讲它的恢复机制。3.3 Agent 跑飞了怎么办工具调用的硬约束多 Agent 最让人头疼的问题就是跑飞——模型不按既定步骤走自己编造工具参数或者拒绝调用工具直接编答案。我的方案是在 Executor 的工具调用层做硬校验。每个工具的参数都有 JSON SchemaAgent 在调用工具前代码先对参数做校验不符合 Schema 就拦截并返回错误信息。模型输出的 if 判断写错了工具层直接报参数类型错误Agent 就必须重新生成参数而不是将错就错。这一步是纯代码逻辑不依赖模型自律。实测下来工具调用失败率从最初的 16% 降到了 2% 左右。剩下的 2% 基本都是模型对参数理解确实有歧义需要规划器重新给出澄清参数。4. 质量评审让结果从能看变能用4.1 评审维度怎么定结果要用起来才算数评审节点最容易犯的错是把标准定成模型自评让生成结果的同一个模型给自己打分。这样评出来的结果基本都是 9 分以上一点参考价值都没有。我的评审 Agent 用的是独立模型而且是配置不同温度、不同提示词的独立模型。评审从五个维度打分结构完整性是否包含标题、摘要、正文、数据附录缺一项就不能通过数据一致性正文引用的数字是否和原始数据表一致抽查三五处逻辑连贯性结论是否由前面的分析推导而来有没有跳跃格式合规性是否符合交付模板比如 Markdown 结构、表格字段可执行性给出的建议是否具体到能直接落地还是正确的废话每个维度都有评分规则和失败示例。评审输出不是简单一个通过/不通过而是一份结构化报告包含每个维度的分数、问题列表、建议修改位置。4.2 评审失败后的迭代策略有限重试不能死循环刚开始时我把评审不通过的节点直接送回生成节点重新生成结果出现了一个经典问题生成节点重新生成的内容和上次几乎一样评审又不通过循环往复白白消耗大量 Token。后来我加了两条限制第一评审不通过时必须把哪里不通过、为什么、期望改成什么样这三条信息结构化回传给生成节点而不是只回传一句请改进。生成节点拿到具体的修改要求后才能有针对性地改。第二设置最大重试次数。我目前设为 2 次也就是说同一个生成节点最多被评审打回 2 次。第 3 次评审还不通过直接走人工兜底流程——把结果标记为需人工审核而不是继续烧钱让模型反复试。核心原则是AI 系统要承认自己能力的边界评审的重试可以暴露问题但不能因为反复重试把线上流程堵死。4.3 评审环节的成本控制评审 Agent 挂的是更强的模型成本比生成节点高。为了控成本我做了两个优化一是在评审之前先跑一个规则过滤器。用正则和代码检查格式合规性、数字一致性这些不需要模型判断的东西先拦下来。规则过滤器能发现的问题不让大模型重复判断能省下不少 Token。二是评审时只输入结果 评分标准 原始数据摘要不输入完整对话历史。评审需要的上下文其实很短完整对话历史里 99% 的信息对评审没有帮助。5. LangGraph 恢复机制与幂等设计可靠交付的底线5.1 为什么运行时恢复是刚需任何一个真实系统都会遇到进程崩溃、网络超时、数据库连接断开。Agent 应用更脆弱因为一次任务要跑十几个节点每个节点还可能调用外部服务任何一环断了整个任务就前功尽弃。没有恢复机制时任务失败只能从头开始。从头开始意味着又要重新调用数据接口、重新跑计算、重新生成内容。我遇到过一个真实案例上游数据服务在任务跑到第 8 个节点时超时了重试只能从节点 1 开始而节点 1 要扫描百万行数据整整跑了 40 分钟。这种体验没人能接受。LangGraph 的恢复机制解决的就是这个问题把每一步执行结果持久化崩溃后从最近的成功节点继续而不是从零开始。5.2 LangGraph 检查点持久化与恢复的实现LangGraph 自带检查点机制核心是checkpointer。我把检查点存储接入了 Redis原因是任务状态不仅要让应用进程自己在崩溃后能读到还需要让多个无状态服务实例共享状态。使用 LangGraph 的方式很简单在编译图的时候传入一个 Checkpointer 实例。每次节点执行完图的状态会自动被持久化。LangGraph 官方支持内存、SQLite、PostgreSQL 等多种存储我这里用 Redis 主要是考虑到部署环境的统一性。from langgraph.graph import StateGraph from langgraph.checkpoint.redis import RedisSaver checkpointer RedisSaver(hostlocalhost, port6379, db0) graph builder.compile(checkpointercheckpointer) config {configurable: {thread_id: task-20240512-001}}关键点在于thread_id。这个 ID 是任务的唯一标识LangGraph 靠它来关联同一个任务的历史状态。恢复执行时传入同样的thread_idLangGraph 会从持久化的状态里恢复中断的节点而不是重新构建一个新的状态。恢复后从哪里开始执行LangGraph 的机制是能恢复的节点直接返回缓存结果跳过不执行中断的节点重新执行下游节点继续往下走。这要求每个节点的执行结果必须是确定性的、可缓存的这正好引出了幂等设计这个话题。5.3 幂等设计让重复执行没有副作用幂等是什么简单说同一个操作执行一次和执行一百次对外部系统产生的结果是一致的。幂等是恢复机制的前置条件。如果节点在恢复后被重复执行而重复执行会产生副作用那系统就完了。5.3.1 接口幂等任务 ID 去重表外部接口调用是副作用的主要来源。我在所有 Agent 调用的外部接口上都加了幂等控制。实现方式很简单客户端在请求头里带一个Idempotency-Key服务端根据这个 key 判断请求是否已经处理过。处理过的直接返回缓存结果没处理过的执行并缓存。POST /api/v1/query-data Idempotency-Key: task-20240512-001-node-03-retry-02 X-Task-ID: task-20240512-001 X-Request-Hash: d41d8cd98f00b204e9800998ecf8427e服务端逻辑分三步先查去重表key 存在则直接返回上一次的响应体。不存在则执行正常逻辑结果写入去重表状态标记为 completed。若执行过程中出现异常写入去重表的记录标记为 failed允许下次重试。去重表我用 Redis 实现key 用 MD5 校验请求内容防止同一个 key 带上不同参数导致缓存错乱。过期时间设为 24 小时长任务也能覆盖。5.3.2 任务级幂等节点重复执行不产生脏数据接口幂等解决的是外部系统的重复调用问题。但任务内部还有一个坑Agent 节点执行时如果输入数据没变输出也应该保持一致。我之前遇到过的场景是生成报告节点被恢复机制重新执行了一次但这一次模型生成的报告内容跟第一次不一样——两篇报告都有价值但和后续的评审、结论都不一致导致整个任务的输出状态混乱。解决办法是在节点入口做状态级幂等判断每个节点的输出都带一个output_hash保存进状态。节点执行前先检查输入是否变化如果输入没变且已有输出直接返回缓存的输出结果。node_output_key f{task_id}:{node_id}:{input_hash} cached redis.get(node_output_key) if cached: return deserialize(cached)这个输入哈希 - 输出缓存的模式是保证恢复后整个工作流状态一致性的关键。5.4 幂等和恢复的配合它们是一套组合拳单独设计幂等或单独设计检查点恢复都解决不了问题。幂等的目的是保证重复执行没有副作用恢复的目标是崩溃后能继续执行。两者结合起来才能做到 at-most-once 语义 变成 effectively-once。实际执行时恢复机制从检查点恢复状态发现节点 3 已经成功执行过就跳过节点 3 直接执行节点 4。但中途可能遇到网络波动节点 6 其实已经执行了只是响应超时没有把结果写回状态此时恢复机制会让节点 6 再执行一次。如果节点 6 没有幂等保护数据就被写了两遍后面的数据汇总就全错了。用代码来表达,我的工作流节点统一封装了一层安全执行器逻辑是def safe_execute(node_name, execute_fn, task_id, input_data): # 1. 检查状态如果该节点已经有成功输出且输入数据未变化直接返回缓存 input_hash compute_hash(input_data) output_cache_key foutput:{task_id}:{node_name}:{input_hash} cached_output redis.get(output_cache_key) if cached_output: return deserialize(cached_output) # 2. 执行真实逻辑 result execute_fn(input_data) # 3. 写入输出缓存这一步保证重试时能拿到一致性结果 output_hash compute_hash(result) redis.set(output_cache_key, serialize(result), ex86400) # 4. 记录执行痕迹供评审和排障使用 redis.rpush(faudit:{task_id}, f{node_name}:{output_hash}) return result有了这个统一封装任何节点在恢复后都不会造成双重副作用整个工作流的状态就是可追溯、可验证的。5.5 关于 LangChain 与 LangGraph 的关系很多人会问 LangChain 和 LangGraph 到底什么关系。个人理解是这样的LangChain 是一套工具集合提供了大量 Tool 和链式调用的封装适合快速搭建串行流水线LangGraph 则是在其基础上提供了有向图的工作流编排核心差异就是状态管理、条件跳转和持久化检查点。标准 LangChain 的 LCELLangChain Expression Language适合固定链路的场景一旦链路分支复杂、需要人工干预、需要可暂停可恢复LCEL 就力不从心了。LangGraph 可以理解为带状态的图执行引擎它把节点、边、状态、检查点这四个概念作为一等公民天然适合我在这个项目里需要的多 Agent 编排和恢复机制。6. 常见问题与排查技巧实录6.1 意图识别被自由发挥改写现象用户输入帮我算一下每个月的增长率意图识别返回的主意图是generate_report实体里却出现了环比这类原文不存在的概念。这种自由发挥会把后续的执行带到完全错误的轨道上。排查思路打开意图识别节点的完整输出对比实体列表和原文。如果模型在抽取时补充了原文没有的信息多半是提示词里没有明确只能基于原文抽取不得推理补充的约束。解决办法在意图识别提示词里加了两条规则一是所有实体必须能在原文中找到对应词找不到就用原文原词二是如果原文没有时间范围不允许猜测本月或上月必须标记为 missing。另外我在解析阶段还会做一个简单的词汇对齐校验模型抽出的实体如果不在原文分词集合里就触发补充提醒。6.2 评审反复不通过的死循环现象生成节点生成报告评审节点不通过生成节点修改后再提交再被拒绝反复消耗大量 Token。排查思路查看评审节点的输出详情确认评审不通过的原因是否相同。如果两次都是同一个原因说明生成节点根本没有理解修改意见或者根本没有能力改到位。解决办法把评审意见从自然语言改为结构化问题列表每个问题带位置信息和建议修改方向。同时设置重试上限为 2 次超过直接转人工。另外还加了一个改良第一次评审不通过时生成节点可以用补丁模式只修改问题位置而不是重新生成全文这样既节省 Token又不容易破坏原本写得好的内容。6.3 恢复之后出现重复副作用现象原来已经执行成功的数据写入节点在恢复后又被执行了一次导致数据重复写入。排查思路先看检查点状态里该节点的执行标记再看输出缓存是否存在。如果状态里标记成功但输出缓存是空的说明该节点执行成功但没有持久化结果这是缓存写入时机的问题。解决办法把步骤拆成先缓存后返回节点执行成功的第一时间就写输出缓存然后再返回给调用方。这个顺序反过来就会造成状态和缓存不一致也是我踩过的最深的坑之一。另外所有写操作的工具调用必须走幂等接口用任务 ID 外加节点标识做去重键。6.4 任务一直重试恢复了还是失败现象任务执行到某个节点时崩溃恢复后还是在这个节点崩溃往复循环任务卡死。排查思路这类问题通常是死节点问题——某个外部服务不稳定或代码逻辑有 bug导致该节点每次都会失败。检查该节点在 Redis 里的执行记录如果失败次数反复增长且失败原因一致基本可以断定是死节点。解决办法加一个断路器逻辑。每个节点连续失败 3 次后自动熔断不再重试直接转入人工处理流程。这个策略看起来很基础但能避免系统在故障状态下继续浪费资源。同时为每个任务节点加上前置条件检查如果上游数据为空或格式不对提前终止而不是带着问题继续往下跑。6.5 幂等键应该选什么这个问题是实践中最高频的疑问。幂等键不是随便一个随机字符串它的选择直接决定了幂等的正确性。幂等键的三个层级层级作用范围推荐键说明任务级整个任务唯一的键业务单号 / 任务标识比如销售分析任务对应task-sales-Q1-20240512节点级某节点执行用任务 ID 节点名节点重试时复用参数级某次具体请求参数 MD5 摘要请求内容变化则视为新请求参数级幂等键有一个特别注意点请求中的时间戳字段会导致同样的业务请求生成不同的幂等键。所以我计算 MD5 时会剔除时间戳一类的非业务字段否则幂等形同虚设。7. 从一句话到可靠交付的完整链路复盘走完这一整套方案之后我最大的体会是AI 项目的大部分工作量不在模型怎么调而在模型外面那层工程骨架。意图识别是入口它的质量决定了后续所有环节的上限。多 Agent 是分工但分工的前提是明确的职责边界和硬约束。任务规划是调度需要把模型自由度约束在封闭集合内。质量评审是底线用独立视角给模型的输出踩一脚刹车。LangGraph 检查点是心脏让执行过程在故障时不至于断崖式归零。幂等设计是免疫系统让重复和重试不再造成二次伤害。这套链路跑通之后一个任务的交付时间从原来平均 30 分钟含人工介入压缩到 8 分钟失败率从 21% 降到 4% 左右。4% 的失败任务里绝大多数也已经能自动定位到具体的失败节点和原因真正需要人工从头干预的情况很少了。从一句话到可靠交付模型是很容易迭代的难的是把想说清楚要做的事和确实做到了的事这两者之间的差距用工程手段填平。如果你也在搭类似的 Agent 工作流我建议的顺序是先做幂等再做恢复最后才优化意图识别。前两者决定了系统的下限意图识别决定了上限。下限不稳上限再高都没意义。