先把 ReAct 那套“while True: 循环调模型”的 Agent 范式拆开看再重构一个基于反馈路由和任务图驱动的架构。这篇文章要讲的不是一个玩具 Demo而是一套能让 Agent 在大规模任务下稳定收敛、可观测、可并行的落地思路。很多人写 Agent第一步就是while True让 LLM 自己决定下一步干什么直到它觉得自己干完了才 break 出来。这个范式很顺也确实是目前多数 Agent 框架包括 ReAct 风格、AutoGPT 风格的核心循环。但我在真实项目里跑了一阵子后逐步意识到这个循环其实是整套系统里最脆弱、最不可控的部分。所以我把 Agent 的 while 循环拆掉了换了一种更像“流程图状态机”的架构。实测下来稳定性和性价比都有明显提升。这篇就把我的设计思路、代码结构、踩坑过程、以及哪些场景不建议拆一次讲清楚。1. 先聊聊 while 循环式 Agent 的问题在哪先说清楚我不是要否定 ReAct 和 AutoGPT 这类方案。它们最大的价值是证明了“让模型自己规划”这条路可行。但如果你把同一个循环放到生产环境里跑会遇到三类问题不确定的收敛、不可控的耗用、以及天然被锁死的并发能力。1.1 传统 Agent 循环长什么样最经典的实现大概是这样的伪代码while True: action agent.think(observation) if action final_answer: return observation observation agent.execute(action)模型在每轮都拿到“当前观察”决策一个动作执行后把新观察喂回去继续循环。听起来很顺。但你有没有想过模型说final_answer就真的收尾吗它可能因为遗漏信息太早退出也可能陷入同一个动作反复执行好几轮的情况。而 while 循环本身没有任何约束来防止这种事情发生——它只是一个无条件的“转圈圈”。注意这里的 while 循环不是语法层面的问题而是架构层面的问题。while True的核心缺陷不是“会无限循环”而是循环条件不携带任何领域语义——它只负责“再来一次”从不负责“为什么再来一次”。1.2 我在生产里遇到的三个痛点第一推理资源的浪费不可控。LLM 每转一圈就是一次 API 调用模型如果绕圈子你的账单也在转圈。我在开发一个基于 Rust 的 AI Agent 时早期测试里出现过同一个子任务反复执行 7 次才最终收敛的案例而第 7 次的判断和第 3 次本质上没有区别——只是换了个说法。七次调用六次是纯浪费。第二超时与失败难以定位。while 循环里所有的上下文都包在同一个堆栈中它不区分当前到底卡在哪一步。线上排查 Agent 问题时日志打出来只能看到“第 N 次调用模型”而不知道这个 Agent 当前到底在解决哪个子问题。没有结构化的任务状态就没有结构化的可观测性。第三并发能力默认被锁死。while 循环是单线程的串行叙事。模型必须在上一轮动作完成后才能思考下一轮。于是分布式架构下其他节点只能等着。扛并发我用 LLMAPI 架构做并行任务评估时发现一个 Agent 处理一个复杂任务可能要 10 轮就算资源池里有 10 个 worker也只受限于“一个任务内部的串行依赖”。1.3 问题本质循环是把“流程控制”交给模型所有 while 循环式 Agent 的本质是把流程控制完全交给了 LLM。模型说“下一步做 A”系统照做模型说“完成了”系统就结束。可 LLM 天然是概率模型它对“是否完成”的判断能力并不可靠也不具备稳定的一致性。在一个开放式长任务里你把罗马帝国的兴衰托付给一个“每轮都重新判断是否继续”的决策者结果就是耗时长、状态散、错误不可复现。所以我想换一种思路不要让模型决定“是否继续循环”而是让模型只回答“当前这一步怎么做”。把“下一步做什么”“什么时候结束”这类控制逻辑从模型的自由意志手里收回到架构代码里。2. 拆掉 while 循环的后台设计把 Agent 变成一张可反馈的任务图这套架构的核心是把 Agent 从“单串循环流程”变成一张带反馈路由的有向任务图并配上有限状态机做收敛控制。简单解释任务不再是由模型自由地“想到哪做到哪”而是在启动前先拆解成明确的子任务结构模型只负责执行子任务并产出结果架构负责判断结果能不能通过验证、该流转到哪个下游子任务。2.1 核心设计反馈驱动的任务图DAG with Feedback原来的 while 循环里任务来源是模型上一轮的输出新架构里任务是一个预先定义好的有向无环图。每个节点是一个子任务每条边定义子任务的依赖关系。Agent 不是自己“走”这个图而是被架构按依赖顺序“调度”着走这个图。这里要特别强调“带反馈路由”这几个字。纯 DAG 是单向无环的而实际执行中会出现“这步结果不达标需要重新做”的情况。我在设计里加了两类反馈边重试边节点自身执行失败回到自身重新执行但带重试次数上限。修订边下游验证节点判定产物不合格让上游产出节点重新执行同时把验证意见作为修订上下文传给上游节点。这样就同时保留了循环的“自愈能力”又不至于让模型自己无限打转。2.2 有限状态机替换“是否结束”的判断原来的while True里模型毛遂自荐说“我完成了”。新架构中每个任务节点周期性汇报状态pending→running→reviewing→done或failed。任何一个节点从running进入reviewing后验证逻辑由代码判定的规则引擎或轻量模型完成而不是由执行模型“自觉”。这是整个设计里最关键的一道变化终止条件不再是模型的自我声明而是结构属性的自然结果。DAG 上所有可执行节点全部进入终态整个任务就结束。类比一下这就好比带一个实习生。原来你让他全权负责一个项目他自己觉得做完就说做完了。现在你给他一张任务清单每个任务做完了必须给你过目你确认没问题才放行到下一步。最后清单上所有任务都打勾了项目才算做完。而“任务清单”是静态的是开工前就定好的。3. 从代码看架构变化while 循环拆掉前 vs 拆掉后不扯太玄的直接上代码对比。先说原来的实现再说拆掉之后的实现以及每个核心模块的职责与细节。3.1 拆之前的代码while 版from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini, temperature0) def run_agent(task: str): context { task: task, history: [], final_answer: None, } while True: response llm.invoke( f当前任务: {task}\n历史步骤: {context[history]}\n 请决策下一步动作仅输出action action_input ) action response.action action_input response.action_input if action final_answer: context[final_answer] action_input return context # 执行动作 result execute_tool(action, action_input) context[history].append((action, action_input, result))这段代码很短也很典型。问题就在while True上模型不断决策、不断执行、不断返回直到它自己说“好了”。执行多少次、什么时候收敛、中间会不会做无用功代码层面完全不可控。你顶多在外面套一个for i in range(MAX_ROUNDS)做粗暴的截断但截断之后任务还是没完成后面的系统也不知道该拿它怎么办。3.2 拆之后的代码任务图 状态机拆掉之后的核心抽象变成了三个东西Task 节点、TaskGraph、executor。from dataclasses import dataclass, field from enum import Enum class TaskState(str, Enum): PENDING pending RUNNING running REVIEWING reviewing DONE done FAILED failed dataclass class Task: task_id: str prompt: str depends_on: list[str] field(default_factorylist) feedback_edges: list[str] field(default_factorylist) # 谁可以反馈修订给我 state: TaskState TaskState.PENDING result: str retry_count: int 0 max_retries: int 2Task 就是有向图里的一个节点。depends_on表示执行前置依赖feedback_edges表示哪些节点可以反馈回来要求修订state是状态机字段retry_count是重试次数上限。这个数据类本身没有一点点 while 循环的影子但它天生具备“可能重来”的弹性。TaskGraph 负责拓扑调度和推进class TaskGraph: def __init__(self, tasks: list[Task]): self.tasks {t.task_id: t for t in tasks} self.ready_queue [] def get_ready_tasks(self): 找出所有依赖已满足、且尚未执行的节点 ready [] for task in self.tasks.values(): if task.state ! TaskState.PENDING: continue deps_met all( self.tasks[dep].state TaskState.DONE for dep in task.depends_on ) if deps_met: ready.append(task) return ready def mark_done(self, task_id: str, result: str): task self.tasks[task_id] task.state TaskState.DONE task.result result def mark_review_failed(self, task_id: str, review_feedback: str): task self.tasks[task_id] if task.retry_count task.max_retries: task.state TaskState.FAILED return task.retry_count 1 task.state TaskState.PENDING task.prompt f\n[修订意见] {review_feedback}注意mark_review_failed的逻辑。它没有说“回到 while 开头从头转”而是把当前节点重置为 PENDING等待调度器重新执行。由于重试次数有硬上限这个“循环”是有界的、理由明确而可审计的。每次修订意见都会追加到 prompt 里意味着模型下一次执行时看到上一次的错误反馈而不是单纯“重试”。调度主循环也非常直观def run_graph(graph: TaskGraph, executor): while True: ready_tasks graph.get_ready_tasks() if not ready_tasks: # 全部节点终态 break for task in ready_tasks: graph.tasks[task.task_id].state TaskState.RUNNING result executor.execute(task) # 验证由独立的reviewer节点判断 review executor.review(task, result) if review.passed: graph.mark_done(task.task_id, result) else: graph.mark_review_failed(task.task_id, review.feedback)等一下这里不是还有while True吗你没看错。但此while非彼while这里的循环条件不再是模型“自我感觉是否完成”而是任务图里还有没有可推进的活。每一轮循环做的事情是确定的、可枚举的检查就绪节点、执行、验证、更新状态。模型不再拥有“决定要不要继续”的自由意志。这个循环的本质是一个调度器它的迭代次数是有限的与任务节点总数和重试上限相关而不是模型规划路径的轮数。3.3 评审节点Reviewer怎么设计拆掉循环后原来“模型自评”那步被独立出来变成评审节点。评审节点可以是规则代码也可以是轻量模型。我通常混用评审类型适用场景实现方式结构化校验输出必须符合 JSON Schema 或包含指定字段用pydantic/ 手写校验逻辑零成本与目标一致性产出要与原始目标一致比如摘要主题贴合度用轻量模型如 gpt-4o-mini做二分类判断事实性抽取验证抽取类任务需要检查内容是否源自原文编辑距离 关键词覆盖规则评审节点和 Swarm/Harness 里的“验证器”概念是相通的把评判职责从执行者身上剥离避免“运动员兼裁判”。我后续会专门写一篇 Rust 语言实现 Agent 的评审节点设计这里先打个标记。提示在使用轻量模型做评审时建议把 prompt 约束成“只需输出 PASS 或 FAIL 一句原因”把 reasoning 隐式化不要让它输出一大段评论性内容。这样能把评审成本控制在一个极低的量级。实测中评审节点的 token 消耗只有执行节点的 10%~15%。4. 负载灵活了拆掉循环意外解锁并发能力这是我拆掉 while 循环之前完全没想到的收益——任务图天然支持并行。之前串行循环里每个子任务必须等上一个完成现在只要depends_on没有依赖关系两个任务就可以同时跑。而这个图本身就是 DAG我可以放心地写一个调度器对所有“就绪的、没有依赖冲突的”节点并发派发。4.1 基于 DAG 的按层并发用上面那个任务图最简单的一版并发调度就是“按层执行”找出当前所有PENDING且依赖已满足的节点用一个线程池并发跑全部跑完再推进下一层。听起来简单但实际测试里已经能带来很大的速度收益。from concurrent.futures import ThreadPoolExecutor, as_completed def run_graph_concurrent(graph: TaskGraph, executor, max_workers4): with ThreadPoolExecutor(max_workersmax_workers) as pool: while True: ready graph.get_ready_tasks() if not ready: break futures { pool.submit(executor.execute, task): task for task in ready } for future in as_completed(futures): task futures[future] result future.result() review executor.review(task, result) if review.passed: graph.mark_done(task.task_id, result) else: graph.mark_review_failed(task.task_id, review.feedback)这里有三个容易踩的坑。线程安全问题如果mark_done和mark_review_failed会被并发调用必须给 TaskGraph 加锁或者在每层执行完之后统一更新状态。上面这个简化版本在“每层跑完再统一更新”的逻辑下是安全的因为同一层的就绪节点之间本来就没有依赖。Executor 的状态污染如果 executor 内部持有上下文比如临时文件、内存缓存、数据库连接并发执行时必须为每个任务准备独立的执行环境。我在 Rust 版本里给每个任务单独的Context就是用作用域隔离来避免这个问题。同类工具调用的限流多个子任务同时调用同一个外部 API比如多个子任务都要调同一个数据库查询接口需要做限流。我一般用一个全局的Semaphore限制同一工具的并发数避免把下游系统打爆。4.2 资源池的参数怎么定并发不是越多越好参数需要根据任务类型来定。拿我自己的一套参数做个参考max_workers对于以 API 调用为主的任务并发上限通常设为 API 限流的 70%~80%。gpt-4o-mini 一般限速 500 RPM 左右我通常把max_workers设为 20。单节点超时每个 Task 节点的执行必须设置超时。LLM 调用本身有超时但工具调用比如爬网页、跑代码也要设置超时上限。否则一个子任务卡死会卡住整个依赖链。我一般设置单任务超时 30s。重试策略代码里的max_retries2并不是随便定的。一次重试的成本大约是一次完整 LLM 调用 一次评审调用。超过两次重试后第三次成功的概率显著下降继续重试就纯是给服务商送钱。如果max_retries2仍失败直接把节点标记为 FAILED让整个任务进入人工兜底分支。4.3 实测数据对比我拿一个具体案例说明收益。这个项目场景是“遍历一批接口文档抽取参数并生成接入方案”。之前用单串 while 循环版 Agent 跑处理 25 个目标接口总共花了 103 秒。换成任务图拆分版后同一批数据同一模型耗时降到 41 秒。token 消耗降了大约 30%——因为不需要重复思考同一个子问题。成功率也从 82% 提升到 93%。成功率提升的核心原因也很实在原来的 while 循环里模型忘记了中间某一步就永远回不去了而任务图里每一步结果都有明确的存储位置下一次执行可以直接取到上一步产出不存在“忘记”。5. 实操怎么把一个现有 Agent 迁移到这个架构你要是已经有一个 while 循环式 Agent迁移不要求推倒重来。我建议按下面四个步骤走每一步都有独立的收益不需要一步到位。5.1 第一步先给任务加“名称”这是最小成本改造。把原来一大段history列表改成带标识的字典{ task_1_extract_entities: ..., task_2_validate_entities: ..., task_3_generate_summary: ... }这样每轮循环执行的到底是什么任务日志里一目了然。加完这一步你的可观测性已经超过一半的 Agent 项目。以后排查问题至少知道是哪一步出的问题。5.2 第二步把“最终答案”改成显式目标节点原来的 while 循环靠模型输出final_answer作为退出信号。迁移时可以先在任务图里加一个显式节点final_answer_node它的依赖是所有前置任务节点它的 prompt 明确写着“基于以下已完成的任务结果生成最终答案”。模型不再需要自己去想“是不是该收尾了”它只需要在看到所有输入后锯成型。这一步改完即使任务图内部仍是串行执行收敛确定性也已经大幅提升。5.3 第三步引入 Reviewer给每个产出加一道验证。先从规则验证开始JSON 格式、字段完整性、长度上限。这些不用模型就能判断成本为零。再把“容易出错但规则能判断”的项目都交给代码。最后再上模型评审此时 review 的 prompt 已经能写得非常稳定。5.4 第四步把串行调度换成并发调度到这一步架构上已经完全脱离 while 循环可以安全地引入并发调度器。并发调度的代码模板就是上面贴的run_graph_concurrent。建议先在离线任务上验证确认没有一致性问题后再推到线上 Agent 服务。整个迁移过程可以按周粒度推进每一步都是可回滚的。我见过很多团队想一步到位重写 Agent大概率中途翻车。渐进迁移在这套架构里特别重要因为每一步都伴随行为变化一次性改动太多出了问题你根本不知道是哪一步引入的。6. 避坑指南与常见问题实录这套架构我前前后后跑了大半年下面这些问题几乎全都踩过。整理成速查表算是给后来者省点学费。问题现象排查思路解决办法任务图中出现环调度器死循环任务永不结束检查depends_on和feedback_edges是否有循环引用启动时构建依赖图做一次拓扑排序发现环直接报错状态丢失并发执行时子任务结果互相覆盖检查 Task 对象是否在多个线程间共享每个任务用任务级 Context线程池每个 worker 独享执行状态评审结果不稳定同样的 output 一会儿 PASS 一会儿 FAIL评审 prompt 太模糊或者带入了主观判断把评审标准写成可枚举的检查项尽量用规则评审修订反馈导致 prompt 爆炸prompt 越来越长token 开销增大每次修订都往 prompt 里追加全文只追加评审意见摘要并对历史修订做滚动窗口压缩单任务重试过多失败节点反复横跳没有控制总重试次数加全局重试预算失败节点达到上限后转人工或标记 FAILED动态子任务依赖子任务数量不固定无法静态建图需要支持运行时追加节点在等待层做二次扫描有新的可执行节点动态加入调度队列6.1 重点说说“评审不稳定”这个坑这是整个架构里最容易导致线上返工的问题。用模型做评审时如果 prompt 是“请判断这个结果是否满足要求”模型通常会给出模棱两可的回答。我的经验是把评审 prompt 改成二分列表请对以下产出逐条判断每条只能回答 YES 或 NO 1. 结果是否是合法的 JSON 2. 是否包含字段 title 3. 是否包含至少一个 reference 条目 4. 是否有任何 reference 条目引用了原文外部内容 如果任一答案为 NO请输出 FAIL 和对应条目标号。 如果全部为 YES输出 PASS。这样模型不需要“思考”太多只是执行一个简单的分类任务稳定性高很多。结构化的评审 prompt 是整个架构里性价比最高的优化点没有之一。6.2 动态任务/动态依赖怎么处理并不是所有场景都能在启动前预定义任务图。比如让 Agent 先调研再根据调研结果生成下一步行动这就是典型的动态任务。我遇到这种情况的做法是让 Agent 的“规划”只做一层不递归。具体来说第一轮让模型生成完整的任务列表生成后立即冻结。后面的执行不再给模型“改计划”的机会最多触发修订边。如果需要多轮调研就把“多轮”拆成多个独立的 Task 节点而不是让模型自己决定循环几轮。这个办法可能有人觉得不够“智能”——毕竟模型没有完全自由。但自由的代价是不可控。我个人在实际项目里更看重收敛性能保证任务图的边界是确定的超时是确定的失败路径是明确的整个系统的 SLA 就能算出来。6.3 Agent 安全与资源滥用防控拆掉循环还有一个安全方面的好处执行路径变窄了。原来 while 循环里模型理论上可以调用任何工具任意次这本身就是很大的安全面。任务图模式下工具调用被固定到指定节点模型只能按图索骥。我在架构里还加了一道权限校验工具执行前检查当前节点是否有该工具的白名单权限。这种方式把 Agent 安全从“事后审计”变成了“事前控制”对生产环境特别重要。7. 什么情况下不建议拆while 循环仍然合理写到这里必须泼一盆冷水。这套任务图架构并不是万能的。至少有两类场景我认为 while 循环依然合理硬拆反而得不偿失。7.1 交互式对话场景当 Agent 需要与用户进行多轮开放式对话比如陪你聊天、辅助创意讨论这本质上就是 while 循环式交互。用户下一轮说什么无法预知预定义任务图根本建不起来。对话场景的循环条件就是“用户继续说话”这不是模型的“自我决定”而是外部输入驱动。这种“外部事件驱动的循环”不能拆拆了会破坏对话的自然体验。7.2 探索性、强上下文依赖的开放任务比如“帮我调研一下当前市场上的主流方案再给一份对比报告”。这类任务的路径完全依赖调研过程中的发现动态性极强。也可以拆——把所有子任务列为“候选规划节点”让 Agent 动态创建依赖但这已经偏离了静态任务图的初衷。真的硬要拆需要引入任务图运行时动态改写的机制复杂度会显著上升。我给这类场景的建议是保留 while 循环但加“结构锚点”。比如在循环里每轮都让模型输出current_step_index把“我现在在第几步”这个状态显式化至少可以保证出问题时有迹可循。随着模型能力提升现在的模型已经能在自由规划的同时输出结构化的状态信息所以现代实现可以把 while 循环和结构化任务图混合使用——循环保留但每轮都同步更新到一个外部的 DAG 状态上。本质上这也是一种妥协但比纯粹的 while 循环可观测性好太多了。7.3 判断标准你的场景是“过程可预测”还是“结果可预测”我做架构选型时用的判断标准很简单如果过程可预测比如“解析文档 → 抽取字段 → 生成 JSON → 校验合法性”用任务图。如果结果可预测但路径不确定比如“找到最优解”用 while 循环 结构化状态追踪。如果过程和结果都不可预测创意协作、开放式聊天别折腾架构先把交互体验做好最重要。8. 后续还能怎么扩展这套架构目前还有几个方向我在持续打磨。第一个是把任务图替换成真正的流式执行引擎让每个 Task 节点一开始执行就能把中间产出以流的形式暴露出来这样用户可以实时看到 Agent 走到了哪一步增强交互感。第二个是给状态机引入持久化层。现在状态都在内存里进程一重启任务就得重来。下一步准备把 Task 状态落库做成可恢复的 Agent 会话这样系统崩溃后可以无缝续跑。第三个方向是我在试的把任务图和 RAG 的记忆系统打通让每个 Task 节点能自动检索历史相似任务的执行经验把 go-to 的模板查询出来提升第一次执行的成功率。这些扩展的详细实现等我跑出更完整的数据再整理一篇分享。回头说一句拆掉 while 循环不是目的把 Agent 的流程控制从模型手里收回来才是目的。你不需要完全放弃模型的自适应能力你只需要把它的自由度限制在“单个子任务的执行”上把“任务全局的推进逻辑”交给代码。实践了这套架构之后我最大的体会是Agent 项目里最容易崩的地方往往不是模型不够聪明而是流程控制的设计不够硬。模型负责创造架构负责收敛只有这两者分清楚Agent 才能从 Demo 走向生产。