长任务 agent 跑到后面容易“越跑越笨”：上下文里塞满了失败的尝试、重复的网页内容和无关对话，真正重要的进度反而被淹没。CORAL 论文（arXiv:2610.40350）给了一个很朴素的思路：让 agent 自己把进度写成检查点，然后清空上下文，从干净状态继续。这个教程不训练模型，只用两个工具和一个循环，把这件事在本地复现出来。

先定清楚“检查点”里放什么

检查点不是聊天摘要，而是一份可继续执行的状态。最少包含四类字段：任务目标、已核实的事实、已完成/待完成的步骤、仍未解决的问题。事实要带来源，避免下一轮把模型自己的猜测当成证据。

checkpoint = {
  "goal": "找出三家供应商中报价最低且支持 SOC2 的一家",
  "facts": [
    {"text": "A 报价 12 万/年，支持 SOC2", "source": "https://a.example/pricing"},
    {"text": "B 报价 9 万/年，未提 SOC2", "source": "https://b.example/pricing"}
  ],
  "done": ["收集 A/B 报价", "核对 A 的合规页"],
  "todo": ["确认 B 是否有 SOC2", "获取 C 报价"],
  "open": ["B 的 SOC2 状态未知"]
}

把两个动作做成工具

不要指望模型“自觉”压缩上下文。给它两个明确的工具：checkpoint 写入状态，reset 清空消息但保留状态。工具描述里写死约束：reset 前必须先 checkpoint；检查点最多 2000 字；事实必须带来源。

TOOLS = [
  {"type":"function","function":{
    "name":"checkpoint",
    "description":"保存任务进度与已核实事实。reset 前必须调用。",
    "parameters":{"type":"object","properties":{
      "goal":{"type":"string"},
      "facts":{"type":"array","items":{"type":"object","properties":{
        "text":{"type":"string"},"source":{"type":"string"}}}},
      "done":{"type":"array","items":{"type":"string"}},
      "todo":{"type":"array","items":{"type":"string"}},
      "open":{"type":"array","items":{"type":"string"}}
    },"required":["goal","facts","done","todo"]}}}},
  {"type":"function","function":{
    "name":"reset",
    "description":"清空当前上下文，只保留最近一次 checkpoint。必须先 checkpoint。",
    "parameters":{"type":"object","properties":{}}}}
]

主循环：到阈值就重置

下面是一段可运行的最小实现，只用标准库。它调用任意 OpenAI 兼容接口（把 BASE_URL、MODEL、API_KEY 换成你的）。每轮把检查点作为 system 消息注入；当消息总长度超过阈值，先要求模型 checkpoint，再清空历史。

import json, os, urllib.request

BASE_URL = os.environ.get("BASE_URL", "http://localhost:8000/v1")
MODEL = os.environ.get("MODEL", "qwen3-8b")
KEY = os.environ.get("API_KEY", "none")

def chat(messages, tools=None):
    body = {"model": MODEL, "messages": messages, "temperature": 0}
    if tools: body["tools"] = tools
    req = urllib.request.Request(
        BASE_URL.rstrip("/") + "/chat/completions",
        data=json.dumps(body).encode(),
        headers={"Content-Type":"application/json","Authorization":"Bearer "+KEY})
    with urllib.request.urlopen(req, timeout=120) as r:
        return json.loads(r.read())["choices"][0]["message"]

def run(task, max_turns=40, char_limit=24000):
    state = None
    messages = [{"role":"system","content":"你是执行任务的 agent。上下文过长时先 checkpoint，再 reset。"},
                {"role":"user","content":task}]
    for turn in range(max_turns):
        msg = chat(messages, TOOLS)
        messages.append(msg)
        if msg.get("tool_calls"):
            for call in msg["tool_calls"]:
                name = call["function"]["name"]
                args = json.loads(call["function"]["arguments"] or "{}")
                if name == "checkpoint":
                    state = args
                    result = "saved"
                elif name == "reset":
                    messages = [
                      {"role":"system","content":"继续任务。检查点："+json.dumps(state, ensure_ascii=False)},
                      {"role":"user","content":"从检查点继续，先列出下一步。"}]
                    result = "reset ok"
                messages.append({"role":"tool","tool_call_id":call["id"],"content":result})
        if sum(len(str(m.get("content") or "")) for m in messages) > char_limit:
            messages.append({"role":"user","content":"上下文接近上限：先调用 checkpoint 保存进度，再调用 reset。"})
        if msg.get("content") and "FINAL" in msg["content"]:
            return msg["content"], state
    return "max turns", state

把 reset 做成硬约束

模型可能一直 checkpoint 却不肯 reset，也可能 reset 后反复重做。两个办法：一是在代码里限制“连续两次 reset 之间必须完成至少一个 todo”；二是 reset 后只给检查点，不给旧消息，避免它从垃圾上下文里恢复旧错误。

def can_reset(state, last_state):
    if not state: return False
    if last_state is None: return True
    return len(state.get("done", [])) > len(last_state.get("done", []))

怎么验证它真的有用

别只看一两个 demo。准备 20 个多步任务（检索、对比、填表各几条），同一模型跑两组：A 组普通 ReAct，B 组加 checkpoint/reset。记录四个数：任务成功率、输入 token、总耗时、重置后关键事实丢失率。CORAL 论文在 GAIA 上报告 8B 模型平均 42.8%、输入 token 比 ReAct 少 6 到 7 倍，但你的任务分布不同，必须自己测。

  • 成功率：按“最终答案是否满足验收条件”判，不看措辞。
  • 输入 token：统计所有请求的 prompt tokens，重置本身也有成本。
  • 事实保留：在检查点里埋 5 个关键事实，重置后逐条问，看是否还在。
  • 错误复现：同一错误在重置后是否再次出现，判断检查点有没有把错误固化成“事实”。

三个常见坑

第一，检查点漂移：模型会把猜测写进 facts。写入前要求每条事实必须有 source，没有来源的放到 open。第二，检查点膨胀：不限制长度，几轮后又变成另一个长上下文；按字数截断，并优先保留未完成事项。第三，工具循环：模型反复 checkpoint 不推进；给每轮设“必须产生一个新事实或完成一个 todo”的检查，否则强制结束。

什么时候不该用

任务很短、或者强依赖逐字上下文（合同审阅、代码 diff 讨论）时，重置会丢细节，不如直接加大窗口。另一个前提是任务可被写成结构化状态；如果进度本身难以言说，先做人工拆解，再谈自动检查点。

这套方法的价值不在“省 token”这一个指标，而在把长任务的失败变得可诊断：是事实没记住、是下一步丢了，还是重置后走错方向。先把检查点做小、做准，再考虑论文里的强化学习训练。