Agent 可靠性工程实战(一):先给智能循环装一只不可失忆的黑匣子

很多 Agent 教程从模型、提示词和工具调用开始,演示效果很热闹,却跳过了最先该解决的问题:一次任务失败以后,我们能否准确回答模型看到了什么、调用了什么、工具返回了什么、为什么停止。普通文本日志只记录人类觉得重要的句子,无法支撑确定性回放。这个系列从一个名为ProofLoop的本地代码修复执行器开始,第一篇只建骨架和事件账本,不接真实模型。

整个项目遵守一个边界:模型可以提出动作,宿主程序决定是否执行,裁判程序决定结果是否合格。三者不能共用一团可随意修改的内存状态。我们先把每一步写成追加式 JSONL 事件,让后续九篇都把runs/<run_id>/events.jsonl当输入。

一、事件不是调试字符串,而是状态事实

传统日志写“正在运行测试”,但这句话没有稳定字段,换一句措辞就无法统计。事件则包含序号、任务编号、事件类型、时间和结构化载荷。序号负责定义因果顺序,任务编号隔离并发执行,事件类型形成可版本化协议。时间只用于观察耗时,不能决定业务顺序,因为系统时钟会回拨,多线程也可能在同一微秒写入。

下面的ledger.py使用标准库实现最小账本。每次追加先序列化为单行 JSON,再刷新并同步文件描述符。它不能让磁盘永不损坏,却把“Python 返回成功但数据仍停在用户态缓冲区”的窗口缩小。载荷只允许 JSON 类型,模型原始回答若包含敏感信息,应在进入账本前脱敏。

from__future__importannotationsimportjsonimportosfromdataclassesimportasdict,dataclassfromdatetimeimportdatetime,timezonefrompathlibimportPathfromtypingimportAny@dataclass(frozen=True)classEvent:seq:intrun_id:strkind:strat:strpayload:dict[str,Any]classLedger:def__init__(self,path:Path,run_id:str)->None:self.path=path self.run_id=run_id self.path.parent.mkdir(parents=True,exist_ok=True)self.next_seq=self._count_lines()+1def_count_lines(self)->int:ifnotself.path.exists():return0withself.path.open(encoding="utf-8")ashandle:returnsum(1forlineinhandleifline.strip())defappend(self,kind:str,payload:dict[str,Any])->Event:event=Event(seq=self.next_seq,run_id=self.run_id,kind=kind,at=datetime.now(timezone.utc).isoformat(),payload=payload,)encoded=json.dumps(asdict(event),ensure_ascii=False,separators=(",",":"))withself.path.open("a",encoding="utf-8")ashandle:handle.write(encoded+"\n")handle.flush()os.fsync(handle.fileno())self.next_seq+=1returneventdefmain()->int:ledger=Ledger(Path("runs/demo-001/events.jsonl"),"demo-001")ledger.append("run_started",{"goal":"修复折扣计算"})ledger.append("model_proposed",{"tool":"read_file","path":"price.py"})print(f"events=2 next_seq={ledger.next_seq}path={ledger.path}")return0if__name__=="__main__":raiseSystemExit(main())

运行输出:

events=2 next_seq=3 path=runs/demo-001/events.jsonl

二、为什么选择 JSONL 而不是一个大 JSON

一个大数组只有在任务结束时才能完整关闭;进程中断时,尾部括号可能缺失,整个文件无法解析。JSONL 每行是独立记录,已经同步的前缀仍然有效。它还允许流式读取,不必把长任务全部装进内存。代价是跨事件约束需要读取器自己检查,因此我们必须验证序号连续、任务编号一致、每行都能解析。

记住一句不太直觉的话:时间戳用于观察,序号才用于证明先后。如果把时间当顺序,NTP 校时或虚拟机暂停恢复会让事件“穿越”;如果只相信文件行号,手工拼接文件又可能隐藏缺口。显式seq让缺失和重复都可检测。

importjsonfrompathlibimportPathdefload_events(path:Path)->list[dict]:events=[]withpath.open(encoding="utf-8")ashandle:forline_no,lineinenumerate(handle,start=1):try:event=json.loads(line)exceptjson.JSONDecodeErrorasexc:raiseValueError(f"broken json at line{line_no}")fromexc expected=len(events)+1ifevent["seq"]!=expected:raiseValueError(f"expected seq{expected}, got{event['seq']}")events.append(event)returnevents events=load_events(Path("runs/demo-001/events.jsonl"))print([event["kind"]foreventinevents])

运行输出:

['run_started', 'model_proposed']

三、追加式不等于绝对不可篡改

文件以追加模式打开,只能防止业务代码无意覆盖,不能防止拥有同一文件权限的进程重写历史。后续会加入摘要链和只读裁判;此时先把所有状态变化变成事件,避免“内存里已经重试三次,磁盘只看到最终成功”的选择性记忆。异常也必须写事件:工具超时、权限拒绝和预算耗尽不是日志噪声,而是任务结论的一部分。

另一个坑是把完整提示词、令牌或客户源码直接塞进账本。审计需要的是可验证引用,不一定是原文。可保存内容摘要、相对路径和脱敏片段,把敏感正文留在权限更严格的工件目录。账本越完整,泄漏后的影响也越大,所以可审计与最小披露必须一起设计。

四、首篇交付什么

本篇产物是ledger.pyruns/demo-001/events.jsonl。验收不是看终端出现两行文字,而是重新启动程序后序号从 3 继续,读取器能发现截断行与跳号,两个任务不会写进同一目录。下一篇将直接读取这份事件文件中的model_proposed,用tools.json判断read_file是否被授权,并把允许或拒绝结果继续追加到同一账本。

参考来源

  • Dev.to|The Dirty Secret Behind AI Agents
  • Python 文档|json — JSON encoder and decoder

👍 觉得有用就点个赞 + 收藏,方便回头查阅;有疑问直接在评论区留言,我看到都会回。

🚀 本文属于《Agent可靠性工程实战》系列,持续更新,关注不迷路。

📌 文章里的代码都能直接跑。想要可直接 clone 的完整工程 + 配套部署脚本 / 踩坑清单?评论一声或发邮件到cj2664@qq.com,我免费发你。
如果你正好在做类似系统、或有工程化难题想找人做,也欢迎邮件聊一句——我按实际情况评估,能落地的就接单或出方案。评论和邮件都能直接找到我,不用跳别的平台。