上一篇把单次工具调用收进白名单和参数校验。本篇继续向前一步让 Agent 在“观察—选择动作—执行—再观察”之间循环同时用状态机、预算和重复检测保证它一定停下来。一、痛点从“能演示”走向“可托管”循环不是一个无限 while而是受控状态机。运行状态至少包含目标、事实、动作历史、剩余步数和终止原因每轮只能产生 tool、final 或 abort 三类决定。执行器先检查预算和重复动作再调用工具并追加观察。模型不能改写历史也不能宣布已经执行了一个实际未执行的动作。真正可迁移的做法是先写任务契约再选模型输入从哪里来成功产物是什么哪些动作禁止自动执行最多允许多少步、多少时间和多少费用失败后由谁接管。契约一旦写进代码和测试模型升级只是实现替换不会悄悄改变业务责任边界。二、原理把概率判断放进确定性边界步数、墙钟时间和费用是三套独立预算任意一项耗尽都要停止。动作去重应基于规范化参数后的哈希否则只改字段顺序就能绕过。终态需要区分 completed、needs_human、budget_exhausted 与 failed调用方才能采取正确后续动作而不是把所有非成功都包装成模糊回复。模型输出、工具数据和记忆内容都按不可信输入处理。每个边界都执行校验、授权、裁剪和审计所有状态变化都有明确原因。这样做看似增加一些代码却把“偶尔答错”拆成可定位的规划错误、参数错误、工具错误或策略错误团队才能针对性改进。三、实现用独立程序跑通最小闭环示例把“查询订单后计算是否满足补偿条件”拆成确定动作序列。计划器在这里用固定规则模拟便于观察循环机制接入模型时只替换decide状态、工具、预算和终态判断无需改变。fromdataclassesimportdataclass,fielddataclassclassState:facts:dict[str,int]field(default_factorydict)history:list[str]field(default_factorylist)remaining:int4status:strrunningdefdecide(state:State)-str:iforder_agenotinstate.facts:returnlookup_orderifmax_agenotinstate.facts:returnread_policyreturnfinishdefexecute(action:str,state:State)-str:ifactionlookup_order:state.facts[order_age]5returnorder_age5ifactionread_policy:state.facts[max_age]7returnmax_age7eligiblestate.facts[order_age]state.facts[max_age]state.statuscompletedreturnfeligible{str(eligible).lower()}stateState()whilestate.statusrunningandstate.remaining:actiondecide(state)observationexecute(action,state)state.history.append(action)state.remaining-1print(fstep{len(state.history)}action{action}observation{observation})ifstate.statusrunning:state.statusbudget_exhaustedprint(fstatus{state.status}remaining{state.remaining})运行输出step1 actionlookup_order observationorder_age5 step2 actionread_policy observationmax_age7 step3 actionfinish observationeligibletrue statuscompleted remaining1可恢复性比循环写法更重要。每完成一步就持久化检查点并为写操作保存幂等键。进程在工具成功后、状态落盘前崩溃是最危险的窗口恢复时必须先查询幂等记录不能盲目重放。观察值要有大小上限和来源标签避免一次网页抓取占满上下文。第二个程序补上生产中最容易遗漏的控制点。它与前一个示例完全独立可单独保存运行不依赖本系列其他文件示例数据是内存假实现因此不会访问真实账户或产生外部副作用。importjsonimporthashlibdeffingerprint(name:str,arguments:dict[str,str])-str:payloadjson.dumps(arguments,ensure_asciiFalse,sort_keysTrue,separators(,,:))digesthashlib.sha256(f{name}:{payload}.encode()).hexdigest()returndigest seen:set[str]set()actions[(search_order,{id:A-7}),(search_order,{id:A-7}),(read_policy,{version:2026-01}),]forname,argumentsinactions:keyfingerprint(name,arguments)normalizedjson.dumps(arguments,ensure_asciiFalse,sort_keysTrue,separators(,,:))ifkeyinseen:print(blockedduplicate_action)continueseen.add(key)print(faccepted{name}:{normalized})print(funique_actions{len(seen)})运行输出acceptedsearch_order:{id:A-7} blockedduplicate_action acceptedread_policy:{version:2026-01} unique_actions2落地时应把示例中的内存状态替换为事务型存储把打印日志替换为结构化事件但不要改变契约。事件至少包含 run_id、步骤、输入摘要、策略结果、耗时、错误类别和版本密钥、完整个人信息及未经脱敏的提示不得进入日志。四、踩坑失败路径决定系统上限循环最常见的故障是振荡Agent 在两个工具间来回切换或者用同一查询反复碰运气。另一个问题是“伪完成”模型生成答案却缺少任务要求的证据。执行器应设置重复上限final 也要通过完成条件校验例如订单与政策两份事实都存在否则转人工而不是继续消耗预算。还要防止把确定逻辑模型化。金额计算、权限判断、枚举路由和状态转移应使用普通代码模型适合处理分类、抽取、歧义消解与证据综合。每减少一个无必要的模型决策点就减少一处延迟、成本和不可复现性同时让测试更容易覆盖。五、验证用轨迹和门槛验收回归用完整轨迹验收而不只看最后一句话。每个案例检查动作是否合法、参数是否稳定、观察是否被正确利用、是否在预算内进入预期终态。加入空结果、超时、重复建议和进程恢复用例。下一篇会把长轨迹中的信息分层保存解决上下文越跑越长的问题。发布顺序固定为离线回放、影子流量、小比例灰度和有限自治。每阶段先定义通过线和回滚条件再看结果不能在看到分数后移动门槛。关键样本要保存初始状态、每步动作、观察、最终产物与终止原因模型、提示和工具契约版本也必须一起记录。循环的状态将直接成为记忆管理的输入哪些是当前步骤临时信息哪些是会话事实哪些才值得跨会话保存。参考来源ReAct 论文Python dataclassesLangGraph overview 觉得有用就点个赞 收藏方便回头查阅有疑问直接在评论区留言我看到都会回。 本文属于《AI Agent 应用实战》系列持续更新关注不迷路。 文章里的代码都能直接跑。想要可直接 clone 的完整工程 配套部署脚本 / 踩坑清单评论一声或发邮件到cj2664qq.com我免费发你。如果你正好在做类似系统、或有工程化难题想找人做也欢迎邮件聊一句——我按实际情况评估能落地的就接单或出方案。评论和邮件都能直接找到我不用跳别的平台。