企业官网建设流程全解析

前两周一个朋友把他们的 Agent 服务代码发给我看，让我帮忙 review。我扫了几百行，最后发现问题不在某个工具函数，也不在 prompt，而在主干逻辑——一段典型的 while 循环：把用户消息塞给 LLM，解析出工具调用，执行工具，把结果拼回上下文，再问 LLM，直到它输出“完成”。我当时的评语很直接：这套代码在 demo 阶段没问题，但如果你想上生产、想抗并发、想排查线上问题，我建议你把 while 循环拆掉。他愣了一下，问我拆掉之后用什么来承接“不断思考下一步”的逻辑。这篇文章就是我对那个问题的完整回答，也是我踩过不少坑之后沉淀下来的一套 Agent 架构思路。

这套思路本质上做一件事：把 Agent 从“一段会自我迭代的循环控制流”变成“一张事件驱动的状态转移表”。循环不再藏在一个个 while 关键字里，而是显式地分布在状态、事件和处理器之间。听起来抽象，但落地之后你会发现，Agent 突然变得像普通后端服务一样可控、可观测、可恢复。这篇文章会讲清楚传统 while 循环到底把债务藏在了哪里，再把拆掉循环后的事件驱动状态机架构一步步拆开，最后把我在迁移过程中踩过的七个坑原原本本写出来。

1. while 循环 Agent：一套看着顺手但很难长大的样板

很多人第一次写 Agent，基本都会落成同一个结构。这不是偶然，因为 ReAct 范式、各类开源框架、甚至大模型厂商的官方示例，都在反复强化这个模式。它足够简单，简单到用一个 while 就能装下“智能”的全部幻觉。

1.1 循环里包的是什么：观察-推理-行动三部曲

传统的 while 循环 Agent 长成这个样子：

history = [user_input] while True: response = llm.chat(history) # 推理 action = parse_action(response) # 解析动作 if action.type == "final_answer": # 退出条件 return action.content result = execute_tool(action) # 执行动作 history.append(result) # 观察结果 # 然后回到循环顶部，再来一轮

每一步的语义非常直白：让 LLM 看当前的上下文，决定下一步要干什么，调用工具，把工具的输出作为新的观察写回历史，然后再让 LLM 继续。整个过程像一个人在黑暗里摸象——摸一块，思考一下，再摸下一块。

这种模式的优点是真的很多。第一，理解成本趋近于零，实习生看五分钟就能照葫芦画瓢。第二，调试很直观，print 一下 history 就能看到完整轨迹。第三，对 LLM 的约束极少，每一步都给足上下文，模型可以自由探索。说白了，这个模式的本质是“把控制权完全交给模型，循环只是一个伺候模型的仆人”。

所以我不反对 while 循环本身。它完全符合“快速验证一个想法的成本最低方案”。问题在于，它把 Agent 的所有状态都隐式地放在了一个局部变量 history 里，把“接下来要干什么”放在了一个隐式的控制流里。当业务复杂度超过某个临界点，这些“隐式”会变成一笔笔需要偿还的债。

1.2 循环的隐性成本：状态、并发、恢复三大债

我们一件一件说。第一笔债是状态不可见。while 循环里面，Agent 当前在哪一步、已经调用过哪些工具、距离最终输出还差什么，全部被压缩在 memory 这个列表里。程序外部完全感知不到。你想做一个“任务详情页”，显示这个 Agent 跑到第几步了？做不到。你想在中途插入一个人工审批？也做不了，因为循环根本不会停在一个可以被外部干预的位置上。

第二笔债是并发能力极差。多用户同时用，就得为每个用户开一个线程或进程，每个线程各自跑一个 while 循环。这个模型下，并发量跟线程数直接挂钩，线程一多，上下文切换开销、共享内存竞争、内存占用都上来了。更麻烦的是，如果中途某个用户的 Agent 卡在等待外部工具响应，这个线程就被白白占住，什么活也干不了。这不叫并发，这叫“并行地各自阻塞”。

第三笔债是最致命的：不可恢复。循环跑着跑着进程崩溃了，你怎么办？从头再来一遍。如果是运行 30 秒的一次性任务，从头再来还能忍；如果是已经运行了几个小时、已经调用了十几个外部 API、已经花掉了大量 token 的长任务，重新跑一遍就是灾难。我之前见过一个数据采集型的 Agent，跑到第 40 步时因为一个网络抖动崩了，用户第二天发现任务卡死，重新发起之后又重复消耗了上千次 LLM 调用。这种问题，while 循环结构本身是解不了的，因为它根本没有“检查点”的概念。

这三个债放在一起，本质上是同一个问题的三个侧面：while 循环把 Agent 的控制流和状态全部私有化、隐藏化了。你把它当脚本用，它什么问题都没有；你把它当服务用，它到处都是坑。

2. 拆掉循环：把 Agent 重新建模成事件驱动状态机

拆掉 while 循环之后，新的主干逻辑不再是“循环 + 局部变量”，而是一个显式的状态机加一张事件表。这是架构层面的转向，从“主动地一轮一轮往下转”变成“对每个到达的事件做一次状态转移”。

2.1 核心转变：从主动轮询到被动响应

如果只挑一个最重要的变化，我会说：原来的 Agent 是主动的，新架构里的 Agent 是被动的。while 循环像一条不断抽打的鞭子，逼着 Agent 往前走；事件驱动的状态机则像一套自动变速箱，整车在什么状态、收到什么信号、该换到什么挡位，全都写在明面上。

一辆车的变速箱肯定不会在代码里写：while (车速还能加) { 换挡(); }。它是由一堆机械结构和电控逻辑组成的状态网：当前在几挡、转速多少、油门踩多深，共同决定下一步怎么走。Agent 也是一样。它的每次动作，都应该由“当前状态 + 到达事件”共同决定，而不是由循环体里的一行行代码隐式驱动。

落到代码上，新架构的核心不再是 while，而是一个分发器：

class Agent: def __init__(self, state_id, snapshot): self.state_id = state_id # 当前状态，比如 "planning" self.snapshot = snapshot # 状态快照，包含历史、上下文、预算等 def handle(self, event: Event) -> AgentResult: handler = self.handlers[self.state_id][event.type] return handler(self, event)

每来一个事件，Agent 就根据当前状态找到对应的事件处理器，执行完逻辑之后，要么迁移到新状态，要么保持当前状态等待下一个事件。循环还在，但它从“while True”变成了“谁都看不见的调度器内部循环”，业务代码里不再有任何循环。

2.2 状态与事件怎么定义：一张能直接抄的转移表

真正动手设计的时候，最大的难点不是写代码，而是定义状态和事件。定义得好，整个系统清晰得像一张地图；定义得不好，状态机比 while 循环还乱。

我常用的定义方法是把 Agent 的整个生命周期拆成几个稳定的阶段：idle（等待任务）、planning（制定或更新计划）、await_tool（等待工具结果）、await_human（等待人工审批或补充信息）、finalized（已完成）、error（出错终止）。每个阶段就是一个状态，状态之间通过事件迁移。

下面这张表是我做过的一个“企业数据分析 Agent”的状态转移表，可以当成一个可参考的起点：

这张表的价值不在于状态名称多么标准，而在于它把 Agent 所有可能的行为路径都显式化出来了。你看一眼就知道：工具超时了会去哪里、人工拒绝了会发生什么、预算超了会怎样。这就是把控制权从“模型自由发挥”收回到“系统明确约束”的过程。LLM 依然负责想“怎么办”，但整个“什么时候想、想多久、想完能干什么”都由状态机说了算。

状态快照则是另一个关键。每个 Agent 实例的 snapshot 通常包含：当前的用户目标、历史消息列表、已经用掉的 token 预算、已经执行过的工具调用记录、当前正在等待的事件编号等。这个快照是一个普通数据结构，可以序列化成 JSON、存进 MySQL、扔进 Redis，随时可以拿出来恢复。快照就是整个状态机对外暴露的“记忆”。

3. 实操落地：从 while 循环迁移到状态机的完整过程

理论讲再多，不如一次真实的重构。这一节我会用一个极简例子，演示一个“搜索资料并撰写摘要”的 Agent 如何从 while 循环版本迁移到状态机版本。

3.1 原循环代码怎么拆：逐步重构示例

原始版本长这样：

async def run_search_agent(query): history = [{"role": "user", "content": query}] final_answer = None while final_answer is None: resp = await llm.chat(history) action = parse_action(resp) if action.type == "search": results = await search_web(action.query) history.append({"role": "tool", "content": results}) elif action.type == "summarize": history.append({"role": "user", "content": f"请用以上搜索结果撰写摘要"}) final_answer = await llm.chat(history) elif action.type == "done": final_answer = action.content else: raise ValueError(f"unknown action: {action.type}") return final_answer

功能上它能跑，但问题也很明显：如果 search_web 要等五秒，整个进程就在等；如果中途进程挂了，上一次的搜索结果全丢；如果 LLM 连续十次都输出 search 而不收敛，用户只能眼睁睁看着 token 烧没。这些痛点的根源都在“循环”两个字上。

迁移的第一步是拆分职责。把 while 循环体里的“一次迭代”提取成一个独立函数，这次迭代的输入是当前快照和当前动作结果，输出是一个事件列表。第二步是把所有的 return 和 break 改写成事件。第三步是把工具调用从同步顺序调用改成异步事件投递。第四步是引入快照的持久化。

重构后的核心代码大致是这个形态：

class SearchAgent: # 状态和事件类型 STATE_PLANNING = "planning" STATE_AWAIT_TOOL = "await_tool" STATE_FINALIZED = "finalized" EVENT_PLAN_READY = "plan_ready" EVENT_TOOL_RESULT = "tool_result" EVENT_TOOL_TIMEOUT = "tool_timeout" EVENT_BUDGET_EXCEEDED = "budget_exceeded" def __init__(self, snapshot: dict): self.state = snapshot.get("state", self.STATE_PLANNING) self.snapshot = snapshot self.snapshot.setdefault("history", []) self.snapshot.setdefault("steps", 0) def handle(self, event: Event) -> AgentResult: if self.state == self.STATE_PLANNING: if event.type == self.EVENT_PLAN_READY: # 根据计划发起工具调用，并投递一个“等待工具结果”的事件 self.state = self.STATE_AWAIT_TOOL return AgentResult( action="call_tool", tool="search_web", args={"query": event.payload["query"]}, on_result=self.EVENT_TOOL_RESULT, ) elif self.state == self.STATE_AWAIT_TOOL: if event.type == self.EVENT_TOOL_RESULT: results = event.payload["data"] self.snapshot["history"].append({"role": "tool", "content": results}) self.state = self.STATE_PLANNING # 继续回到 planning，等待下一次计划事件 return AgentResult(action="llm", prompt="根据工具结果继续规划") if event.type == self.EVENT_TOOL_TIMEOUT: self.snapshot["steps"] += 1 if self.snapshot["steps"] > 10 or self.snapshot["budget"] <= 0: self.state = self.STATE_FINALIZED return AgentResult(action="stop", reason="budget_exceeded") # 否则重试同一工具 return AgentResult(action="call_tool", tool="search_web", args=event.payload.get("args")) return AgentResult(action="idle")

你还是能在这个状态机里找到“循环”的影子：planning 收到 plan_ready 会跳去 await_tool；await_tool 收到 tool_result 会跳回 planning。但这个循环是显式写在状态转移表里的，每一步都是由外部事件驱动的。这个循环虽然逻辑上存在，但它在代码结构上被拆掉了，取而代之的是状态和事件。

3.2 排队与并发：事件泵、消息队列和 Agent 实例

拆掉 while 循环之后的并发模型完全变了。原来是一个线程跑一个循环；现在变成了一个事件泵 + 多个 Agent 实例的模式。

在单机部署里，事件泵可以简单到一个进程内的队列：

async def event_loop(queue): while True: event = await queue.get() agent = load_agent(event.agent_id) # 从存储加载 Agent 快照 result = agent.handle(event) # 执行状态转移 save_agent(agent) # 持久化新快照 for new_event in result.outbox: await queue.put(new_event) # 投递后续事件

注意：这个 while 跟业务无关，它是一个基础设施层的消费循环，跟 Agent 架构解耦了。你把它替换成 RabbitMQ worker、Kafka consumer、SQS listener，完全不影响业务代码。这就是拆掉循环的关键收益之一：Agent 的执行模型不再绑定在某个线程栈上，而是绑定在可路由的事件流上。

并发量上来之后，你会自然地走向分布式版本：事件进 MQ，多个 worker 同时消费，每个事件里带一个 agent_id，同一个 agent_id 的事件必须路由到同一个 worker，不同的 agent_id 可以并行处理。这个约束很关键：状态机一次只能处理一个事件，同一 Agent 实例的事件如果被两个线程并发处理，状态必然打架。所以正确的做法不是让 Agent 内部支持并发，而是让“不同 Agent 并行、同一 Agent 串行”。

3.3 记忆与恢复：状态快照、检查点与事件重放

状态机架构还有一个 while 循环望尘莫及的能力：任意时刻保存现场、随时恢复继续跑。

我的做法是给每个 Agent 实例维护一个持续追加的事件日志，同时每隔 N 个事件生成一个快照。快照是“截至某个时间点的完整状态”，事件日志是“快照之后发生的增量”。重启之后，先读快照，再重放增量事件，就能恢复到一个精确的位置。

这个机制一旦跑通，Agent 的可靠性从“看运气”变成“看设计”。我做系统设计时，凡是涉及长时间运行、外部工具调用、大额 token 消耗的 Agent，都会显式加一层周期快照。之前那个数据采集 Agent，崩溃恢复就变成了打开任务详情页，点一下“恢复”，系统自动从上次快照继续，不用再重新消费 API。

事件日志还有一个附加价值：它是训练和分析的数据金矿。日志里记录的是每一步决策背后的输入输出、工具返回、异常事件。想分析“为什么 Agent 在这个 case 上表现差”，直接查事件日志就行，不用再靠用户截图描述。

4. 踩坑实录：拆掉循环之后我遇到过的 7 个典型问题

拆掉 while 循环绝对不是没有代价。状态机的写法比循环啰嗦至少三倍，而且它把很多以前“靠循环自动地串起来”的东西变成了“靠代码显式地连起来”，漏掉一个事件类型，Agent 就可能卡死在某个状态。这里把我踩过的最典型的七个坑写出来，当成速查表供你参考。

七条问题里，我实际花费时间最多的反而是第二条。快照和事件日志的原子性，一度让我恢复出来的 Agent 回到一个不存在的状态。后来我把“保存快照”和“追加事件日志”放进了同一个数据库事务里，问题才彻底消失。原则就一句话：状态转移必须和状态记录同时提交，不能先改状态再补日志，也不能先记日志再改状态。

另一个让我记忆深刻的是第三条。我在生产环境跑了一段时间之后，发现偶发出现“Agent 明明上下文中已经有工具结果，下一轮却还在等那个结果”的情况。排查半天，发现原因是两个 worker 同时消费了同一个 agent_id 的两个事件，后一个事件处理完之后覆盖了前一个事件的状态。加了 agent_id 的一致性哈希路由之后，这个问题再没出现过。

5. 边界判断：什么场景才值得拆掉 while 循环

如果方案只是“状态机更好”，那我这篇文章就没有太大价值了。现实情况恰恰相反——有一类场景，while 循环依然是最优解，硬套状态机反而自找麻烦。这一节我把判断标准写清楚。

5.1 其实可以不拆的三种情况

第一种是纯教学和验证思路的 demo。目标只是“看 LLM 能不能自主调用工具完成任务”，这时候循环的直观性胜过一切。你不需要它可恢复，也不需要它抗并发，就更不用拆。

第二种是一次性脚本任务。跑完之后就结束，任务生命周期就在几秒到几十秒之间，出错了重跑整个脚本即可。给这种任务专门设计状态快照，属于过度设计。

第三种是单步工具链。也就是 Agent 的路径是完全确定的：先调用 A，再调用 B，再调用 C，最后生成结论。这种场景本质是 pipeline，用一个顺序列表保存步骤就够了，状态机的灵活性反而是多余的。

在这三种场景里强行拆掉 while 循环，只会让代码变长、变绕、变量增多，收益几乎为零。所以我给朋友的建议从来不是“所有 Agent 都必须状态机”，而是“先看清楚你的 Agent 是否会出现不可控的循环、中断、并发、长时运行这四个信号”。

5.2 我的选择标准：四个问题帮你快速决策

我在评估一个新 Agent 项目要不要拆 while 循环时，会问自己四个问题。每个问题只要命中一个，我就会在架构设计里优先考虑事件驱动状态机。

第一个问题：这个 Agent 会运行超过一分钟吗？超过一分钟意味着中途崩溃的概率不是零，而一旦中途崩溃，恢复成本就成了核心指标。第二个问题：这个 Agent 会在运行过程中等待外部资源吗？等待人工审批、等待第三方 API、等待下游工人执行操作，这些都是“外部事件”，while 循环天然处理不好。第三个问题：会不会有多个用户同时使用同一个 Agent 服务？只要涉及“多人使用”，串行循环就一定会被并发问题拖垮。第四个问题：我们能不能接受“从头再来”？如果答案是不能，那说明你需要检查点，而检查点恰恰是循环结构最不擅长提供的东西。

如果四个问题全是“否”，那我还挺乐意在你的代码里看到一个干净的 while 循环。如果有一到两个答案是“是”，我的经验是：提前拆，不要等出线上事故再拆——出事故的时候不仅是代码要改，用户信任、任务数据、团队心态都有折损。

把 while 循环拆成事件驱动状态机，本质上不是一种“更高级”的做法，而是一种“更符合服务化需求”的做法。Agent 在 demo 阶段可以是一个脚本；一旦它要承载真实用户、真实任务、真实成本，它就必须像普通后端服务一样，拥有状态管理、事件路由、故障恢复、并发隔离的能力。这跟是不是用某个框架无关，跟用不用 state machine 库也无关，核心在于你是否愿意把 Agent 的控制流和状态从“隐藏”变成“显式”。

我个人的体会是，状态机迁移最大的收获不是代码架构本身，而是让我终于能在任何时刻回答三个问题：这个 Agent 现在在哪个状态？它正在等什么事件？它已经做了什么？如果哪天你的 Agent 出了问题、而你发现你答不上来这三个问题，那大概率就是 while 循环该拆的时候了。迁移启动前还有一个值得记住的窍门：第一版不要一次性把所有状态都定义完。把 while 循环里已有的分支整理成状态和事件，先跑起来，再逐步扩展，你会发现这套架构的扩展性远比你想象的宽松。