Harness Engineering：Agent 项目跑不稳，问题多半在模型外面

本文是 AI 应用系统学习系列的 L3 实战篇。前置：生产级 Agent 架构：LangGraph、MCP 与多 Agent。 学完可以配合面试题食用：AI 应用的可观测性：Tracing 与 Token 监控、AI 应用与传统后台的落地坑。

为什么 Demo 惊艳，上线就拉胯

看 F1 就懂了。同一个车手，换个车队成绩差一大截：进站 2.3 秒换完四条胎、每圈遥测回传、轮胎衰减建模，这些都不是车手干的，是车队干的。模型就是车手，Harness 就是车队。

写成公式：Agent = Model + Harness。模型只负责一件事——给定上下文，推理出下一步。Harness 是模型外面那层运行环境，管的是其余全部：上下文怎么拼、工具调用怎么执行、报错了怎么兜、干到一半崩了怎么续、哪些动作要人审批。

一个常见的错误归因：线上效果差，第一反应是换模型、改提示词。实测里更常见的真相是，同一个模型，把工具报错兜底加上、上下文裁剪做好，任务成功率就能明显上去；反过来，提示词调得再花，遇到一次工具 500 超时，整条链路照样挂。Demo 与生产的差距，几乎全在 Harness 这层。

Harness 的六层：一层记一个典型事故

graph TB
    M["模型<br/>（只管推理）"]
    subgraph H["Harness：模型外面的六层运行环境"]
        L1["1 模型路由层<br/>选模型、限流、降级"]
        L2["2 上下文装配层<br/>系统提示 / 历史 / 工具说明怎么拼"]
        L3["3 工具执行层<br/>参数校验、超时、报错兜底"]
        L4["4 状态记忆层<br/>checkpoint、断点恢复"]
        L5["5 权限沙箱层<br/>高危审批、隔离执行"]
        L6["6 观测恢复层<br/>trace、指标、事后重放"]
    end
    M --- L1
    L1 --- L2 --- L3 --- L4 --- L5 --- L6

排障时按这六层从上往下过，每层都有一种典型死法：

• 模型路由层：为了省钱，把多步规划和格式化输出都路由给同一个小模型。小模型写周报没问题，做五步以上的规划就开始丢约束、漏条件。路由要分级：格式转换、摘要这类便宜活走小模型，规划、写代码走旗舰模型。
• 上下文装配层：把 20 轮历史、12 个工具说明全量塞进 prompt，关键约束被稀释，模型开始答非所问。装配层要干裁剪和排序：最近的关键约束放前面，老历史压成摘要，用不到的工具说明直接不塞。
• 工具执行层：接口返回 500，错误字符串原样进了对话，模型把报错信息当成事实继续推理，后面全歪。工具层要兜住超时和异常，并把错误翻译成模型能用的信息：「这个调用失败了，原因是 X，建议改用 Y」。
• 状态记忆层：跑了 40 步的数据迁移，第 38 步忘了第 3 步定下的命名规则。长任务的状态必须外置落盘，不能指望模型自己记 40 步之前的事。
• 权限沙箱层：给 Agent 直连生产库的账号，一句「清理脏数据」变成不带 WHERE 的 DELETE。高危动作必须过审批门，执行环境要隔离，账号要最小权限。
• 观测恢复层：凌晨两点任务挂了，日志里只有一行 KeyError，没有 trace 没有现场，只能整批重跑。每一步要留 trace 和 checkpoint，出事能重放到出错那一行。

长任务三件套：checkpoint、幂等、审批门

任务一长（10 步以上），三样东西缺一不可：

1. checkpoint：每步把消息列表、步数、token 花销落盘。崩了从最近一步继续，而不是从零重跑——重跑不只是慢，之前批过的审批、花掉的 token 全得再来一遍。
2. 幂等重试：工具调用带幂等键（用业务 ID 就行）。断点恢复后重放执行过的调用才不会重复下单、重复扣款。支付回调这种天然非幂等的操作，要么改成查询确认，要么加去重表。
3. 审批门：删数据、发邮件、改生产配置，这类动作执行前停下来等人确认。注意，prompt 里写「请不要删除数据」是软约束，模型心情不好就违反；审批门是硬约束，代码层面过不去。

Loop Engineering：先想清楚循环什么时候停

Agent 本质是个 while 循环。触发条件好定：用户消息、定时任务、上游事件。工程重点在停止条件，没有它就是一个烧钱永动机。三种闸门一起上：

• 步数上限：比如 30 步，到了就停，带着当前进度报告出去。
• token 预算：比如单任务 20 万 token。算笔账：一个 50 步的任务，每步上下文 1 万 token，就是 50 万 token；没有预算闸门，一个原地打转的任务能烧掉一整天的 API 预算。
• 单调性检测：连续 3 步状态没变化，或者同一个报错重复出现，判定为原地打转，强制停止并上报。

三个条件是「或」的关系，任何一个触发都停。只设一个的下场：步数设了、预算没设，长上下文任务照样烧穿预算。

动手实操：最小 Harness 骨架

下面是一个能直接跑的骨架：主循环 + 每步 checkpoint 落盘 + token 预算闸门 + 高危操作审批门。llm_call 换成真实模型 API，safe_execute 里接真实工具，就是生产代码的起点。

# harness.py —— 最小 Agent 运行外壳
# 跑一遍：python3 harness.py（演示里含一次高危操作，会触发审批门）
import json, itertools, pathlib

CKPT = pathlib.Path("ckpt.json")
MAX_STEPS = 30            # 停止条件 1：步数上限
TOKEN_BUDGET = 200_000    # 停止条件 2：token 预算
DANGEROUS = ("rm ", "drop ", "delete from", "git push --force")  # 高危关键词

def load_ckpt():
    if CKPT.exists():                      # 断点恢复：从上次的进度继续
        return json.loads(CKPT.read_text())
    return {"step": 0, "tokens": 0, "messages": [], "done": False}

def save_ckpt(state):
    CKPT.write_text(json.dumps(state, ensure_ascii=False))
    # 生产环境：写数据库或对象存储，别放单机磁盘

def needs_approval(name: str, args: str) -> bool:
    return name == "run_sql" or any(d in args.lower() for d in DANGEROUS)

def run(llm_call, goal: str):
    state = load_ckpt()
    if state["done"]:
        return "上次任务已完成"            # 幂等入口：重复触发不重跑
    if not state["messages"]:
        state["messages"] = [{"role": "user", "content": goal}]

    for _ in itertools.count():
        # ---- 两个预算闸门 ----
        if state["step"] >= MAX_STEPS:
            raise RuntimeError(f"步数达到上限 {MAX_STEPS}，转人工")
        if state["tokens"] >= TOKEN_BUDGET:
            raise RuntimeError("token 预算耗尽，强制停止")

        resp = llm_call(state["messages"])
        state["tokens"] += resp["usage"]["total_tokens"]   # 每步记账
        msg = resp["message"]
        state["messages"].append(msg)

        calls = msg.get("tool_calls") or []
        if not calls:                       # 模型不再要工具 = 任务结束
            state["done"] = True
            save_ckpt(state)
            return msg["content"]

        for call in calls:
            # ---- 审批门：高危操作等人确认，prompt 约束不管用 ----
            if needs_approval(call["name"], str(call["arguments"])):
                ans = input(f"高危操作 {call['name']}({call['arguments']})，允许? y/N: ")
                if ans.strip().lower() != "y":
                    state["messages"].append({
                        "role": "tool", "tool_call_id": call["id"],
                        "content": "用户拒绝该操作，请改用别的方案"})
                    continue

            result = safe_execute(call)     # 工具层：报错要兜住，返回给模型重试
            state["messages"].append({
                "role": "tool", "tool_call_id": call["id"], "content": result})

        state["step"] += 1
        save_ckpt(state)                    # 每步落盘，崩了从这续

# ---- 接真实系统时替换这两个函数 ----
def safe_execute(call):
    # 真实实现：参数校验 + 超时控制 + 异常兜底，错误翻译成模型能读的话
    return f"ok: {call['name']} 已执行"

if __name__ == "__main__":
    def fake_llm(messages):
        # 演示：第 1 轮要求调工具（含 delete from，会触发审批门），第 2 轮给结论
        if not any(m["role"] == "tool" for m in messages):
            return {"usage": {"total_tokens": 1200}, "message": {
                "role": "assistant", "content": "",
                "tool_calls": [{"id": "c1", "name": "run_sql",
                                "arguments": "delete from orders where status='test'"}]}}
        return {"usage": {"total_tokens": 800},
                "message": {"role": "assistant", "content": "清理完成，共处理 1 张表"}}
    print(run(fake_llm, goal="清理测试订单数据"))

跑起来后注意两件事：输入 N 拒绝高危操作时，模型收到的是「请改用别的方案」，循环继续而不是中断；Ctrl+C 杀掉进程再重跑，任务从 ckpt.json 里记录的那一步继续，之前的对话不丢。

常见误区与小结

• 只调提示词，不改外壳。工具报错、上下文超长、状态丢失，这三个问题提示词一个都解决不了。
• checkpoint 存单机内存或本地磁盘。机器一重启全没，要写数据库或对象存储。
• 停止条件只设一个。只设步数上限，token 照样烧穿预算；只设预算，任务可能空转一万步。
• 用 prompt 约束高危操作。提示词是软约束，绕过它只需要模型一次发挥；审批门是代码里的硬关卡。
• 出了问题整批重跑。没有 trace 和 checkpoint，重跑等于把审批、token、时间全付两遍。

小结：本文把 38 篇那张架构图拆成了可逐层检查的六层框架，外加长任务三件套和循环停止条件。面试讲 Agent 项目时，按这六层逐条说「模型之外我做了什么」，比背一段提示词工程话术有说服力得多。下一篇讲模型网关：多模型、多 key、限流降级的统一入口怎么搭。

参考

• Anthropic: Building Effective Agents 与 Claude Agent SDK 文档（agent loop 与工具执行设计）
• Anthropic 工程博客：Writing effective tools for agents（工具报错返回的处理思路）
• 本项目实验代码：agent-lab/harness-skeleton/