Agent Plan × DeepSeek Harness：Agent 推理成本 Token 级精细核算与优化策略

1. 为什么 Agent 跑 DeepSeek Harness 时账单总对不上

先说结论：Agent 的推理成本不是"调用次数 × 单价"能算清的，它是一条会自我膨胀的链路。你写一个 ReAct 循环，用户只问了一句"帮我查下上周的订单异常"，Harness 内部可能发生 6 到 12 次模型调用：规划一次、选工具一次、读工具返回一次、判断要不要再调一次、最后总结一次。每一次调用都把前面的上下文重新塞进 prompt，Token 消耗呈阶梯式叠加，而账单上只有一个总数。

我见过最典型的场景：某团队用 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_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 注册后进入控制台，在 API Keys 页面创建密钥。建议按环境分 Key：开发一个、预发一个、生产一个，这样统计脚本按 Key 前缀就能区分环境，不用在业务代码里传 env 参数。

端点信息固定为：

注意 Base URL 结尾不要多加/v1，SDK 内部会自己拼。这一点踩过坑：OpenAI 兼容 SDK 的base_url参数如果写成https://taotoken.net/api/v1，实际请求会变成/api/v1/v1/chat/completions，直接 404。

2.2 环境变量与最小连通验证

先把 Key 放进环境变量，别硬编码：

export TAOTOKEN_API_KEY="sk-你的密钥" export TAOTOKEN_BASE_URL="https://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/api，Key 走环境变量TAOTOKEN_API_KEY，Model 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/summarizer，iteration是 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_path="token_events.jsonl"): self.client = OpenAI( base_url=os.environ["TAOTOKEN_BASE_URL"], api_key=os.environ["TAOTOKEN_API_KEY"], ) self.log_path = log_path def chat(self, role, messages, session_id, trace_id, iteration=1, tool_calls=0, **kwargs): start = time.time() event_id = f"evt_{uuid.uuid4().hex[:12]}" status = "ok" try: resp = self.client.chat.completions.create( model=kwargs.pop("model", "deepseek-chat"), messages=messages, **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 = f"error:{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", encoding="utf-8") as f: f.write(json.dumps(record, ensure_ascii=False) + "\n")

注意finally块：失败请求也要记录，因为重试产生的 Token 同样是成本。很多团队只统计成功调用，结果重试风暴完全不可见。

3.3 在 Harness 循环里接入

Agent 主循环里这样用，把 role 和 iteration 传进去：

client = MeteredClient() def run_agent(user_query, session_id): trace_id = f"trace_{uuid.uuid4().hex[:8]}" history = [{"role": "user", "content": user_query}] total = 0 for it in range(1, 9): resp, rec = client.chat( role="planner" if it == 1 else "executor", messages=history, session_id=session_id, trace_id=trace_id, iteration=it, tool_calls=0, ) 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 或 HTML，Token 会爆炸。第 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", lines=True) 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", ascending=False) 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 关键派生指标

除了绝对值，这几个比率指标更能指导优化：

把这五个指标做成日报，成本异常基本能在发生当天定位到具体环节，而不是月底看账单猜。

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_id=f"bench_{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", index=False)

基线文件存好，优化后跑同一批任务，逐条对比。这样对比的是同一分布下的变化，不受流量结构影响。

4.2 优化动作一：工具返回裁剪

最常见的浪费源。工具返回的 JSON 里往往有大量 Agent 不需要的字段。加一个裁剪层：

def trim_tool_result(tool_name, raw_result, max_chars=800): if isinstance(raw_result, str): text = raw_result else: text = json.dumps(raw_result, ensure_ascii=False) 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_ascii=False)[: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_recent=3): 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 条任务，优化前后对比：

验证方式很直接：优化后重跑baseline.csv的同一批 task_id，生成after.csv，用 pandas merge 后算差值。如果某条任务 Token 反而上升，单独看它的 trace，通常是裁剪把关键信息截掉了导致多跑一轮，这时候要调整max_chars或保留字段列表。

4.6 用模型对话做单点验证

优化过程中如果怀疑某个模型 ID 的行为差异（比如 deepseek-chat 和 deepseek-reasoner 在规划任务上的输出长度差很多），可以到模型对话页面 https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite 手动发几条同样的 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 failed

local 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(f"unexpected 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_url=os.environ["TAOTOKEN_BASE_URL"], api_key=os.environ["TAOTOKEN_API_KEY"]) try: r = c.chat.completions.create( model="deepseek-chat", messages=[{"role": "user", "content": "ping"}], max_tokens=8, ) 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_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite 的额度方案，配合本文的计量脚本做预算预警。Key 管理和用量查看在控制台 https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite ，接入细节和字段说明在文档 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite ，API Keys 创建入口 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite 。

一个实用技巧收尾：把token_events.jsonl按天轮转，保留 90 天，用jq做临时查询比写脚本快。比如查今天 planner 角色的总输入 Token：

jq -s 'map(select(.role=="planner")) | map(.prompt_tokens) | add' \ token_events_2026-09-01.jsonl

这种随手可查的能力，比任何仪表盘都更容易让团队养成看 Token 的习惯。