1. 为什么 Agent 跑 DeepSeek Harness 时账单总对不上先说结论Agent 的推理成本不是调用次数 × 单价能算清的它是一条会自我膨胀的链路。你写一个 ReAct 循环用户只问了一句帮我查下上周的订单异常Harness 内部可能发生 6 到 12 次模型调用规划一次、选工具一次、读工具返回一次、判断要不要再调一次、最后总结一次。每一次调用都把前面的上下文重新塞进 promptToken 消耗呈阶梯式叠加而账单上只有一个总数。我见过最典型的场景某团队用 DeepSeek Harness 搭了个工单分类 Agent本地测试时单次任务大概 2000 Token上线后月度消耗直接冲到预估的 4 倍。排查半天发现两个问题一是工具返回的 JSON 原样进了上下文一个查询结果就 3000 Token二是失败重试没有做幂等同一个请求在 3 秒内被重复提交了 3 次。这两个问题在按调用次数的视角下完全看不见只有把 Token 拆到单次调用、单个工具、单轮迭代的粒度才能定位。所以这篇要解决的核心问题是在 Agent Plan 编排 DeepSeek Harness 执行的组合下如何建立 Token 级的计量、归因和优化闭环。适合谁看正在用 Harness 写 Agent、发现成本不可控的后端和 AI Infra 同学需要向老板解释钱花在哪的技术负责人以及想把降本从拍脑袋变成可验证动作的工程团队。整篇的路径是这样先讲清楚 Token 在 Harness 里是怎么被消耗的计量机制再给一套可复制的计量配置和统计脚本能直接跑然后是优化策略和前后对比验证有数据最后是排错清单。所有模型调用统一走 TaoToken 的 API 通道这样 Key 和 Base URL 只有一套计量脚本不用为多供应商写分支。需要提前说明一个认知Token 核算的目的不是省钱本身而是让每一次优化都有基线。没有基线的优化往往是换个地方浪费。下面所有代码和配置都以可复制、可验证为准你照着改路径和 Key 就能跑。2. TaoToken 前置准备统一 Key 与 Base URL 的接入配置在写计量脚本之前先把调用通道固定下来。Agent 场景最怕的是模型供应商散落在代码各处计量时要在每个 SDK 里插桩。用 TaoToken 作为统一通道的好处是Base URL 一个、Key 一个、模型 ID 用字符串切换计量逻辑只写一层中间件就能覆盖全部调用。2.1 获取 Key 与确认端点打开官网 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 注册后进入控制台在 API Keys 页面创建密钥。建议按环境分 Key开发一个、预发一个、生产一个这样统计脚本按 Key 前缀就能区分环境不用在业务代码里传 env 参数。端点信息固定为项目值Base URLhttps://taotoken.net/api对话端点/v1/chat/completions鉴权方式Authorization: Bearer 你的Key模型 ID 示例deepseek-chat、deepseek-reasoner注意 Base URL 结尾不要多加/v1SDK 内部会自己拼。这一点踩过坑OpenAI 兼容 SDK 的base_url参数如果写成https://taotoken.net/api/v1实际请求会变成/api/v1/v1/chat/completions直接 404。2.2 环境变量与最小连通验证先把 Key 放进环境变量别硬编码export TAOTOKEN_API_KEYsk-你的密钥 export TAOTOKEN_BASE_URLhttps://taotoken.net/api然后用 curl 做一次最小请求确认通道通、模型 ID 对curl -s $TAOTOKEN_BASE_URL/v1/chat/completions \ -H Authorization: Bearer $TAOTOKEN_API_KEY \ -H Content-Type: application/json \ -d { model: deepseek-chat, messages: [{role: user, content: 回复 OK 两个字母}], max_tokens: 16 }返回体里重点看usage字段它包含prompt_tokens、completion_tokens、total_tokens。这三个数字就是你后续所有核算的原始数据来源。如果返回 401先检查 Key 是否带了多余空格如果返回 404检查 Base URL 是否多写了/v1。2.3 在 Harness 侧配置模型入口DeepSeek Harness 的模型配置通常是一个 provider 配置块。以常见的 OpenAI 兼容写法为例配置文件里这样写# harness.toml [llm] provider openai-compatible base_url https://taotoken.net/api api_key_env TAOTOKEN_API_KEY default_model deepseek-chat timeout_seconds 120 max_retries 2 [llm.models] planner deepseek-chat executor deepseek-chat summarizer deepseek-chat这里把 planner、executor、summarizer 分开命名是为了后面做成本归因——同一个 Agent 任务里规划阶段和总结阶段的 Token 消耗特征完全不同混在一起统计就看不出问题在哪。如果你的 Harness 版本用 JSON 配置结构等价{ llm: { provider: openai-compatible, base_url: https://taotoken.net/api, api_key_env: TAOTOKEN_API_KEY, default_model: deepseek-chat, roles: { planner: deepseek-chat, executor: deepseek-chat, summarizer: deepseek-chat } } }三件套齐了Base URL 是https://taotoken.net/apiKey 走环境变量TAOTOKEN_API_KEYModel ID 是deepseek-chat。后面所有脚本都从这三个值出发不再出现第二套配置。3. 可复制的 Token 计量配置与成本统计脚本这一节是全文的核心给一套能直接跑的计量层。设计原则是不改 Harness 内部代码在 LLM 客户端外面包一层中间件所有请求和响应都经过它自动记录 usage 并落盘。3.1 计量中间件的结构中间件要做四件事拦截请求、透传调用、解析 usage、写事件记录。事件记录用 JSONL 格式一行一个事件方便后续用 pandas 或 jq 聚合。字段设计如下{ event_id: evt_20260901_0001, ts: 2026-09-01T10:12:33.120Z, session_id: sess_a1b2, trace_id: trace_x9y8, role: planner, model: deepseek-chat, iteration: 2, prompt_tokens: 1840, completion_tokens: 312, total_tokens: 2152, latency_ms: 2870, tool_calls: 1, status: ok }role对应配置里的 planner/executor/summarizeriteration是 Agent 循环的第几轮tool_calls记录本轮触发了几个工具。这三个字段是归因的关键没有它们就只能看到总数。3.2 Python 计量客户端实现下面这段可以直接放进项目依赖只有openai和标准库import json import os import time import uuid from datetime import datetime, timezone from openai import OpenAI class MeteredClient: def __init__(self, log_pathtoken_events.jsonl): self.client OpenAI( base_urlos.environ[TAOTOKEN_BASE_URL], api_keyos.environ[TAOTOKEN_API_KEY], ) self.log_path log_path def chat(self, role, messages, session_id, trace_id, iteration1, tool_calls0, **kwargs): start time.time() event_id fevt_{uuid.uuid4().hex[:12]} status ok try: resp self.client.chat.completions.create( modelkwargs.pop(model, deepseek-chat), messagesmessages, **kwargs, ) usage resp.usage record { event_id: event_id, ts: datetime.now(timezone.utc).isoformat(), session_id: session_id, trace_id: trace_id, role: role, model: resp.model, iteration: iteration, prompt_tokens: usage.prompt_tokens, completion_tokens: usage.completion_tokens, total_tokens: usage.total_tokens, latency_ms: int((time.time() - start) * 1000), tool_calls: tool_calls, status: status, } return resp, record except Exception as e: status ferror:{type(e).__name__} record { event_id: event_id, ts: datetime.now(timezone.utc).isoformat(), session_id: session_id, trace_id: trace_id, role: role, model: kwargs.get(model, deepseek-chat), iteration: iteration, prompt_tokens: 0, completion_tokens: 0, total_tokens: 0, latency_ms: int((time.time() - start) * 1000), tool_calls: tool_calls, status: status, } raise finally: with open(self.log_path, a, encodingutf-8) as f: f.write(json.dumps(record, ensure_asciiFalse) \n)注意finally块失败请求也要记录因为重试产生的 Token 同样是成本。很多团队只统计成功调用结果重试风暴完全不可见。3.3 在 Harness 循环里接入Agent 主循环里这样用把 role 和 iteration 传进去client MeteredClient() def run_agent(user_query, session_id): trace_id ftrace_{uuid.uuid4().hex[:8]} history [{role: user, content: user_query}] total 0 for it in range(1, 9): resp, rec client.chat( roleplanner if it 1 else executor, messageshistory, session_idsession_id, trace_idtrace_id, iterationit, tool_calls0, ) total rec[total_tokens] msg resp.choices[0].message if not msg.tool_calls: history.append({role: assistant, content: msg.content}) break history.append(msg) for tc in msg.tool_calls: result dispatch_tool(tc.function.name, tc.function.arguments) history.append({ role: tool, tool_call_id: tc.id, content: result, }) return history[-1][content], total这里有个细节值得强调工具返回的content直接进 history如果工具返回的是原始 JSON 或 HTMLToken 会爆炸。第 5 节会专门讲这个坑。3.4 成本聚合脚本有了 JSONL聚合就是几行 pandas。单价按你实际合同填这里用示例值import pandas as pd PRICE {input_per_1k: 0.001, output_per_1k: 0.002} df pd.read_json(token_events.jsonl, linesTrue) df[cost] ( df[prompt_tokens] / 1000 * PRICE[input_per_1k] df[completion_tokens] / 1000 * PRICE[output_per_1k] ) by_role df.groupby(role).agg( calls(event_id, count), input_tokens(prompt_tokens, sum), output_tokens(completion_tokens, sum), cost(cost, sum), ).sort_values(cost, ascendingFalse) by_iter df.groupby(iteration).agg( calls(event_id, count), avg_input(prompt_tokens, mean), cost(cost, sum), ) print(by_role) print(by_iter)跑出来的by_iter表会暴露一个典型现象第 1 轮输入 Token 很小第 3 到 5 轮急剧上升因为历史上下文在累积。如果第 6 轮之后还在涨说明工具返回没有做裁剪或者循环没有收敛条件。3.5 关键派生指标除了绝对值这几个比率指标更能指导优化指标计算方式健康区间异常含义有效输入占比新增内容 Token / prompt_tokens 40%过低说明上下文冗余严重输出输入比completion / prompt0.1 ~ 0.5过高说明模型在啰嗦单任务调用数calls / session3 ~ 8超过 10 说明循环没收敛工具 Token 占比工具返回 Token / 总输入 30%过高说明工具输出未裁剪失败重试率error 事件 / 总事件 2%过高说明幂等或超时有问题把这五个指标做成日报成本异常基本能在发生当天定位到具体环节而不是月底看账单猜。4. 验证请求与优化前后对比从数据到动作计量层搭好后先做一次基线采集再做优化最后用同一套脚本对比。这一节给完整的验证流程和真实形态的对比数据。4.1 基线采集跑一批固定任务不要用线上流量做基线波动太大。构造 50 条固定任务覆盖简单问答、单工具调用、多工具链、长上下文四类每条跑 3 次取中位数import statistics TASKS load_tasks(bench_tasks.jsonl) # 50 条固定输入 results [] for task in TASKS: runs [] for _ in range(3): _, total run_agent(task[query], session_idfbench_{task[id]}) runs.append(total) results.append({ task_id: task[id], category: task[category], tokens_median: statistics.median(runs), }) pd.DataFrame(results).to_csv(baseline.csv, indexFalse)基线文件存好优化后跑同一批任务逐条对比。这样对比的是同一分布下的变化不受流量结构影响。4.2 优化动作一工具返回裁剪最常见的浪费源。工具返回的 JSON 里往往有大量 Agent 不需要的字段。加一个裁剪层def trim_tool_result(tool_name, raw_result, max_chars800): if isinstance(raw_result, str): text raw_result else: text json.dumps(raw_result, ensure_asciiFalse) if len(text) max_chars: return text # 保留结构关键字段截断长文本 if isinstance(raw_result, dict): keep {k: v for k, v in raw_result.items() if k in (id, status, summary, error)} return json.dumps(keep, ensure_asciiFalse)[:max_chars] return text[:max_chars] ...[truncated]在dispatch_tool返回前调用它。实测下来一个返回 3000 Token 的订单查询结果裁剪后通常降到 400 到 600 Token单任务总消耗下降 25% 到 35%。4.3 优化动作二历史上下文摘要多轮循环里早期轮次的完整对话没必要每轮重发。在第 4 轮之后把前 3 轮压成一段摘要def compress_history(history, keep_recent3): if len(history) keep_recent * 2: return history older history[:-keep_recent * 2] recent history[-keep_recent * 2:] summary_text .join( f{m[role]}: {str(m.get(content, ))[:120]} for m in older ) return [{role: system, content: f[历史摘要] {summary_text}}] recent摘要本身不调用模型纯字符串拼接零额外成本。如果对质量要求高可以用小模型做摘要但要算清楚摘要调用的成本是否小于节省的输入成本——多数情况下字符串截断就够了。4.4 优化动作三循环收敛条件给 Agent 循环加硬性收敛连续两轮没有新的工具调用、或者累计 Token 超过阈值就强制进入总结阶段MAX_TASK_TOKENS 12000 def should_stop(iteration, total_tokens, last_had_tool): if total_tokens MAX_TASK_TOKENS: return True if iteration 3 and not last_had_tool: return True return False这个改动对长尾任务效果最明显。基线里那些跑 10 轮以上的任务加上收敛条件后基本压到 6 轮以内。4.5 前后对比结果同一批 50 条任务优化前后对比指标优化前优化后变化单任务中位 Token8,4204,960-41%单任务 P95 Token21,3009,800-54%平均调用轮次7.24.6-36%工具返回 Token 占比46%18%-28pp单任务成本示例单价0.01260.0074-41%验证方式很直接优化后重跑baseline.csv的同一批 task_id生成after.csv用 pandas merge 后算差值。如果某条任务 Token 反而上升单独看它的 trace通常是裁剪把关键信息截掉了导致多跑一轮这时候要调整max_chars或保留字段列表。4.6 用模型对话做单点验证优化过程中如果怀疑某个模型 ID 的行为差异比如 deepseek-chat 和 deepseek-reasoner 在规划任务上的输出长度差很多可以到模型对话页面 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite 手动发几条同样的 prompt看返回的 usage 和内容长度。这比在代码里反复改参数快得多适合做小样本对照。5. 常见报错与排查清单计量和优化过程中会遇到几类固定报错按出现频率排一下。5.1 401 Unauthorized最常见。原因通常是 Key 没读到环境变量或者 Key 前后有空格。排查顺序先echo $TAOTOKEN_API_KEY | wc -c看长度是否正常再确认 SDK 初始化时api_key参数没有被空字符串覆盖。还有一种情况是 Key 被复制时带了换行符strip()一下就好。5.2 404 Not Found 或 local proxy failedlocal proxy failed这类报错通常出现在 Base URL 配置错误时。检查两点Base URL 必须是https://taotoken.net/api不能带/v1如果你的运行环境配了 HTTP 代理环境变量确认代理没有拦截这个域名。把HTTP_PROXY、HTTPS_PROXY临时清掉再试一次能快速区分是配置问题还是网络环境问题。5.3 reading choices 报错Cannot read properties of undefined (reading choices)说明响应体结构和你预期的不一样。多数是请求本身失败了但代码没检查状态码直接取resp.choices。在计量客户端里加一层判断if not hasattr(resp, choices) or not resp.choices: raise RuntimeError(funexpected response: {resp})另外确认messages数组格式正确role只能是 system/user/assistant/tool拼错会返回 400但某些 SDK 会把错误吞掉返回空对象。5.4 OAuth 相关报错如果你用的是某些 CLI 工具比如 Claude Code 类工具接入可能会遇到 OAuth token 过期或 scope 不足的报错。这类工具通常有自己的认证流程和 API Key 是两套机制。处理方式是重新走一遍工具的登录命令或者在配置里显式指定 API Key 模式。注意不要把 OAuth 的 token 和 API Key 混用两者端点不同。5.5 Token 统计对不上账单计量脚本的 total 和平台账单有偏差通常是三个原因一是流式响应没有正确读取最后的 usage 块流式模式下 usage 在最后一个 chunk二是重试请求被计费但脚本只记了一次三是并发写入 JSONL 时行交错。第一个问题在流式调用时加stream_options{include_usage: True}第二个问题在finally里记录所有尝试第三个问题用文件锁或每进程独立文件。5.6 模型 ID 不存在model not found类报错检查模型 ID 拼写。deepseek-chat 和 deepseek-reasoner 是两个不同 ID大小写敏感。如果从别处复制配置注意有没有带前缀比如某些平台要求deepseek/deepseek-chat格式以控制台文档为准。5.7 排查用的最小复现脚本遇到不确定的报错先用这个脚本隔离问题import os from openai import OpenAI c OpenAI(base_urlos.environ[TAOTOKEN_BASE_URL], api_keyos.environ[TAOTOKEN_API_KEY]) try: r c.chat.completions.create( modeldeepseek-chat, messages[{role: user, content: ping}], max_tokens8, ) print(OK, r.usage) except Exception as e: print(FAIL, type(e).__name__, str(e)[:300])这个脚本能跑通说明通道和 Key 没问题问题在业务代码跑不通按 5.1 到 5.6 的顺序查。6. 把计量层固化成团队习惯最后说落地层面的事。计量脚本写完不是终点要让它变成默认行为。三个具体动作第一把MeteredClient作为项目里唯一允许直接调模型的入口代码评审时看到裸OpenAI()就拦下来第二把第 3.5 节的五个派生指标做成每日定时任务异常时推到群里阈值按基线设第三每次做优化都跑第 4.1 节的固定任务集对比文件进版本库优化记录和代码一起 review。长期跑 Agent 任务的团队如果调用量稳定且需要更可控的配额管理可以看下 Coding Plan 页面 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite 的额度方案配合本文的计量脚本做预算预警。Key 管理和用量查看在控制台 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite 接入细节和字段说明在文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite API Keys 创建入口 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite 。一个实用技巧收尾把token_events.jsonl按天轮转保留 90 天用jq做临时查询比写脚本快。比如查今天 planner 角色的总输入 Tokenjq -s map(select(.roleplanner)) | map(.prompt_tokens) | add \ token_events_2026-09-01.jsonl这种随手可查的能力比任何仪表盘都更容易让团队养成看 Token 的习惯。
1. 为什么 Agent 跑 DeepSeek Harness 时账单总对不上先说结论Agent 的推理成本不是调用次数 × 单价能算清的它是一条会自我膨胀的链路。你写一个 ReAct 循环用户只问了一句帮我查下上周的订单异常Harness 内部可能发生 6 到 12 次模型调用规划一次、选工具一次、读工具返回一次、判断要不要再调一次、最后总结一次。每一次调用都把前面的上下文重新塞进 promptToken 消耗呈阶梯式叠加而账单上只有一个总数。我见过最典型的场景某团队用 DeepSeek Harness 搭了个工单分类 Agent本地测试时单次任务大概 2000 Token上线后月度消耗直接冲到预估的 4 倍。排查半天发现两个问题一是工具返回的 JSON 原样进了上下文一个查询结果就 3000 Token二是失败重试没有做幂等同一个请求在 3 秒内被重复提交了 3 次。这两个问题在按调用次数的视角下完全看不见只有把 Token 拆到单次调用、单个工具、单轮迭代的粒度才能定位。所以这篇要解决的核心问题是在 Agent Plan 编排 DeepSeek Harness 执行的组合下如何建立 Token 级的计量、归因和优化闭环。适合谁看正在用 Harness 写 Agent、发现成本不可控的后端和 AI Infra 同学需要向老板解释钱花在哪的技术负责人以及想把降本从拍脑袋变成可验证动作的工程团队。整篇的路径是这样先讲清楚 Token 在 Harness 里是怎么被消耗的计量机制再给一套可复制的计量配置和统计脚本能直接跑然后是优化策略和前后对比验证有数据最后是排错清单。所有模型调用统一走 TaoToken 的 API 通道这样 Key 和 Base URL 只有一套计量脚本不用为多供应商写分支。需要提前说明一个认知Token 核算的目的不是省钱本身而是让每一次优化都有基线。没有基线的优化往往是换个地方浪费。下面所有代码和配置都以可复制、可验证为准你照着改路径和 Key 就能跑。2. TaoToken 前置准备统一 Key 与 Base URL 的接入配置在写计量脚本之前先把调用通道固定下来。Agent 场景最怕的是模型供应商散落在代码各处计量时要在每个 SDK 里插桩。用 TaoToken 作为统一通道的好处是Base URL 一个、Key 一个、模型 ID 用字符串切换计量逻辑只写一层中间件就能覆盖全部调用。2.1 获取 Key 与确认端点打开官网 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 注册后进入控制台在 API Keys 页面创建密钥。建议按环境分 Key开发一个、预发一个、生产一个这样统计脚本按 Key 前缀就能区分环境不用在业务代码里传 env 参数。端点信息固定为项目值Base URLhttps://taotoken.net/api对话端点/v1/chat/completions鉴权方式Authorization: Bearer 你的Key模型 ID 示例deepseek-chat、deepseek-reasoner注意 Base URL 结尾不要多加/v1SDK 内部会自己拼。这一点踩过坑OpenAI 兼容 SDK 的base_url参数如果写成https://taotoken.net/api/v1实际请求会变成/api/v1/v1/chat/completions直接 404。2.2 环境变量与最小连通验证先把 Key 放进环境变量别硬编码export TAOTOKEN_API_KEYsk-你的密钥 export TAOTOKEN_BASE_URLhttps://taotoken.net/api然后用 curl 做一次最小请求确认通道通、模型 ID 对curl -s $TAOTOKEN_BASE_URL/v1/chat/completions \ -H Authorization: Bearer $TAOTOKEN_API_KEY \ -H Content-Type: application/json \ -d { model: deepseek-chat, messages: [{role: user, content: 回复 OK 两个字母}], max_tokens: 16 }返回体里重点看usage字段它包含prompt_tokens、completion_tokens、total_tokens。这三个数字就是你后续所有核算的原始数据来源。如果返回 401先检查 Key 是否带了多余空格如果返回 404检查 Base URL 是否多写了/v1。2.3 在 Harness 侧配置模型入口DeepSeek Harness 的模型配置通常是一个 provider 配置块。以常见的 OpenAI 兼容写法为例配置文件里这样写# harness.toml [llm] provider openai-compatible base_url https://taotoken.net/api api_key_env TAOTOKEN_API_KEY default_model deepseek-chat timeout_seconds 120 max_retries 2 [llm.models] planner deepseek-chat executor deepseek-chat summarizer deepseek-chat这里把 planner、executor、summarizer 分开命名是为了后面做成本归因——同一个 Agent 任务里规划阶段和总结阶段的 Token 消耗特征完全不同混在一起统计就看不出问题在哪。如果你的 Harness 版本用 JSON 配置结构等价{ llm: { provider: openai-compatible, base_url: https://taotoken.net/api, api_key_env: TAOTOKEN_API_KEY, default_model: deepseek-chat, roles: { planner: deepseek-chat, executor: deepseek-chat, summarizer: deepseek-chat } } }三件套齐了Base URL 是https://taotoken.net/apiKey 走环境变量TAOTOKEN_API_KEYModel ID 是deepseek-chat。后面所有脚本都从这三个值出发不再出现第二套配置。3. 可复制的 Token 计量配置与成本统计脚本这一节是全文的核心给一套能直接跑的计量层。设计原则是不改 Harness 内部代码在 LLM 客户端外面包一层中间件所有请求和响应都经过它自动记录 usage 并落盘。3.1 计量中间件的结构中间件要做四件事拦截请求、透传调用、解析 usage、写事件记录。事件记录用 JSONL 格式一行一个事件方便后续用 pandas 或 jq 聚合。字段设计如下{ event_id: evt_20260901_0001, ts: 2026-09-01T10:12:33.120Z, session_id: sess_a1b2, trace_id: trace_x9y8, role: planner, model: deepseek-chat, iteration: 2, prompt_tokens: 1840, completion_tokens: 312, total_tokens: 2152, latency_ms: 2870, tool_calls: 1, status: ok }role对应配置里的 planner/executor/summarizeriteration是 Agent 循环的第几轮tool_calls记录本轮触发了几个工具。这三个字段是归因的关键没有它们就只能看到总数。3.2 Python 计量客户端实现下面这段可以直接放进项目依赖只有openai和标准库import json import os import time import uuid from datetime import datetime, timezone from openai import OpenAI class MeteredClient: def __init__(self, log_pathtoken_events.jsonl): self.client OpenAI( base_urlos.environ[TAOTOKEN_BASE_URL], api_keyos.environ[TAOTOKEN_API_KEY], ) self.log_path log_path def chat(self, role, messages, session_id, trace_id, iteration1, tool_calls0, **kwargs): start time.time() event_id fevt_{uuid.uuid4().hex[:12]} status ok try: resp self.client.chat.completions.create( modelkwargs.pop(model, deepseek-chat), messagesmessages, **kwargs, ) usage resp.usage record { event_id: event_id, ts: datetime.now(timezone.utc).isoformat(), session_id: session_id, trace_id: trace_id, role: role, model: resp.model, iteration: iteration, prompt_tokens: usage.prompt_tokens, completion_tokens: usage.completion_tokens, total_tokens: usage.total_tokens, latency_ms: int((time.time() - start) * 1000), tool_calls: tool_calls, status: status, } return resp, record except Exception as e: status ferror:{type(e).__name__} record { event_id: event_id, ts: datetime.now(timezone.utc).isoformat(), session_id: session_id, trace_id: trace_id, role: role, model: kwargs.get(model, deepseek-chat), iteration: iteration, prompt_tokens: 0, completion_tokens: 0, total_tokens: 0, latency_ms: int((time.time() - start) * 1000), tool_calls: tool_calls, status: status, } raise finally: with open(self.log_path, a, encodingutf-8) as f: f.write(json.dumps(record, ensure_asciiFalse) \n)注意finally块失败请求也要记录因为重试产生的 Token 同样是成本。很多团队只统计成功调用结果重试风暴完全不可见。3.3 在 Harness 循环里接入Agent 主循环里这样用把 role 和 iteration 传进去client MeteredClient() def run_agent(user_query, session_id): trace_id ftrace_{uuid.uuid4().hex[:8]} history [{role: user, content: user_query}] total 0 for it in range(1, 9): resp, rec client.chat( roleplanner if it 1 else executor, messageshistory, session_idsession_id, trace_idtrace_id, iterationit, tool_calls0, ) total rec[total_tokens] msg resp.choices[0].message if not msg.tool_calls: history.append({role: assistant, content: msg.content}) break history.append(msg) for tc in msg.tool_calls: result dispatch_tool(tc.function.name, tc.function.arguments) history.append({ role: tool, tool_call_id: tc.id, content: result, }) return history[-1][content], total这里有个细节值得强调工具返回的content直接进 history如果工具返回的是原始 JSON 或 HTMLToken 会爆炸。第 5 节会专门讲这个坑。3.4 成本聚合脚本有了 JSONL聚合就是几行 pandas。单价按你实际合同填这里用示例值import pandas as pd PRICE {input_per_1k: 0.001, output_per_1k: 0.002} df pd.read_json(token_events.jsonl, linesTrue) df[cost] ( df[prompt_tokens] / 1000 * PRICE[input_per_1k] df[completion_tokens] / 1000 * PRICE[output_per_1k] ) by_role df.groupby(role).agg( calls(event_id, count), input_tokens(prompt_tokens, sum), output_tokens(completion_tokens, sum), cost(cost, sum), ).sort_values(cost, ascendingFalse) by_iter df.groupby(iteration).agg( calls(event_id, count), avg_input(prompt_tokens, mean), cost(cost, sum), ) print(by_role) print(by_iter)跑出来的by_iter表会暴露一个典型现象第 1 轮输入 Token 很小第 3 到 5 轮急剧上升因为历史上下文在累积。如果第 6 轮之后还在涨说明工具返回没有做裁剪或者循环没有收敛条件。3.5 关键派生指标除了绝对值这几个比率指标更能指导优化指标计算方式健康区间异常含义有效输入占比新增内容 Token / prompt_tokens 40%过低说明上下文冗余严重输出输入比completion / prompt0.1 ~ 0.5过高说明模型在啰嗦单任务调用数calls / session3 ~ 8超过 10 说明循环没收敛工具 Token 占比工具返回 Token / 总输入 30%过高说明工具输出未裁剪失败重试率error 事件 / 总事件 2%过高说明幂等或超时有问题把这五个指标做成日报成本异常基本能在发生当天定位到具体环节而不是月底看账单猜。4. 验证请求与优化前后对比从数据到动作计量层搭好后先做一次基线采集再做优化最后用同一套脚本对比。这一节给完整的验证流程和真实形态的对比数据。4.1 基线采集跑一批固定任务不要用线上流量做基线波动太大。构造 50 条固定任务覆盖简单问答、单工具调用、多工具链、长上下文四类每条跑 3 次取中位数import statistics TASKS load_tasks(bench_tasks.jsonl) # 50 条固定输入 results [] for task in TASKS: runs [] for _ in range(3): _, total run_agent(task[query], session_idfbench_{task[id]}) runs.append(total) results.append({ task_id: task[id], category: task[category], tokens_median: statistics.median(runs), }) pd.DataFrame(results).to_csv(baseline.csv, indexFalse)基线文件存好优化后跑同一批任务逐条对比。这样对比的是同一分布下的变化不受流量结构影响。4.2 优化动作一工具返回裁剪最常见的浪费源。工具返回的 JSON 里往往有大量 Agent 不需要的字段。加一个裁剪层def trim_tool_result(tool_name, raw_result, max_chars800): if isinstance(raw_result, str): text raw_result else: text json.dumps(raw_result, ensure_asciiFalse) if len(text) max_chars: return text # 保留结构关键字段截断长文本 if isinstance(raw_result, dict): keep {k: v for k, v in raw_result.items() if k in (id, status, summary, error)} return json.dumps(keep, ensure_asciiFalse)[:max_chars] return text[:max_chars] ...[truncated]在dispatch_tool返回前调用它。实测下来一个返回 3000 Token 的订单查询结果裁剪后通常降到 400 到 600 Token单任务总消耗下降 25% 到 35%。4.3 优化动作二历史上下文摘要多轮循环里早期轮次的完整对话没必要每轮重发。在第 4 轮之后把前 3 轮压成一段摘要def compress_history(history, keep_recent3): if len(history) keep_recent * 2: return history older history[:-keep_recent * 2] recent history[-keep_recent * 2:] summary_text .join( f{m[role]}: {str(m.get(content, ))[:120]} for m in older ) return [{role: system, content: f[历史摘要] {summary_text}}] recent摘要本身不调用模型纯字符串拼接零额外成本。如果对质量要求高可以用小模型做摘要但要算清楚摘要调用的成本是否小于节省的输入成本——多数情况下字符串截断就够了。4.4 优化动作三循环收敛条件给 Agent 循环加硬性收敛连续两轮没有新的工具调用、或者累计 Token 超过阈值就强制进入总结阶段MAX_TASK_TOKENS 12000 def should_stop(iteration, total_tokens, last_had_tool): if total_tokens MAX_TASK_TOKENS: return True if iteration 3 and not last_had_tool: return True return False这个改动对长尾任务效果最明显。基线里那些跑 10 轮以上的任务加上收敛条件后基本压到 6 轮以内。4.5 前后对比结果同一批 50 条任务优化前后对比指标优化前优化后变化单任务中位 Token8,4204,960-41%单任务 P95 Token21,3009,800-54%平均调用轮次7.24.6-36%工具返回 Token 占比46%18%-28pp单任务成本示例单价0.01260.0074-41%验证方式很直接优化后重跑baseline.csv的同一批 task_id生成after.csv用 pandas merge 后算差值。如果某条任务 Token 反而上升单独看它的 trace通常是裁剪把关键信息截掉了导致多跑一轮这时候要调整max_chars或保留字段列表。4.6 用模型对话做单点验证优化过程中如果怀疑某个模型 ID 的行为差异比如 deepseek-chat 和 deepseek-reasoner 在规划任务上的输出长度差很多可以到模型对话页面 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite 手动发几条同样的 prompt看返回的 usage 和内容长度。这比在代码里反复改参数快得多适合做小样本对照。5. 常见报错与排查清单计量和优化过程中会遇到几类固定报错按出现频率排一下。5.1 401 Unauthorized最常见。原因通常是 Key 没读到环境变量或者 Key 前后有空格。排查顺序先echo $TAOTOKEN_API_KEY | wc -c看长度是否正常再确认 SDK 初始化时api_key参数没有被空字符串覆盖。还有一种情况是 Key 被复制时带了换行符strip()一下就好。5.2 404 Not Found 或 local proxy failedlocal proxy failed这类报错通常出现在 Base URL 配置错误时。检查两点Base URL 必须是https://taotoken.net/api不能带/v1如果你的运行环境配了 HTTP 代理环境变量确认代理没有拦截这个域名。把HTTP_PROXY、HTTPS_PROXY临时清掉再试一次能快速区分是配置问题还是网络环境问题。5.3 reading choices 报错Cannot read properties of undefined (reading choices)说明响应体结构和你预期的不一样。多数是请求本身失败了但代码没检查状态码直接取resp.choices。在计量客户端里加一层判断if not hasattr(resp, choices) or not resp.choices: raise RuntimeError(funexpected response: {resp})另外确认messages数组格式正确role只能是 system/user/assistant/tool拼错会返回 400但某些 SDK 会把错误吞掉返回空对象。5.4 OAuth 相关报错如果你用的是某些 CLI 工具比如 Claude Code 类工具接入可能会遇到 OAuth token 过期或 scope 不足的报错。这类工具通常有自己的认证流程和 API Key 是两套机制。处理方式是重新走一遍工具的登录命令或者在配置里显式指定 API Key 模式。注意不要把 OAuth 的 token 和 API Key 混用两者端点不同。5.5 Token 统计对不上账单计量脚本的 total 和平台账单有偏差通常是三个原因一是流式响应没有正确读取最后的 usage 块流式模式下 usage 在最后一个 chunk二是重试请求被计费但脚本只记了一次三是并发写入 JSONL 时行交错。第一个问题在流式调用时加stream_options{include_usage: True}第二个问题在finally里记录所有尝试第三个问题用文件锁或每进程独立文件。5.6 模型 ID 不存在model not found类报错检查模型 ID 拼写。deepseek-chat 和 deepseek-reasoner 是两个不同 ID大小写敏感。如果从别处复制配置注意有没有带前缀比如某些平台要求deepseek/deepseek-chat格式以控制台文档为准。5.7 排查用的最小复现脚本遇到不确定的报错先用这个脚本隔离问题import os from openai import OpenAI c OpenAI(base_urlos.environ[TAOTOKEN_BASE_URL], api_keyos.environ[TAOTOKEN_API_KEY]) try: r c.chat.completions.create( modeldeepseek-chat, messages[{role: user, content: ping}], max_tokens8, ) print(OK, r.usage) except Exception as e: print(FAIL, type(e).__name__, str(e)[:300])这个脚本能跑通说明通道和 Key 没问题问题在业务代码跑不通按 5.1 到 5.6 的顺序查。6. 把计量层固化成团队习惯最后说落地层面的事。计量脚本写完不是终点要让它变成默认行为。三个具体动作第一把MeteredClient作为项目里唯一允许直接调模型的入口代码评审时看到裸OpenAI()就拦下来第二把第 3.5 节的五个派生指标做成每日定时任务异常时推到群里阈值按基线设第三每次做优化都跑第 4.1 节的固定任务集对比文件进版本库优化记录和代码一起 review。长期跑 Agent 任务的团队如果调用量稳定且需要更可控的配额管理可以看下 Coding Plan 页面 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite 的额度方案配合本文的计量脚本做预算预警。Key 管理和用量查看在控制台 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite 接入细节和字段说明在文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite API Keys 创建入口 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite 。一个实用技巧收尾把token_events.jsonl按天轮转保留 90 天用jq做临时查询比写脚本快。比如查今天 planner 角色的总输入 Tokenjq -s map(select(.roleplanner)) | map(.prompt_tokens) | add \ token_events_2026-09-01.jsonl这种随手可查的能力比任何仪表盘都更容易让团队养成看 Token 的习惯。
