摘要很多人第一次写 Agent都会写出类似下面的代码whileTrue:responsellm(messages)ifresponse.final:returnresponse.text resultrun_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状态职责如下状态说明CREATED创建 run_id、预算、权限上下文PREPARE_CONTEXT组装系统提示、历史、记忆、工具说明MODEL_THINKING调用模型DECIDE_ACTION解析模型输出判断是回答、工具、handoff 还是异常WAIT_APPROVAL对高风险动作请求人工或策略审批EXECUTE_TOOL执行工具处理超时、重试、沙箱OBSERVE清洗工具结果并写回上下文COMPACT_OR_SAVE压缩上下文、写 checkpointFINISHED正常结束FAILED异常结束CANCELLED用户或系统取消SUSPENDED暂停等待恢复三、源码示例用状态机重写 Agent Loop下面是一个简化的 Python 实现。它不依赖具体模型厂商重点展示状态如何显式流转。fromdataclassesimportdataclass,fieldfromenumimportEnum,autofromtypingimportAny,LiteralimporttimeimportuuidclassRunState(Enum):CREATEDauto()PREPARE_CONTEXTauto()MODEL_THINKINGauto()DECIDE_ACTIONauto()WAIT_APPROVALauto()EXECUTE_TOOLauto()OBSERVEauto()COMPACT_OR_SAVEauto()FINISHEDauto()FAILEDauto()CANCELLEDauto()SUSPENDEDauto()dataclassclassBudget:max_turns:int20max_seconds:int300max_tool_failures:int3max_tokens:int80_000dataclassclassAgentRun:user_input:strrun_id:strfield(default_factorylambda:str(uuid.uuid4()))state:RunStateRunState.CREATED messages:list[dict[str,Any]]field(default_factorylist)turn_count:int0token_count:int0tool_failures:int0started_at:floatfield(default_factorytime.time)pending_tool_call:dict[str,Any]|NoneNonefinal_answer:str|NoneNoneerror:str|NoneNonecancelled:boolFalse这里的关键点不是代码复杂而是状态被显式建模了。后面所有逻辑都可以围绕RunState做测试。四、源码示例终止条件不要写进 Prompt很多系统会在 prompt 里写“不要无限循环”。这不够。终止条件必须由代码控制。defshould_stop(run:AgentRun,budget:Budget)-tuple[bool,str|None]:ifrun.cancelled:returnTrue,cancelled_by_userifrun.turn_countbudget.max_turns:returnTrue,max_turns_exceedediftime.time()-run.started_atbudget.max_seconds:returnTrue,timeoutifrun.tool_failuresbudget.max_tool_failures:returnTrue,too_many_tool_failuresifrun.token_countbudget.max_tokens:returnTrue,token_budget_exceededreturnFalse,NoneAutoGen 的 termination condition 设计也体现了这个原则终止条件是代码层面的、有状态的、可组合的对象而不是一句提示词。五、源码示例主循环的正确形状下面是一个更完整的执行骨架defrun_agent(user_input:str,llm,tool_registry,policy,store,tracer)-AgentRun:runAgentRun(user_inputuser_input)budgetBudget()try:whilerun.statenotin{RunState.FINISHED,RunState.FAILED,RunState.CANCELLED}:stop,reasonshould_stop(run,budget)ifstop:run.stateRunState.CANCELLEDifreasoncancelled_by_userelseRunState.FAILED run.errorreasonbreaktracer.event(run.run_id,state,{state:run.state.name})ifrun.stateRunState.CREATED:run.messages.append({role:user,content:user_input})run.stateRunState.PREPARE_CONTEXTelifrun.stateRunState.PREPARE_CONTEXT:run.messagesbuild_context(run,store)run.stateRunState.MODEL_THINKINGelifrun.stateRunState.MODEL_THINKING:responsellm.generate(run.messages)run.turn_count1run.token_countresponse.usage.total_tokens run.messages.append(response.to_message())run.stateRunState.DECIDE_ACTIONelifrun.stateRunState.DECIDE_ACTION:actionparse_model_action(run.messages[-1])ifaction.kindfinal:run.final_answeraction.text run.stateRunState.FINISHEDelifaction.kindtool:run.pending_tool_callaction.tool_call run.stateRunState.WAIT_APPROVALelse:run.errorfunsupported_action:{action.kind}run.stateRunState.FAILEDelifrun.stateRunState.WAIT_APPROVAL:decisionpolicy.evaluate(run.pending_tool_call)ifdecisiondeny:run.errortool_deniedrun.stateRunState.FAILEDelifdecisionask:run.stateRunState.SUSPENDED store.save_checkpoint(run)breakelse:run.stateRunState.EXECUTE_TOOLelifrun.stateRunState.EXECUTE_TOOL:try:resulttool_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.stateRunState.OBSERVEexceptExceptionasexc:run.tool_failures1run.messages.append({role:tool,content:ftool_error:{type(exc).__name__}:{exc},})run.stateRunState.OBSERVEelifrun.stateRunState.OBSERVE:run.pending_tool_callNonerun.stateRunState.COMPACT_OR_SAVEelifrun.stateRunState.COMPACT_OR_SAVE:run.messagesmaybe_compact(run.messages,budget)store.save_checkpoint(run)run.stateRunState.MODEL_THINKINGelifrun.stateRunState.SUSPENDED:store.save_checkpoint(run)breakexceptExceptionasexc:run.stateRunState.FAILED run.errorfsystem_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):runstore.load_checkpoint(run_id)ifrun.state!RunState.SUSPENDED:raiseValueError(only suspended runs can be resumed safely)ifapproval_resultallow:run.stateRunState.EXECUTE_TOOLelifapproval_resultdeny:run.stateRunState.FAILED run.errorapproval_deniedelse: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_transitionllm_call_startedllm_call_finishedtool_call_requestedpermission_evaluatedtool_call_finishedcontext_compactedcheckpoint_savedrun_finishedOpenAI Agents SDK 的 tracing 覆盖模型生成、工具调用、handoff、guardrail、自定义事件。自研 Harness 可以参考这种事件结构。九、普通人怎么理解这件事可以把 Agent Loop 想成一辆自动驾驶车模型是司机负责判断下一步。状态机是交通规则决定什么时候能走、什么时候必须停。权限系统是门禁。工具系统是车辆执行机构。Trace 是行车记录仪。Checkpoint 是服务区和救援点。没有状态机的 Agent就像只有司机、没有刹车、没有仪表盘、没有行车记录仪的车。十、落地检查表是否有显式状态枚举是否有代码级终止条件是否支持取消、暂停、恢复是否限制工具并发是否把工具失败转成可处理 Observation是否保存 checkpoint是否记录 trace是否能复现一次失败是否能区分模型错误、工具错误、权限错误、上下文错误结论生产级 Agent Loop 的目标不是让模型多跑几轮而是让系统每一轮都知道自己在做什么、为什么这么做、什么时候该停、失败后从哪里恢复。Agent Loop 的本质是状态机不是循环语句。