Harness 实践：用轨迹时间线找卡点，给工具调用加超时与熔断

原创

用户9746675

发布于 2026-08-07 10:51:30

690

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

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

一、先把每次调用记成 span

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

import time
from dataclasses import dataclass, field
from typing import Any, Callable


@dataclass
class Span:
    name: str
    t0: float
    t1: float = 0.0
    ok: bool = False
    error: str = ""
    meta: dict = field(default_factory=dict)

    @property
    def ms(self) -> float:
        return max(0.0, (self.t1 - self.t0) * 1000)


@dataclass
class 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"] += 1
            row["fail"] += 0 if s.ok else 1
            row["ms"] += s.ms
        return by

一次任务结束后先看 summary()：哪个工具调用次数异常、失败率高、总耗时占比最大，卡点通常就在这三项的交集里。

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

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

import concurrent.futures


class TimeoutError_(TimeoutError):
    pass


def 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 e


TOOL_TIMEOUT = {
    "search": 8.0,
    "http_get": 10.0,
    "run_shell": 30.0,
    "deploy": 120.0,
}

经验值：只读检索类 8–10 秒，本地命令 30 秒，发布类可以更长，但必须有上限。没有上限的工具，等于给生产埋了一颗挂起炸弹。

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

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

@dataclass
class Breaker:
    fail_max: int = 3
    cool_s: float = 60.0
    fails: int = 0
    opened_at: float = 0.0

    def allow(self) -> bool:
        if self.fails < self.fail_max:
            return True
        if time.time() - self.opened_at >= self.cool_s:
            self.fails = 0
            return True
        return False

    def on_result(self, ok: bool) -> None:
        if ok:
            self.fails = 0
            return
        self.fails += 1
        if self.fails >= self.fail_max:
            self.opened_at = time.time()


class ToolHub:
    def __init__(self, tools: dict, timeouts: dict, breakers: dict = None):
        self.tools = tools
        self.timeouts = timeouts
        self.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 = True
            br.on_result(True)
            return out
        except Exception as e:
            span.error = str(e)[:120]
            br.on_result(False)
            raise
        finally:
            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():
        avg = s["ms"] / max(s["n"], 1)
        rows.append((name, avg, 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, avg, fr, n, total in rows:
        reasons = []
        if avg >= max(median * slow_ratio, 50):
            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(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 正常，不会被坏工具拖死。

六、落地时的几条经验

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

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

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

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

AIGC