生产级 Agent Loop 的状态机设计：从 while 循环到可恢复执行引擎

摘要

很多人第一次写 Agent，都会写出类似下面的代码：

whileTrue:response=llm(messages)ifresponse.final:returnresponse.text result=run_tool(response.tool_call)messages.append(result)

这段代码能跑 demo，但很难上生产。真实系统需要处理取消、超时、权限审批、工具失败、上下文压缩、Checkpoint、Trace、成本控制、人工接管等问题。生产级 Agent Loop 的本质不是循环，而是一个显式状态机。

OpenAI Agents SDK 把运行、工具执行、handoff、guardrails、sessions、tracing 作为一套编排能力；AutoGen AgentChat 也把 team、termination condition、state、message flow 放到核心抽象里。这些设计说明：工程化 Agent 必须让每一步“可控制、可恢复、可审计”。

一、为什么while true不够

最小循环的问题有四类。

第一，状态是隐式的。系统不知道自己当前是在“等待模型”“执行工具”“等待审批”还是“恢复中”。

第二，终止条件不完整。模型不返回 final answer，就可能一直跑。

第三，工具执行没有边界。模型说要执行什么，系统就执行什么，风险极高。

第四，失败不可恢复。进程崩溃后，只剩一堆聊天记录，很难知道执行到哪里。

生产级 Agent Loop 要解决的核心问题是：

给定一个不稳定的概率模型，如何用确定性的工程外壳约束它完成任务？

二、最小状态机模型

一个更可靠的 Agent Loop 可以拆成以下状态：

CREATED -> PREPARE_CONTEXT -> MODEL_THINKING -> DECIDE_ACTION -> WAIT_APPROVAL -> EXECUTE_TOOL -> OBSERVE -> COMPACT_OR_SAVE -> MODEL_THINKING -> FINISHED | FAILED | CANCELLED | SUSPENDED

状态职责如下：

三、源码示例：用状态机重写 Agent Loop

下面是一个简化的 Python 实现。它不依赖具体模型厂商，重点展示状态如何显式流转。

fromdataclassesimportdataclass,fieldfromenumimportEnum,autofromtypingimportAny,LiteralimporttimeimportuuidclassRunState(Enum):CREATED=auto()PREPARE_CONTEXT=auto()MODEL_THINKING=auto()DECIDE_ACTION=auto()WAIT_APPROVAL=auto()EXECUTE_TOOL=auto()OBSERVE=auto()COMPACT_OR_SAVE=auto()FINISHED=auto()FAILED=auto()CANCELLED=auto()SUSPENDED=auto()@dataclassclassBudget:max_turns:int=20max_seconds:int=300max_tool_failures:int=3max_tokens:int=80_000@dataclassclassAgentRun:user_input:strrun_id:str=field(default_factory=lambda:str(uuid.uuid4()))state:RunState=RunState.CREATED messages:list[dict[str,Any]]=field(default_factory=list)turn_count:int=0token_count:int=0tool_failures:int=0started_at:float=field(default_factory=time.time)pending_tool_call:dict[str,Any]|None=Nonefinal_answer:str|None=Noneerror:str|None=Nonecancelled:bool=False

这里的关键点不是代码复杂，而是状态被显式建模了。后面所有逻辑都可以围绕RunState做测试。

四、源码示例：终止条件不要写进 Prompt

很多系统会在 prompt 里写“不要无限循环”。这不够。终止条件必须由代码控制。

defshould_stop(run:AgentRun,budget:Budget)->tuple[bool,str|None]:ifrun.cancelled:returnTrue,"cancelled_by_user"ifrun.turn_count>=budget.max_turns:returnTrue,"max_turns_exceeded"iftime.time()-run.started_at>budget.max_seconds:returnTrue,"timeout"ifrun.tool_failures>=budget.max_tool_failures:returnTrue,"too_many_tool_failures"ifrun.token_count>=budget.max_tokens:returnTrue,"token_budget_exceeded"returnFalse,None

AutoGen 的 termination condition 设计也体现了这个原则：终止条件是代码层面的、有状态的、可组合的对象，而不是一句提示词。

五、源码示例：主循环的正确形状

下面是一个更完整的执行骨架：

defrun_agent(user_input:str,llm,tool_registry,policy,store,tracer)->AgentRun:run=AgentRun(user_input=user_input)budget=Budget()try:whilerun.statenotin{RunState.FINISHED,RunState.FAILED,RunState.CANCELLED}:stop,reason=should_stop(run,budget)ifstop:run.state=RunState.CANCELLEDifreason=="cancelled_by_user"elseRunState.FAILED run.error=reasonbreaktracer.event(run.run_id,"state",{"state":run.state.name})ifrun.state==RunState.CREATED:run.messages.append({"role":"user","content":user_input})run.state=RunState.PREPARE_CONTEXTelifrun.state==RunState.PREPARE_CONTEXT:run.messages=build_context(run,store)run.state=RunState.MODEL_THINKINGelifrun.state==RunState.MODEL_THINKING:response=llm.generate(run.messages)run.turn_count+=1run.token_count+=response.usage.total_tokens run.messages.append(response.to_message())run.state=RunState.DECIDE_ACTIONelifrun.state==RunState.DECIDE_ACTION:action=parse_model_action(run.messages[-1])ifaction.kind=="final":run.final_answer=action.text run.state=RunState.FINISHEDelifaction.kind=="tool":run.pending_tool_call=action.tool_call run.state=RunState.WAIT_APPROVALelse:run.error=f"unsupported_action:{action.kind}"run.state=RunState.FAILEDelifrun.state==RunState.WAIT_APPROVAL:decision=policy.evaluate(run.pending_tool_call)ifdecision=="deny":run.error="tool_denied"run.state=RunState.FAILEDelifdecision=="ask":run.state=RunState.SUSPENDED store.save_checkpoint(run)breakelse:run.state=RunState.EXECUTE_TOOLelifrun.state==RunState.EXECUTE_TOOL:try:result=tool_registry.execute(run.pending_tool_call)run.messages.append({"role":"tool","tool_call_id":run.pending_tool_call["id"],"content":clean_tool_output(result),})run.state=RunState.OBSERVEexceptExceptionasexc:run.tool_failures+=1run.messages.append({"role":"tool","content":f"tool_error:{type(exc).__name__}:{exc}",})run.state=RunState.OBSERVEelifrun.state==RunState.OBSERVE:run.pending_tool_call=Nonerun.state=RunState.COMPACT_OR_SAVEelifrun.state==RunState.COMPACT_OR_SAVE:run.messages=maybe_compact(run.messages,budget)store.save_checkpoint(run)run.state=RunState.MODEL_THINKINGelifrun.state==RunState.SUSPENDED:store.save_checkpoint(run)breakexceptExceptionasexc:run.state=RunState.FAILED run.error=f"system_error:{type(exc).__name__}:{exc}"store.save_checkpoint(run)returnrun

这段代码体现了几个生产级原则：

• 模型只提出候选动作，系统决定是否执行。
• 每一步都有状态。
• 每轮都有终止条件检查。
• 工具失败不会直接炸掉进程，而是变成 Observation。
• Checkpoint 不是最后才保存，而是在关键边界保存。
• 高风险工具可以进入SUSPENDED，等待人类批准。

六、工具并发：主状态机串行，工具可控并发

生产系统不应让所有东西都并发。推荐原则是：

主状态机串行。 只读工具可以受控并发。 写工具和外部副作用默认串行。 并发结果必须按确定顺序归并。

示例：

READ_ONLY_TOOLS={"read_file","search_code","list_dir"}defcan_parallelize(tool_calls:list[dict[str,Any]])->bool:returnall(call["name"]inREAD_ONLY_TOOLSforcallintool_calls)defmerge_tool_results(tool_calls,results):by_id={result.tool_call_id:resultforresultinresults}return[by_id[call["id"]]forcallintool_calls]

这里的重点是“归并顺序必须确定”。否则同一次任务可能因为调度顺序不同得到不同上下文。

七、Checkpoint 设计

Checkpoint 不应该只保存聊天记录。至少应保存：

@dataclassclassCheckpoint:run_id:strstate:strmessages:list[dict[str,Any]]turn_count:inttoken_count:inttool_failures:intpending_tool_call:dict[str,Any]|Nonepermission_decisions:list[dict[str,Any]]context_artifacts:list[str]created_at:float

恢复逻辑示例：

defresume_agent(run_id:str,store,approval_result:Literal["allow","deny"]|None):run=store.load_checkpoint(run_id)ifrun.state!=RunState.SUSPENDED:raiseValueError("only suspended runs can be resumed safely")ifapproval_result=="allow":run.state=RunState.EXECUTE_TOOLelifapproval_result=="deny":run.state=RunState.FAILED run.error="approval_denied"else:raiseValueError("approval_result required")returnrun

注意：恢复时不要跳过权限检查。恢复的是运行状态，不是绕过安全流程。

八、Trace：让失败可以被解释

一个 Agent trace 至少应该记录：

classTracer:defevent(self,run_id:str,name:str,payload:dict[str,Any])->None:print({"run_id":run_id,"event":name,"payload":payload,"ts":time.time(),})

建议记录的事件：

• state_transition
• llm_call_started
• llm_call_finished
• tool_call_requested
• permission_evaluated
• tool_call_finished
• context_compacted
• checkpoint_saved
• run_finished

OpenAI Agents SDK 的 tracing 覆盖模型生成、工具调用、handoff、guardrail、自定义事件。自研 Harness 可以参考这种事件结构。

九、普通人怎么理解这件事

可以把 Agent Loop 想成一辆自动驾驶车：

• 模型是司机，负责判断下一步。
• 状态机是交通规则，决定什么时候能走、什么时候必须停。
• 权限系统是门禁。
• 工具系统是车辆执行机构。
• Trace 是行车记录仪。
• Checkpoint 是服务区和救援点。

没有状态机的 Agent，就像只有司机、没有刹车、没有仪表盘、没有行车记录仪的车。

十、落地检查表

• 是否有显式状态枚举？
• 是否有代码级终止条件？
• 是否支持取消、暂停、恢复？
• 是否限制工具并发？
• 是否把工具失败转成可处理 Observation？
• 是否保存 checkpoint？
• 是否记录 trace？
• 是否能复现一次失败？
• 是否能区分模型错误、工具错误、权限错误、上下文错误？

结论

生产级 Agent Loop 的目标不是让模型多跑几轮，而是让系统每一轮都知道自己在做什么、为什么这么做、什么时候该停、失败后从哪里恢复。Agent Loop 的本质是状态机，不是循环语句。