长时智能体的问题往往不在单步能力，而在几百步以后。一次对话里模型能查资料、能调工具，但跑到第 3000 步时，它还记得第 40 步用户说过的"这批订单走海外仓"吗？中途进程被杀掉重启，它是接着干还是从头再来？
单轮 benchmark 测不出这些。这篇教程给出一个可落地的做法：用 ReLiveGym 这类回放式评测框架，把真实操作流录成环境，让智能体在"数周"的虚拟时间里跑，观察记忆衰减、断点恢复率和长时成功率。
> 说明：ReLiveGym 的命令名、配置字段、指标口径以官方文档当前版本为准。下面给的是通用骨架，字段名照着官方文档替换即可，思想不变。
这篇能做出什么
走完全流程，你会得到三样东西：
1. 一条可复现的回放轨迹。把真实操作（点开哪个工单、执行哪条命令、返回什么结果）按步存成 JSONL，评测时按虚拟时钟重放，速度可调。
2. 一次跨数周虚拟时间的评测运行。中间可以暂停、可以杀进程、可以换机器续跑，进度不丢。
3. 三张图三个数：
- 长时成功率曲线（按步数分桶）
- 记忆衰减曲线（早期注入的事实，在多少步之后还答得出）
- 恢复率与丢失步数（被中断后，状态对得上的比例）
具体例子：一个运维巡检智能体，任务是"连续 14 天跟进 30 台机器的告警"，中途故意在第 3000 步和第 7200 步把进程杀掉，看它第 3 天还记得第 1 天临时打的维护窗口标记没有。
前置条件清单
- Python 3.10 以上（具体最低版本以官方文档为准），会用 venv 或 conda
- 你的智能体已经有一个统一的调用入口，最好长这样：给一个 observation，返回一个 action
- 一份真实操作日志，或者能人工造一份。字段至少要有：步骤序号、时间戳、观测、动作、结果
- 能跑长时间任务的机器，或者一台能随时挂起的开发机
- 磁盘空间：20000 步的轨迹，含观测文本，通常几百 MB 到几 GB，视观测体积而定
- 心理预算：长时评测通常以天为单位。别指望一杯咖啡的时间跑完
分步骤
第 1 步：定三条评测轴
动手写代码之前，先写清楚这三个参数，否则后面全是返工。
- 时间跨度：虚拟多少天？决定 replay_speed。
- 记忆压力：什么时候注入事实、什么时候抽问？决定探针配置。
- 中断点：在第几步杀进程？杀几次？决定容错测试强度。
把这三个写进一个 config 文件，后面所有结论都挂在这个 config 的版本上。
第 2 步：搭目录结构
```bash
mkdir -p relive-eval/{configs,recordings,adapters,runs,metrics}
cd relive-eval
python -m venv .venv && source .venv/bin/activate
```
目录职责：
recordings/：回放数据，只读adapters/：你的智能体包装层configs/：评测配置runs/：每次运行的状态、检查点、日志metrics/：分析脚本
第 3 步：把现实录成回放数据
回放数据的核心是"一步一条"，每条自包含。用 JSONL，一行一个 step：
```json
{"episode_id":"ops-2024-03","step":0,"ts":"2024-03-01T09:00:00Z","observation":{"host":"web-01","alert":"disk_80pct","note":null},"action":{"type":"run","cmd":"df -h"},"outcome":{"ok":true,"stdout_tail":"/dev/sda1 81%"},"elapsed_ms":820}
{"episode_id":"ops-2024-03","step":1,"ts":"2024-03-01T09:03:00Z","observation":{"host":"web-01","alert":"disk_80pct","note":"maintenance window 03-05"},"action":{"type":"ack"},"outcome":{"ok":true},"elapsed_ms":140}
```
几个要点：
ts用 UTC，回放时由虚拟时钟推进，别依赖系统时间outcome里保留原始输出，回放时把它当环境反馈喂给智能体- 有外部副作用的动作（发邮件、删文件）单独标
"side_effect": true，评测时走 dry-run
如果只有真实日志没有结构，写个小脚本转换：
```python
tools/to_jsonl.py
import json, sys, re
PAT = re.compile(r"^(?P<ts>\S+)\s+(?P<actor>\w+)\s+(?P<msg>.*)$")
def convert(in_path, out_path, episode_id):
step = 0
with open(in_path, encoding="utf-8") as fin, open(out_path, "w", encoding="utf-8") as fout:
for line in fin:
m = PAT.match(line.strip())
if not m:
continue
rec = {
"episode_id": episode_id,
"step": step,
"ts": m.group("ts"),
"observation": {"raw": m.group("msg")},
"action": {"type": m.group("actor")},
"outcome": {"ok": True},
"elapsed_ms": 0,
}
fout.write(json.dumps(rec, ensure_ascii=False) + "\n")
step += 1
if __name__ == "__main__":
convert(sys.argv[1], sys.argv[2], sys.argv[3])
```
```bash
python tools/to_jsonl.py raw/ops.log recordings/ops-2024-03.jsonl ops-2024-03
wc -l recordings/ops-2024-03.jsonl
```
第 4 步：写适配器，把智能体接进去
适配器要做三件事：接收观测、给出动作、导出可序列化状态。第三件是断点续跑的命脉。
```python
adapters/my_agent.py
from typing import Any, Dict, List, Optional
class MyAgentPolicy:
name = "my-agent"
def __init__(self, keep_last: int = 50, **kwargs):
self.cfg = kwargs
self.keep_last = keep_last
self.history: List[Dict[str, Any]] = []
def reset(self, state: Optional[Dict[str, Any]] = None) -> None:
"""episode 开始或续跑时调用。state 为 None 表示全新开始。"""
self.history = list(state.get("history", [])) if state else []
def act(self, observation: Dict[str, Any]) -> Dict[str, Any]:
self.history.append(observation)
action = self._call_your_agent(observation, self.history)
return action
def dump_state(self) -> Dict[str, Any]:
"""必须可 JSON 序列化。这是断点续跑能对上的前提。"""
return {"history": self.history[-self.keep_last:]}
def _call_your_agent(self, observation, history):
替换成你真实的智能体：SDK 调用、HTTP 请求、本地模型都行
关键是要确定性：temperature 设 0，或固定 seed
return {"type": "noop"}
```
注意 keep_last。很多长时评测的"记忆衰减"其实是适配器自己把历史截断了，不是模型忘了。先确认截断策略，再下结论。
第 5 步：写评测配置
```yaml
configs/long-horizon.yaml
run:
name: long-horizon-v1
seed: 20240301
horizon_steps: 20000
replay_speed: 60 # 1 秒回放 60 秒真实时间
virtual_clock: true
checkpoint:
every_steps: 500
dir: runs/long-horizon-v1/checkpoints
keep: 5
memory_probes:
inject_at_steps: [200, 1000, 5000, 10000]
probe_after_steps: [100, 500, 2000]
facts_file: configs/facts.jsonl
interruptions:
schedule: [3000, 7200]
kill_signal: SIGKILL
grace_seconds: 0
```
replay_speed: 60 意味着 14 天的虚拟时间压到大约 5.5 小时。想更快就把值调大，但太快容易把并发 bug 掩盖掉。
事实文件长这样，用于测记忆：
```json
{"fact_id":"f-001","text":"web-01 在 03-05 有维护窗口","inject_at_step":200}
{"fact_id":"f-002","text":"海外仓订单不走自动对账","inject_at_step":1000}
```
第 6 步：跑起来，然后故意打断它
启动：
```bash
nohup python -m relivegym run \
--config configs/long-horizon.yaml \
--adapter adapters.my_agent:MyAgentPolicy \
--recordings recordings/ \
--out runs/long-horizon-v1 \
> runs/long-horizon-v1.log 2>&1 &
tail -f runs/long-horizon-v1.log
```
如果框架自带中断注入，interruptions.schedule 会自己生效。想手动验证，就在运行到一半时：
```bash
pkill -9 -f "relivegym run"
ls -lh runs/long-horizon-v1/checkpoints/
```
然后续跑：
```bash
python -m relivegym resume \
--run-dir runs/long-horizon-v1 \
--from-checkpoint latest \
--out runs/long-horizon-v1
```
续跑前确认三件事：检查点文件完整（大小不为 0）、适配器代码的 git commit 没变、config 没改。任意一项变了，续跑出来的数据就和前半段不可比。
第 7 步：读指标
先把步骤日志读进来：
```python
metrics/analyze.py
import json
import pandas as pd
def load_steps(path: str) -> pd.DataFrame:
rows = []
with open(path, encoding="utf-8") as f:
for line in f:
r = json.loads(line)
rows.append({
"episode_id": r["episode_id"],
"step": r["step"],
"ts": pd.to_datetime(r["ts"]),
"ok": bool(r.get("outcome", {}).get("ok", False)),
})
return pd.DataFrame(rows)
df = load_steps("runs/long-horizon-v1/steps.jsonl")
df["bucket"] = pd.cut(
df["step"], bins=[0, 500, 2000, 10000, float("inf")],
labels=["0-500", "500-2k", "2k-10k", ">10k"],
)
print(df.groupby("bucket", observed=True)["ok"].mean())
```
长时成功率看的是分桶后的斜率。如果 0-500 是 0.9，>10k 掉到 0.55，说明能力随步数退化，优先查上下文管理而不是工具层。
记忆衰减：
```python
probes = pd.read_json("runs/long-horizon-v1/probes.jsonl", lines=True)
字段：fact_id, injected_at_step, probed_at_step, recalled
probes["gap"] = probes["probed_at_step"] - probes["injected_at_step"]
print(probes.groupby("gap")["recalled"].mean().sort_index())
```
读法要点：区分"忘了"和"没检索到"。如果模型答"没看到过这条信息"，是检索失败；如果答错内容，是记忆污染。前者加检索，后者加去重和摘要。
恢复率：
```python
resume = pd.read_json("runs/long-horizon-v1/interruptions.jsonl", lines=True)
字段：killed_at_step, resumed_at_step, state_match, steps_lost, side_effects_replayed
print("恢复率:", resume["state_match"].mean())
print(resume["steps_lost"].describe())
print("副作用重复次数:", resume["side_effects_replayed"].sum())
```
state_match 为假说明续跑后状态对不上；side_effects_replayed 大于 0 说明幂等没做好，在真实环境里这就是重复发邮件、重复下单。
常见坑与排错
回放踩到真实时间。 代码里只要有一处 datetime.now()，回放就会错位。全部换成虚拟时钟，并在 CI 里加一条 grep 检查。
不确定性没锁死。 模型 temperature 不为 0、并发任务顺序不固定、字典遍历顺序参与决策，都会让两次运行结果不同。固定 seed，关键路径串行化。
续跑后重复副作用。 每个有副作用的动作带一个幂等键（比如 episode_id + step），执行前查一次已执行集合。这是评测环境里最容易被忽略、上线后代价最大的一条。
把截断当成遗忘。 前面提过，先确认适配器保留了多少历史。可以跑一次 keep_last 很大的对照组。
checkpoint 间隔太大。 间隔 500 步、结果丢 400 步，恢复率的结论就不稳。先看 steps_lost 的分布，再调 every_steps。
评测本身跑不完。 20000 步如果每步要调一次模型，时间和费用都不小。先跑 2000 步的小样本，把管道跑通，再放大。
指标口径漂移。 两次运行结果不可比，多半是 config 或适配器代码变了。把 config 哈希、git commit 写进每次运行的元数据文件。
下一步建议
管道跑通后，按这个顺序加码：
1. 把回放数据从 1 条 episode 扩到 10 条以上，覆盖不同任务类型，避免单条轨迹的偶然性主导结论。
2. 加一组消融实验：只改记忆策略（全历史 / 滑动窗口 / 摘要 + 检索），其他不动，看衰减曲线怎么变。
3. 把中断测试从"杀进程"扩展到"网络抖动""工具超时""返回格式错误"，看恢复能力是不是只在一种故障下成立。
4. 把指标脚本接进 CI，每次改智能体提示词或工具定义都跑一遍小样本，防止回归。
5. 记录每次失败的原始轨迹片段，攒成一个"长时失败案例库"。这比任何单一数字都有用。
长时评测的价值不在于跑出一个漂亮分数，而在于你终于能看到智能体在第 3000 步时是什么样子。