如果你正在用大模型做 Agent 或自动化任务,有一个感受应该很熟悉:功能能不能跑通,效果好不好,测试的时候还能说清楚。但问到“刚才这一轮到底花了多少钱”,大多数人只能给一个模糊答案——“应该没多少吧”。
问题恰恰就出在这。Agent 应用的成本画像和普通 API 调用完全不同。一次看似简单的任务,内部可能已经拆成规划、工具调用、结果回填、失败重试等多个环节,每个环节都会发起一次模型请求,而且每一轮请求都会把前面的对话历史重新携带一遍。你看到的是“一个任务完成了”,账单上显示的却是十几笔 token 消耗的累加。
所以,AI 应用进入 Agent 阶段之后,成本监控不再是月底查看账单的“事后审计”,而应该变成开发过程中的实时仪表盘。把 token 用量、每轮费用、实时花费速率摆在眼前,不是炫技,而是控制预算的基础设施。
这篇文章从 Agent Cost Monitor 这类跨平台插件的核心能力出发,拆解三个关键问题:token 费用到底怎么算、每轮费用如何归集、实时花费速率图怎么做。然后我会带你手写一个最小可运行的监控链路,覆盖“记录调用 → 计算费用 → 本地入库 → 绘图展示”的完整过程,最后给出生产环境接入时的工程建议。
读完你可以得到三样东西:一套清晰的花费计算模型;一份可以直接复制的 Python 最小实现;一份避免“账单爆炸”的实操清单。
1. 为什么“看见 AI 花费”正在变成开发刚需
1.1 Agent 让成本从“可估算”变成“不可估算”
传统调用大模型的方式相对可控:一次 prompt,一次 completion,token 数量可以大致根据输入输出长度预估。开发者心里基本有数,这个功能跑一万次大概多少钱。
Agent 把这条路径彻底改变了。以常见的 ReAct 模式为例,模型需要先理解用户目标,然后决定调用哪些工具,观察工具返回结果,再决定下一步动作。中间一旦出现 JSON 解析失败、工具参数错误、外部服务超时,Agent 会自动重试。每一次思考、每一次工具调用、每一次重试,都是一次完整的模型请求。
更隐蔽的成本来自上下文累积。Agent 的每一轮新请求,都需要携带上一轮的全部消息。任务越往后,输入 token 越大。假设第一轮输入是 3000 token,到第 10 轮时输入可能已经膨胀到 20000 token。也就是说,多轮 Agent 对话的累计成本会随轮数加速增长,而不是线性增长。
这就是为什么“凭感觉估算成本”在 Agent 场景下完全失效。你无法再通过“我发了几条消息”来判断花费。
1.2 成本黑盒的三个直接后果
第一,预算失控。团队在开发阶段反复调试 Agent,一次调试可能触发几十次模型调用,但每个人的本地环境里都没有成本反馈。等到季度账单出来,才发现调试成本占了很大比例。
第二,成本归因困难。产品上线后,哪些功能最烧钱?是长文档分析、多工具串联,还是失败重试太多?如果没有按会话、按轮次记录费用,这些问题只能靠猜。
第三,优化失去方向。想降低模型成本,一般有两条路:换更便宜的模型、减少 token 消耗。但如果没有粒度足够细的成本数据,你根本不知道瓶颈是上下文太长、重试太多,还是输出 token 单价太高。没有度量,就没有优化。
1.3 什么样的人最需要这个监控插件
我认为最需要的是三类人:
- 个人开发者:按 token 付费使用大模型 API,希望在调试阶段就看见每一次调用的成本,而不是月底被账单吓一跳。
- 小团队负责人:团队共用多个模型账号或工具,需要判断哪个场景、哪个模型在拉高整体费用。
- 平台/基础设施工程师:正在搭建内部 LLM 网关或多 Agent 平台,需要把成本数据嵌入现有监控体系。
如果你属于其中任何一类,这篇文章的实操部分可以直接迁移到你的项目里。
2. Token 用量与费用计算的核心原理
2.1 token 与计费单位
模型计费的基本单位是 token,而不是字符数。token 是模型处理文本的最小粒度,由分词器决定。英文里一个 token 大约对应 0.75 个单词,中文场景下一个汉字可能对应 0.6 到 1 个 token,具体取决于模型和分词器。
这里有一个新手容易踩的坑:不要试图用“字符数”去估算 token 数。同一段中文,在不同模型的分词器下切分结果可能差一倍。真正可靠的 token 数量,来自 API 响应里返回的usage字段。
大多数 OpenAI 兼容接口的响应都会包含类似这样的结构:
{ "usage": { "prompt_tokens": 3500, "completion_tokens": 800 } }监控插件的第一件事,就是拿到这个usage字段,把它作为所有计算的输入。
2.2 三类计费 token:输入、输出与缓存命中
主流模型的定价策略,通常会把 token 分成输入和输出两类,输入单价低,输出单价高。原因是生成 token 的计算开销比读取已有 token 更大。
部分模型还会进一步区分“缓存命中”的输入 token。如果你的请求前缀和之前某次请求相同,命中了模型的 prompt caching,这部分 token 会以更低的价格计费,甚至免费。这个字段在不同 SDK 里的位置不一样,有的放在usage.prompt_tokens_details.cached_tokens,有的根本不返回。
所以我们计算单次调用的费用时,至少要考虑三个价格项:
| 计费维度 | 含义 | 一般价格关系 |
|---|---|---|
| 输入 token | 请求里 prompt 部分消耗的 token | 通常最低 |
| 输出 token | 模型生成的 completion 部分 | 通常最高 |
| 缓存命中 token | 命中前缀缓存、按更低价计费的输入 token | 通常远低于输入价 |
单次调用费用的通用公式可以写成:
费用 = (输入token - 缓存命中token) / 1000000 * 输入单价 + 缓存命中token / 1000000 * 缓存单价 + 输出token / 1000000 * 输出单价注意,各家的单价普遍以“每百万 token”为单位发布,所以计算时要除以一百万,否则金额会大得离谱。
2.3 每轮费用与上下文累积效应
“每轮费用”是 Agent 场景下比“单次调用费用”更有用的指标。一个 Agent 轮次,指的是从接收用户指令、执行工具调用、到返回结果的一个完整闭环。一个轮次里可能包含一次或多次模型调用。
为什么要按轮次归集?因为 Agent 的成本控制决策通常是按轮做出来的。如果某一轮费用异常高,可能是这个轮次触发了太多次工具调用,也可能是上下文已经太长。只有把模型调用费用累加到轮次维度,才能定位到具体的 Agent 行为。
这里还有一个容易被忽略的效应:上下文累积。假设第 1 轮输入 3000 token,第 2 轮要把第 1 轮的消息全部带上,输入变成 6000 token,第 3 轮变成 9000 token。如果不做任何压缩,第 N 轮的输入 token 大约是N * 平均单轮消息长度。这意味着,Agent 跑得越久,每一轮的实际费用越高。
这也是为什么很多生产级 Agent 会做历史摘要、上下文裁剪或者会话拆分。而这些策略是否有效,恰恰需要每轮费用数据来验证。
2.4 实时花费速率为什么是更敏感的指标
累计费用是一条只增不减的曲线,它适合事后看总量,但不适合实时预警。想象一个失控的 Agent:它陷入了死循环,不断发起工具调用,每分钟消耗大量 token。从累计曲线上看,只是斜率变陡了;如果图表刷新不够快,你甚至察觉不到异常。
花费速率(比如“最近 60 秒新增费用”)是更敏感的指标。一旦速率超过阈值,说明当前正在快速烧钱,需要立刻人工介入或触发自动熔断。
所以一个完整的 Agent Cost Monitor,至少要有两个可视化维度:
- 累计费用曲线:回答“到今天/到现在一共花了多少”。
- 实时花费速率:回答“此刻是不是在异常烧钱”。
3. Agent Cost Monitor 架构与跨平台设计
3.1 四层架构:采集、计算、存储、展示
不管插件以什么形态出现,它的内部架构都可以拆成四层:
- 采集层:拦截对模型 API 的调用,提取
usage字段和模型名、会话 ID、轮次 ID 等上下文信息。 - 计算层:根据价格表,把 token 用量换算成费用,并计算速率、轮次汇总等衍生指标。
- 存储层:把每次调用的明细和费用写入本地数据库,常用的选择是 SQLite。
- 展示层:提供累计费用曲线、实时速率图、按模型/按会话/按轮次的统计表。
四层各司其职,才能做到“采集不影响业务、计算可配置、存储可追溯、展示可实时刷新”。
3.2 采集层:如何在不侵入业务代码的前提下拿到 usage
采集层是整个插件的难点。实现方式通常有三种:
第一,客户端包装器。在 Agent 调用 SDK 的chat.completions.create方法时,套一层代理。调用结束后读取响应里的usage,再执行费用计算。这种方式最通用,对代码侵入最小。
第二,网关中间件。如果你的团队已经有 LLM 网关,可以在网关层统一记录 token 和费用。好处是全局可见,坏处是拿不到 Agent 内部的轮次信息,需要网关自定义字段配合。
第三,日志解析。直接解析模型 API 的访问日志,从 JSON 格式的日志行里提取 usage。这种方式适合已经无法改动调用链路的存量系统。
从实现成本来看,客户端包装器是最适合个人开发者和中小团队的方式,也是本文演示采用的方案。
3.3 跨平台难在“接口不统一”
说一个插件跨平台,很多人第一反应是“Windows、macOS、Linux 都能跑”。实际上操作系统兼容只是基础,真正的跨平台难点在于:
- 不同 Agent 框架的调用入口不一样,有的用 SDK,有的走 HTTP 直连,有的通过命令行工具。
- 不同模型提供商的
usage结构不一样,有的返回prompt_tokens,有的返回input_tokens,缓存字段的位置更是五花八门。 - 流式请求和非流式请求的 usage 获取方式不同,流式响应往往需要额外开启
stream_options: {"include_usage": true}才能拿到统计。
解决思路是统一抽象。插件内部定义一个自己的UsageRecord数据结构,把各家 API 的字段差异在采集层消化掉,计算层只面向统一结构工作。这也是通吃多种模型、多种框架的关键。
3.4 数据落本地:隐私与安全的取舍
成本监控天然会接触到每次模型调用的元数据,比如会话 ID、模型名、token 数、时间戳。如果还记录 prompt 内容,那基本等于复制了一份完整的调用日志。
更稳妥的设计是:插件只记录usage元数据,不记录消息正文;数据默认保存在本地 SQLite,不上传云端。这样既满足了成本分析的需求,又避免把敏感的业务数据交给第三方。团队如果需要汇总,可以后续再通过导出或内部服务完成。
4. 环境准备与项目结构
4.1 运行环境
本文的演示代码使用 Python 编写,原因是可读性最好,便于理解核心链路。你不需要真实的大模型 API Key,因为我们会用模拟数据跑通流程。
建议环境:
- Python 3.9 及以上版本。
- 标准库中的
sqlite3(Python 自带,无需安装)。 matplotlib,用于绘制实时图表。PyYAML,如果你希望从 YAML 配置读取价格表。
如果你只是验证记录和计算逻辑,可以暂时不装 matplotlib。版本以你本机实际可用版本为准,本文不绑定特定版本。
4.2 目录结构
agent-cost-monitor/ ├─ config.yaml # 模型价格表与监控参数 ├─ demo_simulate.py # 模拟数据,跑通记录与计算链路 ├─ run_dashboard.py # 启动实时图表 ├─ requirements.txt # 依赖清单 └─ monitor/ ├─ __init__.py # 标识这是一个 Python 包 ├─ calculator.py # 费用计算器 ├─ recorder.py # SQLite 写入与按轮次查询 ├─ dashboard.py # 实时图表 └─ wrapper.py # 包装 OpenAI 兼容客户端4.3 依赖清单
# 文件路径:requirements.txt PyYAML>=6.0 matplotlib>=3.5安装命令:
pip install -r requirements.txt5. 最小实现:手写一个 Agent Cost Monitor
下面我们按照“计算 → 记录 → 包装 → 查询 → 绘图”的顺序,逐个文件实现。代码保持最小化,重点展示链路,不堆砌无关功能。
5.1 第 1 步:价格表配置
价格表用于描述“每个模型每百万 token 多少钱”。注意,以下只是配置结构示例,数值不代表任何模型的真实报价,实际使用时要替换成你的接入渠道的实时价格。
# 文件路径:config.yaml models: my-model-a: input_per_million: 2.5 output_per_million: 10.0 cache_hit_per_million: 0.5 my-model-b: input_per_million: 1.0 output_per_million: 3.0 cache_hit_per_million: 0.5 monitor: db_path: agent_cost.db # SQLite 数据库路径 window_seconds: 300 # 图表展示最近 5 分钟 rate_window_seconds: 60 # 花费速率统计窗口把价格表放在配置里的好处是:模型降价、渠道调价时,不需要改代码,改配置即可。
5.2 第 2 步:费用计算器
计算器接收一个统一的UsageRecord,根据配置的价格表输出费用。它只负责纯计算,不关心数据从哪来。
# 文件路径:monitor/calculator.py # -*- coding: utf-8 -*- """根据模型价格表,把一次调用的 token 用量换算成费用。""" from dataclasses import dataclass @dataclass class ModelPrice: input_per_million: float output_per_million: float cache_hit_per_million: float = 0.0 @dataclass class UsageRecord: model: str input_tokens: int output_tokens: int cached_tokens: int = 0 class CostCalculator: def __init__(self, price_table: dict[str, ModelPrice]): self.price_table = price_table def compute(self, record: UsageRecord) -> float: price = self.price_table.get(record.model) if price is None: return 0.0 cache_cost = ( record.cached_tokens / 1_000_000 * price.cache_hit_per_million ) input_cost = ( (record.input_tokens - record.cached_tokens) / 1_000_000 * price.input_per_million ) output_cost = ( record.output_tokens / 1_000_000 * price.output_per_million ) return round(cache_cost + max(input_cost, 0.0) + output_cost, 8)关键逻辑:
cache_cost计算命中缓存的那部分输入 token,使用更低单价。input_cost使用“总输入 token 减去缓存命中 token”的部分,按正常输入单价计算。max(input_cost, 0.0)避免缓存字段异常导致负数。- 金额保留 8 位小数,防止浮点误差在累加时被放大。
如果模型名不在价格表里,计算器返回 0。这是故意设计的:宁可漏记,也不能因为一个未知模型报错中断 Agent 主流程。
5.3 第 3 步:记录到 SQLite
记录层负责两件事:把每次调用明细写入数据库;提供按轮次汇总的查询方法。
# 文件路径:monitor/recorder.py # -*- coding: utf-8 -*- """把每次 LLM 调用的用量写入 SQLite,并提供按轮次聚合查询。""" import sqlite3 import time CREATE_TABLE_SQL = """ CREATE TABLE IF NOT EXISTS usage_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, round_id TEXT NOT NULL, model TEXT NOT NULL, ts_ms INTEGER NOT NULL, latency_ms REAL, input_tokens INTEGER NOT NULL, output_tokens INTEGER NOT NULL, cached_tokens INTEGER NOT NULL DEFAULT 0, cost REAL NOT NULL DEFAULT 0 ); CREATE INDEX IF NOT EXISTS idx_usage_time ON usage_log(session_id, ts_ms); """ class UsageRecorder: def __init__(self, db_path: str): self.db_path = db_path conn = sqlite3.connect(db_path) conn.executescript(CREATE_TABLE_SQL) conn.commit() conn.close() def record( self, session_id: str, round_id: str, model: str, input_tokens: int, output_tokens: int, cached_tokens: int = 0, cost: float = 0.0, latency_ms: float = 0.0, ): conn = sqlite3.connect(self.db_path) conn.execute( """ INSERT INTO usage_log( session_id, round_id, model, ts_ms, latency_ms, input_tokens, output_tokens, cached_tokens, cost ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( session_id, round_id, model, int(time.time() * 1000), latency_ms, input_tokens, output_tokens, cached_tokens, cost, ), ) conn.commit() conn.close() def round_costs(self, session_id: str): """按轮次汇总费用,便于核对每一轮 Agent 决策的花费。""" conn = sqlite3.connect(self.db_path) rows = conn.execute( """ SELECT round_id, COUNT(*) AS call_count, SUM(input_tokens) AS input_tokens, SUM(output_tokens) AS output_tokens, SUM(cached_tokens) AS cached_tokens, ROUND(SUM(cost), 8) AS round_cost, MAX(ts_ms) AS last_ts_ms FROM usage_log WHERE session_id = ? GROUP BY round_id ORDER BY MAX(ts_ms) ASC """, (session_id,), ).fetchall() conn.close() return rows从工程角度看,每次record都新建 SQLite 连接,性能一般。真实插件会使用连接复用或批量写入。这里保持简单,是为了让核心逻辑不被连接管理干扰。
5.4 第 4 步:包装 OpenAI 兼容客户端
真正接入 Agent 时,我们需要自动拦截模型调用。下面这段代码演示如何给 OpenAI 兼容客户端套一层“隐形账本”。
# 文件路径:monitor/wrapper.py # -*- coding: utf-8 -*- """给 OpenAI 兼容客户端加一层调用拦截,自动记录 token 和费用。""" import time from .calculator import CostCalculator, UsageRecord from .recorder import UsageRecorder def attach_cost_monitor( client, recorder: UsageRecorder, calculator: CostCalculator, session_id: str = "default", ): """ 替换 client.chat.completions.create 方法,在每次调用后读取 usage 字段,计算费用并写入数据库。 调用时可以通过 monitor_session、monitor_round 两个额外参数 标记当前会话和轮次,这两个参数会被本包装器消费,不会传给 SDK。 """ original_create = client.chat.completions.create def patched_create(*args, **kwargs): current_session = kwargs.pop("monitor_session", session_id) current_round = kwargs.pop("monitor_round", "round-1") t0 = time.time() response = original_create(*args, **kwargs) latency_ms = (time.time() - t0) * 1000.0 usage = getattr(response, "usage", None) if usage is not None: details = getattr(usage, "prompt_tokens_details", None) cached_tokens = ( getattr(details, "cached_tokens", 0) if details else 0 ) record = UsageRecord( model=response.model, input_tokens=usage.prompt_tokens, output_tokens=usage.completion_tokens, cached_tokens=cached_tokens, ) cost = calculator.compute(record) recorder.record( session_id=current_session, round_id=current_round, model=record.model, input_tokens=record.input_tokens, output_tokens=record.output_tokens, cached_tokens=record.cached_tokens, cost=cost, latency_ms=latency_ms, ) return response client.chat.completions.create = patched_create return client使用时,Agent 代码只需要在调用处多传两个参数:
response = client.chat.completions.create( model="my-model-a", messages=..., monitor_session="session-001", monitor_round="round-3", )这两个参数会被包装器消费掉,不会真的发送给模型 API。这个设计让 Agent 内部代码几乎不需要改动,只增加调用点的标记即可。
5.5 第 5 步:实时花费速率图
实时图表用 matplotlib 的FuncAnimation每两秒刷新一次。图表包含一条累计费用曲线,并在标题上显示最近 60 秒的花费速率。
# 文件路径:monitor/dashboard.py # -*- coding: utf-8 -*- """实时绘制累计费用曲线和最近 1 分钟花费速率。""" import sqlite3 import time import matplotlib.pyplot as plt from matplotlib.animation import FuncAnimation class CostDashboard: def __init__( self, db_path: str, window_seconds: int = 300, rate_window_seconds: int = 60, refresh_ms: int = 2000, ): self.db_path = db_path self.window_seconds = window_seconds self.rate_window_seconds = rate_window_seconds self.refresh_ms = refresh_ms def _load_points(self): conn = sqlite3.connect(self.db_path) cutoff = int(time.time() * 1000) - self.window_seconds * 1000 rows = conn.execute( """ SELECT ts_ms, cost FROM usage_log WHERE ts_ms >= ? ORDER BY ts_ms """, (cutoff,), ).fetchall() conn.close() return rows def _update(self, frame): rows = self._load_points() if not rows: return times = [ts / 1000.0 for ts, _ in rows] costs = [cost for _, cost in rows] cumulative = [] total = 0.0 for item in costs: total += item cumulative.append(total) now = time.time() recent = sum( cost for ts, cost in zip(times, costs) if now - ts <= self.rate_window_seconds ) self.ax.clear() self.ax.plot(times, cumulative, marker="o", label="累计费用") self.ax.set_xlabel("时间") self.ax.set_ylabel("累计费用") self.ax.set_title( f"最近 {self.rate_window_seconds}s 花费速率: {recent:.6f}" ) self.ax.legend() self.ax.grid(True) def run(self): fig, self.ax = plt.subplots(figsize=(10, 5)) FuncAnimation(fig, self._update, interval=self.refresh_ms) plt.show()运行入口:
# 文件路径:run_dashboard.py from monitor.dashboard import CostDashboard if __name__ == "__main__": CostDashboard( db_path="agent_cost.db", window_seconds=300, rate_window_seconds=60, ).run()这里有一点需要说明:如果图表跑在没有图形界面的服务器上,plt.show()不会弹出窗口。生产环境更推荐换成 Web 图表方案(比如基于 ECharts 的本地页面,前端轮询同一个查询接口),本文的 matplotlib 版本目的是演示指标如何计算和刷新。
6. 运行验证:记录、计算、绘图的完整链路
6.1 模拟数据跑通
为了避免读者在没有 API Key 的情况下卡住,我们写一个模拟脚本。它不调用任何真实模型,而是随机生成 token 数据,走一遍“计算费用 → 写入数据库 → 按轮次汇总”的完整链路。
# 文件路径:demo_simulate.py # -*- coding: utf-8 -*- """离线模拟:不调用真实 LLM,直接生成一批模拟用量数据。""" import random import time from monitor.calculator import CostCalculator, ModelPrice, UsageRecord from monitor.recorder import UsageRecorder def main(): price_table = { "my-model-a": ModelPrice( input_per_million=2.5, output_per_million=10.0, cache_hit_per_million=0.5, ), } calculator = CostCalculator(price_table) recorder = UsageRecorder("agent_cost.db") for round_no in range(1, 11): input_tokens = random.randint(2000, 8000) output_tokens = random.randint(300, 1500) cached_tokens = random.randint(0, input_tokens // 2) record = UsageRecord( model="my-model-a", input_tokens=input_tokens, output_tokens=output_tokens, cached_tokens=cached_tokens, ) cost = calculator.compute(record) recorder.record( session_id="session-demo", round_id=f"round-{round_no}", model="my-model-a", input_tokens=input_tokens, output_tokens=output_tokens, cached_tokens=cached_tokens, cost=cost, latency_ms=random.uniform(300, 1200), ) print(f"[round-{round_no}] cost={cost:.6f}") time.sleep(0.1) print("\n按轮次汇总:") for row in recorder.round_costs("session-demo"): print(row) if __name__ == "__main__": main()运行:
python demo_simulate.py由于随机数存在,每次输出会不同。下面是某一次运行的示例输出,重点看格式和字段:
[round-1] cost=0.014618 [round-2] cost=0.011042 [round-3] cost=0.018373 [round-4] cost=0.012007 [round-5] cost=0.009685 ... 按轮次汇总: ('round-1', 1, 6538, 1272, 0, 0.014618, 1738...) ('round-2', 1, 5214, 858, 1000, 0.011042, 1738...)每个元组的字段顺序是:round_id、调用次数、输入 token 总量、输出 token 总量、缓存 token 总量、轮次费用、最后调用时间戳。
6.2 SQL 核对
除了代码查询,你还可以用 sqlite3 命令行直接核对数据:
sqlite3 agent_cost.db "SELECT COUNT(*), ROUND(SUM(cost), 8) FROM usage_log;"预期结果类似:
10|0.129905再按模型统计,确认费用归集正确:
sqlite3 agent_cost.db "SELECT model, COUNT(*), ROUND(SUM(cost), 8) FROM usage_log GROUP BY model;"如果输出里模型名为my-model-a,且总费用约等于前面脚本打印的十轮费用之和,说明记录和计算链路都正常。
6.3 判断成功的标准
- 每次运行打印一行
[round-N] cost=...,说明费用计算器工作正常。 - 数据库里出现 10 条
usage_log,说明记录层写入成功。 round_costs按轮次汇总出 10 行,说明每轮费用归集正确。- 启动
run_dashboard.py后,图表绘制出一条持续上升的累计费用曲线,标题上的速率能够随新数据刷新,说明实时可视化链路正常。
最后启动图表:
python run_dashboard.py如果本机是图形环境,会弹出一个窗口,曲线每两秒刷新一次。
7. 常见问题与排查思路
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
| 图表窗口不弹出 | 当前环境无图形界面,或 matplotlib 后端不匹配 | 检查是否在服务器/容器中运行 | 换本机运行,或改用 Agg 后端导出 PNG |
| 数据库没有新记录 | 包装器没有被执行,或调用未走被替换的方法 | 在original_create前后打印日志 | 确认attach_cost_monitor在客户端初始化后调用 |
| 费用始终为 0 | 模型名不在价格表,或价格表配置字段名不一致 | 打印record.model与价格表 key | 修正配置里的模型名,注意大小写 |
| 流式响应拿不到 usage | 流式请求默认不返回 usage 统计 | 检查 SDK 流式响应文档 | 开启stream_options={"include_usage": True}或改用非流式 |
| cached_tokens 一直为 0 | 模型或网关不支持缓存统计,或字段位置不同 | 打印usage原始结构 | 根据实际响应调整字段解析 |
| 金额和账单对不上 | 使用的是渠道价、折扣价,或单位没换算 | 核对单价单位是否为“每百万 token” | 在配置中维护真实成交价格 |
| 数据库文件无限膨胀 | 记录所有历史调用从未清理 | 查看usage_log表大小 | 按会话归档或定期清理旧数据 |
再补充两个容易混淆的细节。
第一个是关于流式请求。很多 Agent 框架为了提高响应速度,默认开启流式输出。流式响应默认不携带 usage,或者只在最后一个 chunk 里携带。如果你在包装器里没有拿到 usage,先检查请求参数里有没有开启 usage 统计,再检查你是否读取的是最后一个 chunk。
第二个是关于多模型网关。如果你的请求经过统一网关转发,网关可能重写了模型名,也可能不返回 usage。这种情况下,网关层是更合适的埋点位置。本地插件只能做到“尽量记”,无法保证网关吞掉字段之后还能还原。
8. 生产环境最佳实践与工程建议
8.1 价格表不要手写在代码里
价格是会变的,渠道商和模型厂商经常调整价格、推出缓存折扣。把价格表放在独立配置文件里,并且纳入版本管理。每次更新价格表,都记录一条变更说明,方便日后核对历史账单。
如果你的费用数据要用于对账,最好同时保存“当时使用的价格版本”。简单做法是在usage_log表里增加一列price_version,写入记录时带上当前价格表版本号。
8.2 警惕缓存 token 的统计差异
缓存 token 是成本计算里最容易出错的地方。有的模型把缓存命中 token 计入prompt_tokens,有的则单独返回。如果你漏算,会把费用记高;如果你把缓存部分也按正常输入价计算,同样会记高。
最稳妥的方案是:先打印一次完整的usage对象,看清楚你接入的模型和 SDK 版本到底返回哪些字段,再写解析逻辑。不要照搬别人的代码。
8.3 给花费设置阈值和告警
监控的价值在于触发行动。给监控插件加上两个阈值:
- 单轮费用阈值:某一轮 Agent 决策费用超过阈值,说明上下文过长或重试过多。
- 花费速率阈值:最近 60 秒新增费用超过阈值,说明可能已经陷入异常循环。
触发告警后的动作,可以从保守到激进分三档:打印警告日志、发送通知、主动熔断当前 Agent 会话。生产系统建议先加前两档,观察一段时间后再决定是否启用自动熔断。
8.4 密钥与数据安全边界
插件如果作为 IDE 扩展或本地工具分发,必须遵守最小权限原则。它只需要读取模型 API 返回的响应,不需要读取磁盘上的其他项目文件,也不需要把数据上传到任何第三方服务。
默认情况下,token 记录和费用数据都应该在本地保存。如果要团队汇总,建议只上传脱敏后的指标数据,不包含 prompt 正文。涉及认证信息时,优先使用系统环境变量或系统钥匙串,不要把 API Key 写进配置文件。
8.5 Agent 本身的上下文瘦身
成本监控只能帮你发现问题,真正省钱还要靠改造 Agent。上下文累积是最主要的成本来源,可以尝试:
- 给历史对话做滚动摘要,用一小段摘要替换前面的完整消息。
- 按主题拆分会话,避免一个会话无限变长。
- 对工具返回的大段文本先压缩再喂给模型。
- 设置最大轮数或最大上下文 token 上限,超过即终止。
这些手段的效果,都可以用本文的每轮费用数据来验证。改之前看每轮费用曲线,改之后再看曲线,降幅一目了然。
9. 总结与下一步建议
这篇文章的核心判断是:Agent 时代,成本监控不是可选优化项,而是开发基础设施。你无法控制看不见的支出。
我们拆解了 Agent Cost Monitor 这类跨平台插件的四个关键能力:采集调用用量、按价格表计算费用、按轮次归集成本、实时绘制花费速率。然后用手写的最小 Python 示例,把“记录 → 计算 → 入库 → 查询 → 绘图”的链路完整跑通了一遍。所有代码都可以直接复制到你的项目里作为骨架。
下一步如果你要继续深入,有三个方向值得优先探索:
- 把包装器接到你实际使用的 Agent 框架上,跑一两个真实任务,看看每轮费用分布是否符合预期。
- 把 matplotlib 图表替换成一个本地 Web 面板,用轮询方式读取 SQLite 数据,界面会更适合团队使用。
- 增加价格表版本管理和告警机制,让监控真正具备生产可用性。
建议把本文的代码骨架保存下来,下次调试 Agent 时直接跑一遍。看见每一轮的花费之后,你会发现很多之前“没想到”的成本,其实早就存在,只是你一直没有去看。