本篇复用上一篇的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}运行输出limits4 attempts5 same_failure3from__future__importannotationsimportjsonfromcollectionsimportCounterfromdataclassesimportasdict,dataclassfrompathlibimportPathfromledgerimportLedger,load_eventsdataclass(frozenTrue)classState:name:strattempts:inttool_calls:intcost_units:intreason:strdefderive(events:list[dict],budget:dict)-State:kindsCounter(event[kind]foreventinevents)failures[event[payload].get(output_sha256)foreventineventsifevent[kind]check_finishedandnotevent[payload].get(passed)]ifany(event[kind]check_finishedandevent[payload].get(passed)foreventinevents):returnState(succeeded,kinds[attempt_started],kinds[tool_started],kinds[model_finished],checks_passed)repeatedmax(Counter(failures).values(),default0)limitsbudget[limits]ifrepeatedbudget[stop_on_same_failure]:reasonsame_failure_repeatednameblockedelifkinds[attempt_started]limits[attempts]:reason,nameattempts_exhausted,exhaustedelifkinds[tool_started]limits[tool_calls]:reason,nametool_calls_exhausted,exhaustedelifkinds[model_finished]limits[cost_units]:reason,namecost_exhausted,exhaustedelse:reason,namebudget_available,retryingreturnState(name,kinds[attempt_started],kinds[tool_started],kinds[model_finished],reason)defmain()-int:rootPath(runs/demo-001)eventsload_events(root/events.jsonl)budgetjson.loads(Path(budget.json).read_text(encodingutf-8))statederive(events,budget)(root/state.json).write_text(json.dumps(asdict(state),indent2),encodingutf-8)Ledger(root/events.jsonl,demo-001).append(state_changed,asdict(state))print(fstate{state.name}reason{state.reason}attempts{state.attempts})return0if__name____main__:raiseSystemExit(main())运行输出stateretrying reasonbudget_available attempts1二、状态应从事件推导而不是只信 state.json进程可能在写完事件后、更新状态文件前崩溃。如果恢复时只读state.json会少算一次调用。事件账本是事实源状态文件只是缓存启动时重新推导并覆盖缓存。这个设计叫事件溯源但不必引入框架关键是明确哪份数据拥有最终解释权。墙钟预算不能简单累加事件时间戳差因为机器休眠和时钟调整会产生异常。运行中用单调时钟计量恢复后用已提交的分段耗时求和。费用也不应拿“模型返回的估算值”直接做结算宿主按供应商响应或本地计数器记录。记忆点是预算不是为了省钱才停止而是为了让停止成为正确结果。无限重试会把不可解问题伪装成“还在努力”占用资源且无法交接。exhausted是带证据的业务状态不是异常堆栈。三、重复失败为什么比轮数更有信息五次不同失败说明循环可能在推进三次相同摘要则说明它原地踏步。不过只比较完整输出摘要会被时间戳和临时路径扰动。检查器应先规范化非语义字段再计算故障指纹同时保留原始输出摘要用于审计。相同失败也不总该停止。模型第一次读文件、第二次申请新工具虽然检查结果相同但信息状态发生变化。可以把“故障指纹 已获取证据集合”组成进展键。只有两者都不变才算无进展。这个权衡避免既烧无限预算也过早结束有希望的探索。四、恢复与人工接管状态为blocked时交接包应包含目标摘要、最后失败证据、已尝试补丁摘要、拒绝工具和剩余预算。人工补充权限或新证据后不直接修改旧事件而是追加human_intervention状态机再决定能否回到retrying。历史仍然完整。本篇产物state.json明确给出下一步是否合法。下一篇会读取其中的retrying与steer.json为每个工具调用生成稳定call_id和回执文件即使进程在工具执行后崩溃恢复也不会重复产生外部副作用。并发执行器还需要用任务级锁保护“推导后追加”这一小段否则两个进程可能同时看到剩余一次预算并各启动一次。锁超时不等于可以强抢恢复者必须核对持有进程与最后事件再决定接管并记录lease_recovered。预算边界本身也是共享状态必须像业务写入一样处理竞态。参考来源Dev.toThe Dirty Secret Behind AI AgentsPython 文档time — Time access and conversions 觉得有用就点个赞 收藏方便回头查阅有疑问直接在评论区留言我看到都会回。 本文属于《Agent可靠性工程实战》系列持续更新关注不迷路。 文章里的代码都能直接跑。想要可直接 clone 的完整工程 配套部署脚本 / 踩坑清单评论一声或发邮件到cj2664qq.com我免费发你。如果你正好在做类似系统、或有工程化难题想找人做也欢迎邮件聊一句——我按实际情况评估能落地的就接单或出方案。评论和邮件都能直接找到我不用跳别的平台。