给长程 Agent 加一层 harness：能力清单、幂等重放与可回滚检查点

原创

用户9746675

修改于 2026-08-03 18:57:46

最近两个开源项目让我把自己 Agent 项目里的执行层重写了一遍。清华 SIGS 开源的 VeriLoop Coder-E1 给了权重、tokenizer 和部分 PEFT 适配器，但生产级的 Self-Harness 运行时没有放出来；同一天发布的 JarvisHub 则相反，把画布形态的项目状态和每轮下发的 Capability Manifest 整套开源了。两个项目一正一反，指向的是同一个工程事实：长程 Agent 在第三十轮掉链子，多数时候不是模型能力不够，而是外面缺一层能约束它、能记住它、能回滚它的运行时。

这篇不聊趋势，直接给一个能跑的最小 harness 实现。Python 3.10+ 标准库即可运行，四个部件：每轮重算的能力清单、带幂等键的工具调用、证据绑定的自动回滚、原子写入的检查点。文末有完整的串联示例和运行输出。

一、能力清单：把这一轮允许做什么变成显式约束

常见做法是启动时把全量工具注册给模型，之后每轮都把同一份工具列表塞进 prompt。问题在于动作的合法性是随状态变化的：仓库还没克隆就不该允许 run_tests，构建产物还没生成就不该允许 deploy。指望模型自己记住这些前置条件，等于把一致性约束交给概率生成。

更稳的做法是每一轮根据当前状态重算一份可用动作集合，既下发给模型，也用于执行前的服务端校验。

from dataclasses import dataclass
from typing import Any, Callable


@dataclass(frozen=True)
class Capability:
    name: str
    handler: Callable[..., Any]
    required_state: frozenset = frozenset()   # 依赖的前置状态键
    mutating: bool = False                    # 是否写外部系统
    max_calls: int = 5                        # 单次运行的调用上限


class CapabilityRegistry:
    def __init__(self, caps: list):
        self._caps = {c.name: c for c in caps}

    def manifest(self, state: dict, counts: dict) -> list:
        """每轮重算：只返回前置状态齐备且未超配额的动作"""
        allowed = []
        for name, cap in self._caps.items():
            if not cap.required_state <= state.keys():
                continue
            if counts.get(name, 0) >= cap.max_calls:
                continue
            allowed.append(name)
        return sorted(allowed)

    def check(self, name: str, state: dict, counts: dict) -> Capability:
        if name not in self.manifest(state, counts):
            raise PermissionError(f"动作 {name} 在当前状态下不可用")
        return self._caps[name]

里面关键的一行是 cap.required_state <= state.keys()，集合包含判断，前置键不齐的动作直接从清单里消失。模型返回越界动作时，harness 抛 PermissionError 并把这条错误写回上下文，而不是先执行再指望后面能纯正。

二、幂等键：恢复不等于重复写入

检查点恢复最容易踩的坑是重复副作用。任务在第 12 步崩溃，从第 10 步的检查点恢复后，第 10、11 步的工具调用会被再执行一遍。如果这两步里有建分支、发消息、写数据库，业务侧就会看到重复记录。

解法是给每次调用算一个只依赖动作名和参数的幂等键，命中就直接返回上次结果，跳过真实副作用。

import hashlib
import json


def idem_key(cap_name: str, args: dict) -> str:
    payload = json.dumps({"cap": cap_name, "args": args},
                         sort_keys=True, ensure_ascii=False)
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:16]


class CallLog:
    def __init__(self, results: dict = None):
        self.results = results if results is not None else {}

    def run_once(self, key: str, fn: Callable[[], Any]):
        if key in self.results:
            return self.results[key], True     # 命中重放，不再产生副作用
        value = fn()
        self.results[key] = value
        return value, False

这里有个容易翻车的细节：参数里千万别掺入时间戳、随机 request_id 或者带毫秒的 trace_id，否则每次算出来的键都不一样，幂等等于没做。比较稳的做法是显式声明哪些参数参与计算键，其余的走一个 meta 字段旁路传递。

三、证据绑定：变更必须带校验结果，否则回滚

VeriLoop 那套循证螺旋落到工程上很朴素：不要问模型你做完了吗，而是每次 mutating 动作之后立刻跑一次独立校验，校验器给出可复核的证据；不通过就回滚这一步，并把失败证据写进上下文让模型换个方案。

import copy


@dataclass
class StepResult:
    ok: bool
    output: Any = None
    evidence: str = ""      # 校验器给出的可复核证据，不是 bool


class Executor:
    def __init__(self, registry: CapabilityRegistry, log: CallLog, verify: Callable):
        self.registry = registry
        self.log = log
        self.verify = verify

    def step(self, cap_name: str, args: dict, ctx: dict) -> StepResult:
        state, counts = ctx["state"], ctx["counts"]
        cap = self.registry.check(cap_name, state, counts)
        key = idem_key(cap_name, args)
        before = copy.deepcopy(state) if cap.mutating else None

        value, replayed = self.log.run_once(
            key, lambda: cap.handler(state=state, **args))
        counts[cap_name] = counts.get(cap_name, 0) + 1

        result = self.verify(cap_name, value, state)
        if not result.ok and before is not None:
            state.clear()
            state.update(before)             # 回滚到动作执行前
            self.log.results.pop(key, None)  # 清掉幂等键，允许换参数重试
        result.output = value
        ctx["trace"].append({"cap": cap_name, "ok": result.ok,
                             "replayed": replayed, "evidence": result.evidence})
        return result

注意 evidence 是字符串而不是布尔值。让校验器回一句 pytest 收集到 42 个用例、失败 1 个：test_parse_utf8，比回一个 False 有用得多。这句话可以直接进下一轮 prompt，模型知道往哪改；而 False 只会让它重试同样的方案。

四、检查点：存什么，存在哪一层

检查点有两种做法，取舍很实际：

维度	沙箱层整机快照	编排层状态序列化
恢复保真度	高，进程和文件系统一起回来	中，只恢复显式声明的键
单点存储成本	GB 级	KB 到 MB 级
跳版本兼容	差，镜像和运行时耦合	好，就是一份 JSON
适合场景	代码仓库、编译产物这类重环境	检索、审批、内容生产这类轻状态

多数业务用编排层就够了，重点是写入要原子：

import os

CKPT_KEYS = ("state", "counts", "results", "cursor")


def save(path: str, ctx: dict) -> None:
    tmp = path + ".tmp"
    with open(tmp, "w", encoding="utf-8") as f:
        json.dump({k: ctx[k] for k in CKPT_KEYS}, f, ensure_ascii=False)
        f.flush()
        os.fsync(f.fileno())
    os.replace(tmp, path)      # 原子替换，避免半截检查点


def load(path: str) -> dict:
    if not os.path.exists(path):
        return {"state": {}, "counts": {}, "results": {}, "cursor": 0}
    with open(path, encoding="utf-8") as f:
        return json.load(f)

os.fsync 和 os.replace 这两行别省。写检查点的过程中崩溃，留下一个语法完整但内容截断的 JSON，比完全没有检查点更难排查。

五、串起来跑一遍

def clone(state, repo):
    state["repo"] = repo
    return f"cloned {repo}"


def patch(state, path, content):
    state.setdefault("files", {})[path] = content
    return f"wrote {path}"


def run_tests(state):
    failed = [p for p, c in state.get("files", {}).items() if "TODO" in c]
    return {"failed": failed}


def verify(cap_name, value, state) -> StepResult:
    if cap_name == "run_tests":
        failed = value["failed"]
        return StepResult(ok=not failed,
                          evidence=f"失败用例：{failed}" if failed else "全部通过")
    return StepResult(ok=True, evidence="无需校验")


registry = CapabilityRegistry([
    Capability("clone", clone),
    Capability("patch", patch, required_state=frozenset({"repo"}), mutating=True),
    Capability("run_tests", run_tests, required_state=frozenset({"files"})),
])

ctx = load("ckpt.json")
ctx["trace"] = []
ex = Executor(registry, CallLog(ctx["results"]), verify)

plan = [
    ("clone", {"repo": "demo"}),
    ("patch", {"path": "a.py", "content": "# TODO"}),
    ("run_tests", {}),
    ("patch", {"path": "a.py", "content": "print(1)"}),
    ("run_tests", {}),
]

for i, (cap, args) in enumerate(plan[ctx["cursor"]:], start=ctx["cursor"]):
    print("manifest:", registry.manifest(ctx["state"], ctx["counts"]))
    r = ex.step(cap, args, ctx)
    print(f"  step{i} {cap} ok={r.ok} evidence={r.evidence}")
    ctx["cursor"] = i + 1
    save("ckpt.json", ctx)

运行输出：

manifest: [clone]
  step0 clone ok=True evidence=无需校验
manifest: [clone, patch]
  step1 patch ok=True evidence=无需校验
manifest: [clone, patch, run_tests]
  step2 run_tests ok=False evidence=失败用例：[a.py]
manifest: [clone, patch, run_tests]
  step3 patch ok=True evidence=无需校验
manifest: [clone, patch, run_tests]
  step4 run_tests ok=True evidence=全部通过

输出里能看到三件事。第一轮的清单里只有 clone，patch 和 run_tests 因为前置状态不齐而不可见；step2 校验失败后 state 被回滚，幂等键也被清掉，所以 step3 换内容重写同一个文件不会被当成重放；每一步结束都落了一次检查点，中途 kill 掉进程再启动，会从 cursor 记录的位置继续，已完成的调用直接命中重放。

六、落地时的几个经验

manifest 要同时下发给模型和用于服务端校验，只做一边等于没做。
幂等键的输入要显式白名单，别把随机 id 和时间戳算进去。
校验器返回可复核证据而不是布尔值，这是模型能自我纯正的前提。
监控里加两个指标：单位任务成本和平均恢复次数，它们比准确率更能反映系统是否真的可用。

这套东西加起来不到三百行，但它是 Agent 项目里唯一不会因为换模型而作废的部分。模型权重在快速变成公共品，harness 才是各自团队真正需要沉淀的资产。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

AIGC