给 3.8 Live / 3.8 Live Extended Thinking 这类语音智能体记账，第一步不是加埋点，而是把出口收拢：先到 TaoToken 官网（https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_content=intro_voice_metering）取一个 Key，再把所有客户端的 Base URL 指向 https://taotoken.net/api。语音场景的成本黑洞从来不是单价，而是一屋子长连接：会话是流式的，turn 是交错的，思考块是隐式的，断线重连后你甚至不知道该把哪一段算给谁。作为可观测性工程师，我的做法很朴素——先把计量下沉到「turn 事件」这一层，再统一字段、统一指标、统一看板。下面这套方案已经在近实时语音对话模型的生产链路上跑通，包含可直接复制的日志字段规范、Token 指标采集代码和看板查询语句。

1. 为什么 3.8 Live Extended Thinking 的账最难记

近实时语音对话模型的计量难点，和传统文本 API 完全不是一个量级。文本请求是「一问一答一次结算」，语音会话是「一条长连接里跑了几十个 turn」。

具体来说，有五个坑：

第一，三类 Token 混在同一条流里。 输入侧有音频 token 和文本 token（系统提示、工具描述、历史转写），输出侧有文本 token、音频 token，再加上 Extended Thinking 产生的思考 token。它们在计费口径上属于不同模态，如果只看一个总数，你永远说不清成本涨在哪。

第二，usage 只在终结帧出现。 流式响应里，增量帧携带的是内容，不携带用量。用量对象通常在流的最后一个 chunk 才补齐。一旦客户端在收到终结帧前就断开，这次推理的 usage 就直接丢了，账单上却照样有消耗。

第三，turn 边界由语音活动决定，不由 HTTP 请求决定。 用户插话（barge-in）、静音超时、模型主动追问，都会切出一个新的 turn。如果按「一次 WebSocket 连接 = 一次调用」来记账，粒度太粗，做不了归因。

第四，重连与重试会造成重复计数。 移动端切网、网关超时重连，同一个 turn 可能被发两次。没有幂等键，看板上的 token 会凭空翻倍。

第五，客户端口径不一致。 Web 端、移动端、服务端 SDK 各自埋点，有的上报 total_tokens，有的只上报音频时长。最后汇总时你只能靠猜。

结论只有一句：记账的基本单位必须是 turn，而不是请求，也不是连接。

2. 统一出口：在 TaoToken 取 Key，把 Base URL 收成一条线

可观测性的前提是可归因。如果每个业务线各自持有不同来源的 Key，你就无法回答「这个月的语音成本谁占的」。我这边做的是把出口统一到 TaoToken。

操作路径很直接：先访问 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_content=key_setup ，在控制台创建 API Key，然后把所有客户端的 Base URL 收敛为：

https://taotoken.net/api

这个地址会贯穿后面所有配置：Claude Code 用 ANTHROPIC_BASE_URL 指向它，Codex 用 model_providers.*.base_url 指向它，自研的语音网关也是一样，只是协议头不同。

几个容易被忽略的工程细节：

1. Key 命名即成本中心。 建议按 voice-agent-<env>-<region> 命名，例如 voice-agent-prod-cn、voice-agent-staging-cn。这样在用量报表里可以直接按 Key 维度做归属，不用事后猜。
2. 不要把 Key 下发到端侧。 语音智能体常见架构是端侧采集音频、服务端扇出推理。Key 只放在服务端或边缘网关，前端拿到的应该是一个短时效的会话凭证。
3. 按环境分 Key。 灰度环境和生产环境用不同 Key，否则压测流量会污染成本曲线，看板上的突刺全部变成噪声。
4. Key 只当凭证，不当路由。 具体走哪个模型由请求体里的模型 ID 决定，不要在 Key 上做模型绑定，否则后期换模型要重新发一遍 Key。

3. 客户端配置：Claude Code / Codex / CC Switch 三件套

同一套 Key 和 Base URL，不同客户端的写法差别很大，尤其是协议层。这里必须强调一句：ANTHROPIC_ 系列环境变量只属于 Claude Code 这一类走 Anthropic 协议的客户端，不要套到 Codex 上。

3.1 Claude Code：settings.json + ANTHROPIC_*

Claude Code 读取 ~/.claude/settings.json，把供应商信息放在 env 段里即可：

{
  "env": {
    "ANTHROPIC_BASE_URL": "https://taotoken.net/api",
    "ANTHROPIC_AUTH_TOKEN": "YOUR_API_KEY",
    "ANTHROPIC_MODEL": "<TaoToken 控制台中可用的模型 ID>",
    "ANTHROPIC_SMALL_FAST_MODEL": "<TaoToken 控制台中可用的轻量模型 ID>",
    "CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC": "1"
  },
  "includeCoAuthoredBy": false
}

如果你更习惯用 shell 环境变量，等价写法是：

export ANTHROPIC_BASE_URL="https://taotoken.net/api"
export ANTHROPIC_AUTH_TOKEN="YOUR_API_KEY"
export ANTHROPIC_MODEL="<模型 ID>"

注意 ANTHROPIC_AUTH_TOKEN 对应的是 Anthropic 风格的鉴权头，Claude Code 会自己组装，你不需要手写 x-api-key。模型 ID 请以控制台实际展示为准，不要照抄别人博客里的名字——模型清单是会变的，写死一个过期 ID 的典型症状就是启动即 404。

3.2 Codex：config.toml

Codex 走的是另一套协议，配置文件在 ~/.codex/config.toml：

model = "<模型 ID>"
model_provider = "taotoken"
approval_policy = "on-request"

[model_providers.taotoken]
name = "TaoToken"
base_url = "https://taotoken.net/api/v1"
env_key = "TAOTOKEN_API_KEY"
wire_api = "responses"

[history]
persistence = "save-all"

对应环境变量只有一个：

export TAOTOKEN_API_KEY="YOUR_API_KEY"

这里有两个细节值得记一笔。其一，OpenAI 兼容风格的客户端一般要求 base_url 带上 /v1，而 Anthropic 风格客户端通常不需要；两者指向的是同一个服务，只是路径拼接习惯不同，具体以官方文档给出的写法为准。其二，env_key 只是告诉 Codex 去哪个环境变量里读 Key，它不会帮你把 Key 写进配置文件，这正好符合「凭证不落盘 Git」的要求。

3.3 CC Switch 三件套：让切换不靠手改文件

多环境、多供应商来回切的时候，手工改配置文件迟早出错。我这边用 CC Switch 管三份东西，把它们当成一套「三件套」同步维护：

{
  "providers": [
    {
      "id": "taotoken-voice-prod",
      "name": "TaoToken / voice-prod",
      "baseUrl": "https://taotoken.net/api",
      "authToken": "YOUR_API_KEY",
      "model": "<模型 ID>"
    },
    {
      "id": "taotoken-voice-staging",
      "name": "TaoToken / voice-staging",
      "baseUrl": "https://taotoken.net/api",
      "authToken": "YOUR_API_KEY",
      "model": "<模型 ID>"
    }
  ],
  "current": "taotoken-voice-prod"
}

三件套分别是：供应商条目（上面的 JSON，字段名以你本机 CC Switch 版本为准）、Claude Code 的 settings.json（由切换动作写入，保证 ANTHROPIC_* 与当前供应商一致）、项目根目录的 .env（给自研采集脚本和本地调试用）。切换完成后，用一条命令验一下当前生效值：

env | grep -E 'ANTHROPIC_BASE_URL|ANTHROPIC_AUTH_TOKEN' | sed 's/\(TOKEN=\).*/\1***/'

打印时务必把 Key 打码，日志里出现完整 Key 是安全事故，不是小疏忽。

4. 日志字段规范：把一次 turn 变成一行结构化事件

指标是从日志里长出来的，所以字段规范必须先行。我采用「一个 turn 一行事件 + 多行用量明细」的写法：事件行负责描述上下文，明细行负责描述 token。

事件类型枚举控制得很死，只有六个：

session_start / turn_start / turn_first_token / turn_end / tool_call / session_end

事件行的 JSON 长这样：

{
  "event_type": "turn_end",
  "ts": "2026-01-15T10:22:31.482Z",
  "session_id": "sess_7f3a91",
  "turn_id": "sess_7f3a91_t018",
  "tenant": "voice-prod",
  "model": "<模型 ID>",
  "api_key_alias": "voice-agent-prod-cn",
  "transport": "websocket",
  "thinking_enabled": true,
  "thinking_budget_tokens": 4096,
  "usage_missing": false,
  "ttfb_ms": 412,
  "duration_ms": 3180,
  "audio_in_ms": 2600,
  "audio_out_ms": 2900,
  "retry_count": 0,
  "request_fingerprint": "sha1:9c1b0d5e..."
}

用量明细行则是拆开后的模态粒度：

{
  "event_type": "turn_usage",
  "session_id": "sess_7f3a91",
  "turn_id": "sess_7f3a91_t018",
  "model": "<模型 ID>",
  "direction": "output",
  "modality": "thinking",
  "tokens": 1180
}

字段设计上我坚持四条规则：

• turn_id 必须是单调递增的服务端序号，不能是时间戳。 时间戳在并发插话时可能撞车，撞车就意味着去重失败。
• 音频只记时长和 token 数，绝不记原始音频与转写文本。 计量不需要内容，把内容写进日志等于给自己造一个合规雷区。
• usage_missing 是显式字段，不是靠字段缺失推断。 采集器发现终结帧没到，就主动打上这个标记，看板才能单独统计「丢账率」。
• api_key_alias 只记别名不记 Key 本身。 别名到真实 Key 的映射放在密钥管理系统里。

5. Token 指标采集：从流式响应里抠出 usage

采集器要解决三件事：把 usage 拆成模态、给重连去重、把结果变成指标。

下面是核心的数据结构，思路是把原始 usage 对象「拍平」成一张明细表：

import hashlib
import time
from dataclasses import dataclass, asdict, field

MODALITY_MAP = {
    "input_audio_tokens":  ("input",  "audio"),
    "input_text_tokens":   ("input",  "text"),
    "cached_input_tokens": ("input",  "cached"),
    "output_text_tokens":  ("output", "text"),
    "output_audio_tokens": ("output", "audio"),
    "thinking_tokens":     ("output", "thinking"),
}

@dataclass
class TurnUsage:
    session_id: str
    turn_id: str
    model: str
    tenant: str = "voice-prod"
    ts: float = field(default_factory=time.time)
    ttfb_ms: int = 0
    duration_ms: int = 0
    retry_count: int = 0
    usage_missing: bool = False
    raw_usage: dict = field(default_factory=dict)

    def fingerprint(self) -> str:
        raw = f"{self.session_id}|{self.turn_id}|{self.retry_count}"
        return "sha1:" + hashlib.sha1(raw.encode("utf-8")).hexdigest()

    def flatten(self):
        rows = []
        for key, (direction, modality) in MODALITY_MAP.items():
            value = int(self.raw_usage.get(key) or 0)
            if value <= 0:
                continue
            rows.append({
                "ts": self.ts,
                "session_id": self.session_id,
                "turn_id": self.turn_id,
                "model": self.model,
                "tenant": self.tenant,
                "direction": direction,
                "modality": modality,
                "tokens": value,
                "ttfb_ms": self.ttfb_ms,
                "duration_ms": self.duration_ms,
                "fingerprint": self.fingerprint(),
            })
        return rows

在流式回调里，把它挂到「终结帧」上：

def on_stream_chunk(chunk, state: TurnUsage):
    if state.ttfb_ms == 0 and chunk_has_content(chunk):
        state.ttfb_ms = int((time.time() - state.start_at) * 1000)

    # usage 通常在最后一个 chunk 才补齐，中途不要急着落库
    if getattr(chunk, "usage", None):
        state.raw_usage.update(chunk.usage)

    if getattr(chunk, "done", False):
        state.duration_ms = int((time.time() - state.start_at) * 1000)
        if not state.raw_usage:
            state.usage_missing = True     # 连接中途断开，账必须留痕
        emit_rows(state.flatten())

有了明细行，指标层就只是聚合。我用的命名约定是：

llm_tokens_total{direction, modality, model, tenant}      # counter
llm_turn_duration_ms_bucket{model, tenant}                 # histogram
llm_ttfb_ms_bucket{model, tenant}                          # histogram
llm_usage_missing_total{model, tenant}                     # counter

llm_usage_missing_total 是最容易被跳过、但最有价值的一个指标：它直接告诉你「有多少次推理的账丢了」。这个数字只要不是零，就说明网关或客户端存在提前断开的问题。

6. 看板查询语句：三套查询覆盖 90% 的问题

看板不需要花哨，能回答四个问题就够：谁在烧 token、烧在哪个模态、延迟是否影响体验、有没有丢账。

问题一：整体输出速率与模态占比（PromQL）

# 每秒输出 token 速率，按模态拆分
sum by (modality) (
  rate(llm_tokens_total{direction="output"}[5m])
)

# Extended Thinking 在输出中的占比
sum(rate(llm_tokens_total{direction="output", modality="thinking"}[15m]))
/
sum(rate(llm_tokens_total{direction="output"}[15m]))

第二条比值一旦持续走高，通常意味着任务变复杂或思考预算给大了，这时候调 thinking_budget_tokens 比换模型更划算。

问题二：找出最贵的会话（ClickHouse，本地可跑）

SELECT
    session_id,
    countDistinct(turn_id)                      AS turns,
    sum(tokens)                                 AS total_tokens,
    sumIf(tokens, modality = 'thinking')        AS thinking_tokens,
    sumIf(tokens, modality = 'audio')           AS audio_tokens,
    round(sum(tokens * price_per_1k) / 1000, 4) AS cost,
    max(ttfb_ms)                                AS worst_ttfb_ms
FROM llm.usage_events
WHERE ts >= now() - INTERVAL 24 HOUR
  AND tenant = 'voice-prod'
GROUP BY session_id
ORDER BY cost DESC
LIMIT 20;

这张表是我每周必看的一页。语音场景的分布极不均匀，往往是少数几个长会话贡献了大部分成本，而这些会话多半来自同一个测试账号或一段被循环播放的音频。

问题三：按分钟看趋势与缓存命中

SELECT
    toStartOfMinute(ts)                                       AS minute,
    model,
    sumIf(tokens, direction = 'input'  AND modality = 'audio') AS audio_in,
    sumIf(tokens, direction = 'input'  AND modality = 'cached') AS cached_in,
    sumIf(tokens, direction = 'input')                         AS input_total,
    sumIf(tokens, direction = 'output' AND modality = 'thinking') AS thinking_out,
    uniqExact(session_id)                                      AS sessions
FROM llm.usage_events
WHERE ts >= now() - INTERVAL 6 HOUR
GROUP BY minute, model
ORDER BY minute DESC
LIMIT 200;

其中 cached_in / input_total 就是缓存命中率。它掉下来的时候，通常是系统提示词被改动过，导致前缀不再稳定——这是一个典型的「改一行文案、成本涨一截」的案例。

问题四：丢账排查（LogQL）

{app="voice-agent"} | json
  | event_type = "turn_end"
  | usage_missing = "true"
  | line_format "{{.session_id}} {{.model}} {{.ttfb_ms}}ms"

按 session_id 聚一下，如果集中在某个客户端版本，基本可以确定是端侧提前断开，而不是服务端问题。

7. 排障清单：账对不上时按顺序查这八条

1. 401 / 403。 先确认协议：Anthropic 风格走 ANTHROPIC_AUTH_TOKEN，OpenAI 兼容风格走环境变量里的 env_key。把 Anthropic 的变量名套到 Codex 上是最高频的错误。
2. 404 模型不存在。 模型 ID 写的是别的平台的名字。回到控制台复制当前可用的 ID。
3. usage 全为空。 检查采集器是否在终结帧之前就关闭了流，或者用了非流式回落路径而没接 usage。
4. token 数是实际的两倍左右。 大概率是重连导致的重复上报。用 fingerprint 做幂等去重，落库时按主键覆盖。
5. thinking token 看起来是零。 确认 thinking_enabled 是否真的打开了，以及采集器有没有把思考 token 单独映射成 modality=thinking。
6. 音频 token 与时长比例异常。 检查是不是有静音段被持续送入。语音活动检测的阈值太灵敏，会悄悄抬高输入音频 token。
7. 延迟毛刺。 看 llm_ttfb_ms_bucket 的 p95，而不是平均值。语音体验的劣化几乎总是先体现在尾部延迟上。
8. 成本曲线突变。 先看缓存命中率，再看思考 token 占比，最后才怀疑单价。绝大多数突变来自提示词前缀变化，而不是计费调整。

8. 落地顺序建议与下一步

如果你今天就要开始做，建议按这个顺序推进：先把所有客户端的 Base URL 收敛到 https://taotoken.net/api 并统一 Key 命名；再补 turn_id 与终结帧 usage 的采集；然后是明细表与四个指标；最后才做看板。顺序反过来的话，你会得到一堆漂亮但无法归因的图表。

需要动手的时候，可以从这几个入口进去：

• 先和模型对话，确认 Key 与 Base URL 是否生效：模型对话
• 需要长期跑语音智能体与编码任务，看套餐规划：Coding Plan
• 正式环境按环境拆分凭证，先建 Key：API Keys
• 配置 Claude Code 的完整字段说明：Claude Code 文档

记账这件事的收益不会立刻出现在仪表盘上，但当某天语音成本突然翻倍、而你能在五分钟内定位到是哪个会话、哪个模态、哪次提示词改动造成的，你就会明白这套字段规范值回票价。