Harness是什么-工程外壳和流程控制

Harness 在 AI Agent 工程里不是模型本身，而是把提示、工具调用、状态记录、权限和执行流程串起来的一层工程外壳。

Python 3.10 标准库可跑，文末有串联示例和输出。

一、先把每次调用记成 span

没有轨迹，优化全靠猜。最小字段只要五样：工具名、开始时间、耗时、成功与否、错误摘要。

代码语言：python

按路线处理时，fieldfrom typing import Any复制import timefrom dataclasses import dataclassCallable@dataclassclass Span:name: strt0: floatt1: float = 0.0ok: bool = Falseerror: str = ""meta: dict = field(default_factory=dict)@propertydef ms(self) -> float:return max(0.0(self.t1 - self.t0) * 1000)@dataclassclass Trace:spans: list = field(default_factory=list)def add(selfspan: Span) -> None:self.spans.append(span)def summary(self) -> dict:by = {}for s in self.spans:row = by.setdefault(s.name{"n": 0"fail": 0"ms": 0.0})row["n"] = 1row["fail"] = 0 if s.ok else 1row["ms"] = s.msreturn by一次任务结束后先看 summary()：哪个工具调用次数异常、失败率高、总耗时占比最大卡点通常就在这三项的交集里。

二、超时必须写在 harness，不能指望模型

模型不会乖乖在 prompt 里遵守“30 秒还没返回就停”。超时是执行层的责任，按工具风险分级就能。

代码语言：python

没有上限的工具，等于给生产埋了一颗挂起炸弹。复制import concurrent.futuresclass TimeoutError_(TimeoutError):passdef call_with_timeout(fn: Callable, timeout_s: float, kwargs):with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:fut = pool.submit(fn, kwargs)try:return fut.result(timeout=timeout_s)except concurrent.futures.TimeoutError as e:raise TimeoutError_(f"超时 {timeout_s}s") from eTOOL_TIMEOUT = {"search": 8.0,"http_get": 10.0,"run_shell": 30.0,"deploy": 120.0,}经验值：只读检索类 8–10 秒，本地命令 30 秒，发布类可以更长，但必须有上限。

三、失败累积就熔断，别让重试风暴烧预算

同一工具连续失败时，继续重试往往只是在重复付费。给每个工具维护窗口内的失败计数，达到阈值就短开。

代码语言：python

ok: bool) -> None:if ok:self.fails = 0returnself.fails = 1if self.fails >= self.fail_max:self.opened_at = time.time()class ToolHub:def __init__(self复制@dataclassclass Breaker:fail_max: int = 3cool_s: float = 60.0fails: int = 0opened_at: float = 0.0def allow(self) -> bool:if self.fails < self.fail_max:return Trueif time.time() - self.opened_at >= self.cool_s:self.fails = 0return Truereturn Falsedef on_result(selftools: dicttimeouts: dictbreakers: dict = None):self.tools = toolsself.timeouts = timeoutsself.breakers = breakers or {k: Breaker() for k in tools}def run(selfname: strtrace: Tracekwargs) -> Any:span = Span(name=namet0=time.time())br = self.breakers[name]if not br.allow():span.t1 = time.time()span.error = "熔断中"trace.add(span)raise RuntimeError(f"{name} 熔断中冷却后重试")try:out = call_with_timeout(self.tools[name]self.timeouts.get(name15.0)kwargs)span.ok = Truebr.on_result(True)return outexcept Exception as e:span.error = str(e)[:120]br.on_result(False)raisefinally:span.t1 = time.time()trace.add(span)熔断后不要静默吞掉要把 熔断中 写回轨迹让编排层改走降级路径比如，换只读缓存或请求人工接管。

四、用时间线自动标出卡点要先看什么

有了 span，卡点可以规则化：耗时超过中位数若干倍，或失败率超过阈值。

代码语言：python

s["ms"]))if not rows:return []median = sorted(r[1] for r in rows)[len(rows) // 2]hits = []for nameavgfrntotal in rows:reasons = []if avg >= max(median * slow_ratio50):reasons.append(f"均耗时 {avg:.0f}ms")if fr >= fail_rate and n >= 2:reasons.append(f"失败率 {fr:.0%}")if n >= 8:reasons.append(f"调用过密 {n} 次")if reasons:hits.append({"tool": name"reasons": reasons"total_ms": total})return sorted(hitskey=lambda x: -x["total_ms"])这三个信号覆盖了我线上最常见的三类问题：慢工具、坏工具、被模型疯狂重试的工具。复制def find_bottlenecks(trace: Traceslow_ratio: float = 3.0fail_rate: float = 0.5):rows = []for names in trace.summary().items():avg = s["ms"] / max(s["n"]1)rows.append((nameavgs["fail"] / max(s["n"]1)s["n"]

五、串起来跑一遍要先看什么

代码语言：python

"http_get": http_get复制def search(q):time.sleep(0.05)return ["doc1"]def http_get(url):time.sleep(0.2)if "bad" in url:raise ConnectionError("连接失败")return {"ok": True}def run_shell(cmd):time.sleep(0.01)return "done"hub = ToolHub({"search": search"run_shell": run_shell}TOOL_TIMEOUT)trace = Trace()for i in range(4):try:hub.run("http_get"traceurl="https://bad.example")except Exception as e:print("http_get"e)hub.run("search"traceq="harness")hub.run("run_shell"tracecmd="echo 1")hub.run("search"traceq="breaker")print("summary"trace.summary())print("bottlenecks"find_bottlenecks(traceslow_ratio=1.5fail_rate=0.5))运行输出大致如下：

代码语言：bash

search 和 run_shell 正常，不会被坏工具拖死。复制http_get 连接失败http_get 连接失败http_get 连接失败http_get http_get 熔断中，冷却后重试summary {'http_get': {'n': 4, 'fail': 4, 'ms': ...}, 'search': {...}, 'run_shell': {...}}bottlenecks [{'tool': 'http_get', 'reasons': ['均耗时 ...', '失败率 100%'], ...}]第四次 http_get 已被熔断拦截，时间线里能直接看到它是失败率与总耗时的双料卡点；

六、落地时的几条经验容易忽略的点

span 必记，哪怕先落本地 JSONL，没有轨迹就不要谈优化。超时按工具分级写死在 harness，不要写在提示词里。熔断阈值建议从 3 次起，冷却 30–60 秒，按业务再调。每天看一眼 find_bottlenecks 的结果，比看模型分数更能发现回归。和能力清单、检查点可以叠用：熔断后走降级动作，检查点负责把任务接到安全点。

Harness 的价值不在“又包了一层 Agent 框架”，而在于把超时、失败和成本变成可观测、可熔断、可恢复的工程量。轨迹时间线是第一步，也是最便宜的一步。