没人给 Agent 编过预算：跨部门的 AI 用量到底怎么记帐

10 月 30 日，Cognition 发布了面向 Devin 的企业治理方案，与银行、零售、电信的几家大客户共同打磨。把公告读细一点会发现它的范围很窄也很具体：解决的是大规模使用 AI Agent 时的资金归属与监督问题，对象是那种「几千名工程师散在不同业务单元里、消费的是按用量计价的 Agent」的组织。各家报道反复用的词是「谁付钱」问题，这个词不错，因为它点到了真正的摩擦点。用 Agent 的人是一个拿着工单的开发者；对那个数字负责的人，是隔着两个层级、在另一个成本中心里的经理。中间那层工具，过去谁也不管这件事。

用 Agent 的人和付钱的人，几乎从来不是同一个。

一、公告说了什么，以及它没说什么

已披露的事实是这些：产品面向的是「Agent 在多个部门、数千个工程座位上被消耗」的大型组织；与银行、零售、电信客户共同开发，这几个行业监管要求重、内部结构复杂；对外描述是「让 AI 投入的资金归属透明、能追踪谁在付钱、并对天然分散的用量做预算控制」。公告里没有的，是任何机制层面的东西：没有计量单位、没有策略模型、没有 API、也没有一句话说明预算用完了会怎样。我不打算替它猜。我能做的是描述这类产品必然具备的设计，因为不管你买还是自己造，底层问题是一样的。

# Attribution has to exist at the call site, or it is a guess. Wrap the agent's
# entry point once, and let the run id travel through every model and tool call.
import contextvars, json, time, uuid
from dataclasses import dataclass, asdict, field

RUN = contextvars.ContextVar("agent_run")


@dataclass
class RunContext:
    run_id: str
    cost_centre: str          # the thing finance reports on
    project: str
    ticket: str
    model: str = ""
    input_tokens: int = 0
    cached_tokens: int = 0
    output_tokens: int = 0
    tool_calls: int = 0
    started: float = field(default_factory=time.time)
    outcome: str = "open"

    def event(self, kind: str, **fields):
        # One line per billable step. Append to your warehouse, not the vendor's.
        print(json.dumps({"run": asdict(self), "kind": kind, **fields}))


def start_run(cost_centre: str, project: str, ticket: str) -> RunContext:
    ctx = RunContext(str(uuid.uuid4()), cost_centre, project, ticket)
    RUN.set(ctx)
    return ctx


def current() -> RunContext:
    ctx = RUN.get(None)
    if ctx is None:
        # Untagged runs are the leading cause of unexplained spend. Fail loudly
        # in staging and tag them as "unattributed" in production.
        ctx = RunContext(str(uuid.uuid4()), "unattributed", "unknown", "unknown")
        RUN.set(ctx)
    return ctx

# The test that matters: start a run, let the agent call a tool, and assert that
# the tool's record carries the same run_id. If it does not, that work is
# billable and invisible, which is the worst combination.

二、按量计费把「按座位」的预算打碎了

建立在座位上的软件预算有一个财务部门默默依赖的属性：它事先就是已知的。买一百个 licence，数字就定了，偏差是零。Agent 的定价不是这样。Cognition 用 ACU 计量 Devin，也就是归一化的 Agent 算力单位；其他厂商按 token、按分钟、按任务。不管这个单位叫什么，花的钱随「委派出去多少工作」而增长，而这件事是由离问题最近的那个人、一张工单一张工单地决定的。乘上几千名工程师，你得到的就是一个没人预测过的数字，用一种没人做过预算的计价方式，并且归不到任何人头上。这不是采购的失职，这是把一件变动成本的工具，塞进一套为固定成本设计的流程里必然发生的事。

# The ledger. One row per run, event-sourced from the per-step events, closed
# daily. Small enough to read in a review, which is the point.
import json, sqlite3
from pathlib import Path

db = sqlite3.connect("agent_usage.db")
db.executescript(
    "CREATE TABLE IF NOT EXISTS runs ("
    "  run_id TEXT PRIMARY KEY,"
    "  cost_centre TEXT NOT NULL,"
    "  project TEXT NOT NULL,"
    "  ticket TEXT NOT NULL,"
    "  model TEXT,"
    "  day TEXT NOT NULL,"
    "  input_tokens INTEGER DEFAULT 0,"
    "  cached_tokens INTEGER DEFAULT 0,"
    "  output_tokens INTEGER DEFAULT 0,"
    "  tool_calls INTEGER DEFAULT 0,"
    "  outcome TEXT DEFAULT 'open'"
    ");"
    "CREATE INDEX IF NOT EXISTS runs_day ON runs(day, cost_centre);"
)

def ingest(path: str):
    with open(path, encoding="utf-8") as fh:
        for line in fh:
            ev = json.loads(line)
            r = ev["run"]
            db.execute(
              "INSERT INTO runs (run_id, cost_centre, project, ticket, model, day,"
              " input_tokens, cached_tokens, output_tokens, tool_calls, outcome)"
              " VALUES (?,?,?,?,?,?,?,?,?,?,?)"
              " ON CONFLICT(run_id) DO UPDATE SET"
              "   input_tokens = excluded.input_tokens,"
              "   cached_tokens = excluded.cached_tokens,"
              "   output_tokens = excluded.output_tokens,"
              "   tool_calls = excluded.tool_calls,"
              "   outcome = excluded.outcome",
              (r["run_id"], r["cost_centre"], r["project"], r["ticket"],
               r["model"], r.get("day", "1970-01-01"),
               r["input_tokens"], r["cached_tokens"], r["output_tokens"],
               r["tool_calls"], r["outcome"]))
    db.commit()

def monthly_statement(month: str, rate_in: float, rate_cached: float, rate_out: float):
    rows = db.execute("""
      SELECT cost_centre, SUM(input_tokens), SUM(cached_tokens),
             SUM(output_tokens), COUNT(*)
      FROM runs WHERE day LIKE ? GROUP BY cost_centre ORDER BY 2 DESC
    """, (month + "%",)).fetchall()
    for centre, tin, tcached, tout, runs in rows:
        billed = (tin - tcached) / 1e6 * rate_in + tcached / 1e6 * rate_cached \
                 + tout / 1e6 * rate_out
        print(f"{centre:<24} runs={runs:<6} ${billed:>10,.2f}")

# Every row here is a claim you may have to defend. Keep the raw events too, so
# the claim can be re-derived by somebody who does not trust your rollup.

三、归集是工程问题，而且从调用点就开始了

任何一套内部计费系统，最后都归结成一个问题：对这一份花费，你能不能说清它属于谁。对 Agent 来说，这意味着归集信息必须存在于发起调用的地方，而不是事后从那些根本没打算承载它的日志里反推出来。实际可用的最低标准，是一条运行记录：一个稳定的 id、一个成本中心、一个项目、一个工单号，并且贯穿这次运行所发的每一次模型调用和每一次工具调用——包括 Agent 自己决定额外发出的那些，而那恰恰是最让预算主人意外的部分。计量单位的选择也很重要。token 好测，难向经理解释；「等效人力小时」好解释，难诚实地测。选你能站得住的那个，但原始计数器无论如何都要存，这样以后还能把另一个推导出来。

# Reserve before the spend. The point is not to be strict, it is to make an
# over-budget state a product decision with a defined cheaper path.
import time
from dataclasses import dataclass


@dataclass
class Budget:
    cost_centre: str
    limit_usd: float
    spent_usd: float = 0.0
    reserved_usd: float = 0.0

    def available(self) -> float:
        return self.limit_usd - self.spent_usd - self.reserved_usd


BUDGETS = {"payments-platform": Budget("payments-platform", 4_000.0)}


def estimate(prompt_tokens: int, max_output_tokens: int, rate_in: float,
             rate_out: float, cached_ratio: float = 0.0) -> float:
    fresh = prompt_tokens * (1 - cached_ratio)
    return fresh / 1e6 * rate_in + max_output_tokens / 1e6 * rate_out


def admit(cost_centre: str, prompt_tokens: int, max_output_tokens: int):
    """Return the plan this run may use, or refuse with a defined fallback."""
    b = BUDGETS[cost_centre]
    want = estimate(prompt_tokens, max_output_tokens, 1.0, 4.0, cached_ratio=0.6)
    if want <= b.available():
        b.reserved_usd += want
        return {"tier": "standard", "reservation": want}

    # Degrade before denying: a cheaper model, a shorter completion, a later slot.
    cheap = estimate(prompt_tokens, 2_000, 0.15, 0.6, cached_ratio=0.6)
    if cheap <= b.available():
        b.reserved_usd += cheap
        return {"tier": "degraded", "reservation": cheap,
                "note": "budget low: cheaper model, 2k completion cap"}

    return {"tier": "queued", "reservation": 0.0,
            "note": "over budget: queued until period close or an approved override"}

# Two properties worth testing: concurrent admissions must not each receive the
# full remaining amount, and the refusal path must be a documented behaviour
# rather than whatever the exception handler happened to do.

四、只会出报表的预算是账单，会拦人的预算才是控制

第一代 Agent 成本工具的通病，是它只是个看板。它会在月初三号告诉你：某个团队已经把整个季度的额度花掉了。到那个时候，选择早就做完了。真正有效的是在花钱之前、在请求路径上就执行：按这次运行的预估用量做预留，预留拿不到就拒绝或降级，而降级要做成产品决策而不是一个错误页——换便宜模型、缩小上下文、排队。预留还能顺手修掉一个很隐蔽的记帐问题：并发运行共享同一份剩余预算时，不能每一个都按全额放行。另外一定要留一条明确的例外通道，因为「发版周谁也没法override」这种治理工具，第二周就会被人关掉。

# The number a manager will ask for: cost per unit of work that actually
# landed. Compute it from the ledger joined to your delivery data.
import csv, sqlite3

db = sqlite3.connect("agent_usage.db")
db.execute("CREATE TEMP TABLE merged AS SELECT 1")  # placeholder for clarity

def outcomes(path: str) -> dict:
    # ticket -> (merged_prs, swallowed) from your VCS export
    out = {}
    with open(path, newline="", encoding="utf-8") as fh:
        for row in csv.DictReader(fh):
            out[row["ticket"]] = (int(row.get("merged_prs") or 0),
                                 row.get("swallowed", "no") == "yes")
    return out

def report(vcs_csv: str, rate_in=1.0, rate_out=4.0, rate_cached=0.1):
    deliv = outcomes(vcs_csv)
    rows = db.execute("""
      SELECT ticket, SUM(input_tokens), SUM(cached_tokens), SUM(output_tokens),
             COUNT(*)
      FROM runs GROUP BY ticket
    """).fetchall()

    for ticket, tin, tcached, tout, runs in rows:
        cost = ((tin - tcached) / 1e6 * rate_in
                + tcached / 1e6 * rate_cached
                + tout / 1e6 * rate_out)
        prs, swallowed = deliv.get(ticket, (0, False))
        if prs:
            print(f"{ticket:<18} ${cost:>8,.2f}  prs={prs}  ${cost/prs:>8,.2f}/pr")
        elif swallowed:
            # Spend with no artefact is not waste — it is the honest cost of
            # exploration, and the only way to keep that line visible is to log it.
            print(f"{ticket:<18} ${cost:>8,.2f}  abandoned (keep, in a separate line)")

# Two rules that keep this honest: never divide by zero-merges, and never
# average abandoned work into the cost per PR. Both flatter the number and
# both get caught the first time somebody reads the raw rows.

五、对账，是治理赢得信任或失去信任的地方

你自己的账本和厂商的发票一定对不上，问题只在于你是从自己的月结里发现的，还是从财务来问你「这个数字为什么动了」的时候发现的。把对账做成月度任务：按成本中心汇总你的用量事件，与发票条目做关联，按团队报出差异，并且阈值要设得有意义——在团队层面，三个百分点不是噪声，通常意味着计量有缺口。两个最常见的原因是：新的入口绕过了打标层，导致有些运行从来没被归集；以及厂商计费了重试、而你的封装把重试折叠成了一个事件。这两种都不稀奇，但除非有人专门把两份数据 join 一次，它们都不会自己冒出来。

#!/usr/bin/env python3
# Reconcile your ledger against the vendor invoice, monthly. Variance at team
# level is the signal; a single total that roughly matches is not.
import csv, sqlite3, sys
from collections import defaultdict

THRESHOLD = 0.03  # 3% at team level is not noise

db = sqlite3.connect("agent_usage.db")
ours = dict(db.execute("""
  SELECT cost_centre, SUM(input_tokens - cached_tokens) / 1e6 * 1.0
       + SUM(cached_tokens) / 1e6 * 0.1 + SUM(output_tokens) / 1e6 * 4.0
  FROM runs WHERE day LIKE ? GROUP BY cost_centre
""", (sys.argv[1] + "%",)).fetchall())

theirs = defaultdict(float)
with open(sys.argv[2], newline="", encoding="utf-8") as fh:
    for row in csv.DictReader(fh):
        theirs[row["cost_centre"]] += float(row["amount_usd"])

bad = 0
for centre in sorted(set(ours) | set(theirs)):
    a, b = ours.get(centre, 0.0), theirs.get(centre, 0.0)
    delta = (b - a) / a if a else float("inf")
    flag = "  CHECK" if abs(delta) > THRESHOLD else ""
    print(f"{centre:<24} ours={a:>9,.2f} invoice={b:>9,.2f} delta={delta:>7.1%}{flag}")
    bad += bool(flag)

raise SystemExit(1 if bad else 0)

# When a line is flagged, look at two things first: runs tagged 'unattributed'
# (a new entry point skipped the wrapper) and retry counts on your side versus
# billable attempts on theirs. Those two explain most first-pass variances.

六、买产品还是自己写那三十行，但要在第一张发票之前决定

如果厂商把归集、拦截、报表打成一个产品卖你，而且能对接你的身份系统和成本中心层级，那买它很合理。不该外包的是那份 schema，因为换厂商的时候，你留下的是它：运行 id、成本中心、项目、工单、模型、输入 token、缓存 token、输出 token、工具调用次数、耗时、以及这次工作的结果。把它存进你自己的数仓，而不是只存在厂商控制台里。这样到续约的时候，那是一个工程决策而不是一个迁移项目；而月末那个问题——这笔钱谁付的——有一个不依赖任何人家的看板能否打开的答案。