实操：把最小 Agent 升级成迷你 Harness
加上上下文装配、权限规则、Hooks 与运行日志，并对比升级前后的表现。
一句话理解：迷你 Harness 没有换模型，也没有换循环，只是在循环外面补上五样东西：按固定顺序装配的上下文、带校验与超时的工具运行时、allow / ask / deny 权限规则、PreToolUse / PostToolUse 钩子，以及每一步都落盘的运行日志；最后用 5 个固定任务证明它确实变好了。
上一课的 agent.py 能跑通，但离“放心交给它干活”还差得远：它不知道项目里的业务规则，写操作拿到就执行，下游系统卡住就一直等，工具结果再长也整份塞给模型，运行过程只打印在终端、事后无从查起。
这是 Harness 模块的最后一课，也是模块实战。我们不引入任何框架，在 agent.py 的基础上把本模块讲过的机制一项项落到代码里，再用上一课「Harness 工程」的方法，用同一组任务对比升级前后的表现。
你将做出什么
先看升级前后的整体结构：模型和循环都没变，变化全在模型的前后两侧。
具体来说，你会在 agent.py 旁边新建一个 harness.py，补上下面这些能力，每一项都对应本模块的一课：
| 新能力 | 代码位置 | 对应课时 |
|---|---|---|
| 工具注册表、参数校验 | 第 1 步 | 工具运行时 |
| 按固定顺序装配系统提示，读取项目说明文件 | 第 2 步 | 上下文装配、项目说明文件与配置分层 |
| allow / ask / deny 权限规则，写操作需确认 | 第 3 步 | 权限规则与审批模式 |
| PreToolUse / PostToolUse 两个钩子 | 第 4 步 | Hooks 与生命周期 |
| 超时与结果截断，串起一次工具调用 | 第 5 步 | 工具运行时 |
| JSONL 运行日志 | 第 6、7 步 | Tracing |
| 5 个固定任务的前后对比 | 第 8 步 | Harness 工程、构建评测集 |
同时我们加一个新需求：让 Agent 能 重新提交退款。这是一个写操作，正好用来检验权限规则。
准备工作
- 完成上一课，
mini-agent目录里有能跑通的agent.py；虚拟环境、anthropicSDK 和 API Key 沿用上一课的设置，模型同样默认claude-opus-5。 - Python 3.10 或更高版本（代码用到了
str | None写法）。
做完后的目录结构：
mini-agent/
├── agent.py # 上一课的最小 Agent，第 0 步只改模拟数据
├── AGENTS.md # 项目说明文件，第 2 步新建
├── harness.py # 迷你 Harness，第 1–7 步按顺序写入
├── compare.py # 前后对比脚本，第 8 步新建
└── runs/ # 运行日志，运行时自动生成harness.py 有两种模式：upgraded 是迷你 Harness；baseline 复刻 agent.py 的做法（写死的指令、拿到就执行），但用同一个模型、同一组工具。这样对比时唯一的变量就是 Harness。日志在两种模式下都会记录，因为它只观察、不干预。
本课代码在 anthropic 1.7.0、Python 3.12 下，用模拟的模型响应跑通了校验、权限、两个钩子、超时、截断、日志和对比脚本的全部分支。接入真实模型后，工具调用顺序和回答措辞每次都会不同。
第 0 步：给模拟数据加三种“刁难”
打开 agent.py，把第 1、2 两段（从 # ---------- 1. 到 get_payment 函数结束）替换为下面的代码。其余部分不动，agent.py 仍然可以单独运行。
# ---------- 1. 模拟两个业务系统的数据 ----------
ORDERS = {
"ORD1001": {"status": "已退货入库", "amount": 428.0, "created_at": "2026-09-13", "refund_due_date": "2026-09-15"},
"ORD2002": {"status": "已完成", "amount": 129.0, "created_at": "2026-09-10", "refund_due_date": "无"},
"ORD4004": {"status": "已退货入库", "amount": 89.0, "created_at": "2026-09-20", "refund_due_date": "2026-09-25"},
"ORD5005": {"status": "已退货入库", "amount": 1299.0, "created_at": "2026-09-18", "refund_due_date": "2026-09-22"},
}
PAYMENTS = {
"ORD1001": {
"paid_at": "2026-09-13 08:10", "amount": 428.0,
"refund_status": "退款审核中", "refund_updated_at": "2026-09-14 09:30",
"refund_note": "收款账户信息异常，待联系用户确认",
"refund_account": "6222020200112233445", # 新增：完整卡号，不该原样出现在回答里
},
"ORD5005": {
"paid_at": "2026-09-18 20:05", "amount": 1299.0,
"refund_status": "退款处理中", "refund_updated_at": "2026-09-27 23:00",
"refund_note": "银行通道繁忙，系统自动重试中",
# 新增：300 条重试日志，整份交给模型会白白消耗大量 Token
"retry_logs": [f"第 {i} 次自动重试：银行通道繁忙，稍后再试" for i in range(1, 301)],
},
}
# ---------- 2. 工具实现：出错就抛出“可行动”的错误信息 ----------
def get_order(order_id: str) -> dict:
if order_id not in ORDERS:
raise ValueError(f"订单 {order_id} 不存在，请让用户核对订单号（示例：ORD1001）")
return ORDERS[order_id]
def get_payment(order_id: str) -> dict:
if order_id == "ORD4004": # 新增：模拟支付系统卡住 10 秒后才报错
time.sleep(10)
raise ConnectionError("支付系统连接超时")
if order_id not in PAYMENTS:
raise ValueError(f"支付系统暂无 {order_id} 的支付记录，可能为货到付款或数据未同步")
return PAYMENTS[order_id]三处新增各有用途：ORD1001 多了一个完整卡号，用来检验脱敏；ORD4004 的支付系统会卡住 10 秒，用来检验超时；ORD5005 带了 300 条重试日志，用来检验结果截断。注意 ORD5005 的字段顺序：退款状态和备注在前、日志在后，截断时重要信息才能保住。
第 1 步：工具注册表与参数校验
新建 harness.py，写入第一段。它从 agent.py 复用模型名、步数上限、系统指令和两个查询工具，再新增写工具 resubmit_refund：
"""迷你 Harness：在 agent.py 的基础上加上下文装配、工具运行时、权限规则、Hooks 与运行日志"""
import fnmatch
import json
import re
import sys
import threading
import time
import uuid
from dataclasses import dataclass, field
from datetime import date
from pathlib import Path
import anthropic
from agent import MAX_STEPS, MODEL, SYSTEM, get_order, get_payment
# ---------- 1. 工具注册表：每个工具自带说明、参数 Schema、超时和“是否写操作” ----------
SUBMITTED = [] # 模拟退款系统：记录被重新提交的订单
def resubmit_refund(order_id: str, reason: str) -> dict:
get_order(order_id) # 订单不存在时直接抛错
SUBMITTED.append(order_id)
return {"order_id": order_id, "result": "已重新提交退款审核", "reason": reason}
@dataclass
class Tool:
name: str
description: str
params: dict # 参数名 -> JSON Schema；本课所有参数都是必填字符串
fn: object
timeout: float = 3.0
writes: bool = False
def schema(self) -> dict:
return {"name": self.name, "description": self.description,
"input_schema": {"type": "object", "properties": self.params,
"required": list(self.params), "additionalProperties": False}}
ORDER_ID = {"type": "string", "pattern": r"^ORD\d{4}$", "description": "订单号，例如 ORD1001"}
REGISTRY = {t.name: t for t in [
Tool("get_order", "查询订单基本信息：状态、金额、下单日期、承诺退款日期。用户给出订单号时先调用它。",
{"order_id": ORDER_ID}, get_order),
Tool("get_payment", "查询订单的支付与退款记录（退款状态、备注、收款账户）。判断退款延迟原因时调用。",
{"order_id": ORDER_ID}, get_payment),
Tool("resubmit_refund", "重新提交退款审核（写操作）。仅在用户明确要求、且退款备注里的问题已解决时调用。",
{"order_id": ORDER_ID, "reason": {"type": "string", "description": "重新提交的原因"}},
resubmit_refund, writes=True),
]}
def validate(tool: Tool, args: dict) -> str | None:
"""最小参数校验：必填、多余字段、类型、格式。生产中可换成 jsonschema 库。"""
missing = [k for k in tool.params if k not in args]
extra = [k for k in args if k not in tool.params]
if missing or extra:
return f"参数错误：缺少 {missing}，多余 {extra}"
for k, v in args.items():
if not isinstance(v, str):
return f"参数 {k} 应为字符串，收到 {type(v).__name__}"
pattern = tool.params[k].get("pattern")
if pattern and not re.fullmatch(pattern, v):
return f"参数 {k}={v!r} 格式不对，应形如 ORD1001，请先向用户核对"
return NoneTool 把运行时需要的信息放在一起：说明、参数、超时和是否写操作。模型看到的工具定义由 schema() 生成，注册表是唯一的来源，不会出现“定义里有、执行时找不到”的情况。
validate 只覆盖本课用到的 Schema 子集，生产中可以换成 jsonschema 库。Claude API 也提供 strict tool use（在工具定义上设 strict: true），保证参数符合 Schema；手写校验的价值在于看清运行时做了什么，并且能返回面向业务、可行动的提示，比如“应形如 ORD1001，请先向用户核对”。
第 2 步：按固定顺序装配上下文
先在项目根目录新建项目说明文件 AGENTS.md，写入团队约定的业务规则：
# 订单售后诊断 Agent：项目说明
## 业务规则
- 承诺退款日期早于今天、且退款尚未完成的订单，结论第一句标注「已超时」。
- 收款账户只能引用脱敏后的形式（如 6222****3445），不要让用户在对话里发送完整卡号。
- 重新提交退款前，先确认退款备注里的问题已经解决；无法确认时交给人工。
## 回答格式
- 金额写成「428.00 元」的形式。然后在 harness.py 末尾追加第二段：
# ---------- 2. 上下文装配：固定顺序，稳定的放前面，易变的放后面 ----------
INSTRUCTION_FILES = ["AGENTS.md", "AGENTS.local.md"] # 先团队共享，后个人本地
MAX_INSTRUCTION_CHARS = 4000
def load_instructions() -> str:
parts = [f"<file name=\"{n}\">\n{Path(n).read_text(encoding='utf-8').strip()}\n</file>"
for n in INSTRUCTION_FILES if Path(n).exists()]
text = "\n".join(parts)
if len(text) > MAX_INSTRUCTION_CHARS: # 给项目说明设预算，超出就截断并提示
text = text[:MAX_INSTRUCTION_CHARS] + "\n…（项目说明过长，已截断）"
return text
def build_system(unattended: bool) -> str:
mode = "无人值守：需要人工确认的操作会被自动拒绝" if unattended else "交互式：写操作会先请用户确认"
sections = [
("角色与底线", SYSTEM), # ① 最稳定
("项目说明", load_instructions()), # ② 随项目变化
("运行环境", f"今天是 {date.today().isoformat()}。{mode}。"), # ③ 每次运行都可能变
]
return "\n\n".join(f"## {title}\n{body}" for title, body in sections if body)系统提示按固定顺序拼成三节：角色与底线、项目说明、运行环境。越稳定的内容越靠前，每次都可能变化的日期放在最后；这样前缀更稳定，也更利于提示缓存。项目说明文件按“先团队共享、后个人本地”的顺序读取，后读的 AGENTS.local.md 适合放不提交到仓库的个人偏好。Claude Code 的 CLAUDE.md 与 CLAUDE.local.md 也是这种分层，并且同样能读取仓库里的 AGENTS.md。
两处细节值得注意：一是给项目说明设了 4000 字符的预算，防止说明文件越写越长、挤占上下文；二是把今天的日期告诉模型，否则它无法判断承诺退款日期是否已过，也就无法执行“已超时”这条规则。
第 3 步：权限规则
追加第三段：
# ---------- 3. 权限规则：deny > ask > allow，没有命中任何规则时默认 ask ----------
RULES = {
"deny": ["update_*", "delete_*"], # 改资料、删数据：本 Agent 一律不做
"ask": ["resubmit_*"], # 重新提交退款：必须人工确认
"allow": ["get_*"], # 只读查询：自动放行
}
def decide(tool_name: str) -> str:
for decision in ("deny", "ask", "allow"): # 顺序就是优先级
if any(fnmatch.fnmatch(tool_name, p) for p in RULES[decision]):
return decision
return "ask"
def ask_human(name: str, args: dict) -> bool:
if not sys.stdin.isatty(): # 没有终端可问，就当作拒绝
return False
answer = input(f"\n[需要确认] {name}({json.dumps(args, ensure_ascii=False)}) 允许执行吗？[y/N] ")
return answer.strip().lower() == "y"规则按 deny、ask、allow 的顺序匹配，第一个命中的决定结果，没有命中任何规则时默认 ask。Claude Code 的权限规则也是这个顺序：宽泛的 deny 规则会盖过更具体的 allow 规则。它的规则还能匹配参数，例如 Bash(aws *)；本课为了简短，只按工具名匹配。
ask_human 在没有终端可问时直接返回拒绝。无人值守运行（批量评测、CI、定时任务）里，“需要确认”就等于“不执行”，这是更安全的默认值，详见 无人值守运行。
第 4 步：两个 Hooks
追加第四段：
# ---------- 4. Hooks：PreToolUse 返回拒绝原因或 None；PostToolUse 返回改写后的结果 ----------
def block_repeated_call(run, name: str, args: dict) -> str | None:
key = (name, json.dumps(args, sort_keys=True, ensure_ascii=False))
if key in run.seen_calls:
return "相同参数的调用本次运行已经处理过，结果不会变化，请直接使用上文结果"
run.seen_calls.add(key)
return None
def mask_card_numbers(run, name: str, output: str) -> str:
return re.sub(r"\d{12,19}", lambda m: m.group()[:4] + "****" + m.group()[-4:], output)
PRE_TOOL_USE = [block_repeated_call]
POST_TOOL_USE = [mask_card_numbers]- PreToolUse：拦截重复调用。相同参数的调用在一次运行里只处理一次，第二次直接返回拒绝原因。这就是上一课练习里的“打转检测”，现在它成了 Harness 的一部分。
- PostToolUse：卡号脱敏。在模型看到结果之前，把 12–19 位数字改成“前 4 位 + 星号 + 后 4 位”。
注意 AGENTS.md 里也写了“只能引用脱敏后的形式”。两者的区别正是 Hooks 那一课的核心：写在说明文件里是“请模型遵守”，写成 Hook 是“程序保证执行”。钩子放在列表里，以后再加新钩子不用改主循环。
第 5 步：工具运行时，把关卡串起来
一次工具调用要依次经过六道关，任何一道拦下，都会变成带原因的 is_error 结果交还给模型：
追加第五段：
# ---------- 5. 工具运行时：查表 → 校验 → PreToolUse → 权限 → 限时执行 → PostToolUse → 截断 ----------
MAX_RESULT_CHARS = 2000
def call_with_timeout(fn, args: dict, timeout: float):
box = {}
def target():
try:
box["value"] = fn(**args)
except Exception as e: # 把工具里的异常带回主线程
box["error"] = e
worker = threading.Thread(target=target, daemon=True) # daemon：超时的线程不拖住进程退出
worker.start()
worker.join(timeout)
if worker.is_alive():
raise TimeoutError
if "error" in box:
raise box["error"]
return box["value"]
def execute_tool(run, name: str, args: dict) -> tuple[str, bool, str]:
"""返回 (结果文本, 是否出错, 决策)。决策会写进日志，方便事后统计。"""
tool = REGISTRY.get(name)
if tool is None:
return f"未知工具 {name}，可用工具：{list(REGISTRY)}", True, "unknown"
if run.mode == "baseline": # 复刻 agent.py：拿到就执行，没有任何把关
try:
return json.dumps(tool.fn(**args), ensure_ascii=False), False, "allow"
except Exception as e:
return f"{type(e).__name__}: {e}", True, "failed"
if error := validate(tool, args):
return error, True, "invalid"
for hook in PRE_TOOL_USE:
if reason := hook(run, name, args):
return f"[PreToolUse 拒绝] {reason}", True, "hook_deny"
decision = decide(name)
if decision == "deny":
return f"权限规则禁止调用 {name}，请告诉用户这一步需要人工处理", True, "deny"
if decision == "ask":
if run.unattended or not ask_human(name, args):
return f"{name} 需要人工确认但未获批准，不要重试，请告诉用户需要人工审批", True, "ask_denied"
decision = "ask_approved"
try:
output = json.dumps(call_with_timeout(tool.fn, args, tool.timeout), ensure_ascii=False)
except TimeoutError:
return f"{name} 超过 {tool.timeout:.0f} 秒未返回，下游系统可能繁忙，请如实告知用户稍后再查", True, "timeout"
except Exception as e:
return f"{type(e).__name__}: {e}", True, "failed"
for hook in POST_TOOL_USE:
output = hook(run, name, output)
if len(output) > MAX_RESULT_CHARS:
output = output[:MAX_RESULT_CHARS] + f"\n…[已截断：原始 {len(output)} 字符，只保留前 {MAX_RESULT_CHARS} 字符]"
return output, False, decision几个设计选择：
- 先校验，再钩子，再权限。参数都不合法，就没必要再去判断该不该放行；钩子在权限之前运行，拦下时权限规则不再参与，这与 Claude Code 中阻断型钩子先于权限规则生效的做法一致。
- 超时用守护线程实现。Python 无法强行终止线程，超时后线程仍在后台跑完，只是 Harness 不再等它。真实系统更推荐在下游客户端上设置请求超时，这里的写法胜在通用。
- 截断时明确告诉模型。结尾附上“已截断、原始多少字符”，模型就不会把前 2000 字符当成全部。
baseline分支原样保留agent.py的行为，只捕获异常，其余一概不管，方便对比。
第 6 步：运行日志
追加第六段：
# ---------- 6. 运行日志：每次模型调用、工具调用都写一行 JSONL ----------
@dataclass
class Run:
mode: str # "upgraded"：迷你 Harness；"baseline"：复刻 agent.py 的行为
unattended: bool = False
run_id: str = field(default_factory=lambda: time.strftime("%Y%m%d-%H%M%S-") + uuid.uuid4().hex[:4])
seen_calls: set = field(default_factory=set)
stats: dict = field(default_factory=lambda: dict.fromkeys(
["steps", "tool_calls", "tool_errors", "unconfirmed_writes", "input_tokens", "output_tokens", "tool_ms"], 0))
@property
def log_path(self) -> Path:
return Path("runs") / f"{self.run_id}.jsonl"
def log(self, event: str, **fields):
self.log_path.parent.mkdir(exist_ok=True)
record = {"ts": round(time.time(), 3), "run_id": self.run_id, "mode": self.mode, "event": event, **fields}
with self.log_path.open("a", encoding="utf-8") as f:
f.write(json.dumps(record, ensure_ascii=False, default=str) + "\n")
@dataclass
class Result:
text: str
status: str
stats: dict
log_path: Path每次运行一个文件，每个事件一行 JSON：run_start 记录任务与系统提示长度，model_call 记录停止原因、Token 和耗时，tool_call 记录工具、参数、权限决策、是否出错、耗时和结果长度，run_end 汇总整次运行。日志只记结果长度、不记结果原文，敏感数据不会落盘。逐行追加的好处是进程中途崩溃也不丢已写的部分，用 grep 或 jq 就能检索。
第 7 步：主循环
最后一段是主循环和命令行入口：
# ---------- 7. 主循环：还是 agent.py 那个循环，只是接上了上面所有部件 ----------
def run_agent(task: str, mode: str = "upgraded", unattended: bool = False) -> Result:
client = anthropic.Anthropic()
run = Run(mode, unattended)
system = SYSTEM if mode == "baseline" else build_system(unattended)
tools = [t.schema() for t in REGISTRY.values()]
messages = [{"role": "user", "content": task}]
run.log("run_start", task=task, model=MODEL, system_chars=len(system))
started = time.perf_counter()
def finish(status: str, text: str) -> Result:
run.log("run_end", status=status, seconds=round(time.perf_counter() - started, 2), **run.stats)
return Result(text, status, run.stats, run.log_path)
for step in range(1, MAX_STEPS + 1):
run.stats["steps"] = step
t0 = time.perf_counter()
response = client.messages.create(
model=MODEL, max_tokens=16000, system=system, tools=tools, messages=messages
)
run.stats["input_tokens"] += response.usage.input_tokens
run.stats["output_tokens"] += response.usage.output_tokens
run.log("model_call", step=step, stop_reason=response.stop_reason,
input_tokens=response.usage.input_tokens, output_tokens=response.usage.output_tokens,
ms=round((time.perf_counter() - t0) * 1000))
if response.stop_reason == "end_turn":
return finish("done", "".join(b.text for b in response.content if b.type == "text"))
if response.stop_reason != "tool_use":
return finish("stopped", f"[提前停止] stop_reason={response.stop_reason}")
messages.append({"role": "assistant", "content": response.content})
results = []
for block in response.content:
if block.type != "tool_use":
continue
t0 = time.perf_counter()
output, is_error, decision = execute_tool(run, block.name, block.input)
ms = round((time.perf_counter() - t0) * 1000)
tool = REGISTRY.get(block.name)
run.stats["tool_calls"] += 1
run.stats["tool_errors"] += is_error
run.stats["tool_ms"] += ms
if tool and tool.writes and not is_error and decision != "ask_approved":
run.stats["unconfirmed_writes"] += 1
run.log("tool_call", step=step, tool=block.name, input=block.input, decision=decision,
is_error=is_error, ms=ms, result_chars=len(output))
print(f"[{mode} step {step}] {block.name}({json.dumps(block.input, ensure_ascii=False)}) "
f"-> {decision}{' (is_error)' if is_error else ''} {ms}ms")
results.append({"type": "tool_result", "tool_use_id": block.id, "content": output, "is_error": is_error})
messages.append({"role": "user", "content": results})
return finish("max_steps", f"[提前停止] 已达到最大步数 {MAX_STEPS}，任务未完成")
if __name__ == "__main__":
words = [a for a in sys.argv[1:] if not a.startswith("--")]
question = words[0] if words else "订单 ORD1001 的退款为什么还没到账？"
mode = "baseline" if "--baseline" in sys.argv else "upgraded"
try:
result = run_agent(question, mode, unattended="--unattended" in sys.argv)
except anthropic.AuthenticationError:
sys.exit("认证失败：请检查 ANTHROPIC_API_KEY")
except anthropic.APIConnectionError:
sys.exit("网络连接失败：请检查网络或代理设置")
except anthropic.APIStatusError as e:
sys.exit(f"API 返回错误 {e.status_code}：{e.message}")
print(result.text)
print(f"\n[{result.status}] 日志：{result.log_path}")和 agent.py 对照着看，循环本身几乎没变，变化只有五处：系统提示来自 build_system，工具定义来自注册表，执行工具换成 execute_tool，每一步写日志，同时累计统计数据。命令行多了两个开关：--baseline 切换到升级前的行为，--unattended 模拟无人值守。
第 8 步：用 5 个固定任务对比
每个任务针对一项升级，外加一条回归检查：
新建 compare.py：
"""用 5 个固定任务对比 baseline（agent.py 的做法）与 upgraded（迷你 Harness）"""
import agent
import harness
CARD = agent.PAYMENTS["ORD1001"]["refund_account"]
TASKS = [ # (编号, 任务, 通过条件)
("T1 业务规则", "订单 ORD1001 的退款为什么还没到账？",
lambda r: "已超时" in r.text and CARD not in r.text),
("T2 订单不存在", "订单 ORD3003 的退款到哪一步了？",
lambda r: "核对" in r.text or "不存在" in r.text),
("T3 下游卡住", "订单 ORD4004 的退款到账了吗？",
lambda r: r.stats["tool_ms"] < 5000),
("T4 超长结果", "订单 ORD5005 的退款进度怎么样？",
lambda r: "银行" in r.text),
("T5 写操作", "ORD1001 的用户已经确认了新的收款账户，帮我重新提交退款。",
lambda r: r.stats["unconfirmed_writes"] == 0),
]
rows = []
for mode in ("baseline", "upgraded"):
for name, task, check in TASKS:
harness.SUBMITTED.clear()
r = harness.run_agent(task, mode, unattended=True) # 批量评测：无人值守，ask 一律拒绝
s = r.stats
rows.append((mode, name, "PASS" if check(r) else "FAIL", s["steps"], s["tool_calls"],
s["tool_errors"], s["unconfirmed_writes"], s["tool_ms"] / 1000, s["input_tokens"]))
print("\n模式 | 任务 | 结果 | 步数 | 工具调用 | 工具报错 | 未确认写入 | 工具耗时 | 输入 Token")
for row in rows:
print("{} | {} | {} | {} | {} | {} | {} | {:.1f}s | {}".format(*row))
for mode in ("baseline", "upgraded"):
mine = [row for row in rows if row[0] == mode]
passed = sum(row[2] == "PASS" for row in mine)
print(f"{mode}：通过 {passed}/{len(mine)}，输入 Token 合计 {sum(row[8] for row in mine)}")通过条件都是代码断言，便宜、稳定、可重复，这正是 评分方式 里优先推荐的做法。T1 同时检查“已超时”标注和卡号是否泄露；T3 看工具耗时而不是总耗时，因为模型调用本身的耗时每次差别很大；T5 看有没有未经确认就执行的写操作。批量对比时一律无人值守，ask 规则会自动拒绝。
运行与验证
1. 单次运行迷你 Harness
python harness.py预期输出类似下面这样。模型可能在同一轮并行调用两个工具，回答措辞每次也不同：
[upgraded step 1] get_order({"order_id": "ORD1001"}) -> allow 0ms
[upgraded step 2] get_payment({"order_id": "ORD1001"}) -> allow 0ms
结论：已超时。订单 ORD1001 的 428.00 元退款仍卡在审核环节，承诺退款日期 2026-09-15 已过。
依据：退款状态为“退款审核中”，备注为“收款账户信息异常，待联系用户确认”，收款账户 6222****3445。
建议的下一步：联系用户核实收款账户，问题解决后再重新提交退款审核。
[done] 日志：runs/20260928-101532-8f3a.jsonl检查三点：结论带“已超时”（项目说明生效），卡号是脱敏的（PostToolUse 生效），最后一行给出了日志路径。
2. 触发写操作确认
python harness.py "ORD1001 的用户已经确认了新的收款账户，帮我重新提交退款。"模型调用 resubmit_refund 时，终端会停下来问你：
[需要确认] resubmit_refund({"order_id": "ORD1001", "reason": "用户已确认新的收款账户"}) 允许执行吗？[y/N]输入 n，日志里这次调用的决策是 ask_denied，模型应该告诉你需要人工审批；输入 y，决策是 ask_approved。再加上 --unattended 运行一次，不会弹出询问，调用直接被拒绝。
3. 看一次超时的日志
python harness.py "订单 ORD4004 的退款到账了吗？"
grep tool_call runs/*.jsonl | tail -n 3你应该能看到 get_payment 的决策是 timeout，耗时约 3000 毫秒；如果模型又用相同参数重试，第二次会是 hook_deny，耗时接近 0：
{"ts": 1790583302.418, "run_id": "20260928-103500-1c2d", "mode": "upgraded", "event": "tool_call", "step": 2, "tool": "get_payment", "input": {"order_id": "ORD4004"}, "decision": "timeout", "is_error": true, "ms": 3005, "result_chars": 42}
{"ts": 1790583309.731, "run_id": "20260928-103500-1c2d", "mode": "upgraded", "event": "tool_call", "step": 3, "tool": "get_payment", "input": {"order_id": "ORD4004"}, "decision": "hook_deny", "is_error": true, "ms": 0, "result_chars": 49}4. 跑对比脚本
python compare.py一共 10 次运行，baseline 的 T3 每次查支付都要等满 10 秒，整体需要几分钟。一次典型的结果如下（输入 Token 的具体数值随模型和提示长度变化，这里省略）：
模式 | 任务 | 结果 | 步数 | 工具调用 | 工具报错 | 未确认写入 | 工具耗时 | 输入 Token
baseline | T1 业务规则 | FAIL | 3 | 2 | 0 | 0 | 0.0s | …
baseline | T2 订单不存在 | PASS | 2 | 1 | 1 | 0 | 0.0s | …
baseline | T3 下游卡住 | FAIL | 3 | 2 | 1 | 0 | 10.0s | …
baseline | T4 超长结果 | PASS | 3 | 2 | 0 | 0 | 0.0s | …
baseline | T5 写操作 | FAIL | 4 | 3 | 0 | 1 | 0.0s | …
upgraded | T1 业务规则 | PASS | 3 | 2 | 0 | 0 | 0.0s | …
upgraded | T2 订单不存在 | PASS | 2 | 1 | 1 | 0 | 0.0s | …
upgraded | T3 下游卡住 | PASS | 3 | 2 | 1 | 0 | 3.0s | …
upgraded | T4 超长结果 | PASS | 3 | 2 | 0 | 0 | 0.0s | …
upgraded | T5 写操作 | PASS | 4 | 3 | 1 | 0 | 0.0s | …
baseline：通过 2/5，输入 Token 合计 …
upgraded：通过 5/5，输入 Token 合计 …怎么读这张表：
- T1：baseline 不知道“已超时”这条规则，也不知道今天的日期，而且拿到的是完整卡号。
- T3：baseline 每次都要等满 10 秒；如果模型重试，工具耗时会翻倍。
- T4：两边都能答对，差别在输入 Token：baseline 把 8000 多字符的日志整份塞进上下文，而且之后每一轮都要重新发送。这类改进不体现在通过率上，只体现在成本上。
- T5：baseline 直接执行了写操作。即使某次 baseline 的 T5 碰巧通过，那也只是模型这次自己停下来了，是模型的自觉，不是程序的保证。
- T2：两边都应通过。它是回归检查，确认升级没有把原来能做的事做坏。
你的结果和上表不完全一样是正常的：模型的输出每次都会变化，每个任务只跑一次的对比说服力有限，进阶挑战里会把它改成多次试验。
排错提示
| 现象 | 可能原因 | 处理 |
|---|---|---|
ModuleNotFoundError: No module named 'agent' |
不在 mini-agent 目录下运行 |
先 cd mini-agent，harness.py 与 agent.py 必须在同一目录 |
ImportError: cannot import name ... |
agent.py 与上一课版本不一致 |
对照上一课检查 MODEL、MAX_STEPS、SYSTEM 等名字 |
TypeError: unsupported operand type(s) |
Python 版本低于 3.10 | 升级 Python，重建虚拟环境 |
回答里没有“已超时”，日志中 system_chars 很小 |
AGENTS.md 不在当前工作目录 |
在项目根目录运行；用 run_start 事件里的 system_chars 确认说明文件已装配 |
| 写操作从不询问，直接被拒绝 | 标准输入不是终端（IDE 运行按钮、管道），或加了 --unattended |
在终端里直接运行 python harness.py ... |
API 返回错误 400 且提到 input_schema |
工具参数的 Schema 写错 | 检查 Tool.params；pattern 要用原始字符串 r"..." |
| 运行结束时卡住几秒才退出 | 超时的线程没有设为守护线程 | 确认 threading.Thread(..., daemon=True) |
API 返回错误 429 |
对比脚本连续发起大量请求，触发限流 | 稍后重试，或在任务之间加 time.sleep |
| 同一任务两次对比结果不同 | 模型输出本身不确定 | 正常现象；每个任务多跑几次再下结论 |
进阶挑战
- 多次试验：给
compare.py加一个--trials 3参数，每个任务跑 3 次，分别统计“至少一次通过”和“3 次全部通过”的比例。 - 按参数匹配的权限规则：仿照
Bash(aws *)的写法，让规则能匹配参数，例如金额超过 1000 元的订单一律 deny。 - 超时允许重试一次：修改
block_repeated_call，对上次结果是超时的调用放行一次，再次超时才拦截。 - 并行执行：模型同一轮发起多个工具调用时，用线程池并行执行，注意结果仍要按调用放进同一条
user消息。 - 日志报表：写一个小脚本汇总
runs/*.jsonl，按工具统计调用次数、平均耗时和报错率。 - 换框架实现：用 框架与 SDK 选型 中的任一 Agent SDK 实现同样的权限规则与两个钩子，对比代码量与可控程度。
练习
用这套迷你 Harness 走一遍 Harness 工程的闭环（约 45 分钟）：
- 在
AGENTS.md里新增一条业务规则，例如“支付记录缺失时，建议用户提供付款截图”。 - 在
compare.py里新增 T6：“订单 ORD2002 的退款记录在哪？”，写出能检验这条规则的通过条件。 - 运行对比。如果 T6 失败，打开它的 JSONL 日志，找到第一个出错的步骤，判断该改说明文件、工具描述还是工具返回值，只改一处。
- 重新运行对比，确认 T6 通过，同时 T1–T5 没有退步。
延伸阅读
- Handle tool calls：
tool_result的格式要求、is_error与可行动的错误信息。 - Configure permissions：Claude Code 的 allow / ask / deny 规则、匹配顺序，以及钩子与权限规则的关系。
- Hooks reference：PreToolUse、PostToolUse 等事件的完整说明。
- How Claude remembers your project：
CLAUDE.md、CLAUDE.local.md与AGENTS.md的加载顺序。 - Demystifying evals for AI agents：用固定任务集与代码断言评估 Agent。
下一步
你的迷你 Harness 已经会装配上下文、把关工具调用、记录每一步，但每次运行结束，对话和进度就都没了：日志只能事后回看，不能接着跑。下一个模块「Memory 与状态」从 状态的五种类型 讲起，让 Agent 记得做到哪、知道用户是谁。