朋友们，你有没有过这种经历:让 Agent 跑一个长任务,比如给整个项目补测试、照着设计文档批量重构,跑到四十分钟,进程崩了,或者手滑关了终端。重开之后,之前建立的上下文、读过的文件、做了一半的修改,全部归零,从头再来。Agent 越能干,这件事越疼。
我最近两天写了一个 Rust 的 Agent 运行时(rust-deepseek-harness,参考的 DeepSeek 的那个 harness),
是的，DeepSeek 的 harness 这个思路就把这件事解决了,顺手把另外几件让人不放心的事也一起解决了:
- • 对话进度跟进程解绑。 任何时刻杀掉进程,一条 resume 命令从断点续跑,连你崩溃前一秒发出去、还没被处理的消息都不丢。
- • 全程可回放。 模型每一步看见了什么、调了什么工具、工具返回了什么,都在一份只追加的日志里。它改错文件的时候,你可以像查飞行记录仪一样查它当时到底看到了什么,而不是靠猜。
- • 出事不连坐。 某个工具崩了,只是那一次调用失败,会话还活着;Ctrl+C 取消当前这轮,不是丢掉一切。
- • 能力是插件。 联网搜索、跨会话记忆、测试运行器都是注册进去的,想接内部 API 或自己的知识库,写一个 struct 注册进去就行,不用改框架一行代码。
这篇文章讲它是怎么做到的:跟着一个回合,从用户按下回车走到 turn/end 落盘。文中所有代码来自真实仓库,实验也都是现场跑的。
先把地基讲清楚:所有发生过的事,都写进一个只追加的日志
大多数 Agent 的起点都是一个二十行的循环:调模型,遇到工具调用就执行,把结果拼回消息列表。它能跑,但对话状态散落在内存里的几个变量中,没有任何一方拥有完整的一份。
上面那四个卖点,缺的其实是同一个地基:一份权威的、只增不改的记录。框架的中心就是它,一个事件日志,只追加,不修改。用户消息、模型输出的每个增量、每次工具调用和结果、每次发往模型的完整请求,各占一条。平时大家熟悉的消息列表并不单独存在,它是重放这份日志算出来的结果,需要发给模型时才算一份,用完就扔。
这样设计直接换来两个能力。第一个是可审计:想知道模型当时到底看见了什么,查日志就有答案。日志里有专门的 request/header 事件,把系统提示、工具清单、模型参数原样落盘,没有靠猜的成分。第二个是可恢复:进程随时可以杀,重启后重放日志,对话从断点继续。
光有日志还不够,得有人盯着它和实际发出的请求不脱钩。所以每次发请求前,框架做一次自检:
pub async fn assert_request_reconstructable(&self, request: &LlmRequest) -> Result
let expected_messages = self.derive_messages().await;
if expected_messages != request.messages {
return Err(HarnessError::Invariant(
"request messages diverge from session projection".into(),
));
}
let Some(header) = self.last_request_header().await else {
return Err(HarnessError::Invariant("no request/header event".into()));
};
if header.config != request.config || header.system != request.system {
return Err(HarnessError::Invariant("request header diverges".into()));
}
Ok(())
}
即将发出的请求,必须能从日志重放出来,一字不差。任何代码路径想偷偷往消息里塞东西,或者改了系统提示却没记事件,这里直接炸掉。这条自检在开发期拦下过好几个 bug,成本几乎为零,我认为是整个设计里性价比最高的一行代码。
第一个卖点不是吹的,现场做一遍。我让 Agent 从 1 数到 300,数到第 1 个数的时候用 kill -9 杀掉进程,注意是 SIGKILL,进程连清理代码都没机会跑,那一轮连收尾事件都没写进日志。然后 resume 同一个会话,补一句继续:
▸ shell
1
2
3
4
…(一路数到 300)
它接着被杀的位置往下数,全程 1 到 300 一条不缺,日志里最后一条事件是 turn/end。加载器做的事也不复杂:把那条写了一半的损坏尾巴截掉,给中断的回合补一个 Interrupted 的结束标记,现场就干净了。
这就是对话跟进程解绑的意思:进程变成了一次性的,对话不是。这个实验建议你拿到代码后亲手跑一次,比看十篇架构文章都有说服力。
那么,我们来看一个完整的回合是怎么跑的
先看全景图。左边五个参与者,青色箭头是写入事件日志的内容,灰色箭头是运行时的动作;虚线大框是一个循环,模型还在调工具就回到发请求那一步。
下面按顺序讲六个步骤,重点讲每一步为什么必须这样设计。
第一步,消息进收件箱。 用户输入不直接进对话,先进一个带状态的 Inbox。它内部只有两个槽位:下个回合处理的 NextTurn,和当前回合下一步处理的 NextStep,由此派生出三种投递语义:followup 把消息排进下一回合,steer 插队到当前回合的下一步,inject 安静地等着,直到来了一条会唤醒对话的消息。其实 Claude Code、Codex 这些 Agent 也都具备这样的能力,这个设计合理性毋庸置疑。收件箱每次变动都以 splice 事件落盘,记下插入位置、删除数量和消息全文,这一点在后面讲恢复时会用到。
第二步,组装上下文。 系统提示由注册表拼出来,静态部分(身份、规则)加上动态的运行时上下文(工作区路径、召回的记忆)。动态部分有条纪律:内容没变就不重发。
let retained = self.session.last_runtime_context_text(RUNTIME_CONTEXT_PLUGIN).await;
match (&assembly.runtime, retained.as_deref()) {
(Some(runtime), current) if current != Some(runtime.text.as_str()) => {
messages.push(Message::plugin_text(
RUNTIME_CONTEXT_PLUGIN, runtime.text.clone(), Some(ContextForm::Snapshot),
));
}
(None, Some(_)) => { /* 通知运行时上下文已消失 */ }
_ => {}
}
计较这个的原因很实际:运行时上下文每一步都可能重组,不做变化检测,同一份工作区说明会被反复塞进对话,白白吃掉上下文窗口。
第三步,发请求前先落盘。 请求头作为事件写进日志,然后适配器发起真正的 SSE 流。适配器把 provider 的增量流映射成框架内部的块协议:每个内容块以 BlockStart 开场,中间是若干 TextDelta 或 ToolCallDelta,收尾的 BlockEnd 事件带着装配完成的完整块。装配器长这样:
pub fn push(&mut self, chunk: &StreamChunk) {
match chunk {
StreamChunk::BlockStart { index, block_type } => { /* 按类型建部分块 */ }
StreamChunk::TextDelta { index, text } => { /* 追加进对应 index 的部分块 */ }
StreamChunk::BlockEnd { index, block } => {
self.completed.insert(*index, block.clone());
}
StreamChunk::Usage { usage } => self.usage = Some(usage.clone()),
StreamChunk::Finish { reason } => self.finish = Some(reason.clone()),
/* … */
}
}
增量用来实时渲染,完整块用来组装最终消息,同一个流服务两个消费方。顺带说一个判断流式真假的土办法:看日志里增量事件的时间戳分布。实测一个长回复产生 54 个增量,分布在 1806 毫秒里,相邻间隔从 5 到 117 毫秒,收完再一次性吐出的假流式给不出这个分布。
第四步,工具调度,重点在保序。 模型一次可能发起多个调用,框架先按工具自报的执行模式分类:只读的(read、search)进并行批,互斥的(shell、写文件)排队串行。并行批用信号量限流,用 FuturesUnordered 收集,谁先完成就先回填自己的坑位:
let mut all_results: Vec
执行可以乱序,提交必须有序:工具结果按模型发起调用的顺序写进日志、拼进对话。如果按完成顺序提交,同一段代码在不同机器上会产生不同的对话历史,日志重放也就不可复现了。
第五步,闭合括号。 回合的骨架是 turn/start 到 turn/end 的一对事件,中间嵌着若干 step/start、step/end,闭合逻辑集中在一个函数里:
async fn run_turn(&self, turn: u64, cancel: CancellationToken) -> Result
self.session.append(SessionEventData::TurnStart { turn }).await?;
let reason = match self.run_turn_body(turn, cancel.clone()).await {
Ok(reason) => reason,
Err(HarnessError::Cancelled) => {
TurnEndReason::Aborted { cause: self.cancel_cause() }
}
Err(error) => TurnEndReason::Error {
code: error_code(&error), message: error.to_string(),
},
};
self.session.append(SessionEventData::TurnEnd { turn, reason }).await?;
Ok(())
}
无论体面结束、被取消还是出错,turn/end 都带着原因写进日志。括号里的内容可以有残缺(比如取消时某个工具没跑完),括号本身永远闭合。
这样做的好处显而易见:恢复逻辑靠它判断回合边界,审计靠它回答这一轮是怎么结束的。
第六步,回到循环。 模型调了工具,工具结果进入下一步的上下文,回到第二步;模型给出了最终回答,回合结束,收件箱又没有新消息,驱动器转入空闲。此刻会话的全部状态都在日志里,进程可以随时消失。
进程被杀掉之后,怎么把现场恢复出来
重启后拿到的是一个 JSONL 文件,恢复分三件事。
先修尾巴。最后一行物理写入不完整(进程死在写一半),允许截掉这条损坏行;序列号在中间出现断层,直接报错拒载,因为那意味着日志本身出了更严重的问题,猜着补是危险的。
再重建投影。Session 重放事件流,生成消息视图,模型看到的对话和被杀之前完全一致。
最后重建收件箱。前面说过每次收件箱变动都是一条 splice 事件,重放它们就能还原两个队列当时的模样,连用户在崩溃前一秒发来、还没被处理的消息都不会丢:
pub async fn restore(session: Arc -> Result
恢复代码有一条原则:数据对不上宁可报错,绝不猜测。我开发时手写过一条不合规的测试数据(没套 Turn 括号的用户消息),加载器直接甩回来一句 user/message outside a step。愣了一下才反应过来,这正是想要的行为:静默修复出一份看似可用实则错位的历史,比明确拒载糟糕得多。
新功能是怎么加进来的:注册进去,用完自动拆掉
前面全程没有出现"如果工具是 web_search 就如何"这样的分支。工具、钩子、适配器、提示词段落、子代理,五类能力全部通过注册表挂载,这就是 DeepSeek 说的,在 Agent 的世界里,万物皆为插件,主循环对具体能力一无所知。每个注册动作返回一个 EffectGuard,析构时自动卸载:
pub fn register(self: &Arc
弱引用保证卸载动作不会把注册表本身救活,RAII 保证作用域结束效果一定回收。合起来,能力就变成可以试错的东西:注册一个工具,效果不理想,作用域一退,系统回到原样。这次的联网搜索、测试运行器、跨会话记忆,全是这么挂上去的,主循环一行没动。
最后,如果只记三条
第一,把事件日志当成唯一的记录,消息列表重放时才算,并且用 assert_request_reconstructable 这样的自检守住两边不脱钩,这比任何花哨的记忆方案都可靠。第二,给每个回合一副必须闭合的括号,成功、失败、取消、panic 一视同仁,恢复和审计都建立在它上面。第三,新能力只通过注册表进入系统,注册可逆,主循环不动,系统的复杂度才有机会停在一个常数上。
这套骨架真正给我的,是敢让它跑长任务的底气:崩了能续,错了一眼能查出来,试新东西不留渣。、多模型故障转移、定时唤醒、上下文自动压缩,这些还没做,但在当前架构下它们都只是新的接缝。
粤ICP备14082021号
免费POC，
零成本试错
