最近把一个能跑通 Demo 的 Agent 往生产里推时，最先崩的不是模型，而是 harness。任务跑到第三十步，有时卡在一次没设超时的 HTTP 工具上，有时同一工具连续失败十几次还在重试，token 账单比结果先出来。这篇文章不谈概念，只给一套我正在用的最小实践：给每次工具调用打 span、用时间线找卡点、按工具配置超时，失败累积后熔断。

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

一、先把每次调用记成 span

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

复制

import timefrom dataclasses import dataclass, fieldfrom typing import Any, Callable@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(self, span: 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 秒还没返回就停”。超时是执行层的责任，按工具风险分级即可。

复制

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 秒，发布类可以更长，但必须有上限。没有上限的工具，等于给生产埋了一颗挂起冲击波。

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

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

@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(self, 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, tools: dict, timeouts: dict, breakers: dict = None):self.tools = toolsself.timeouts = timeoutsself.breakers = breakers or {k: Breaker() for k in tools}def run(self, name: str, trace: Trace, **kwargs) -> Any:span = Span(name=name, t0=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(name, 15.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，卡点可以规则化：耗时超过中位数若干倍，或失败率超过阈值。

复制

def find_bottlenecks(trace: Trace, slow_ratio: float = 3.0, fail_rate: float = 0.5):rows = []for name, s in trace.summary().items():a vg = s["ms"] / max(s["n"], 1)rows.append((name, a vg, s["fail"] / max(s["n"], 1), s["n"], s["ms"]))if not rows:return []median = sorted(r[1] for r in rows)[len(rows) // 2]hits = []for name, a vg, fr, n, total in rows:reasons = []if a vg >= max(median * slow_ratio, 50):reasons.append(f"均耗时 {a vg:.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(hits, key=lambda x: -x["total_ms"])

这三个信号覆盖了我线上最常见的三类问题：慢工具、坏工具、被模型疯狂重试的工具。

五、串起来跑一遍

复制

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, "http_get": http_get, "run_shell": run_shell},TOOL_TIMEOUT,)trace = Trace()for i in range(4):try:hub.run("http_get", trace, url="https://bad.example")except Exception as e:print("http_get", e)hub.run("search", trace, q="harness")hub.run("run_shell", trace, cmd="echo 1")hub.run("search", trace, q="breaker")print("summary", trace.summary())print("bottlenecks", find_bottlenecks(trace, slow_ratio=1.5, fail_rate=0.5))

运行输出大致如下：

复制

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 已被熔断拦截，时间线里能直接看到它是失败率与总耗时的双料卡点；search 和 run_shell 正常，不会被坏工具拖死。

六、落地时的几条经验

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