先声明一下立场我不是说所有 Agent 都必须把 while 循环拆掉而是我在生产环境里被这种“标准架构”坑了太多次之后才下定决心换掉它。如果你还没有遇到过 Agent 卡死、无法续跑、没法审计、一步出错只能从头再来的问题那这篇文章可能对你来说还只是个“方案预览”但如果你已经开始把 Agent 往业务系统里接了我强烈建议你花十分钟看完这个架构思路。1. 先说痛点Agent 为什么被 while 循环捆死了1.1 教科书式的 Agent 实现市面上 90% 的 Agent 框架或教程核心逻辑都长一个样def run_agent(goal: str): context create_initial_context(goal) while not context.finished: thought llm.call(context.build_prompt(think)) action llm.call(context.build_prompt(act)) result execute_action(action) context.update(thought, action, result) return context.final_answer这个伪代码你应该不陌生。不管是 ReAct 还是 Function Calling 方案只要把思路套进去最终都会收敛成同一个形态Agent 内部有一个“思考→行动→观察→再思考”的循环循环条件就是“没结束”。LLM 的每次调用都靠这个循环驱动Agent 在循环里自行决定下一步做什么直到它认为自己完成了任务。新手阶段这么写完全没问题。在小规模的脚本、演示项目、个人工具里这个 while 循环简单、直观、够用。我在早期做 Agent 原型验证的时候也是这么写的一天就能跑通一个能上网搜索、能调计算器、能回答问题的玩具 Agent。1.2 把 while 循环搬到生产环境后问题全来了但等我真正把这套逻辑接到业务系统里问题就开始一个一个冒头了。不是说循环本身写错了而是Agent 内部自旋转的这种架构跟真实业务系统的协作模型根本对不上。第一个坑不可观测。Agent 在 while 循环里自主运转跑得对不对、目前在哪个阶段、已经耗费多少 token、执行过几个工具调用全都黑盒。你只能在外面看它“有没有结束”中间过程全靠日志拼凑。一旦它做出意外决策你连干预的入口都没有。第二个坑不可挂起。业务里的 Agent 经常需要等人。比如写邮件的 Agent 需要用户确认措辞或者执行转账前需要审批。while 循环版本碰见这种场景只能“阻塞等”或者“轮询查”——阻塞会让服务线程一直被占着轮询查本质上又回到循环里出不来了。第三个坑不可恢复。只要进程崩了、容器重启了或者 LLM 调用超时异常没处理好整个 Agent 的运行状态就全部归零。而 Agent 任务常常是高成本的已经烧掉的 token、已经执行完的副作用动作比如已经发出去的邮件、已经累积到一半的推理上下文全没了。从头再来一次成本直接翻倍。第四个坑不可扩展。多个 Agent 任务并发跑的时候while 循环版本只能靠“多线程 多个进程”硬扛。Agent 不是无状态服务它的全部状态都锁在自己的内存里既没法横向扩容也没法把一个任务从一个节点转移到另一个节点。所以我在重构自己的 Agent 平台时下了个决心把 Agent 这个“内部 while 循环的自主机器”改造成“外部事件驱动的状态机”。这个改动看起来只是代码结构变了实际上它把 Agent 从“自己转个没完的程序”变成了“被外部一步步推着走的服务”。下面我把整个思路、改造步骤、代码实现和踩坑记录都展开讲一讲。2. 拆掉 while 循环的核心设计思路状态机 事件驱动2.1 核心思想把“循环”摊平成“状态迁移”很多人第一次听到这个方案第一反应是拆掉 while 循环那 Agent 下一次该做什么谁来决定答案是由外部事件来决定。比较直观的类比是电饭煲。传统 while 循环的 Agent 是一个自己决定“什么时候再想一想、什么时候再动一动”的机器人它会一直盯着锅里的饭自己判断“再煮两分钟”还是“可以出锅了”。而事件驱动状态机的 Agent 是它不会自己持续转它每次只执行一个步骤执行完就把“我下一步该做什么”交还给外部调度器等调度器发一个事件过来比如“LLM 已经返回了思考结果”“工具已经执行完毕”它再处理下一个状态。每个 Agent 实例的内部不再是无限循环而是一张状态迁移表当前状态触发事件处理动作迁移到状态INITSTART构造初始上下文调用 LLM 生成思考THINKINGTHINKINGLLM_RESPONSE解析思考结果决定下一步 actionACTION / DONEACTIONTOOL_FINISHED把工具结果写回上下文再次调用 LLMTHINKING / DONEWAITINGUSER_APPROVAL收到人工确认结果继续执行THINKING / DONEDONE-输出最终结果释放资源-这张表怎么理解你可以把 Agent 的每个阶段都看成“状态”把“触发下一步”的动作看成“事件”。Agent 的功能从“循环执行全部步骤”变成了“根据当前状态和外部传入的事件完成一次迁移”。这种架构在成熟后台系统中叫事件驱动状态机不是什么新鲜概念但用在 Agent 上恰好解决了一个关键问题LLM 的推理本身是异步的、不可控的与其用循环强行把它扭成同步不如顺着它的异步特性来做设计。2.2 为什么状态机方案比循环方案更稳相比 while 循环状态机方案带来几个结构性的优点。第一Agent 变成无状态服务。如果把 Agent 的上下文对话历史、工具结果、任务目标单独存到一个持久化存储里Agent 本身就可以做到“不保留任何状态”。同一个任务A 节点处理完第一步之后停了B 节点可以从存储里把上下文捞出来继续处理第二步。这对容器环境、分布式部署非常友好。第二人工审批终于好做了。业务系统里最怕的就是 Agent “自作主张”。状态机天然允许一个中间状态叫 WAITINGAgent 在需要人工确认时可以主动进入这个状态然后释放线程资源。审批接口一回调事件传入状态机Agent 从 WAITING 被唤醒继续往下走。整个过程没有阻塞、没有轮询。第三审计和回放极其舒服。每一次状态迁移背后都有触发事件而每个事件里携带着 LLM 的输入输出、工具调用的参数和结果。这意味着你可以在事后拿到一条完整的时间线这个 Agent 是什么时候开始思考的、它当时看到了什么、它为什么决定调用这个工具、工具返回了什么。出问题的时候不是看“它最后输出错了”而是能精确回溯到哪一步开始错的。2.3 状态机 外部队列 持久化 一套可跑在生产环境的 Agent 架构拆掉 while 循环不只是把 while 改写成 switch-case 那么简单。我的完整架构包含四个部分状态机核心Agent 的每一步逻辑接收“当前状态 事件”返回“新状态 动作指令”。外部调度器负责接收事件、把事件分发到对应对的 Agent 实例上。持久化存储保存 Agent 的状态机和上下文可以是 Redis、PostgreSQL也可以是专门的遍历存储。事件总线/消息队列在 LLM 服务、工具执行器、人工审批后台与调度器之间传递事件。这套架构最大的变化是Agent 自身不再“主动跑”而是“被动响应”。你甚至可以说这个 Agent 已经不是传统意义上的“一个会自主循环的程序”了而是一个“每次被外部插一下才动一步的乐高积木”。注意一个容易混淆的点调度器底层可能还会有一个 while True 的消息监听循环但那属于基础设施层的消息循环跟 Agent 业务层的“思考-行动循环”是两码事。拆掉的是后者前者相当于 Web 服务器的事件循环你没必要也没办法全拆掉。3. 具体改造步骤一个 while 循环 Agent 到状态机 Agent 的完整迁移3.1 第一步先定义状态枚举迁移第一步是把你代码里隐式的“阶段”显式化。大多数 while 版本的 Agent阶段是靠 context 里的字段隐式表达的比如waiting_for_approval是一个布尔值。这种设计在状态一多就容易乱。我的做法是定义一个枚举类from enum import Enum class AgentState(Enum): INIT init THINKING thinking ACTION action WAITING waiting DONE done FAILED failed这六个状态就够覆盖绝大多数场景了。核心是THINKING正在调用 LLM 推理、ACTION正在执行工具、WAITING等待外部输入这三个DONE 和 FAILED 都属于终态。状态定太多反而会让迁移逻辑变复杂建议先从少的状态开始后面按需再加。3.2 第二步把上下文单独抽离出来while 版本里上下文是存在 Agent 实例内存里的。状态机版必须把上下文抽成一个独立对象并且支持序列化、反序列化from dataclasses import dataclass, field from typing import Any dataclass class AgentContext: agent_id: str goal: str state: AgentState history: list[dict] field(default_factorylist) tool_results: dict[str, Any] field(default_factorydict) final_answer: str def append_event(self, role: str, content: str): self.history.append({role: role, content: content}) def to_prompt(self) - list[dict]: system_prompt { role: system, content: self.goal, } return [system_prompt] self.history[-20:] # 窗口截断示例这里有个重要的设计细节上下文对象不依赖任何运行时资源。它没有 LLM client 的引用也没有数据库连接的引用这样它才能被自由地存进 Redis、传上消息队列、在另一个节点上恢复执行。3.3 第三步把每个处理步骤改写成纯函数状态机的核心函数是这样一个东西def handle_event(ctx: AgentContext, event: AgentEvent, services: AgentServices) - AgentContext: if ctx.state AgentState.INIT: ctx.state AgentState.THINKING ctx.append_event(user, event.payload) return ctx if ctx.state AgentState.THINKING: # 这里假定外部已经帮我们调用了 LLM结果放在 event 里 thought event.payload[thought] action event.payload[action] ctx.append_event(assistant, thought) if action finish: ctx.state AgentState.DONE ctx.final_answer event.payload[answer] else: ctx.state AgentState.ACTION ctx.tool_results[pending_action] action return ctx if ctx.state AgentState.ACTION: result event.payload[result] ctx.tool_results[last_result] result ctx.append_event(tool, result) ctx.state AgentState.THINKING return ctx return ctx看着是不是特别简单这就是状态机的本质——根据当前状态和事件返回一个更新后的状态。它不自己循环它只处理“一步”。关键的决策逻辑完全暴露在外层谁拿到这个新的上下文谁就可以决定下一步要调用什么服务、发什么事件。3.4 第四步设计事件结构事件是这个架构的血液。我给 Agent 系统里的事件定义了一个通用结构from dataclasses import dataclass from typing import Any dataclass class AgentEvent: agent_id: str event_type: str # START, LLM_RESPONSE, TOOL_FINISHED, USER_APPROVED payload: Any # LLM 输出、工具结果、用户输入等 timestamp: float为什么要单独抽出事件结构而不是直接调用函数因为在异步、分布式的环境下“LLM 返回结果”这个动作可能发生在一个完全不同的节点上——比如某个 worker 进程调用了 LLM API拿到返回结果之后它需要把结果发回给状态机。这时候你面对的就是一个跨进程通信问题而不是简单的函数调用。事件结构就是让这种通信标准化、可序列化的基础。3.5 第五步实现调度器调度器负责接收事件、找到事件对应的 Agent 上下文、执行一次状态迁移然后把迁移后的上下文存回存储里import asyncio class AgentEngine: def __init__(self, store: ContextStore): self.store store self.queue asyncio.Queue() async def submit(self, agent_id: str, event_type: str, payload: dict): await self.queue.put(AgentEvent(agent_idagent_id, event_typeevent_type, payloadpayload)) async def run(self): while True: event await self.queue.get() try: await self._process(event) except Exception as e: print(fFailed to process event {event}: {e}) async def _process(self, event: AgentEvent): async with self.store.lock(event.agent_id): # 防止并发事件打架 ctx await self.store.load(event.agent_id) if ctx is None: ctx AgentContext(agent_idevent.agent_id, goalevent.payload.get(goal, )) ctx handle_event(ctx, event, services) if ctx.state AgentState.THINKING: # 通知 LLM 调用器去生成思考结果 event_bus.publish(llm.invoke, ctx) elif ctx.state AgentState.ACTION: # 通知工具执行器 event_bus.publish(tool.invoke, ctx) await self.store.save(ctx)注意这段代码里的while True这是引擎的消息循环不是 Agent 的思考-行动循环。它本质上是一个消费者从队列里取事件执行一次状态迁移然后再取下一个事件。Agent 的“自主性”已经不在这个循环里了而是在事件流里——拆掉 while 循环指的就是拆掉 Agent 内部那个负责推理决策的循环而引擎层的消息循环只是基础设施。3.6 while 循环版 与 状态机版 差异对比维度while 循环版状态机事件驱动版执行模型同步自旋转异步事件驱动中断恢复崩溃即丢失状态持久化后可恢复人工介入难需轮询自然支持 WAITING 状态并发扩容难状态在内存易状态在存储层审计追踪依赖日志拼凑天然生成事件流调试回放困难支持事件流重放实现复杂度低中高需要基础组件4. 实操落地一个通过 HTTP 驱动的事件驱动版 Agent看完原理直接拿一个可运行的例子来感受一下。我用 FastAPI 做一个最简的 HTTP 驱动版 Agent每次 Agent 没结束外部都可以发一个事件进来推动它往前跳一步。4.1 定义上下文和状态常量先定义一个最小上下文对象为了演示清晰直接兼容 Python 字典序列化from enum import Enum class S(str, Enum): INIT init THINKING thinking ACTION action WAITING waiting DONE done上下文直接用字典存方便直接丢进 Redisdef new_context(agent_id: str, goal: str): return { agent_id: agent_id, goal: goal, state: S.INIT, history: [], pending_action: None, last_result: None, final_answer: None, }4.2 状态迁移函数Agent 大脑的单一入口def step(ctx: dict, event_type: str, payload: dict) - dict: state ctx[state] if state S.INIT and event_type START: ctx[history].append({role: user, content: payload.get(input, )}) ctx[state] S.THINKING elif state S.THINKING and event_type LLM_RESPONSE: thought payload.get(thought, ) action payload.get(action, ) ctx[history].append({role: assistant, content: thought}) if action finish: ctx[state] S.DONE ctx[final_answer] payload.get(answer, ) else: ctx[state] S.ACTION ctx[pending_action] action elif state S.ACTION and event_type TOOL_FINISHED: result payload.get(result, ) ctx[history].append({role: tool, content: str(result)}) ctx[last_result] result ctx[pending_action] None ctx[state] S.THINKING elif state S.WAITING and event_type USER_APPROVED: decision payload.get(approved, False) ctx[history].append({role: user, content: f审批结果: {decision}}) ctx[state] S.THINKING return ctx这其实就是前面伪代码的直接落地版。你可以看到它完全没有循环每次调用只走一步。想让它走下一步外部必须再向它发一个事件。4.3 模拟 LLM 调用器和工具执行器LLM 调用器在外部模拟一个异步进程——收到 THINKING 状态的上下文调用 LLM 返回“思考/行动决策”。这里用一个随机策略模拟 LLM 的输出import random, asyncio async def fake_llm_invoke(ctx: dict): await asyncio.sleep(0.5) if len(ctx[history]) 4: return { thought: 我需要调用一个工具来获得更多信息, action: search, query: ctx[history][-1][content], } else: return { thought: 我已经拿到足够的信息可以总结了, action: finish, answer: 模拟总结 str(ctx[last_result]), }工具执行器也很简单async def fake_tool_execute(ctx: dict): action ctx[pending_action] await asyncio.sleep(0.3) return f{action} 的执行结果是: 一些模拟数据4.4 控制器循环HTTP 驱动即可关键来了。你怎么推动这个 Agent 往下走答案是 HTTP 接口。Agent 不再自己跑循环而是每次由外部调用一个 HTTP 接口“踢一脚”。我们用 FastAPI 把 Agent 的核心暴露成两个接口创建 Agent、向 Agent 推送事件。from fastapi import FastAPI, HTTPException app FastAPI() contexts {} app.post(/agents/{agent_id}) async def start_agent(agent_id: str, goal: str): contexts[agent_id] new_context(agent_id, goal) return step(contexts[agent_id], START, {input: goal}) app.post(/agents/{agent_id}/events) async def push_event(agent_id: str, event_type: str, payload: dict): if agent_id not in contexts: raise HTTPException(status_code404, detailagent not found) ctx step(contexts[agent_id], event_type, payload) if ctx[state] S.THINKING: # 在真实环境里这里可以往消息队列丢事件让 worker 去调 LLM llm_result await fake_llm_invoke(ctx) ctx step(ctx, LLM_RESPONSE, llm_result) if ctx[state] S.ACTION: result await fake_tool_execute(ctx) ctx step(ctx, TOOL_FINISHED, {result: result}) return {state: ctx[state], context: ctx}这个例子虽然简化了但骨架已经出来了你每调用一次/agents/{agent_id}/eventsAgent 就走一步走完一步后如果它还想继续它会停在新的状态上等你发下一个事件。你完全可以在两次事件之间等很久、插入人工审核、甚至重启整个服务——只要上下文还在Agent 就能接着走。4.5 模拟一次完整对话# 1. 创建 Agent curl -X POST http://localhost:8000/agents/agent-001?goal帮我调研一下xxx # 2. 手动推动给它一个初始输入 curl -X POST http://localhost:8000/agents/agent-001/events?event_typeLLM_RESPONSE \ -H Content-Type: application/json \ -d {thought:先搜索看看, action:search, query:xxx} # 3. 等它自己走完 THINKING - ACTION # 4. 再推一个 TOOL_FINISHED 事件 curl -X POST http://localhost:8000/agents/agent-001/events?event_typeTOOL_FINISHED \ -H Content-Type: application/json \ -d {result:搜索到了一些内容}在真实系统里第二步和第四步通常是由后台 worker 自动完成的——LLM 返回结果后worker 自动把事件发回来工具执行完工具服务也会发事件回来。人工只需要在关键的审批节点介入即可。Agent 整体的行为从一个“蒙着头自己跑完的程序”变成了一个“每一步都有迹可循、每步都可以被控制的服务”。5. 拆掉循环后架构上多出来的那些红利5.1 可观测性和审计能力大幅提升while 循环版你要看 Agent 在干嘛只能到处打日志。事件驱动状态机版不一样每个事件本身就是一条结构化日志。把事件持久化到数据库你就拥有了一张完整的 Agent 运行时间线START任务目标是什么LLM_RESPONSE模型当时想了什么、决定做什么TOOL_FINISHED工具返回了什么USER_APPROVED谁在什么时候做了审批这对排查问题几乎是降维打击。有一次我们的 Agent 在线上执行了一笔错误的操作我用 while 循环版的时候只能翻半天日志去猜是模型幻觉还是工具返回了脏数据换成状态机版后直接按 agent_id 拉出事件流十分钟就定位到了是工具服务在高负载下返回了一个超时错误被 Agent 误当成了“正常空结果”处理了。5.2 暂停、恢复、人工审批成为一等公民事件驱动架构强制你把“运行中”和“等待中”分开。while 循环版最怕的就是 Agent 执行到一半需要人来确认现在这只是一个状态转换Agent 在 THINKING 阶段判断出“这个动作需要人工审批”然后迁移到 WAITING 状态该发通知发通知该释放资源释放资源。整个服务没有任何线程被占用。人的审批动作只是一个新的 HTTP 事件或消息队列事件被推进来Agent 从 WAITING 恢复成 THINKING继续干活。这在比如自动发邮件、自动下单、自动修改线上配置等场景下非常关键我甚至敢说没有这个能力Agent 根本不敢真正接入高权限业务动作。5.3 水平扩展不再是噩梦while 循环版的 Agent 状态在内存里这意味着两个 Agent 之间不能互相接管任务一个节点故障就意味着它承载的所有 Agent 全部死亡。状态机版把状态下沉到 Redis 或数据库后Agent 的处理逻辑变成纯函数——只有“当前状态 事件 → 新状态”跟哪台机器执行完全无关。这样一来天然可以做 N 个 worker 进程消费事件队列同一个 Agent 的后续事件完全可以被另一个 worker 处理。需要扩并发多挂几个 worker 就行需要缩容等队列里的任务处理完即可不用像以前那样一台机器跑着几十个 Agent 只能硬扛。5.4 多 Agent 编排终于有了抓手做复杂任务的时候我经常需要一个“主编排 Agent”管理多个子 Agent一个负责搜索一个负责总结一个负责质检。while 循环版去做编排只能一套循环套另一套循环嵌套深了以后代码根本没法维护任何一个子 Agent 卡住都会把整条链卡死。状态机版就不一样了。每个 Agent 可以独立处于不同的状态编排器只需要根据各个子 Agent 发过来的事件决定下一步推动谁。子 Agent A 发来DONE事件编排器就给子 Agent B 发一个START事件。Agent 之间完全解耦整个流程变成了可编排、可观测的“事件流水线”。这也是我后来把系统往工作流引擎方向演进的基础。6. 常见问题与排查技巧实录6.1 状态机卡在某个状态不往前走怎么排查我的排查顺序是先查事件是否发出再查事件是否被处理最后查状态迁移逻辑。事件驱动架构有个经典坑状态机本身没坏是事件丢了。比如 LLM 调用超时了但没有产生任何失败事件这时候 Agent 就会永远停在 THINKING 状态。我的建议是每个状态迁移都要有配套的超时兜底机制。可以给 Agent 加一个last_updated_at时间戳外部巡检任务扫描所有“卡住”的 Agent比如在 THINKING 状态超过 30 秒自动重发事件或者标记失败。6.2 上下文越攒越大token 消耗爆炸怎么办while 循环版里每次循环都会把全部历史交给 LLM状态机版也一样有这个隐患。但状态机版因为每一步都是独立的更容易做上下文裁剪和摘要。我的经验是在 THINKING 状态触发 LLM 调用之前先对 history 做一轮压缩。比如超过 20 轮对话后把最早的历史交给一个“摘要 LLM”生成一段概括再拼进 prompt。同时工具结果在进入历史前做字段裁剪——只保留关键的返回值不要一股脑把整个 JSON 倒进去。这招能把 token 成本降低 30% 到 50%响应速度也快不少。6.3 事件被重复投递状态迁移被做了两次要不要紧这是事件驱动系统最经典的幂等性问题。HTTP 超时会触发客户端重试消息队列有 at-least-once 语义这些都会导致同一个事件被投递两次。状态机如果没有幂等处理就可能出现“工具执行了两次”的严重事故。我的处理办法是给每个 Agent 实例维护一个processed_event_ids集合在处理事件前先检查事件 ID 是否已经处理过同时在存储层给上下文加重试锁前面的示例里用了store.lock来防并发。这两个保险叠加之后重复事件的危害就能基本消除。6.4 开发调试体验会不会变差这是我一开始最担心的地方。担心状态机版“不像 while 循环那样全局都在一个函数栈里跑打断点就能看清所有变量”。实测下来发现状态机版调试其实更舒服。因为每次事件处理都是纯函数我可以用“事件流回放”的方式调试把一个生产环境出错的 agent_id 的所有事件拉出来在本地按事件流顺序重放一遍每一步的输入输出都清清楚楚。相比 while 循环时代靠“重新跑一遍加日志”的调试方式效率高太多了。本地开发时我建议搞一个可视化的“推进工具”一条条手动触发事件看 Agent 的状态迁移是否符合预期。既是调试工具也是验收工具。6.5 从一个 while 循环 Agent 迁移过来建议从哪个场景开始不要一上来就重构整个复杂 Agent。我先推荐三个适合切入的场景需要人工审批的 Agent。这是收益最明显的场景因为你立刻就能感受到 WAITING 状态带来的便利。长耗时任务 Agent。比如调研、报告生成可能要跑十几分钟。状态机版可以随时暂停、恢复容错能力远超 while 循环。多 Agent 协作场景。把每个 Agent 拆成状态机之后编排逻辑会清爽一大截。先在一个小场景里跑通“事件驱动 状态持久化”这套模式攒足了经验和信心再推广到其他 Agent 上。我自己就是这么一步步迁移过来的第一次成功改造一个需要人工审批的 Agent 之后后面所有的 Agent 我都是直接按新架构写了。我个人在实际操作中体会最深的一点是拆 while 循环这个动作表面上是改代码结构实质上是强迫你把 Agent 当成一个“可以被管理的事务”而不是一个“会自己跑的程序”。如果你只想写个一次性脚本这个改造纯属浪费但如果你要让 Agent 稳定跑在业务系统里这套架构的投资回报率非常高。把 Agent 从“自转的陀螺”变成“被有序推动的齿轮”你会发现之前那些不敢做的业务场景突然全都敞开了。