拆掉Agent的while循环：用状态机架构提升稳定性与可恢复性

如果你现在去 GitHub 上随便翻几个 Agent 项目，十个里面有八个主循环长这样：while True里头装着 LLM 调用、工具执行、结果判断，直到模型说一句"任务完成"才break。这个写法的好处是极其直观，ReAct 范式就是靠这个模式火起来的。但我在生产环境里被它坑了不止一次——不是死循环就是烧 token，最头疼的是任务跑到一半崩了，整个状态全丢，一点恢复的余地都没有。这篇文章就聊聊我后来怎么把 Agent 的 while 循环拆掉的，换成一种状态机驱动的架构，顺便把事件驱动和任务队列两种替代方案也一并讲清楚。如果你正在设计一个要上线的 Agent 系统，尤其是要做成多 Agent 协作或者异步任务平台的，这篇应该能帮你少踩几个大坑。

先说清楚一个容易被误解的点。拆掉 while 循环不是说代码里一个循环都不留，那不可能，也不应该。事件循环、消息队列消费循环这种基础设施层面的循环是必须保留的。真正要拆掉的，是那个把"模型推理、动作执行、结果反馈"焊死在一起、由 Agent 自己说了算的决策循环。核心区别在于：谁控制循环的边界，谁决定循环什么时候停下来。

1. 典型 Agent 架构为什么都喜欢用 while 循环

1.1 ReAct 范式的天然产物

现在市面上绝大多数 Agent 框架，骨子里都是 ReAct（Reasoning and Acting）范式。这个范式来自 2022 年 Google 那篇论文，核心思想很朴素：让模型交替进行推理和行动。模型先想一步（Thought），再决定调用什么工具（Action），工具执行完返回结果（Observation），模型看到结果后再继续想下一步。这个"想→做→看"的过程天然就是一个循环，所以开发者最自然的实现方式就是用 while 包住整个过程。

这种实现方式的代码结构其实非常优雅，它把 Agent 的执行流程压缩成了一个极简的多步交互过程。我见过很多初版 Agent 代码就是这么干的，写起来快、逻辑直观、演示效果还特别好。你给用户看的时候，模型每一步都在"思考"，每一步都有工具调用的实况转播，视觉效果拉满。

1.2 while 循环 Agent 的典型代码长什么样

大部分 Agent 项目的主循环都可以抽象成下面这个伪代码：

def run_agent(task): messages = [system_prompt, task] steps = 0 while steps < max_steps: # 1. 让 LLM 基于当前对话历史决策 response = llm.chat(messages) # 2. 解析模型输出，提取动作 action = parse_action(response) # 3. 如果模型说任务完成了，就退出 if action.type == "finish": return action.output # 4. 执行模型指定的工具调用 tool_result = execute_tool(action) # 5. 把模型输出和工具结果都追加到对话历史 messages.append(response) messages.append(tool_result) steps += 1 return "达到最大步数，任务未完成"

这个逻辑链条环环相扣，看起来无懈可击。但问题恰恰出在这个"环环相扣"上：每一步都依赖前一步的输出，整个流程是一个单一的执行链，中间没有任何外部干预的入口。你没法在某个步骤之间插入人工审批，没法让另一个 Agent 参与协作，更没法在流程跑了一半时把它持久化保存下来。

1.3 这个模式在 Demo 里没问题，上生产就出事

我最早用这种架构写 Agent 服务的时候，本地上跑几个测试用例完全正常。但一旦放到线上，让真实的用户输入去驱动它，各种幺蛾子就全冒出来了。模型输出一个格式错误的 JSON，解析器直接抛异常，整个任务就崩了。模型在某一步反复调用同一个工具，陷入死循环，把 token 预算全部烧光。任务跑了一百多步，用户等得不耐烦直接关闭了页面，但是后端那个 while 还傻傻地跑着，根本没意识到用户已经走了。

这些问题的根源不是模型不够聪明，而是架构设计上把 Agent 的生命周期和单次进程的执行绑死了。while 循环内部是一个封闭的、不可观测、不可干预的自治过程。进程一挂，循环就断，所有中间状态灰飞烟灭。这种架构天生就不适合需要可靠性、可观测性、可恢复性的生产环境。

2. 为什么必须拆掉 while 循环：四个致命伤

2.1 成本不可控：Token 消耗是二次方增长的

这是最直接的痛点。在 while 循环架构里，每一轮循环都会把模型回复和工具结果追加到 messages 列表里。第 n 轮循环时，发送给 LLM 的消息长度大约是初始长度加上 n 轮迭代的累积内容。而 LLM 的计费是按输入 token 数算的，每次调用都要把完整的对话历史发送一遍。这意味着总 token 消耗近似等于 1 + 2 + 3 + ... + n，也就是 O(n²)。

举个例子，假设初始系统提示和用户任务一共 1500 token，每轮循环平均新增 800 token（模型回复加工具结果）。跑到第 20 轮时，单次调用的输入已经接近 17500 token，而过去 20 轮累计的输入消耗大约是 20 × 1500 + 800 × (1+2+...+20) ≈ 198000 token。如果 Agent 跑了 50 轮，这个数字直接飙到上百万。你以为是在跑 Agent，其实是在烧钱。

2.2 模型幻觉会导致失控：循环空转是常态

LLM 不是确定性程序，它在做工具选择时偶尔会抽风。最常见的情况是：模型反复调用同一个失败的函数，每次拿到相同的错误信息，然后又一次选择调用同一个函数，陷入一个完全空转的循环。while 循环对这种局面几乎没有抵抗力，只能靠 max_steps 做最后一道闸门。

有个更隐蔽的场景：模型"觉得"自己已经完成了任务，但实际上结果根本没有落库。它基于幻觉的自信输出了一个 finish 动作，任务实际是失败的。传统 while 循环只认模型单方面的判断，缺乏独立验证机制，这个问题会被静默吞掉。

2.3 无法断点续跑：中间状态全部丢失

while 循环的所有状态都保存在进程内存里：对话历史、已执行的动作序列、中间产物。一旦进程崩溃、容器重启、代码发布，这些状态就全部没有了。对于简单的问答 Agent 可能无所谓，但对于一个跑了二十分钟、已经完成了十步操作的复杂任务，状态丢失就意味着从头再来。

我在实际运维中遇到过这种场景：一个数据分析 Agent 已经跑完了数据清洗、特征工程、模型训练三步，在生成报告那一步因为上游 API 超时崩了。重启之后所有中间结果全部丢失，又得重新跑一遍。而前几步的耗时占了整个任务的三分之二，重跑的成本和等待时间都非常离谱。

2.4 并发能力为零：无法多 Agent 协作

while 循环是典型的同步阻塞模型，一个循环实例就是一个完整的执行单元。要让多个 Agent 协同完成一个任务，你只能在外面再套一层循环去调度多个 while 循环——这本质上还是在用进程级并发解决问题，Agent 之间没有任何原生的通信机制。

更要命的是，每个 while 循环实例的执行路径完全由模型自由发挥，你没法预判它下一步要干什么。这就导致你没法做资源预留、没法做步骤规划、没法做依赖编排。多个 Agent 并发操作同一个共享资源时，也没法在架构层面做互斥控制。这些能力在 while 循环模式下全都得靠外部补丁，而且补得特别别扭。

3. 三种替代架构：状态机、事件驱动、任务队列

3.1 状态机驱动：把"过程"变成"状态"

拆掉 while 循环的第一个思路，是把 Agent 的每一步执行建模成有限状态机（Finite State Machine）。Agent 的整个生命周期被拆成一组有限的、定义明确的状态，每个状态对应一个处理函数。模型不再自己决定"下一步做什么"，而是根据当前状态和外部事件，通过预设的转移规则跳转到下一个状态。

一个标准的 Agent 状态机至少需要这几个状态：初始（INIT）、规划（PLANNING）、执行（EXECUTING）、验证（VERIFYING）、完成（DONE）、失败（FAILED）。模型在规划状态负责拆解任务，在执行状态负责调用工具，在验证状态负责检查执行结果，最后根据验证结果决定跳到 DONE 还是回到 EXECUTING。

状态机和 while 循环最大的区别在于：状态的每一步转移都是可观测、可记录、可控制的。你随时可以知道 Agent 当前处于哪个状态，可以在转移前后插入钩子函数，可以把状态持久化到数据库，可以在任意状态恢复执行。

3.2 事件驱动：Agent 变成响应式组件

第二种思路是彻底抛弃"循环拉取"（pull）的模式，改成事件驱动（push）。Agent 不再主动去"下一步做什么"，而是响应外部事件。你向 Agent 发布一个task_received事件，它执行相应的处理器；处理器执行完产生一个新的tool_result事件，再由下一个处理器消费。

事件驱动架构的好处是解耦。Agent 的各个处理环节被拆成独立的事件处理器，每个处理器只负责自己那一亩三分地。这些处理器可以分布在不同的进程、不同的服务上，通过消息队列连接。Agent 变成了一个响应式组件，你说"有事件来了它就有反应，没事件它就安静待着"。

事件驱动架构天然支持并行。同一个 Agent 的多个事件可以交给多个处理器并发处理，不同 Agent 之间通过事件总线解耦，互不干扰。这个模式特别适合做多 Agent 协作系统，你在 agent A 的工具处理器里发布一个task_assigned事件，agent B 的监听器收到后就自动启动执行。

3.3 任务队列：把 Agent 步骤持久化成任务

第三种思路更像传统后端架构的 Agent 化改造。把 Agent 的每一步执行（一次 LLM 调用、一次工具调用、一次验证判断）都封装成一个独立的"任务"，任务的元数据、输入参数、执行状态都持久化到数据库或消息队列里。由一个外部调度器统一负责任务的分发、重试、超时处理。

这种方案下，Agent 本身不再有"连续执行"的概念。Agent 进程只是一个 Worker，从队列里取一个任务，执行一步，把结果写回任务状态，然后等下一个任务。任务和任务之间的流转逻辑（上一步的结果应该触发哪一步）由工作流引擎决定，而不是由 Agent 的内循环决定。

任务队列方案是三种方案里最接近生产标准的，它直接复用了后端领域成熟的任务编排经验。Temporal 就是这么干的，LangGraph 也在往这个方向走。这套架构的可靠性天花板很高，因为它本质上就是"把 Agent 当成一种特殊的后台任务系统"来设计。

4. 实操：把现有 while 循环 Agent 改造成状态机

4.1 第一步：梳理循环体里的动作类型

改造第一步不是写代码，而是梳理现有循环体里到底有哪几类动作。我通常把所有动作归纳成四类：推理决策型（需要 LLM 参与）、工具执行型（需要调用外部 API）、结果判断型（决定是否完成任务）、终止退出型（任务成功或失败）。

这一步的核心任务是画出当前的执行流程图，标注清楚每个节点的输入、输出和跳转条件。我见过很多人直接上手改代码，结果改到一半被复杂的分支逻辑绕晕了。先把动作类型梳理清楚，后面的状态定义就是水到渠成的事。

以我之前做过的一个文档摘要 Agent 为例，它的动作类型是这样的：

• 读取文档（工具执行型）
• 分析文档结构（推理决策型）
• 生成摘要（推理决策型）
• 检查摘要质量（结果判断型）
• 输出最终摘要（终止退出型）

4.2 第二步：定义状态枚举与转移表

动作类型梳理清楚之后，定义状态其实就是把动作类型一一映射成状态节点，然后再加上始末状态。这一步的关键是设计好转移条件。我一般用一张转移表来表达状态机，表格里每一行表示一条转移规则。

还是上面那个文档摘要 Agent，它的状态转移表长这样：

这张表的价值在于它把 Agent 的行为规则从代码里抽离出来了。新增一个状态只需要加一行配置和一个处理函数，删除一个状态只需要删除对应行，完全不需要动主流程代码。这也让非工程师角色能够看懂 Agent 的决策逻辑，方便团队评审。

4.3 第三步：实现状态处理器与外部调度循环

状态机的主执行器可以做得非常简洁。每个状态对应一个纯函数：输入是当前上下文对象，输出是转移事件。主执行器只负责查表跳转，不包含任何业务逻辑。

from enum import Enum class AgentState(Enum): INIT = "init" READING = "reading" ANALYZING = "analyzing" SUMMARIZING = "summarizing" VERIFYING = "verifying" DONE = "done" FAILED = "failed" class AgentContext: def __init__(self, task): self.task = task self.doc_content = None self.analysis = None self.summary = None self.retry_count = 0 TRANSITIONS = { (AgentState.INIT, "task_ready"): AgentState.READING, (AgentState.READING, "read_done"): AgentState.ANALYZING, (AgentState.READING, "read_fail"): AgentState.FAILED, (AgentState.ANALYZING, "analysis_done"): AgentState.SUMMARIZING, (AgentState.SUMMARIZING, "summary_done"): AgentState.VERIFYING, (AgentState.VERIFYING, "verify_pass"): AgentState.DONE, (AgentState.VERIFYING, "verify_fail"): AgentState.SUMMARIZING, } def handle_init(ctx): return "task_ready" def handle_reading(ctx): content = load_document(ctx.task["doc_id"]) if content is None: return "read_fail" ctx.doc_content = content return "read_done" def handle_analyzing(ctx): ctx.analysis = analyze_structure(ctx.doc_content) return "analysis_done" def handle_summarizing(ctx): ctx.summary = generate_summary(ctx.doc_content, ctx.analysis) return "summary_done" def handle_verifying(ctx): if check_quality(ctx.summary): return "verify_pass" ctx.retry_count += 1 return "verify_fail" if ctx.retry_count < 3 else "force_fail" HANDLERS = { AgentState.INIT: handle_init, AgentState.READING: handle_reading, AgentState.ANALYZING: handle_analyzing, AgentState.SUMMARIZING: handle_summarizing, AgentState.VERIFYING: handle_verifying, } def run_agent_state_machine(task, max_visits=50): ctx = AgentContext(task) state = AgentState.INIT state_visits = {} while state not in (AgentState.DONE, AgentState.FAILED): state_visits[state] = state_visits.get(state, 0) + 1 if state_visits[state] > max_visits: return "状态访问超限，疑似状态死循环" handler = HANDLERS[state] event = handler(ctx) if event == "force_fail": return "重试次数用完，任务失败" next_state = TRANSITIONS.get((state, event)) if next_state is None: return f"非法转移：{state} + {event}" state = next_state return ctx.summary if state == AgentState.DONE else None

你别看这个外部循环还带个while，它和最初的 while 循环本质上是两回事。第一，这个循环每次迭代都会触发状态转移，而转移路径是预定义好的，不会出现模型幻觉导致的随机跳转。第二，循环体里的state_visits计数是对状态机本身的保护，防止我自己的转移表写错了导致状态之间来回打转。第三，最重要的是 Agent 内部已经不存在循环了，每个 handler 都是单步执行，你可以随时在循环外部保存 ctx 对象，实现断点续跑。

4.4 第四步：让状态可持久化，实现断点续跑

状态机架构带来的一个直接好处就是状态持久化变得特别简单——你只需要把AgentContext对象序列化保存下来就行。AgentContext 里面记录了任务 ID、当前状态、已完成的中间结果、重试次数，这些信息足够你随时恢复一个任务的执行。

我之前把这个能力应用在了一个多步骤数据处理 Agent 上。这个 Agent 要从数据库里拉数据、清洗、转换、加载，整个流程要跑三四分钟。现在我会在 AgentContext 的每次转移之后把它存到 Redis，key 是 task_id，value 是序列化的上下文对象。如果 Worker 进程崩溃了，新的 Worker 拉起时从 Redis 加载上下文，直接从上一次转移后的状态继续执行。

这里有个细节：存储上下文时一定要存储下一个转移的触发事件是哪一个，这样恢复执行时才能精确知道从哪里继续。

5. 常见问题与排障实录

5.1 状态粒度怎么定？太粗太细都会出问题

状态机改造最容易踩的坑是状态粒度设计失衡。我见过有人把每一个工具调用都定义成一个独立状态，结果状态枚举列表拉出来四五十个，转移表维护起来崩溃。也见过有人把整个工具调用链黑盒化成一个状态，状态机退化成不定式循环，白白做了改造。

我的经验是，状态的粒度应该对齐"业务审批节点"，而不是"底层函数调用"。五个串行的数据清洗步骤，如果没有中间检查点，就合并成一个 CLEANING 状态；如果每一步的结果都需要人工确认或有独立的失败处理策略，那就拆成多个状态。判断标准非常简单：如果你不需要在某个步骤之间进行干预、检查、或者恢复，它就不需要是一个独立状态。

5.2 状态机自己陷入死循环怎么办

转移表写错真的会让状态机在几个状态之间无限打转。比如 A 状态产生done事件跳到 B，B 状态看到条件不满足又产生retry事件跳回 A，两边互相踢皮球，配合得天衣无缝。

我在实现上加了双重保险。第一道硬性保险就是上面代码里的state_visits计数，每个状态的访问次数加上限值，超过了就强制终止。第二道保险是给整个状态机的运行加上最大转移次数限制，比如一个任务最多只允许流转 15 次，超过就判定为异常。这两道保险的意义不在于解决死循环，而在于让死循环变成可观测的失败事件，方便排障。

5.3 持久化上下文什么时候写入

状态持久化的时机需要谨慎设计。太频繁写入，高频状态转移时性能下降严重；太稀疏写入，进程崩溃时丢状态的概率变大。我实测下来，对于大多数 Agent 任务，在每个需要调用外部工具的状态处理函数之前和之后各持久化一次是性价比最高的方案。

执行前持久化是为了防幂等性问题——如果 Worker 在处理状态时崩溃了，恢复时重跑一遍该状态的处理是安全的。执行后持久化是为了保存工具调用的结果，避免重复调用付费接口。对于 LLM 调用这种又贵又不确定的外部依赖，我还会单独存一份工具调用的响应哈希，恢复时先查缓存，命中就直接复用，不再重复请求模型。

5.4 从事件驱动和任务队列方案里吸取的教训

状态机不是唯一解，我在实践过程中也试过事件驱动和任务队列方案。事件驱动架构最大的问题是排查问题时链路追踪特别麻烦，一个事件经过多个处理器流转，你很难直观地看到 Agent 当前的完整状态。后来我引入了 corrlation_id 贯穿全链路，才算解决这个问题。

任务队列方案的鲁棒性最好，但开发成本也最高。为了两个 Agent 协作场景引入一套完整的工作流引擎，确实有点牛刀杀鸡的感觉。我的最终做法是把三种方案融合在了一起：核心流程用状态机驱动，状态转移产生的事件异步发到消息队列，分布式场景下用任务队列做跨进程调度。这套组合架构在稳定性和开发效率之间找到了一个比较舒服的平衡点。

根据我个人的实操体验，拆掉 while 循环不是一次简单的代码重构，而是对 Agent 系统边界的一次重新定义。你不再把 Agent 当作一个"会自己跑到终点的自动程序"，而是把它当成一组"可以被观察、被控制、被恢复的状态节点"。这个思维转换带来的收益是全方位的，稳定性能提升一个量级，排查问题时也不再像以前那样两眼一抹黑。

如果你正准备动手改自己项目里的 Agent 主循环，我最后再分享一个实操技巧：不要一次性重写所有逻辑。先把原有的 while 循环完整跑一遍，把所有真实场景下的转移路径记录下来，确认这些路径覆盖了大部分线上案例，再开始设计状态机。否则你会发现自己精心设计的状态表，在最正常的业务链路里就卡住了。