本篇复用上一篇的runs/demo-001/steer.json和此前累积的events.jsonl。转向文件提供下一次尝试的目标与失败摘要，账本提供已花费次数和重复证据；本篇新增budget.json、machine.py，输出state_changed事件与state.json。下一篇只会执行状态机批准的工具调用。
一、一次重试其实消耗多种预算
只设置最大轮数会掩盖不同动作的成本：读一个小文件与启动十分钟集成测试不等价，模型调用还有金额和上下文上限。预算至少分为尝试次数、墙钟时间、工具调用数和费用单位。它们任一耗尽都应停止，而且停止原因要进入账本。
状态不能由循环里几个布尔变量拼凑。显式状态包括ready、generating、checking、retrying、succeeded、blocked、exhausted。每个事件只允许触发有限转换，非法顺序立即暴露。例如没有check_finished就不能宣布成功，权限拒绝也不能伪装成测试失败。
{"schema_version":1,"limits":{"attempts":5,"tool_calls":20,"wall_seconds":900,"cost_units":120},"stop_on_same_failure":3}运行输出：
limits=4 attempts=5 same_failure=3from__future__importannotationsimportjsonfromcollectionsimportCounterfromdataclassesimportasdict,dataclassfrompathlibimportPathfromledgerimportLedger,load_events@dataclass(frozen=True)classState:name:strattempts:inttool_calls:intcost_units:intreason:strdefderive(events:list[dict],budget:dict)->State:kinds=Counter(event["kind"]foreventinevents)failures=[event["payload"].get("output_sha256")foreventineventsifevent["kind"]=="check_finished"andnotevent["payload"].get("passed")]ifany(event["kind"]=="check_finished"andevent["payload"].get("passed")foreventinevents):returnState("succeeded",kinds["attempt_started"],kinds["tool_started"],kinds["model_finished"],"checks_passed")repeated=max(Counter(failures).values(),default=0)limits=budget["limits"]ifrepeated>=budget["stop_on_same_failure"]:reason="same_failure_repeated"name="blocked"elifkinds["attempt_started"]>=limits["attempts"]:reason,name="attempts_exhausted","exhausted"elifkinds["tool_started"]>=limits["tool_calls"]:reason,name="tool_calls_exhausted","exhausted"elifkinds["model_finished"]>=limits["cost_units"]:reason,name="cost_exhausted","exhausted"else:reason,name="budget_available","retrying"returnState(name,kinds["attempt_started"],kinds["tool_started"],kinds["model_finished"],reason)defmain()->int:root=Path("runs/demo-001")events=load_events(root/"events.jsonl")budget=json.loads(Path("budget.json").read_text(encoding="utf-8"))state=derive(events,budget)(root/"state.json").write_text(json.dumps(asdict(state),indent=2),encoding="utf-8")Ledger(root/"events.jsonl","demo-001").append("state_changed",asdict(state))print(f"state={state.name}reason={state.reason}attempts={state.attempts}")return0if__name__=="__main__":raiseSystemExit(main())运行输出：
state=retrying reason=budget_available attempts=1二、状态应从事件推导，而不是只信 state.json
进程可能在写完事件后、更新状态文件前崩溃。如果恢复时只读state.json，会少算一次调用。事件账本是事实源，状态文件只是缓存；启动时重新推导并覆盖缓存。这个设计叫事件溯源，但不必引入框架，关键是明确哪份数据拥有最终解释权。
墙钟预算不能简单累加事件时间戳差，因为机器休眠和时钟调整会产生异常。运行中用单调时钟计量，恢复后用已提交的分段耗时求和。费用也不应拿“模型返回的估算值”直接做结算，宿主按供应商响应或本地计数器记录。
记忆点是：预算不是为了省钱才停止，而是为了让停止成为正确结果。无限重试会把不可解问题伪装成“还在努力”，占用资源且无法交接。exhausted是带证据的业务状态，不是异常堆栈。
三、重复失败为什么比轮数更有信息
五次不同失败说明循环可能在推进，三次相同摘要则说明它原地踏步。不过只比较完整输出摘要会被时间戳和临时路径扰动。检查器应先规范化非语义字段，再计算故障指纹；同时保留原始输出摘要用于审计。
相同失败也不总该停止。模型第一次读文件、第二次申请新工具，虽然检查结果相同，但信息状态发生变化。可以把“故障指纹 + 已获取证据集合”组成进展键。只有两者都不变才算无进展。这个权衡避免既烧无限预算，也过早结束有希望的探索。
四、恢复与人工接管
状态为blocked时，交接包应包含目标摘要、最后失败证据、已尝试补丁摘要、拒绝工具和剩余预算。人工补充权限或新证据后，不直接修改旧事件，而是追加human_intervention，状态机再决定能否回到retrying。历史仍然完整。
本篇产物state.json明确给出下一步是否合法。下一篇会读取其中的retrying与steer.json，为每个工具调用生成稳定call_id和回执文件；即使进程在工具执行后崩溃，恢复也不会重复产生外部副作用。
并发执行器还需要用任务级锁保护“推导后追加”这一小段，否则两个进程可能同时看到剩余一次预算并各启动一次。锁超时不等于可以强抢，恢复者必须核对持有进程与最后事件，再决定接管并记录lease_recovered。预算边界本身也是共享状态，必须像业务写入一样处理竞态。
参考来源
- Dev.to｜The Dirty Secret Behind AI Agents
- Python 文档｜time — Time access and conversions
👍 觉得有用就点个赞 + 收藏，方便回头查阅；有疑问直接在评论区留言，我看到都会回。
🚀 本文属于《Agent可靠性工程实战》系列，持续更新，关注不迷路。
📌 文章里的代码都能直接跑。想要可直接 clone 的完整工程 + 配套部署脚本 / 踩坑清单？评论一声或发邮件到cj2664@qq.com，我免费发你。
如果你正好在做类似系统、或有工程化难题想找人做，也欢迎邮件聊一句——我按实际情况评估，能落地的就接单或出方案。评论和邮件都能直接找到我，不用跳别的平台。
