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