1. 从 CronJob 的 401 开始:把 csdn_ugc 汇总任务拆成三步
Kubernetes 里 CronJob 跑不起来,日志停在401 Unauthorized,这是把 TaoToken 接入定时汇总任务时最典型的断点。先去 TaoToken 官网 获取 Key,再把 Base URL 设为https://taotoken.net/api。很多团队一开始会把问题归因到集群网络、RBAC 或镜像拉取,但真正的原因是 Secret 没注入、环境变量名写错,或者脚本请求路径和 Base URL 拼接不一致。
最近行业里在讨论 Token 调用规模与数据服务用电成本,但落到平台工程侧,更实际的问题是:如何按来源平台稳定统计 Token 调用量。本文以csdn_ugc这个来源平台为例,用 Kubernetes CronJob 每天定时跑一次汇总 Job,产出可审计的 JSON 报告和 Job 日志。整个方案拆成三步:
- 在 TaoToken 准备 API Key、确认 Base URL 和可用模型名;
- 准备一份只读的用量明细 JSONL,字段里必须包含
platform、model、prompt_tokens、completion_tokens、total_tokens; - 部署 CronJob,让 Job 在固定时区聚合
csdn_ugc,并可选调用 TaoToken 生成一段中文摘要。
这里刻意不让汇总任务直连生产库。平台侧更稳的做法是:由日志管道或离线导出任务,把调用明细以只读方式落到 PVC、对象存储或数据湖,再让 CronJob 读取样本文件。下面的 SQL 和命令都只建议在本地或离线副本上执行,不要在生产库上直接跑。
2. 在 TaoToken 准备 Key、Base URL 与模型名
先打开 TaoToken 官网,完成登录后进入控制台。创建 API Key 的入口在这里: API Keys。Key 通常只在创建时完整显示一次,复制后立刻写入 Kubernetes Secret,不要提交到 Git。
本文所有配置里的 Key 占位符统一写成:
YOUR_API_KEYBase URL 固定为:
https://taotoken.net/api注意,Base URL 不加 UTM 参数。UTM 只用于官网、控制台和文档链接,方便做来源追踪;真正给 SDK、CronJob、Claude Code、Codex 使用的 API 基址保持干净。
模型名不要凭感觉写。打开 模型对话,选一个可用模型,复制模型名,后面填入 CronJob 的TAOTOKEN_MODEL。如果你的汇总任务只需要本地聚合,模型名可以不填;如果要让 Job 额外生成一段中文日报,再填模型名。
在 Kubernetes 里创建 Secret:
kubectl create namespace platform-observability --dry-run=client -o yaml | kubectl apply -f - kubectl create secret generic taotoken-api \ -n platform-observability \ --from-literal=api-key=YOUR_API_KEY \ --dry-run=client -o yaml | kubectl apply -f -不要把 Key 放到 ConfigMap。ConfigMap 适合放脚本、普通环境变量和配置片段,Secret 才适合放 API Key。CronJob 的 Pod 模板里通过secretKeyRef注入,这样 Job 日志不会打印明文 Key。
3. 用量明细怎么来:给 CronJob 准备只读 JSONL
汇总任务能不能复现,关键不在 CronJob 本身,而在输入数据是否稳定。建议准备一份按天导出的 JSONL,每行一个调用记录。最小字段如下:
{"ts":"2026-07-15T01:12:03+08:00","platform":"csdn_ugc","model":"YOUR_MODEL","prompt_tokens":320,"completion_tokens":48,"total_tokens":368,"request_id":"req_001"} {"ts":"2026-07-15T01:13:44+08:00","platform":"csdn_ugc","model":"YOUR_MODEL","prompt_tokens":510,"completion_tokens":72,"total_tokens":582,"request_id":"req_002"} {"ts":"2026-07-15T01:15:09+08:00","platform":"other_platform","model":"YOUR_MODEL","prompt_tokens":100,"completion_tokens":20,"total_tokens":120,"request_id":"req_003"}其中platform是来源平台。本文只汇总:
csdn_ugc不要写成CSDN_UGC、csdn-ugc或csdn ugc。大小写、连字符、下划线不一致,是造成汇总结果为空的高频原因。上线前先在本地样本文件里确认字段值。
本地校验命令可以这样执行:
jq -c 'select(.platform=="csdn_ugc")' usage.jsonl | head -n 5统计样本里csdn_ugc的记录数:
jq -r 'select(.platform=="csdn_ugc") | .request_id' usage.jsonl | wc -l如果已经有离线 SQLite 副本,也可以在本地执行 SQL,而不是连生产库:
SELECT platform, model, COUNT(*) AS requests, SUM(prompt_tokens) AS prompt_tokens, SUM(completion_tokens) AS completion_tokens, SUM(total_tokens) AS total_tokens FROM token_usage WHERE platform = 'csdn_ugc' AND ts >= date('now', '-1 day') GROUP BY platform, model ORDER BY total_tokens DESC;这段 SQL 只用于本地离线表。平台工程里一个常见反模式,是让定时任务直接连生产库做聚合,甚至让 Agent 自己决定查询语句。正确做法是先把明细导出为只读 JSONL、CSV 或 Parquet,再让 CronJob 消费导出文件。这样即使汇总脚本写错,也不会影响线上库。
把 JSONL 放到 Kubernetes 可读的 PVC 中。假设 PVC 名为usage-export-pvc,挂载路径为/data,文件名为usage.jsonl。CronJob 每次启动时读取/data/usage.jsonl,按platform=csdn_ugc聚合,输出到/report。
4. 汇总脚本 summarize.py:聚合 + 可选调用 TaoToken 生成日报
下面这个脚本只依赖 Python 标准库,适合放进 ConfigMap。它做两件事:
- 读取
/data/usage.jsonl,过滤csdn_ugc,按模型聚合 token; - 如果提供了
OPENAI_API_KEY和TAOTOKEN_MODEL,就调用 TaoToken 的兼容接口生成一段中文摘要。
import collections import datetime import json import os import pathlib import urllib.request PLATFORM = os.getenv("SUMMARY_PLATFORM", "csdn_ugc") USAGE_FILE = os.getenv("USAGE_FILE", "/data/usage.jsonl") REPORT_DIR = pathlib.Path(os.getenv("REPORT_DIR", "/report")) BASE_URL = os.getenv("OPENAI_BASE_URL", "https://taotoken.net/api").rstrip("/") API_KEY = os.getenv("OPENAI_API_KEY", "") MODEL = os.getenv("TAOTOKEN_MODEL", "") today = datetime.date.today().isoformat() stats = { "date": today, "platform": PLATFORM, "requests": 0, "prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0, "models": collections.Counter(), } with open(USAGE_FILE, "r", encoding="utf-8") as f: for line in f: line = line.strip() if not line: continue try: row = json.loads(line) except json.JSONDecodeError: continue if row.get("platform") != PLATFORM: continue stats["requests"] += 1 stats["prompt_tokens"] += int(row.get("prompt_tokens", 0)) stats["completion_tokens"] += int(row.get("completion_tokens", 0)) stats["total_tokens"] += int(row.get("total_tokens", 0)) model = row.get("model", "unknown") stats["models"][model] += int(row.get("total_tokens", 0)) top_models = stats["models"].most_common(5) stats["top_models"] = [{"model": m, "total_tokens": t} for m, t in top_models] print( f"[summary] platform={stats['platform']} " f"requests={stats['requests']} " f"prompt_tokens={stats['prompt_tokens']} " f"completion_tokens={stats['completion_tokens']} " f"total_tokens={stats['total_tokens']}" ) print(f"[summary] top_models={json.dumps(stats['top_models'], ensure_ascii=False)}") REPORT_DIR.mkdir(parents=True, exist_ok=True) report_path = REPORT_DIR / f"token-summary-{today}.json" report_path.write_text(json.dumps(stats, ensure_ascii=False, indent=2), encoding="utf-8") print(f"[report] wrote {report_path}") if API_KEY and MODEL: prompt = ( f"请用三句话总结以下 Token 用量汇总,不要编造数据:" f"平台={stats['platform']},请求数={stats['requests']}," f"输入Token={stats['prompt_tokens']},输出Token={stats['completion_tokens']}," f"总Token={stats['total_tokens']},Top模型={stats['top_models']}。" ) payload = { "model": MODEL, "messages": [ {"role": "system", "content": "你是平台可观测性助手,只根据给定数据输出简短摘要。"}, {"role": "user", "content": prompt}, ], "temperature": 0.2, } req = urllib.request.Request( f"{BASE_URL}/chat/completions", data=json.dumps(payload).encode("utf-8"), headers={ "Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json", }, method="POST", ) try: with urllib.request.urlopen(req, timeout=30) as resp: body = json.loads(resp.read().decode("utf-8")) content = body["choices"][0]["message"]["content"] print(f"[taotoken-summary] {content}") except Exception as exc: print(f"[taotoken-summary][warn] {type(exc).__name__}: {exc}") else: print("[taotoken-summary][skip] OPENAI_API_KEY 或 TAOTOKEN_MODEL 未设置")这个脚本的路径拼接是BASE_URL/chat/completions。由于 Base URL 是https://taotoken.net/api,最终请求地址就是https://taotoken.net/api/chat/completions。不要在不同脚本里一半写/v1、一半不写,路径不一致会直接导致 404。
把脚本放进 ConfigMap:
kubectl create configmap token-summary-script \ -n platform-observability \ --from-file=summarize.py=./summarize.py \ --dry-run=client -o yaml | kubectl apply -f -5. CronJob YAML:每天 02:10 汇总 csdn_ugc 并留存报告
下面这份 CronJob 每天凌晨 02:10 执行,时区设置为Asia/Shanghai。如果集群版本较低不支持timeZone,可以去掉该字段,并让集群节点使用统一时区,或在脚本里显式处理时区。
apiVersion: batch/v1 kind: CronJob metadata: name: token-usage-summary namespace: platform-observability spec: schedule: "10 2 * * *" timeZone: "Asia/Shanghai" concurrencyPolicy: Forbid successfulJobsHistoryLimit: 3 failedJobsHistoryLimit: 3 jobTemplate: spec: backoffLimit: 2 ttlSecondsAfterFinished: 86400 template: spec: restartPolicy: OnFailure containers: - name: summarizer image: python:3.12-slim command: ["python", "/opt/scripts/summarize.py"] env: - name: OPENAI_BASE_URL value: "https://taotoken.net/api" - name: OPENAI_API_KEY valueFrom: secretKeyRef: name: taotoken-api key: api-key - name: SUMMARY_PLATFORM value: "csdn_ugc" - name: USAGE_FILE value: "/data/usage.jsonl" - name: REPORT_DIR value: "/report" - name: TAOTOKEN_MODEL value: "YOUR_MODEL" volumeMounts: - name: script mountPath: /opt/scripts - name: usage-data mountPath: /data readOnly: true - name: report mountPath: /report volumes: - name: script configMap: name: token-summary-script - name: usage-data persistentVolumeClaim: claimName: usage-export-pvc - name: report persistentVolumeClaim: claimName: token-report-pvcTAOTOKEN_MODEL的YOUR_MODEL要替换成你在模型对话页里确认的模型名。如果暂时不做模型摘要,可以留空或不设置该环境变量,脚本仍然会完成本地聚合。
应用 YAML:
kubectl apply -f cronjob-token-summary.yaml查看 CronJob:
kubectl get cronjob -n platform-observability手动触发一次,不用等凌晨:
kubectl create job \ --from=cronjob/token-usage-summary \ token-usage-summary-manual \ -n platform-observability查看 Job:
kubectl get jobs -n platform-observability kubectl get pods -n platform-observability6. 查看 Job 日志与排障:401、404、超时、时区、空数据
正常汇总 Job 的日志应该类似下面这样。注意,下面的数字只是示例,不是任何行业统计数据。
[summary] platform=csdn_ugc requests=1842 prompt_tokens=623401 completion_tokens=88123 total_tokens=711524 [summary] top_models=[{"model":"YOUR_MODEL","total_tokens":711524}] [taotoken-summary] 今日 csdn_ugc 来源的 Token 调用主要集中在默认模型,总 Token 量为 711524,其中输出 Token 占比约 12%,整体请求数与昨日样本基本一致。 [report] wrote /report/token-summary-2026-07-15.json如果日志里没有[summary],先看 Pod 是否处于CrashLoopBackOff或Error:
kubectl logs job/token-usage-summary-manual -n platform-observability kubectl describe pod -n platform-observability -l job-name=token-usage-summary-manual常见问题按下面顺序排查:
401 Unauthorized:Secret 里的api-key是否和 TaoToken 控制台创建的一致;CronJob 的secretKeyRef.name和key是否写对;Key 是否被误删或过期。404 Not Found:检查OPENAI_BASE_URL是否为https://taotoken.net/api,脚本拼接是否为/chat/completions。不要在 Base URL 后面重复加/v1,也不要手动拼/api/api。Connection timed out:检查集群出网策略、DNS、NetworkPolicy。超时时间可以在脚本里从 30 秒调到 60 秒,但不要无限等待。platform=csdn_ugc requests=0:检查 JSONL 中platform字段值是否严格等于csdn_ugc;检查/data/usage.jsonl是否挂载到了容器内;检查文件是否为空。- 报告没有写入
/report:检查token-report-pvc是否存在、是否有写权限、REPORT_DIR是否指向挂载路径。 - 时间不对:检查 CronJob 的
timeZone,同时检查 JSONL 中的ts是 UTC 还是本地时间。建议存储统一用 UTC,展示时再转成Asia/Shanghai。 - 模型摘要失败但聚合成功:这是可接受的降级。脚本已经把模型调用放在 try/except 中,摘要失败不会影响 JSON 报告落盘。
如果你要直接验证 TaoToken 接口连通性,可以在本地用 curl 测一次,命令只在你本机执行:
curl -sS https://taotoken.net/api/chat/completions \ -H "Authorization: Bearer YOUR_API_KEY" \ -H "Content-Type: application/json" \ -d '{ "model": "YOUR_MODEL", "messages": [ {"role": "user", "content": "只回复 ok"} ], "temperature": 0 }'返回体里如果出现choices,说明 Key、Base URL、模型名三者基本匹配。如果返回 401,先查 Key;返回 404,先查路径和 Base URL;返回模型不存在,回到 模型对话 重新确认模型名。
7. 本地工具链对齐:Claude Code、Codex、CC Switch 填 TaoToken 的不同写法
CronJob 跑通后,平台工程师通常还要让本地开发工具也走同一套 TaoToken 配置。这里最容易出错的地方,是把 Claude Code 的ANTHROPIC_*环境变量套到 Codex 上。两者配置格式不同,不要混用。
Claude Code 使用settings.json和ANTHROPIC_*。示例:
{ "env": { "ANTHROPIC_BASE_URL": "https://taotoken.net/api", "ANTHROPIC_AUTH_TOKEN": "YOUR_API_KEY", "ANTHROPIC_MODEL": "YOUR_MODEL" } }Claude Code 的详细配置可以看官方文档: Claude Code 文档。注意,ANTHROPIC_BASE_URL同样填https://taotoken.net/api,不要加 UTM,也不要写成别的路径。
Codex 使用config.toml,不要写ANTHROPIC_*。示例:
model = "YOUR_MODEL" model_provider = "taotoken" [model_providers.taotoken] name = "TaoToken" base_url = "https://taotoken.net/api" env_key = "TAOTOKEN_API_KEY" wire_api = "chat"然后在本地环境变量里设置:
export TAOTOKEN_API_KEY=YOUR_API_KEY这样 Codex 读取的是自己的model_providers配置,不会和 Claude Code 的ANTHROPIC_*冲突。
CC Switch 可以理解成三件套管理:Provider、Base URL、API Key。推荐填法:
Provider: 自定义 / TaoToken Base URL: https://taotoken.net/api API Key: YOUR_API_KEY如果你在 CC Switch 里维护多个供应商,建议给 TaoToken 单独建一个配置项,命名成taotoken-chat、taotoken-codex或taotoken-claude,避免和默认官方供应商互相覆盖。再次提醒:Codex 配置不要套ANTHROPIC_*,Claude Code 配置不要套TAOTOKEN_API_KEY作为ANTHROPIC_AUTH_TOKEN之外的变量。
8. 文末 CTA:从模型对话到 Coding Plan,再到 API Keys
到这里,Kubernetes CronJob 已经可以每天定时汇总csdn_ugc来源的 Token 调用量,产出 JSON 报告和 Job 日志。建议按下面的路径继续落地:
- 先在 模型对话 里确认可用模型名和返回格式;
- 如果要在本地开发工具里长期使用,查看 Coding Plan;
- 需要给 CronJob 或本地工具创建独立 Key,去 API Keys;
- Claude Code 用户再对照 Claude Code 文档 检查
settings.json; - 所有入口都可以从 TaoToken 官网 进入,Base URL 记住保持为
https://taotoken.net/api。
最后给一份上线检查清单:
- CronJob 的
schedule、timeZone、concurrencyPolicy是否符合团队结算周期; - Secret 是否使用
taotoken-api/api-key,而不是 ConfigMap; OPENAI_BASE_URL是否为https://taotoken.net/api;- 脚本过滤值是否严格为
csdn_ugc; /data/usage.jsonl是否由只读导出任务生成,而不是定时任务直连生产库;/reportPVC 是否有足够空间和保留策略;- 模型摘要失败时,聚合报告是否仍然落盘;
- 手动触发一次 Job,确认日志里同时出现
[summary]、[report],以及需要时的[taotoken-summary]。
把 CronJob YAML、ConfigMap 脚本和 Secret 创建命令放进同一套 GitOps 目录后,以后新增来源平台只需要改SUMMARY_PLATFORM,或者在脚本里扩展成平台列表。这样,Token 调用量汇总不再是临时脚本,而是 Kubernetes 里可调度、可重试、可审计的定时任务。