ClickHouse 存 TaoToken 调用记录，分析千倍级增长

1. 从 Claude Code 的 404 到 ClickHouse 明细表TaoToken 调用元数据怎么定你在 Claude Code 里把ANTHROPIC_BASE_URL写成https://taotoken.net/api/v1结果请求直接 404或者在 Codex 的config.toml里误填了ANTHROPIC_*provider 一直不生效。这类配置问题最后都会落到同一个盲区TaoToken 调用记录没有统一进仓。建议先去 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_contentclickhouse_intro 获取 Key再把 Base URL 设为https://taotoken.net/api。只要调用链路统一后面才有资格谈趋势分析。近期公开报道在讨论 Token 调用量与数据服务业用电量增长作为数据分析工程师我更关心的是这些调用到底以什么粒度被记录下来。如果每次请求只在日志里留一行文本你很难回答“哪个模型在涨”“哪类 Key 增长最快”“错误率是否同步上升”。ClickHouse 适合做这件事因为它对高吞吐写入、宽表明细、聚合查询和 TTL 生命周期都比较友好而且 SQL 表达能力强物化视图可以把小时级聚合提前算好。本文不把 TaoToken 当成一个黑盒。我们把 TaoToken 调用记录拆成可观测字段用 ClickHouse 建一张token_usage_detail明细表再写调用趋势查询。所有 SQL 建议在本地 ClickHouse 或开发集群执行不要让 Agent/MCP 直连生产库生产环境变更走审批和变更窗口。先定义字段。一次模型调用至少应记录时间event_time精确到毫秒统一 UTC。标识request_id、trace_id、api_key_id。来源source_platform、app_name、endpoint。模型model用于区分不同模型或同一模型的不同版本。用量prompt_tokens、completion_tokens、total_tokens。性能latency_ms、first_token_ms。状态status_code、error_code。成本辅助cached_tokens、request_bytes、response_bytes。业务维度is_stream、country、province、metadata。如果你的来源平台是csdn_ugc就把source_platform写成csdn_ugc。后续所有趋势查询都可以按这个字段切分避免把不同平台、不同应用、不同 Key 的调用混在一起。2. 建表 SQL按天分区、按来源平台和模型排序的 token_usage_detail下面 SQL 在本地 ClickHouse 执行。建议先建数据库再建明细表。分区按月排序键把日期、来源平台、模型、时间和请求 ID 放在前面兼顾范围扫描和去重排查。CREATE DATABASE IF NOT EXISTS ai_ops; CREATE TABLE IF NOT EXISTS ai_ops.token_usage_detail ( event_time DateTime64(3, UTC), event_date Date DEFAULT toDate(event_time), request_id String, trace_id String DEFAULT , api_key_id String DEFAULT , source_platform LowCardinality(String), app_name LowCardinality(String) DEFAULT , model LowCardinality(String), endpoint LowCardinality(String) DEFAULT /v1/chat/completions, status_code UInt16, error_code LowCardinality(String) DEFAULT , prompt_tokens UInt32 DEFAULT 0, completion_tokens UInt32 DEFAULT 0, total_tokens UInt32 MATERIALIZED prompt_tokens completion_tokens, cached_tokens UInt32 DEFAULT 0, latency_ms UInt32 DEFAULT 0, first_token_ms UInt32 DEFAULT 0, request_bytes UInt32 DEFAULT 0, response_bytes UInt32 DEFAULT 0, is_stream UInt8 DEFAULT 0, client_ip_hash UInt64 DEFAULT 0, country LowCardinality(String) DEFAULT , province LowCardinality(String) DEFAULT , metadata String DEFAULT {} ) ENGINE MergeTree PARTITION BY toYYYYMM(event_date) ORDER BY (event_date, source_platform, model, event_time, request_id) TTL event_date INTERVAL 180 DAY SETTINGS index_granularity 8192;几个设计点值得说明。第一total_tokens用MATERIALIZED自动计算写入时不用传避免 prompt token 和 completion token 已写、total 却忘记更新。第二source_platform、model、app_name用LowCardinality(String)在基数可控时能降低存储并提升聚合效率。第三ORDER BY把event_date放最前日常查询大多带时间范围。第四TTL 设为 180 天如果合规要求更长可以改成 365 天或把冷数据下沉到对象存储。第五client_ip_hash只存哈希不存明文 IP避免把分析表变成隐私风险点。如果数据量继续增长可以再加投影或聚合表。但在明细表阶段先把字段和写入链路稳定下来比过早优化更重要。3. 写入链路TaoToken 返回 usage 后如何批量落 ClickHouse调用记录可以来自客户端埋点、网关层或本地代理但核心原则一致请求完成后把 usage、延迟、状态码和业务维度一起写入。下面示例用 Python 调用 TaoToken 兼容接口再把记录插入本地 ClickHouse。Key 从环境变量读取不硬编码。import os import time import uuid from datetime import datetime, timezone import requests import clickhouse_connect API_KEY os.environ[TAOTOKEN_API_KEY] BASE_URL https://taotoken.net/api CH_CLIENT clickhouse_connect.get_client( hostlocalhost, port8123, usernamedefault, password, databaseai_ops, ) def record_chat(model: str, messages: list): request_id str(uuid.uuid4()) started time.time() resp requests.post( f{BASE_URL}/v1/chat/completions, headers{ Authorization: fBearer {API_KEY}, Content-Type: application/json, }, json{ model: model, messages: messages, stream: False, }, timeout60, ) latency_ms int((time.time() - started) * 1000) body resp.json() usage body.get(usage, {}) event_time datetime.now(timezone.utc).strftime(%Y-%m-%d %H:%M:%S.%f)[:-3] row [ event_time, request_id, csdn_ugc, demo_app, model, resp.status_code, usage.get(prompt_tokens, 0), usage.get(completion_tokens, 0), latency_ms, 0, 0, len(resp.content), ] CH_CLIENT.insert( token_usage_detail, [row], column_names[ event_time, request_id, source_platform, app_name, model, status_code, prompt_tokens, completion_tokens, latency_ms, first_token_ms, is_stream, response_bytes, ], ) return body这段代码故意没有插入total_tokens因为它在表里是物化列。生产环境不建议逐行 insert可以改成批量缓冲例如每 500 行或每 2 秒 flush 一次。对于流式响应first_token_ms应在收到第一个 chunk 时记录response_bytes累加 chunk 长度。不要把完整 prompt 或 completion 原文写进 ClickHouse除非有明确合规授权分析表只存元数据即可。如果你是通过网关统一转发也可以在网关层记录。网关比客户端更可靠因为它能覆盖 Claude Code、Codex、脚本和 Web 调用。只要最终写入字段一致ClickHouse 并不关心中间是 Python、Go 还是 Nginx 插件。4. 调用趋势查询小时、日、错误率与 P95 延迟一起看最常用的查询是按小时看调用量、token 量和错误率。下面查询只看最近 7 天并按来源平台等于csdn_ugc过滤。你可以把时间范围改成 30 天把分组改成天。SELECT toStartOfHour(event_time) AS ts, source_platform, count() AS calls, sum(total_tokens) AS tokens, round(avg(latency_ms), 1) AS avg_latency_ms, quantile(0.95)(latency_ms) AS p95_latency_ms, round(countIf(status_code 400) / count(), 4) AS error_rate FROM ai_ops.token_usage_detail WHERE event_time now() - INTERVAL 7 DAY AND source_platform csdn_ugc GROUP BY ts, source_platform ORDER BY ts ASC;这条 SQL 能回答几个问题调用量是否在特定小时突然抬升token 增长是否快于调用次数增长延迟 P95 是否恶化错误率是否和流量峰值同步。很多“调用量增长”其实不是请求数增长而是单次请求 token 变多。把calls和tokens放在同一张结果里能避免误判。按天看趋势可以这样写SELECT event_date, source_platform, count() AS calls, sum(total_tokens) AS tokens, round(sum(total_tokens) / nullIf(count(), 0), 1) AS avg_tokens_per_call, round(avg(latency_ms), 1) AS avg_latency_ms, round(countIf(status_code 400) / count(), 4) AS error_rate FROM ai_ops.token_usage_detail WHERE event_date today() - 30 GROUP BY event_date, source_platform ORDER BY event_date ASC, source_platform;如果只看总量可能会被大模型、长上下文或某个批量任务拉高。继续按模型拆SELECT event_date, model, count() AS calls, sum(total_tokens) AS tokens, round(sum(total_tokens) / nullIf(count(), 0), 1) AS avg_tokens_per_call FROM ai_ops.token_usage_detail WHERE event_date today() - 30 AND source_platform csdn_ugc GROUP BY event_date, model ORDER BY event_date ASC, tokens DESC;这类查询建议都带时间范围。ClickHouse 虽然快但全表扫描没有意义。若你发现小时查询频繁可以提前建物化视图。5. 增长归因模型、Key、来源平台三维拆解与物化视图调用量出现数量级变化时不要只看总量。至少拆三个维度模型、API Key、来源平台。模型回答“是不是换了更贵的模型”Key 回答“是不是某个业务方在放量”来源平台回答“是不是某个渠道突然接入”。按 Key 拆SELECT api_key_id, model, count() AS calls, sum(total_tokens) AS tokens, round(avg(latency_ms), 1) AS avg_latency_ms, round(countIf(status_code 400) / count(), 4) AS error_rate FROM ai_ops.token_usage_detail WHERE event_date today() - 7 GROUP BY api_key_id, model ORDER BY tokens DESC LIMIT 50;按来源平台和模型拆SELECT source_platform, model, count() AS calls, sum(total_tokens) AS tokens, uniqExact(request_id) AS unique_requests, round(countIf(status_code 400) / count(), 4) AS error_rate FROM ai_ops.token_usage_detail WHERE event_date today() - 30 GROUP BY source_platform, model ORDER BY tokens DESC LIMIT 100;如果想看周环比可以用窗口函数或neighbor。下面用聚合后的日表计算 7 天前对比注意nullIf防止除零WITH daily AS ( SELECT event_date, model, source_platform, sum(total_tokens) AS tokens, count() AS calls FROM ai_ops.token_usage_detail WHERE event_date today() - 60 GROUP BY event_date, model, source_platform ) SELECT event_date, model, source_platform, tokens, calls, round(tokens / nullIf(neighbor(tokens, 7), 0), 2) AS wow_ratio FROM daily ORDER BY event_date DESC, tokens DESC LIMIT 100;neighbor依赖排序生产查询建议先按维度排序。更稳妥的方式是建小时或天级聚合表然后用物化视图维护。CREATE TABLE IF NOT EXISTS ai_ops.token_usage_hourly ( hour DateTime, source_platform LowCardinality(String), model LowCardinality(String), status_code UInt16, requests UInt64, total_tokens UInt64, prompt_tokens UInt64, completion_tokens UInt64, latency_ms_sum UInt64, latency_ms_max UInt32, error_requests UInt64 ) ENGINE SummingMergeTree PARTITION BY toYYYYMM(hour) ORDER BY (hour, source_platform, model, status_code); CREATE MATERIALIZED VIEW IF NOT EXISTS ai_ops.mv_token_usage_hourly TO ai_ops.token_usage_hourly AS SELECT toStartOfHour(event_time) AS hour, source_platform, model, status_code, count() AS requests, sum(total_tokens) AS total_tokens, sum(prompt_tokens) AS prompt_tokens, sum(completion_tokens) AS completion_tokens, sum(latency_ms) AS latency_ms_sum, max(latency_ms) AS latency_ms_max, countIf(status_code 400) AS error_requests FROM ai_ops.token_usage_detail GROUP BY hour, source_platform, model, status_code;查询时记得用sum合并SELECT hour, source_platform, sum(requests) AS calls, sum(total_tokens) AS tokens, round(sum(latency_ms_sum) / nullIf(sum(requests), 0), 1) AS avg_latency_ms, max(latency_ms_max) AS max_latency_ms, round(sum(error_requests) / nullIf(sum(requests), 0), 4) AS error_rate FROM ai_ops.token_usage_hourly WHERE hour now() - INTERVAL 7 DAY GROUP BY hour, source_platform ORDER BY hour ASC;这样趋势看板不需要每次扫明细表响应会更稳。明细表继续保留用于排障、审计和重新聚合。6. 配置排障Claude Code、Codex、CC Switch 的 Base URL 与 Key 写法分析调用记录之前先保证调用本身配置正确。TaoToken 的 Base URL 是https://taotoken.net/api。注意它不加 UTMUTM 只用于官网跳转统计。Key 使用占位符YOUR_API_KEY实际使用时替换成你在官网创建的值。Claude Code 用settings.json核心是ANTHROPIC_*系列变量。示例{ env: { ANTHROPIC_BASE_URL: https://taotoken.net/api, ANTHROPIC_AUTH_TOKEN: YOUR_API_KEY, ANTHROPIC_MODEL: claude-sonnet-4-5 } }把ANTHROPIC_MODEL替换成你在 TaoToken 模型列表中实际可用的模型 ID。如果 Claude Code 报 404优先检查 Base URL 是否误加了/v1或尾部斜杠如果报鉴权错误检查ANTHROPIC_AUTH_TOKEN是否用了 Key而不是官网登录态。Codex 用config.toml不要套用ANTHROPIC_*。示例model_provider taotoken model your-model-id [model_providers.taotoken] name TaoToken base_url https://taotoken.net/api env_key TAOTOKEN_API_KEY wire_api chat环境变量单独设置export TAOTOKEN_API_KEYYOUR_API_KEYCodex 的env_key指向TAOTOKEN_API_KEY不是ANTHROPIC_AUTH_TOKEN。把 Claude Code 的环境变量复制到 Codex通常会出现 provider 不识别或鉴权失败。CC Switch 可以理解成三件套配置供应商名称、Base URL、API Key。新增供应商时填供应商名称TaoTokenBase URLhttps://taotoken.net/apiAPI KeyYOUR_API_KEY然后在 CC Switch 里切换 Claude Code 配置。切换后建议重启终端或相关进程让环境变量重新加载。如果你在多个工具之间来回切最好把每个工具的配置分开维护Claude Code 管settings.jsonCodex 管config.tomlCC Switch 管供应商切换。不要混写。如果你还没有 Key可以直接去 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_contentclickhouse_setup 进入官网在控制台创建 API Key。创建后先跑一条最小请求确认status_code是 200再看 ClickHouse 里是否出现记录。7. 容量与成本TTL、冷热分层、采样表让 ClickHouse 扛住增长当调用量持续增长ClickHouse 表会从 GB 级走向 TB 级。数据分析工程师不能只写查询还要提前设计生命周期。几个实用策略第一TTL 分层。明细表 180 天聚合表 365 天。明细表过期后趋势分析仍可用小时表或天表。示例ALTER TABLE ai_ops.token_usage_detail MODIFY TTL event_date INTERVAL 180 DAY; ALTER TABLE ai_ops.token_usage_hourly MODIFY TTL hour INTERVAL 365 DAY;第二冷热分离。如果本地磁盘有限可以把老分区移动到对象存储或冷盘。ClickHouse 支持存储策略但生产变更前先在测试环境验证。不要直接对生产表执行未经演练的ALTER。第三采样表。对于超大量的探索性查询可以建采样表例如每 10 条采 1 条用于快速看趋势。但成本核算、错误率统计仍应走全量聚合表。采样表不能替代明细。第四控制字段基数。不要把完整 URL、完整 user-agent、完整 prompt 放进LowCardinality或排序键。高基数字符串会拖慢合并和查询。metadata只放必要 JSON查询时用JSONExtractString按需取字段。第五写入批量化。逐行 insert 会产生大量小 part后台合并压力大。客户端可以每 500 到 5000 行批量写入或者用 ClickHouse 的异步插入。批量大小根据延迟要求调整。第六监控 part 数量和合并队列。调用量增长时写入吞吐和查询延迟会同时变化。可以定期查询系统表SELECT database, table, sum(rows) AS rows, formatReadableSize(sum(bytes_on_disk)) AS disk_size, count() AS parts FROM system.parts WHERE active GROUP BY database, table ORDER BY sum(bytes_on_disk) DESC;这类查询也在本地或只读副本执行。生产库上的系统表查询要评估频率。8. 从模型对话到 Coding Plan把调用记录闭环回 TaoToken 控制台ClickHouse 分析不是终点。调用趋势最终要反哺三件事模型选择、Key 管理、成本控制。你可以在 ClickHouse 里发现某个模型的平均 token 消耗突然升高然后回到 TaoToken 控制台检查 Key 和用量也可以发现某类错误集中出现然后调整客户端超时和重试策略。如果你还没开始接入建议按这个顺序走先体验模型对话确认模型输出和延迟符合预期https://taotoken.net/models/detail/chat?utm_sourcetaotoken_aicg_blog_endutm_contentclickhouse_chat如果 Coding 场景用量稳定查看 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentclickhouse_plan在控制台创建 API Key把YOUR_API_KEY替换成真实值https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentclickhouse_keys配置 Claude Code 时对照官方文档避免 Base URL 和鉴权字段写错https://taotoken.net/doc/ClaudeCodeAnthropic?utm_sourcetaotoken_aicg_blog_endutm_contentclickhouse_cc_doc官网入口也放在这里方便统一获取 Key 和查看控制台https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_contentclickhouse_cta回到 ClickHouse 侧建议你至少保留三个查询小时级调用趋势、按模型 token 占比、按 Key 错误率。把这三张结果接进看板再设置阈值告警。这样当调用量出现新的数量级变化时你不是凭感觉判断而是能从token_usage_detail和token_usage_hourly里定位到具体来源平台、具体模型和具体 Key。对数据分析工程师来说这比只看到一个总量数字更有价值。