基于执行跨度建立明细账,分别排序令牌消耗和运行耗时,避免把最慢步骤误认成成本最高步骤。方案还覆盖重放去重、子代理支出归属及可测试的预算熔断。
一个 Agent 跑了一整夜。早上起来,免费额度已经耗尽,日志里只写着循环已结束。没人知道预算究竟花在了哪一步。
trace 里其实已经有答案,只是它的排序方式不符合我们的需求。大多数 trace 查看器按时间排列 span。成本是另一个维度,而且它给出的排序往往不同。
这篇文章会构建一个简单的 span 台账,回答每次运行中的两个问题:token 花到哪里去了,实际耗时又花到哪里去了?然后再加上一个可以测试、行为确定的预算熔断器。
trace UI 按 started_at 排序。这个顺序没有错,但对分析成本毫无帮助。
一次耗时 180 秒的测试运行,看起来像是问题所在,但它可能一个 token 都没消耗。
一次仅耗时 4 秒的模型调用,却可能用掉整次运行一半的 token。
一次重试的工具调用,会让某个看起来免费的条目的调用量翻倍。
你需要两份排名,而不是一份排好序的列表。两份排名都应该基于同一组 span 生成。
六个字段就能支撑整个台账。其他字段都是可选的。
如果你的 tracer 无法输出每个 span 的 token 数,到这里就可以停下了。只有每次运行的汇总用量,不足以做任何归因。
把测试数据和脚本保存在同一个目录中。这份测试数据包含六个 span、一次重试和一个耗时较长的工具调用。
{"span_id":"s1","parent_id":null,"seq":1,"name":"planner","kind":"model","started_at":"2026-10-18T01:00:00Z","ended_at":"2026-10-18T01:00:06Z","prompt_tokens":900,"completion_tokens":300,"status":"ok","retry_of":null}
{"span_id":"s2","parent_id":"s1","seq":2,"name":"repo.search","kind":"tool","started_at":"2026-10-18T01:00:06Z","ended_at":"2026-10-18T01:00:07Z","prompt_tokens":0,"completion_tokens":0,"status":"ok","retry_of":null}
{"span_id":"s3","parent_id":"s2","seq":3,"name":"reader","kind":"model","started_at":"2026-10-18T01:00:07Z","ended_at":"2026-10-18T01:00:15Z","prompt_tokens":3200,"completion_tokens":800,"status":"ok"}
{"span_id":"s4","parent_id":"s2","seq":4,"name":"repo.search","kind":"tool","started_at":"2026-10-18T01:00:15Z","ended_at":"2026-10-18T01:00:16Z","prompt_tokens":0,"completion_tokens":0,"status":"ok","retry_of":"s2"}
{"span_id":"s5","parent_id":null,"seq":5,"name":"test.run","kind":"tool","started_at":"2026-10-18T01:00:16Z","ended_at":"2026-10-18T01:03:16Z","prompt_tokens":0,"completion_tokens":0,"status":"ok"}
{"span_id":"s6","parent_id":"s3","seq":6,"name":"summarize","kind":"model","started_at":"2026-10-18T01:03:16Z","ended_at":"2026-10-18T01:03:21Z","prompt_tokens":2600,"completion_tokens":400,"status":"ok"}
# ledger.py
import json
from collections import defaultdict
from datetime import datetime
from pathlib import Path
def ts(value: str) -> datetime:
return datetime.fromisoformat(value.replace("Z", "+00:00"))
def read_spans(path) -> list:
spans = []
for lineno, raw in enumerate(Path(path).read_text().splitlines(), 1):
raw = raw.strip()
if not raw:
continue
try:
spans.append(json.loads(raw))
except json.JSONDecodeError as exc:
raise ValueError(f"{path}:{lineno}: {exc}") from exc
return sorted(spans, key=lambda s: s["seq"])
def span_row(span) -> dict:
ms = int((ts(span["ended_at"]) - ts(span["started_at"])).total_seconds() * 1000)
tokens = span.get("prompt_tokens", 0) + span.get("completion_tokens", 0)
return {
"name": span["name"],
"kind": span["kind"],
"ms": ms,
"tokens": tokens,
"failed": span.get("status", "ok") != "ok",
"retry": bool(span.get("retry_of")),
}
def ledger(path):
rows = [span_row(s) for s in read_spans(path)]
total_ms = sum(r["ms"] for r in rows) or 1
total_tokens = sum(r["tokens"] for r in rows) or 1
groups = defaultdict(
lambda: {"calls": 0, "ms": 0, "tokens": 0, "retries": 0, "failed": 0}
)
for r in rows:
g = groups[(r["kind"], r["name"])]
g["calls"] += 1
g["ms"] += r["ms"]
g["tokens"] += r["tokens"]
g["retries"] += int(r["retry"])
g["failed"] += int(r["failed"])
out = []
for (kind, name), g in groups.items():
out.append({**g, "kind": kind, "name": name,
"token_share": round(100 * g["tokens"] / total_tokens, 1),
"time_share": round(100 * g["ms"] / total_ms, 1)})
return sorted(out, key=lambda r: r["tokens"], reverse=True), total_ms, total_tokens
if __name__ == "__main__":
rows, total_ms, total_tokens = ledger("trace.jsonl")
print(f"run total: {total_tokens} tokens, {total_ms} ms")
for r in rows:
print(
f"{r['kind']:<5} {r['name']:<12} calls={r['calls']} "
f"tokens={r['tokens']} ({r['token_share']}%) "
f"ms={r['ms']} ({r['time_share']}%) retries={r['retries']}"
)
上述测试数据的预期输出如下(数值由这六个 span 得出;运行脚本即可确认):
run total: 8200 tokens, 201000 ms
model reader calls=1 tokens=4000 (48.8%) ms=8000 (4.0%) retries=0
tool repo.search calls=2 tokens=0 (0.0%) ms=2000 (1.0%) retries=1
model summarize calls=1 tokens=3000 (36.6%) ms=5000 (2.5%) retries=0
model planner calls=1 tokens=1200 (14.6%) ms=6000 (3.0%) retries=0
tool test.run calls=1 tokens=0 (0.0%) ms=180000 (89.6%) retries=0
两份排名指向不同的 span。reader 消耗了 48.8% 的 token。test.run 占用了 89.6% 的实际耗时,却没有消耗任何 token。削减这个慢工具的耗时,并不能节省 token 成本。
台账用于事后分析。接下来,在派发下一个任务之前,加上一条面向后续执行的规则。
规则:如果 committed + estimate > cap,就不要派发这个 span。记录决策,然后停止运行。
# breaker.py
import json
from pathlib import Path
from ledger import read_spans
def next_span_allowed(committed: int, cap: int, est: int):
projected = committed + est
if projected > cap:
return False, f"projected {projected} > cap {cap}"
return True, f"projected {projected} <= cap {cap}"
def walk(path, cap: int, out_path):
committed, last_model = 0, 0
lines = []
for span in read_spans(path):
tokens = span.get("prompt_tokens", 0) + span.get("completion_tokens", 0)
est = span.get("est_tokens", last_model if span["kind"] == "model" else 0)
ok, reason = next_span_allowed(committed, cap, est)
lines.append({"seq": span["seq"], "name": span["name"],
"action": "dispatch" if ok else "abort",
"committed": committed, "est": est, "reason": reason})
if not ok:
break
committed += tokens
if span["kind"] == "model":
last_model = tokens
Path(out_path).write_text("\n".join(json.dumps(l) for l in lines) + "\n")
return lines
这个估算器是刻意做得简单的。它使用上一个模型 span 的 token 用量;如果调用方有更准确的估算,也可以通过 est_tokens 字段明确提供。简单但行为确定,胜过聪明却无法测试。
五个测试确保台账结果可靠。它们不到一秒就能跑完,也不需要联网。
# test_ledger.py
import json
import pytest
from ledger import ledger, read_spans
from breaker import walk
def test_totals():
_, total_ms, total_tokens = ledger("trace.jsonl")
assert (total_tokens, total_ms) == (8200, 201_000)
def test_rankings_differ():
rows, _, _ = ledger("trace.jsonl")
assert rows[0]["name"] == "reader"
slowest = max(rows, key=lambda r: r["ms"])
assert slowest["name"] == "test.run"
assert slowest["tokens"] == 0
def test_retry_is_counted_not_hidden():
rows, _, _ = ledger("trace.jsonl")
search = next(r for r in rows if r["name"] == "repo.search")
assert search["calls"] == 2 and search["retries"] == 1
def test_breaker_aborts_before_dispatch(tmp_path):
out = tmp_path / "budget.jsonl"
lines = walk("trace.jsonl", cap=6000, out_path=out)
assert lines[-1]["action"] == "abort"
assert lines[-1]["seq"] == 6
assert all(l["action"] == "dispatch" for l in lines[:-1])
assert json.loads(out.read_text().splitlines()[-1])["action"] == "abort"
def test_bad_line_names_the_line(tmp_path):
bad = tmp_path / "bad.jsonl"
bad.write_text('{"seq": 1}\nnot json\n')
with pytest.raises(ValueError, match="bad.jsonl:2"):
read_spans(bad)
当 cap=6000 时,遍历会在 seq 为 6 的位置中止,原因是 projected 9200 > cap 6000。这次运行已消耗的 token 保持在 5200,并跳过原本会消耗 3000 个 token 的总结步骤。
采集 trace,记录 seq、kind,以及每个 span 的 token 用量。
运行台账。在提出假设之前,先看完两份排名。
明确 token 消耗最多的 span 和耗时最长的 span。如果两者不同,你面对的就是两个问题,而不是一个。
只改一处:缩短 prompt、缓存检索结果,或者去掉重试。
重新运行,对比两份台账。差异就是证据。
大多数团队在第 5 步还没做完时就停下了。一次运行变便宜了,却出现了另一种失败模式,这算不上胜利。也要比较 failed 列。
上面的闭环需要反复运行。每做一次削减,都意味着要用同一份测试数据再回放一次。这类工作恰好适合用免费模型访问和免费服务器方案来降低操作负担:你睡觉时回放照常运行,醒来就能看到台账差异。
根据 MonkeyCode 项目自己的介绍,它提供免费模型访问和免费服务器方案。在据此规划批量运行规模之前,请先到其网站确认当前限制;可用性和额度上限应由运营方说明,博客文章说了不算。
利益披露:本文是 MonkeyCode 产品推广活动的一部分。
还有一点需要注意:不要直接把这个熔断器接到使用生产预算的付费密钥上。先在沙箱中运行,并把决策日志保存在 trace 旁边。
如果你想先试试这个闭环,又不想一开始就购买 token,免费额度是风险较低的起点。
只提供汇总用量的服务商。如果你的 API 对整棵请求树只返回一个用量数字,那么所谓的逐 span 归因就是凭空编造。
只有一次调用的脚本。只有一次模型调用,就没有排名可供比较。可以跳过台账。
流式输出的未完成片段。一个始终没有结束的 span 不会有 ended_at。超时时要输出一条哨兵记录,否则表格就会失真。
嵌套的子 Agent。使用 parent_id 构建树结构。如果父 span 的 token 用量已经包含子 span,直接平铺求和就会重复计算。
耗时不足一秒的工具。毫秒取整带来的误差会占主导。应该跟踪调用次数,而不是耗时占比。
台账不是仪表盘。它只是用五行结果回答“该削减哪个 span?”通常这就够了。
一份 trace,两份排名。token 占比和耗时占比出现分歧的频率,比直觉以为的更高。
为每个 span 记录 seq、kind 和 token 用量。剩下的都是四十行代码就能写完的算术。
用测试数据验证熔断器,不要拿真实预算做实验。一条无法复现的规则,也就无法让人信任。
如果还想采取进一步行动,可以考虑屏蔽此人,和/或举报滥用行为。