聚合 DSH 模型调用次数，TaoToken 的 Input/Output 怎么分开统计-平芜编程栈

1. 从 pnpm 构建脚本报错和 cordis.patch.yml 说起：模型出口没切到 TaoToken，Token 数就只是一笔糊涂账

如果你在 DSH 里装完可观测插件后，pnpm v9提示依赖的 install 脚本被忽略，或者cordis.patch.yml里的模型适配器仍指向旧端点，那么会话轨迹里看到的 Token 用量往往只能说明“这次任务花了多少”，无法回答“哪个模型被调了多少次、Input 和 Output 各占多少”。TaoToken 官网入口先放这里：https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_content=dsh_cost_intro。去那里拿到 Key 后，把 DSH 的 Base URL 设为https://taotoken.net/api，再通过事件流与可观测插件重建 chat Span，才能按模型聚合调用次数，把 Input/Output 分列成报表，并产出 Token 消耗排行。

DeepSeek Harness 本身并不难用，难的是规模上来之后的成本归因。单会话、单机器的轨迹视图能看一轮任务用了多少 Token，但团队真正要管的是：昨天deepseek-chat被调了多少次，claude系列被调了多少次，输入 Token 和输出 Token 分别是多少，哪几个会话消耗最多，哪些重试在悄悄放大成本。这些问题的答案不在一个 turn 的总数里，而在每一次真实模型调用里。

DSH 的执行模型是 ReAct 循环：一次用户任务可以理解为一个 turn，turn 内部有若干 step，每个 step 可能触发模型推理，也可能触发工具调用。模型调用和工具调用混在同一条时间线上，如果只按 turn 或 step 统计，就会出现两个偏差：第一，把工具执行时间算进模型耗时；第二，把重试合并成一次调用，导致调用次数偏低、单次成本偏高。要解决这个问题，先要统一模型出口，再让每一次模型调用都带上可识别的模型名、Input/Output Token 和尝试序号。

TaoToken 在这里承担的是统一入口角色。DSH、Claude Code、Codex、CC Switch 都可以把 Base URL 指到https://taotoken.net/api，模型请求经过同一个聚合层后，返回的 usage 字段会包含prompt_tokens、completion_tokens、total_tokens。其中prompt_tokens就是 Input，completion_tokens就是 Output。后续所有报表，本质上都是把这两个值从响应里拆出来，再按模型、会话、时间窗口聚合。

2. 把 DSH 的模型出口切到 TaoToken：Key、Base URL 与 profile 配置

DSH 采用 Cordis 微内核加插件化装配，模型适配器、工具集、沙箱策略、会话持久化都在插件层。不同 profile 可以使用不同模型配置。要做成本统计，第一步不是先写报表，而是先让 DSH 的模型请求稳定地走 TaoToken。建议用环境变量保存 Key，不要把YOUR_API_KEY写进仓库。

# 终端环境，建议写入 ~/.bashrc 或 ~/.zshrc export DSH_HOME="$HOME/.dsh" export TAOTOKEN_API_KEY="YOUR_API_KEY" export OPENAI_API_KEY="$TAOTOKEN_API_KEY" export OPENAI_BASE_URL="https://taotoken.net/api"

然后在 DSH 的 profile 补丁文件里显式指定 OpenAI 兼容适配器。下面是一个示意配置，插件名与字段名请按你当前安装的适配器版本调整，核心是baseURL、apiKey、model和流式 usage 开关：

# ~/.dsh/profiles/default/cordis.patch.yml plugins: model-openai-compatible: baseURL: https://taotoken.net/api apiKey: ${TAOTOKEN_API_KEY} model: deepseek-chat stream: true streamOptions: includeUsage: true

如果你的 DSH 版本使用别的模型插件名，不要硬套上面的插件名。保持三件事正确即可：请求地址是https://taotoken.net/api，鉴权使用YOUR_API_KEY，流式响应开启 usage 返回。非流式请求通常会在响应体里直接带 usage；流式请求如果客户端不主动请求 usage，最后一个 chunk 可能没有 Token 数，报表就会缺行。

配置完成后，先用一次最小请求验证连通性。可以用 curl 在本地测试，不要连生产库，也不要在 Agent 里直接执行高风险命令：

curl https://taotoken.net/api/chat/completions \ -H "Authorization: Bearer YOUR_API_KEY" \ -H "Content-Type: application/json" \ -d '{ "model": "deepseek-chat", "messages": [{"role": "user", "content": "只回复 ok"}], "stream": false }'

返回体里如果能看到类似下面的结构，就说明 Input/Output 已经可拆：

{ "usage": { "prompt_tokens": 12, "completion_tokens": 3, "total_tokens": 15 } }

同一套口径可以延伸到其他编码工具。Claude Code 用settings.json或ANTHROPIC_*环境变量，注意不要把它套到 Codex 上：

{ "env": { "ANTHROPIC_BASE_URL": "https://taotoken.net/api", "ANTHROPIC_AUTH_TOKEN": "YOUR_API_KEY", "ANTHROPIC_MODEL": "YOUR_CLAUDE_MODEL_ID" } }

Codex 使用config.toml，鉴权字段和 Claude Code 不同：

model_provider = "taotoken" model = "YOUR_MODEL_ID" [model_providers.taotoken] name = "TaoToken" base_url = "https://taotoken.net/api" env_key = "TAOTOKEN_API_KEY"

如果使用 CC Switch 管理多套配置，三件套要填清楚，不要混用 Anthropic 与 OpenAI 字段：

供应商名称：TaoToken Base URL：https://taotoken.net/api API Key：YOUR_API_KEY

到这里，模型出口已经统一。接下来要解决的是：DSH 的事件流里，哪些记录算一次模型调用，哪些字段能拆出 Input/Output。

3. DSH 事件流里哪些字段能拆出 Input/Output：不要拿 turn 当模型调用

DSH 原生会把 Session 事件流以 zstd 压缩的 JSONL 落盘在$DSH_HOME/sessions/下。这个数据很完整，但它是按时间排列的事件序列，不是天然的成本报表。一个 turn 里可能有多个 step，一个 step 里可能有模型调用，也可能有工具调用；模型调用失败后还可能重试。如果只数 turn，就会漏掉重试；如果只数 step，就会把工具调用也算进去。

更稳的做法是把每一次真实模型请求还原成一条llm_spans记录。判断依据不是 turn 编号，而是这次事件是否真正向模型发起了请求，并且是否拿到了 usage。一次成功调用是一条；一次失败后重试，重试也是独立一条；流式响应如果被用户中断，只要服务端返回了 usage 或插件补发了 usage，也应该保留下来。这样统计调用次数时，COUNT(*)才接近真实请求次数。

可观测插件通常会把链路建成五层：entry / agent / step / chat / tool。其中 chat 层对应模型调用，tool 层对应工具执行。做模型成本报表时，只筛 chat 层，不筛 tool 层。chat 层里需要关注这些字段：

{ "type": "chat", "model": "deepseek-chat", "trace_id": "trace-xxx", "session_id": "session-xxx", "attempt": 1, "usage": { "prompt_tokens": 2310, "completion_tokens": 512, "total_tokens": 2822 }, "finish_reason": "stop", "status": "OK" }

不同插件版本字段名可能略有差异，例如input_tokens/output_tokens，或者嵌套在response.usage下。解析时建议做兼容映射：

prompt_tokens或input_tokens→ Input；
completion_tokens或output_tokens→ Output；
model或request.model→ 模型名；
attempt或dsh.llm.attempt→ 尝试序号；
status或status_code→ 成功、失败、中断；
finish_reason→ 结束原因，用于区分正常结束、长度截断、工具调用结束。

这里有一个容易忽略的点：工具调用的参数和返回结果也会进入上下文，下一轮模型调用的 Input Token 会因此变大。也就是说，Output Token 高通常意味着模型生成内容多，Input Token 高往往意味着上下文膨胀。两者分开统计后，才能判断成本到底花在“模型说了很多”还是“上下文带了很多”。

4. 本地落库：把 DSH 的 JSONL 事件流还原成 llm_spans 表

如果你暂时不想接完整可观测平台，先用本地 SQLite 也能快速产出报表。思路是：读取$DSH_HOME/sessions/下的 JSONL 或 zstd JSONL，筛出模型调用事件，把模型名、Input、Output、状态、会话 ID 写入 SQLite。下面脚本是可直接修改运行的示例，字段映射按你的事件结构调整：

import json import sqlite3 import zstandard as zstd from pathlib import Path DB = "dsh_tokens.db" ROOT = Path.home() / ".dsh" / "sessions" conn = sqlite3.connect(DB) conn.execute(""" CREATE TABLE IF NOT EXISTS llm_spans ( trace_id TEXT, session_id TEXT, span_id TEXT, model TEXT, input_tokens INTEGER, output_tokens INTEGER, status_code TEXT, start_time TEXT, end_time TEXT, attempt INTEGER, finish_reason TEXT ) """) def iter_lines(path: Path): if path.suffix == ".zst": with open(path, "rb") as f: reader = zstd.ZstdDecompressor().stream_reader(f) for line in reader: yield line else: with open(path, "r", encoding="utf-8") as f: for line in f: yield line def get_usage(ev): u = ev.get("usage") or {} if not u and isinstance(ev.get("response"), dict): u = ev["response"].get("usage") or {} input_tokens = int(u.get("prompt_tokens") or u.get("input_tokens") or 0) output_tokens = int(u.get("completion_tokens") or u.get("output_tokens") or 0) return input_tokens, output_tokens for path in ROOT.rglob("*.jsonl*"): for raw in iter_lines(path): try: ev = json.loads(raw) except Exception: continue event_type = ev.get("type") or ev.get("event") or "" if event_type not in ("chat", "llm", "model_call", "generation"): continue input_tokens, output_tokens = get_usage(ev) model = ev.get("model") or ev.get("request", {}).get("model") or "unknown" conn.execute( "INSERT INTO llm_spans VALUES (?,?,?,?,?,?,?,?,?,?,?)", ( ev.get("trace_id") or ev.get("traceId") or "", ev.get("session_id") or ev.get("sessionId") or path.stem, ev.get("span_id") or ev.get("spanId") or "", model, input_tokens, output_tokens, ev.get("status") or ev.get("status_code") or "OK", ev.get("start_time") or ev.get("startTime") or "", ev.get("end_time") or ev.get("endTime") or "", int(ev.get("attempt") or ev.get("dsh.llm.attempt") or 1), ev.get("finish_reason") or ev.get("finishReason") or "", ), ) conn.commit() conn.close() print("done:", DB)

运行前先安装依赖：

pip install zstandard python parse_dsh_tokens.py

如果数据量很大，不要每次全量重跑。可以记录已处理的文件 mtime，或者用trace_id + span_id做唯一索引，避免重复插入。SQL 和命令都在本地执行，不要把它接到生产数据库上。

5. 三张报表：模型调用次数、Input/Output 分列、Token 消耗排行

到了这一步，报表其实就是 SQL 聚合。第一张表看模型调用次数与 Input/Output 分列，这是成本统计的地基：

SELECT model, COUNT(*) AS calls, SUM(input_tokens) AS input_tokens, SUM(output_tokens) AS output_tokens, SUM(input_tokens + output_tokens) AS total_tokens, ROUND(AVG(input_tokens + output_tokens), 2) AS avg_tokens_per_call FROM llm_spans WHERE status_code = 'OK' GROUP BY model ORDER BY total_tokens DESC;

结果可以整理成这样：

model	calls	input_tokens	output_tokens	total_tokens	avg_tokens_per_call
deepseek-chat	128	456000	78000	534000	4171.88
claude-sonnet	42	210000	96000	306000	7285.71
其他	15	62000	18000	80000	5333.33

第二张表看 Token 消耗排行。可以按会话排，找出最贵的会话：

SELECT session_id, COUNT(*) AS calls, SUM(input_tokens) AS input_tokens, SUM(output_tokens) AS output_tokens, SUM(input_tokens + output_tokens) AS total_tokens FROM llm_spans GROUP BY session_id ORDER BY total_tokens DESC LIMIT 20;

第三张表看按天趋势，判断成本是否随着任务量线性增长：

SELECT substr(start_time, 1, 10) AS day, model, COUNT(*) AS calls, SUM(input_tokens) AS input_tokens, SUM(output_tokens) AS output_tokens, SUM(input_tokens + output_tokens) AS total_tokens FROM llm_spans GROUP BY day, model ORDER BY day DESC, total_tokens DESC;

如果只想看模型调用次数排行，不关心 Token：

SELECT model, COUNT(*) AS calls FROM llm_spans GROUP BY model ORDER BY calls DESC;

如果还要看失败与重试：

SELECT model, COUNT(*) AS total_calls, SUM(CASE WHEN status_code NOT IN ('OK', 'UNSET', '') THEN 1 ELSE 0 END) AS error_calls, SUM(CASE WHEN attempt > 1 THEN 1 ELSE 0 END) AS retry_calls FROM llm_spans GROUP BY model ORDER BY total_calls DESC;

当 Input 和 Output 分开后，你会发现很多“成本异常”并不是模型调用次数暴涨，而是输入上下文膨胀。比如工具返回了大量 JSON、日志或文件内容，被原样塞进下一轮请求，Input Token 就会快速上升。另一种情况是输出被截断后自动续写，Output Token 反复累加。只有分列报表才能把这两类问题区分开。

6. 接入可观测插件时的坑：pnpm v9、captureContent、批量上报与重试

如果你希望不自己维护解析脚本，可以接入 DSH 可观测插件，把运行时事件还原为结构化调用链。插件通常挂载在 DSH 的插件层，订阅生命周期事件，并在模型流式管道上旁路观察，不改业务代码。它最终会生成带父子关系的 Span，并批量上报。配置模型出口仍然走 TaoToken：https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_content=dsh_cost_plugin。

安装或更新插件后，先检查 DSH 和 Node 版本。版本不匹配时，插件可能能被加载，但事件字段缺失：

dsh --version node --version pnpm --version

pnpm v9 及以上默认会限制依赖包的 install 脚本。如果安装时看到Ignored build scripts或类似提示，可在对应 profile 目录处理一次：

cd ~/.dsh/profiles/default pnpm approve-builds # 或将相关依赖加入允许列表后执行 pnpm rebuild

headless / harness profile 同理：

cd ~/.dsh/profiles/headless pnpm approve-builds cd ~/.dsh/profiles/harness pnpm approve-builds

插件配置一般会涉及接收端 topic、批量大小、刷新间隔和内容捕获。默认情况下，prompts、responses、tool arguments/results 可能被附到 span 上，成本排障阶段很方便，但涉及敏感内容时建议关闭：

# ~/.dsh/profiles/default/cordis.patch.yml plugins: tencentcloud-agentobs-sdk-dsh: topicId: ${CLS_TOPIC_ID} captureContent: false batch: maxSize: 100 flushIntervalMs: 3000

显式插件配置通常优先于环境变量。访问凭证属于敏感信息，建议通过环境变量或密钥管理工具注入，不要把真实 SecretId、SecretKey 或 TaoToken Key 提交到代码仓库。模型侧 Key 用YOUR_API_KEY占位，接收端凭证单独管理，两者不要混用。

安装或更新插件后需要重启 DSH 服务。重启后发起一次测试任务，至少触发一次模型调用，然后在调用链视图里确认 chat Span 是否存在。检查项包括：

模型名是否与实际请求一致；
Input/Output Token 是否非零；
重试是否生成了独立 chat Span；
中断或失败是否带错误状态；
同一会话的多条 trace 是否通过 session id 关联。

如果报表里模型名是unknown，优先检查模型适配器是否把 model 字段传入事件；如果 Input/Output 都是 0，优先检查流式 usage 是否开启；如果调用次数明显低于实际请求数，优先检查是否把重试合并了。

7. 成本视角的告警与优化：Input/Output 分开后先看什么

有了按模型聚合的调用次数和 Input/Output 分列，成本优化就不再是拍脑袋。第一优先级看 Input Token 排行，因为输入通常占总 Token 的大头。重点排查三类会话：

工具结果超长：工具返回的 JSON、日志、目录列表没有裁剪，直接进入下一轮上下文；
多轮会话没有摘要：长会话每轮都携带全部历史，Input 随轮次线性增长；
系统提示与规则过长：每次调用都重复携带大量固定文本，调用次数越多浪费越大。

第二优先级看 Output Token 排行。输出高不一定是坏事，可能任务确实需要生成代码或长文档。但要看finish_reason，如果大量是长度截断，说明模型在反复续写，成本可能翻倍。第三优先级看重试。重试次数不一定要告警，但重试率对应模型侧失败或超时，应该和错误率一起看。

可以配置几类简单规则：

单会话总 Token 超过阈值时提醒；
某模型 Input Token 环比增长超过固定比例时提醒；
模型调用失败率或重试率超过阈值时提醒；
单次调用 Input Token 超过上下文预算时提醒。

这些规则不需要一开始就很复杂。先有按模型、按会话、按天的三张报表，再根据真实分布设阈值。DSH 的失败可能来自模型侧、工具侧或循环中断，调用链里每层 Span 有独立状态，后续可以把模型失败和工具失败分开告警，避免所有问题都算到模型成本上。

8. 总结与 CTA

DSH 原生的轨迹视图和事件流落盘，解决的是本机、单会话、实时调试问题。要把模型成本管起来，需要先统一模型出口：去 TaoToken 官网获取 Key，把 Base URL 设为https://taotoken.net/api，让每次模型调用都返回可拆分的 usage。然后用事件流或可观测插件把 chat 层单独抽出来，按每次真实模型调用落库，最终产出模型调用次数、Input/Output 分列报表和 Token 消耗排行。

这条链路的好处是可验证：调用次数来自 chat Span，Input 来自prompt_tokens，Output 来自completion_tokens，排行来自 SQL 聚合。它不依赖单会话视图，也不把工具耗时混进模型耗时。规模越大，这种结构化口径越重要。

如果你准备把 DSH 接到 TaoToken，可以从下面顺序开始：