1. 从“每秒成本”倒推Agentic RL 后训练里谁在消耗 Token在 LlamaIndex 里把OpenAILike的api_base留空或者在 Codex 的config.toml里误用ANTHROPIC_*通常会先看到 401 或 404。排查前先去 TaoToken 官网 获取 Key 并查看可用模型请求 Base URL 填https://taotoken.net/api。外部直播里讨论小米 mimo-v2.6-pro 与 flash 的 Agentic RL 后训练有作者按训练资源推算每秒约 10 美元mimo-v2.6-pro 的 DeepSWE 为 62低于 DeepSeek-v4.1-flash 的 74.2仍处训练早期。这里不评价模型排名也不把外部推算当成结论只借这个现象说明工程侧真正缺的东西如果没有逐节点 Token 账本你无法知道成本花在检索、工具调用、重试还是长上下文拼接。本文用 TaoToken 的 Key 和 Base URL 把推理入口统一起来再用 LlamaIndex 回调审计代码把每个节点的 prompt token、completion token、总 token、耗时和重试次数输出成表。这样再看“谁在消耗 Token”就不是凭感觉猜而是能按 event_id、parent_id、节点类型逐条对账。Agentic RL 后训练和普通问答调用最大的区别是推理链路不是一次请求就结束。一个 rollout 里可能包含规划、检索、工具调用、环境反馈、结果校验、失败重试、轨迹拼接、奖励计算等多个阶段每个阶段都可能向模型发请求。外部讨论里“每秒烧多少钱”之所以容易引发焦虑是因为成本被压成了一个总账但总账无法回答三个问题第一消耗来自长上下文输入还是长输出第二消耗来自模型本身还是重试策略第三消耗来自并行 rollout 还是串行工具链。推理链路审计的目标就是把总账拆成节点账。可以把常见消耗点先列出来消耗点典型表现审计字段长上下文拼接prompt_tokens 远高于 completion_tokensprompt_tokens、上下文长度、节点 parent_id工具调用多轮单次任务有多个 LLM 子事件event_type、tool_name、调用次数失败重试相同 query 出现多次 LLM 事件retry、error、parent_id流式响应响应快但 usage 缺失usage_source、chunk 累积并行 rollout总 token 高但单节点正常trace_id、并发数、批次模型切换pro 与 flash 混用model、单节点成本系数如果 Base URL 分散在多个脚本、多个环境变量、多个工具里审计就会断链。把 LlamaIndex、Claude Code、Codex、CC Switch 的推理入口都指向 TaoToken 的https://taotoken.net/api至少能保证 Key 来源、请求域名、模型名可追踪。注意Base URL 不加 UTM 参数Key 使用占位符YOUR_API_KEY。接下来先从 Key 和 Base URL 开始。2. TaoToken Key 与 Base URL先把推理入口统一到可审计通道要审计推理链路第一步不是写回调而是统一入口。如果你在多个地方分别配置 Key最后导出 token 表时会发现无法判断某个请求来自 LlamaIndex 还是编码工具。建议去 TaoToken 官网 创建 Key 并确认模型列表然后只维护一份环境变量。以下命令在读者本地执行export TAOTOKEN_API_KEYYOUR_API_KEY export TAOTOKEN_BASE_URLhttps://taotoken.net/api如果你使用 OpenAI 兼容客户端也可以临时设置export OPENAI_API_KEYYOUR_API_KEY export OPENAI_API_BASEhttps://taotoken.net/api这里的https://taotoken.net/api是工具配置用的 Base URL不要在后面拼接 UTM 参数。Key 占位符统一写成YOUR_API_KEY避免把真实 Key 写进代码仓库。LlamaIndex 侧可以用OpenAILike接入from llama_index.llms.openai_like import OpenAILike llm OpenAILike( model控制台可用模型名, api_keyYOUR_API_KEY, api_basehttps://taotoken.net/api, is_chat_modelTrue, timeout120, )这段配置的重点不是模型名而是api_base必须与审计链路一致。很多 404 并不是 Key 错而是 Base URL 写成了别的域名或者多了、少了一层路径。排查时可以先用一个最小请求确认通道curl -s https://taotoken.net/api/chat/completions \ -H Authorization: Bearer YOUR_API_KEY \ -H Content-Type: application/json \ -d { model: 控制台可用模型名, messages: [ {role: user, content: 只回复 ok} ], stream: false }如果返回 401优先检查 Key 是否完整、是否带了Bearer、环境变量是否在当前 shell 生效。如果返回 404优先检查 Base URL 是否被客户端二次拼接例如有的客户端会自己加/v1这时应以客户端要求为准但域名和 Key 仍保持 TaoToken 这一套。如果返回 429说明并发或频率触发限制Agentic RL 场景下要特别小心重试会放大 token而且会在审计表里表现为同一 parent_id 下多个相似 LLM 事件。统一入口后LlamaIndex 的每次模型调用都会经过同一个 Base URL。这样后续回调里拿到的 usage 才能和 Key、模型、时间线对应起来。不要把 Claude Code 的ANTHROPIC_*环境变量搬到 Codex也不要把 Codex 的config.toml格式套到 Claude Code。下面是本文的审计主线用 TaoToken Key 发起 LlamaIndex 调用用回调记录事件用 CSV 或 Markdown 表输出逐节点 Token。3. LlamaIndex 回调审计逐节点记录 prompt/completion usageLlamaIndex 的回调机制适合做推理链路审计因为它能按事件类型触发并且事件带有event_id和parent_id。我们不需要改业务 query只需要注册一个全局 CallbackManager。下面这段代码可以在本地运行目标是记录 LLM、Embedding、Retrieve、Synthesize 等事件的耗时和 token 使用情况。不同版本的 LlamaIndex 事件枚举可能有差异所以代码里对未知事件类型做了字符串化处理。import csv import json import time from typing import Any, Dict, Optional from llama_index.core.callbacks import ( BaseCallbackHandler, CBEventType, CallbackManager, ) from llama_index.core.callbacks.schema import EventPayload class TokenAuditHandler(BaseCallbackHandler): 记录每个 LlamaIndex 事件的 token 与耗时。 def __init__(self): super().__init__() self.stack: Dict[str, Dict[str, Any]] {} self.records: list[Dict[str, Any]] [] def on_event_start( self, event_type: CBEventType, payload: Optional[Dict[str, Any]] None, event_id: str , parent_id: str , **kwargs: Any, ) - str: payload payload or {} self.stack[event_id] { event_id: event_id, parent_id: parent_id, event_type: str(event_type), start_time: time.time(), payload_keys: ,.join(sorted(payload.keys())), retry: 0, error: , } return event_id def on_event_end( self, event_type: CBEventType, payload: Optional[Dict[str, Any]] None, event_id: str , parent_id: str , **kwargs: Any, ) - None: payload payload or {} record self.stack.pop(event_id, { event_id: event_id, parent_id: parent_id, event_type: str(event_type), start_time: time.time(), payload_keys: , retry: 0, error: , }) elapsed_ms round((time.time() - record.pop(start_time)) * 1000, 2) usage self._extract_usage(payload) record.update({ elapsed_ms: elapsed_ms, model: usage.get(model, ), prompt_tokens: usage.get(prompt_tokens, 0), completion_tokens: usage.get(completion_tokens, 0), total_tokens: usage.get(total_tokens, 0), usage_source: usage.get(usage_source, missing), }) self.records.append(record) staticmethod def _pick_usage(obj: Any) - Optional[Dict[str, Any]]: if obj is None: return None if isinstance(obj, dict): usage obj.get(usage) or obj.get(token_usage) if isinstance(usage, dict): return usage raw obj.get(raw) if raw is not None: return TokenAuditHandler._pick_usage(raw) return None raw getattr(obj, raw, None) if raw is not None: picked TokenAuditHandler._pick_usage(raw) if picked: return picked usage getattr(obj, usage, None) if usage is None: return None if isinstance(usage, dict): return usage return { prompt_tokens: getattr(usage, prompt_tokens, 0), completion_tokens: getattr(usage, completion_tokens, 0), total_tokens: getattr(usage, total_tokens, 0), } staticmethod def _extract_usage(payload: Dict[str, Any]) - Dict[str, Any]: response ( payload.get(response) or payload.get(chunk) or payload.get(message) or payload.get(llm_response) ) usage TokenAuditHandler._pick_usage(response) if not usage: return { prompt_tokens: 0, completion_tokens: 0, total_tokens: 0, usage_source: missing, } prompt ( usage.get(prompt_tokens) or usage.get(input_tokens) or usage.get(promptTokens) or 0 ) completion ( usage.get(completion_tokens) or usage.get(output_tokens) or usage.get(completionTokens) or 0 ) total usage.get(total_tokens) or usage.get(totalTokens) or (prompt completion) return { model: usage.get(model, ), prompt_tokens: int(prompt), completion_tokens: int(completion), total_tokens: int(total), usage_source: api, } def export_csv(self, path: str token_audit.csv) - None: fieldnames [ event_id, parent_id, event_type, model, prompt_tokens, completion_tokens, total_tokens, elapsed_ms, usage_source, retry, error, payload_keys, ] with open(path, w, newline, encodingutf-8) as f: writer csv.DictWriter(f, fieldnamesfieldnames) writer.writeheader() for row in self.records: writer.writerow({k: row.get(k, ) for k in fieldnames})注册回调并调用 LlamaIndexfrom llama_index.core import Settings, SimpleDirectoryReader, VectorStoreIndex from llama_index.llms.openai_like import OpenAILike audit TokenAuditHandler() Settings.callback_manager CallbackManager([audit]) Settings.llm OpenAILike( model控制台可用模型名, api_keyYOUR_API_KEY, api_basehttps://taotoken.net/api, is_chat_modelTrue, timeout120, ) documents SimpleDirectoryReader(./data).load_data() index VectorStoreIndex.from_documents(documents) query_engine index.as_query_engine() response query_engine.query(把当前任务拆成可执行步骤并记录每次工具调用) print(response) audit.export_csv(token_audit.csv)这段代码运行后token_audit.csv会包含每个事件的 token 记录。如果某个事件的usage_source是missing说明该响应没有返回 usage可能发生在流式响应、Embedding 调用或部分兼容接口中。不要直接把 0 当成没有消耗下一节会给出兜底估算和逐节点输出表。4. 逐节点 Token 输出表把工具调用、重试、长上下文拆开审计的价值不在于“总 token 是多少”而在于定位谁消耗得最多。建议把 CSV 再整理成逐节点 Token 输出表至少包含event_id、parent_id、节点类型、模型、prompt_tokens、completion_tokens、total_tokens、elapsed_ms、usage_source、备注。下面是一个示例表数据仅用于说明结构实际以你的审计结果为准event_idparent_id节点类型模型prompt_tokenscompletion_tokenstotal_tokenselapsed_msusage_source备注root-QUERY-00015.2none查询入口e1rootRETRIEVE-00042.8none本地向量检索e2rootLLMmimo-v2.6-flash182042022401350.4api生成计划e3e2FUNCTION_CALLmimo-v2.6-flash64088728410.7api工具调用e4e2LLMmimo-v2.6-pro4120102451443820.1api长上下文首次尝试e5e2LLMmimo-v2.6-pro408098050603600.5api失败重试e6rootSYNTHESIZEmimo-v2.6-pro236064030001820.9api汇总输出total---1302031521617211060.6-重试占 LLM token 约 31%这个表能直接回答几个问题。第一如果RETRIEVE节点 token 为 0但后续LLM的 prompt_tokens 很高说明成本主要来自上下文拼接而不是检索本身。第二如果同一个parent_id下出现多次相似LLM事件且retry标记增加说明失败重试在放大成本。第三如果elapsed_ms很高但 token 很低可能是排队、网络或工具等待如果 token 很高但elapsed_ms不高通常是输入太长。第四如果 pro 和 flash 混用逐节点表能看出 pro 被用在了哪些节点是否可以用 flash 替换。当 usage 缺失时可以本地估算。以下函数只做粗略兜底不能替代 API usage但能帮助发现异常def estimate_tokens(text: str) - int: if not text: return 0 # 粗略估算英文约 4 字符/token中文按 1.5-2 字符/token 估 return max(1, int(len(text) / 3.5))也可以使用 tiktoken 做更细的估算import tiktoken def estimate_with_tiktoken(text: str, model: str cl100k_base) - int: encoding tiktoken.get_encoding(model) return len(encoding.encode(text or ))然后把估算值写回审计记录def backfill_missing_usage(records, text_getter): for row in records: if row.get(usage_source) missing and row.get(event_type) CBEventType.LLM: text text_getter(row) row[prompt_tokens] estimate_tokens(text) row[completion_tokens] 0 row[total_tokens] row[prompt_tokens] row[usage_source] estimated return records注意审计日志建议落在本地 CSV、SQLite 或表格文件中由读者本地执行分析。不要从 Agent 直接连接生产库也不要把 Key 写进审计日志。你可以在本地用 Python 或 SQL 查询 CSV但 SQL 命令必须由你在本地环境执行。5. Claude Code / Codex / CC Switch 三件套配置不要混用环境变量推理链路审计不只发生在 LlamaIndex。如果你还用 Claude Code、Codex 或 CC Switch 做编码任务同样需要把供应商入口统一到 TaoToken。这里最容易犯的错误是环境变量串用Claude Code 用ANTHROPIC_*Codex 用config.tomlCC Switch 则填 Base URL、API Key、Model 三件套。不要把ANTHROPIC_*写进 Codex 的config.toml也不要把 Codex 的env_key套到 Claude Code。Claude Code 可以在~/.claude/settings.json中配置{ env: { ANTHROPIC_BASE_URL: https://taotoken.net/api, ANTHROPIC_AUTH_TOKEN: YOUR_API_KEY, ANTHROPIC_MODEL: 控制台可用模型名, ANTHROPIC_SMALL_FAST_MODEL: 控制台可用轻量模型名 } }如果你的 Claude Code 版本使用ANTHROPIC_API_KEY也可以按版本文档调整但核心仍然是 Base URL 指向https://taotoken.net/apiKey 使用YOUR_API_KEY。修改后重启终端或重新加载配置再用一个简单任务验证是否走到了 TaoToken。Claude Code 文档入口会在文末给出。Codex 使用~/.codex/config.toml配置方式与 Claude Code 不同model_provider taotoken model 控制台可用模型名 model_reasoning_effort medium [model_providers.taotoken] name TaoToken base_url https://taotoken.net/api env_key TAOTOKEN_API_KEY wire_api chat然后在本地环境变量中设置export TAOTOKEN_API_KEYYOUR_API_KEY这里env_key写的是环境变量名不是 Key 本身。Codex 会读取TAOTOKEN_API_KEY的值。不要把ANTHROPIC_AUTH_TOKEN填进env_key否则会出现找不到 Key 或 401。CC Switch 可以理解成供应商切换器配置时重点检查三件套配置项填什么说明Base URLhttps://taotoken.net/api不加 UTM不加多余斜杠API KeyYOUR_API_KEY从 TaoToken 控制台创建不要复用旧 KeyModel控制台模型名不要手写不存在的别名如果你在 CC Switch 里同时维护多个供应商建议给 TaoToken 单独命名例如taotoken-pro、taotoken-flash这样切换后能从模型名反推审计记录。更多入口和配置说明可以在 TaoToken 官网 查看工具接入说明 获取。注意编码工具本身不一定输出逐节点 token usage所以真正的细粒度审计仍建议放在 LlamaIndex 或你自己的请求网关层。6. 排障Base URL、401、404、流式响应无 usage 的审计兜底把入口统一到 TaoToken 后常见问题集中在四类401、404、429 和流式响应无 usage。下面按症状给出排查顺序。第一401 未授权。先确认 Key 是否完整再确认请求头格式是否为Authorization: Bearer YOUR_API_KEY。如果你用的是环境变量执行env | grep TAOTOKEN或env | grep ANTHROPIC检查当前 shell 是否真的加载。Claude Code 的settings.json修改后需要重启工具Codex 的TAOTOKEN_API_KEY需要在启动前导出。第二404 路径错误。最常见原因是 Base URL 被客户端二次拼接。TaoToken 的 Base URL 是https://taotoken.net/api。某些 OpenAI 兼容客户端会自己追加/v1这时你应以客户端实际请求路径为准但不要同时保留两个不同的域名。排查时可以先用最小 curl 请求确认通道。第三429 限流。Agentic RL 或批量审计场景容易并发过高触发限制后如果代码自动重试会产生额外 token。审计表里应记录重试次数和错误信息def mark_retry(record: dict, error: str) - None: record[retry] int(record.get(retry, 0)) 1 record[error] error[:200]然后在导出表里把retry 0的行标红或单独汇总。重试成本经常被低估尤其是长上下文请求失败后重试prompt_tokens 会重复计入。第四流式响应无 usage。某些兼容接口只在最后一个 chunk 返回 usage或者完全不返回。可以在回调里累积 chunkclass StreamingUsageMerger: def __init__(self): self.prompt_tokens 0 self.completion_tokens 0 self.total_tokens 0 self.text_parts [] def feed(self, chunk): if chunk is None: return if isinstance(chunk, dict): usage chunk.get(usage) delta chunk.get(choices, [{}])[0].get(delta, {}) content delta.get(content) else: usage getattr(chunk, usage, None) content getattr(chunk, delta, None) or getattr(chunk, text, None) if content: self.text_parts.append(str(content)) if usage: if isinstance(usage, dict): self.prompt_tokens usage.get(prompt_tokens, self.prompt_tokens) self.completion_tokens usage.get(completion_tokens, self.completion_tokens) self.total_tokens usage.get(total_tokens, self.total_tokens) else: self.prompt_tokens getattr(usage, prompt_tokens, self.prompt_tokens) self.completion_tokens getattr(usage, completion_tokens, self.completion_tokens) self.total_tokens getattr(usage, total_tokens, self.total_tokens) def fallback(self): if self.total_tokens 0: self.prompt_tokens 0 self.completion_tokens estimate_tokens(.join(self.text_parts)) self.total_tokens self.prompt_tokens self.completion_tokens return { prompt_tokens: self.prompt_tokens, completion_tokens: self.completion_tokens, total_tokens: self.total_tokens, usage_source: stream-merged if self.total_tokens else missing, }把流式合并结果写回审计记录再导出表格。这样即使接口不返回完整 usage你也能看到输出侧的 token 量级。不过要明确估算只用于发现异常和排障不能替代账单数据。除了错误排查还要看时间线。elapsed_ms和 token 数交叉分析很有用现象可能原因处理token 高、耗时低上下文太长模型快速消费压缩检索结果限制历史消息token 低、耗时长排队、网络、工具等待检查并发和超时同一 parent_id 多次 LLM重试或递归调用增加重试上限记录错误pro 模型集中在某个节点模型路由过重评估 flash 替换usage_source 多为 missing流式未合并或接口不返回启用合并与估算兜底最后审计表要能按trace_id聚合。虽然上面的 LlamaIndex 回调记录了event_id和parent_id但你可以在业务层给每次查询或 rollout 加一个外部 trace_id写进 payload 或日志。这样在 Agentic RL 后训练场景里可以把一条轨迹下的所有模型调用串起来看看究竟是规划阶段、工具阶段还是汇总阶段消耗最多。7. 文末 CTA模型对话、Coding Plan、创建 Key、Claude Code 文档如果你已经准备把推理链路审计落到可复现的流程里可以按下面的顺序操作。先验证模型对话确认 Base URL 和 Key 可用再决定是否把编码工具接入 Coding Plan然后创建独立 API Key避免和旧 Key 混用最后按 Claude Code 文档完成settings.json配置。以下入口都带 UTM便于区分来源模型对话先用一个短请求确认https://taotoken.net/api和YOUR_API_KEY可用。Coding Plan如果要把 Claude Code、Codex、CC Switch 一起接入先确认编码场景的模型与额度策略。创建 API Key为审计环境单独创建一个 Key不要和测试 Key 混用。Claude Code 文档按文档配置settings.json中的ANTHROPIC_*不要套用到 Codex。最后给一个本地行动清单。第一步在 TaoToken 官网 获取 Key 并确认可用模型名。第二步把 LlamaIndex 的api_base设为https://taotoken.net/apiKey 使用YOUR_API_KEY。第三步注册TokenAuditHandler运行一次带检索和工具调用的 query。第四步导出token_audit.csv整理成逐节点 Token 输出表重点看重试和长上下文节点。第五步如果还要接 Claude Code 或 Codex分别按settings.json和config.toml配置不要混用环境变量。做到这五步再讨论“谁在消耗 Token”你手里就有账本而不是只有一个总价。
推理链路审计：Agentic RL 后训练用 TaoToken 的 Key
1. 从“每秒成本”倒推Agentic RL 后训练里谁在消耗 Token在 LlamaIndex 里把OpenAILike的api_base留空或者在 Codex 的config.toml里误用ANTHROPIC_*通常会先看到 401 或 404。排查前先去 TaoToken 官网 获取 Key 并查看可用模型请求 Base URL 填https://taotoken.net/api。外部直播里讨论小米 mimo-v2.6-pro 与 flash 的 Agentic RL 后训练有作者按训练资源推算每秒约 10 美元mimo-v2.6-pro 的 DeepSWE 为 62低于 DeepSeek-v4.1-flash 的 74.2仍处训练早期。这里不评价模型排名也不把外部推算当成结论只借这个现象说明工程侧真正缺的东西如果没有逐节点 Token 账本你无法知道成本花在检索、工具调用、重试还是长上下文拼接。本文用 TaoToken 的 Key 和 Base URL 把推理入口统一起来再用 LlamaIndex 回调审计代码把每个节点的 prompt token、completion token、总 token、耗时和重试次数输出成表。这样再看“谁在消耗 Token”就不是凭感觉猜而是能按 event_id、parent_id、节点类型逐条对账。Agentic RL 后训练和普通问答调用最大的区别是推理链路不是一次请求就结束。一个 rollout 里可能包含规划、检索、工具调用、环境反馈、结果校验、失败重试、轨迹拼接、奖励计算等多个阶段每个阶段都可能向模型发请求。外部讨论里“每秒烧多少钱”之所以容易引发焦虑是因为成本被压成了一个总账但总账无法回答三个问题第一消耗来自长上下文输入还是长输出第二消耗来自模型本身还是重试策略第三消耗来自并行 rollout 还是串行工具链。推理链路审计的目标就是把总账拆成节点账。可以把常见消耗点先列出来消耗点典型表现审计字段长上下文拼接prompt_tokens 远高于 completion_tokensprompt_tokens、上下文长度、节点 parent_id工具调用多轮单次任务有多个 LLM 子事件event_type、tool_name、调用次数失败重试相同 query 出现多次 LLM 事件retry、error、parent_id流式响应响应快但 usage 缺失usage_source、chunk 累积并行 rollout总 token 高但单节点正常trace_id、并发数、批次模型切换pro 与 flash 混用model、单节点成本系数如果 Base URL 分散在多个脚本、多个环境变量、多个工具里审计就会断链。把 LlamaIndex、Claude Code、Codex、CC Switch 的推理入口都指向 TaoToken 的https://taotoken.net/api至少能保证 Key 来源、请求域名、模型名可追踪。注意Base URL 不加 UTM 参数Key 使用占位符YOUR_API_KEY。接下来先从 Key 和 Base URL 开始。2. TaoToken Key 与 Base URL先把推理入口统一到可审计通道要审计推理链路第一步不是写回调而是统一入口。如果你在多个地方分别配置 Key最后导出 token 表时会发现无法判断某个请求来自 LlamaIndex 还是编码工具。建议去 TaoToken 官网 创建 Key 并确认模型列表然后只维护一份环境变量。以下命令在读者本地执行export TAOTOKEN_API_KEYYOUR_API_KEY export TAOTOKEN_BASE_URLhttps://taotoken.net/api如果你使用 OpenAI 兼容客户端也可以临时设置export OPENAI_API_KEYYOUR_API_KEY export OPENAI_API_BASEhttps://taotoken.net/api这里的https://taotoken.net/api是工具配置用的 Base URL不要在后面拼接 UTM 参数。Key 占位符统一写成YOUR_API_KEY避免把真实 Key 写进代码仓库。LlamaIndex 侧可以用OpenAILike接入from llama_index.llms.openai_like import OpenAILike llm OpenAILike( model控制台可用模型名, api_keyYOUR_API_KEY, api_basehttps://taotoken.net/api, is_chat_modelTrue, timeout120, )这段配置的重点不是模型名而是api_base必须与审计链路一致。很多 404 并不是 Key 错而是 Base URL 写成了别的域名或者多了、少了一层路径。排查时可以先用一个最小请求确认通道curl -s https://taotoken.net/api/chat/completions \ -H Authorization: Bearer YOUR_API_KEY \ -H Content-Type: application/json \ -d { model: 控制台可用模型名, messages: [ {role: user, content: 只回复 ok} ], stream: false }如果返回 401优先检查 Key 是否完整、是否带了Bearer、环境变量是否在当前 shell 生效。如果返回 404优先检查 Base URL 是否被客户端二次拼接例如有的客户端会自己加/v1这时应以客户端要求为准但域名和 Key 仍保持 TaoToken 这一套。如果返回 429说明并发或频率触发限制Agentic RL 场景下要特别小心重试会放大 token而且会在审计表里表现为同一 parent_id 下多个相似 LLM 事件。统一入口后LlamaIndex 的每次模型调用都会经过同一个 Base URL。这样后续回调里拿到的 usage 才能和 Key、模型、时间线对应起来。不要把 Claude Code 的ANTHROPIC_*环境变量搬到 Codex也不要把 Codex 的config.toml格式套到 Claude Code。下面是本文的审计主线用 TaoToken Key 发起 LlamaIndex 调用用回调记录事件用 CSV 或 Markdown 表输出逐节点 Token。3. LlamaIndex 回调审计逐节点记录 prompt/completion usageLlamaIndex 的回调机制适合做推理链路审计因为它能按事件类型触发并且事件带有event_id和parent_id。我们不需要改业务 query只需要注册一个全局 CallbackManager。下面这段代码可以在本地运行目标是记录 LLM、Embedding、Retrieve、Synthesize 等事件的耗时和 token 使用情况。不同版本的 LlamaIndex 事件枚举可能有差异所以代码里对未知事件类型做了字符串化处理。import csv import json import time from typing import Any, Dict, Optional from llama_index.core.callbacks import ( BaseCallbackHandler, CBEventType, CallbackManager, ) from llama_index.core.callbacks.schema import EventPayload class TokenAuditHandler(BaseCallbackHandler): 记录每个 LlamaIndex 事件的 token 与耗时。 def __init__(self): super().__init__() self.stack: Dict[str, Dict[str, Any]] {} self.records: list[Dict[str, Any]] [] def on_event_start( self, event_type: CBEventType, payload: Optional[Dict[str, Any]] None, event_id: str , parent_id: str , **kwargs: Any, ) - str: payload payload or {} self.stack[event_id] { event_id: event_id, parent_id: parent_id, event_type: str(event_type), start_time: time.time(), payload_keys: ,.join(sorted(payload.keys())), retry: 0, error: , } return event_id def on_event_end( self, event_type: CBEventType, payload: Optional[Dict[str, Any]] None, event_id: str , parent_id: str , **kwargs: Any, ) - None: payload payload or {} record self.stack.pop(event_id, { event_id: event_id, parent_id: parent_id, event_type: str(event_type), start_time: time.time(), payload_keys: , retry: 0, error: , }) elapsed_ms round((time.time() - record.pop(start_time)) * 1000, 2) usage self._extract_usage(payload) record.update({ elapsed_ms: elapsed_ms, model: usage.get(model, ), prompt_tokens: usage.get(prompt_tokens, 0), completion_tokens: usage.get(completion_tokens, 0), total_tokens: usage.get(total_tokens, 0), usage_source: usage.get(usage_source, missing), }) self.records.append(record) staticmethod def _pick_usage(obj: Any) - Optional[Dict[str, Any]]: if obj is None: return None if isinstance(obj, dict): usage obj.get(usage) or obj.get(token_usage) if isinstance(usage, dict): return usage raw obj.get(raw) if raw is not None: return TokenAuditHandler._pick_usage(raw) return None raw getattr(obj, raw, None) if raw is not None: picked TokenAuditHandler._pick_usage(raw) if picked: return picked usage getattr(obj, usage, None) if usage is None: return None if isinstance(usage, dict): return usage return { prompt_tokens: getattr(usage, prompt_tokens, 0), completion_tokens: getattr(usage, completion_tokens, 0), total_tokens: getattr(usage, total_tokens, 0), } staticmethod def _extract_usage(payload: Dict[str, Any]) - Dict[str, Any]: response ( payload.get(response) or payload.get(chunk) or payload.get(message) or payload.get(llm_response) ) usage TokenAuditHandler._pick_usage(response) if not usage: return { prompt_tokens: 0, completion_tokens: 0, total_tokens: 0, usage_source: missing, } prompt ( usage.get(prompt_tokens) or usage.get(input_tokens) or usage.get(promptTokens) or 0 ) completion ( usage.get(completion_tokens) or usage.get(output_tokens) or usage.get(completionTokens) or 0 ) total usage.get(total_tokens) or usage.get(totalTokens) or (prompt completion) return { model: usage.get(model, ), prompt_tokens: int(prompt), completion_tokens: int(completion), total_tokens: int(total), usage_source: api, } def export_csv(self, path: str token_audit.csv) - None: fieldnames [ event_id, parent_id, event_type, model, prompt_tokens, completion_tokens, total_tokens, elapsed_ms, usage_source, retry, error, payload_keys, ] with open(path, w, newline, encodingutf-8) as f: writer csv.DictWriter(f, fieldnamesfieldnames) writer.writeheader() for row in self.records: writer.writerow({k: row.get(k, ) for k in fieldnames})注册回调并调用 LlamaIndexfrom llama_index.core import Settings, SimpleDirectoryReader, VectorStoreIndex from llama_index.llms.openai_like import OpenAILike audit TokenAuditHandler() Settings.callback_manager CallbackManager([audit]) Settings.llm OpenAILike( model控制台可用模型名, api_keyYOUR_API_KEY, api_basehttps://taotoken.net/api, is_chat_modelTrue, timeout120, ) documents SimpleDirectoryReader(./data).load_data() index VectorStoreIndex.from_documents(documents) query_engine index.as_query_engine() response query_engine.query(把当前任务拆成可执行步骤并记录每次工具调用) print(response) audit.export_csv(token_audit.csv)这段代码运行后token_audit.csv会包含每个事件的 token 记录。如果某个事件的usage_source是missing说明该响应没有返回 usage可能发生在流式响应、Embedding 调用或部分兼容接口中。不要直接把 0 当成没有消耗下一节会给出兜底估算和逐节点输出表。4. 逐节点 Token 输出表把工具调用、重试、长上下文拆开审计的价值不在于“总 token 是多少”而在于定位谁消耗得最多。建议把 CSV 再整理成逐节点 Token 输出表至少包含event_id、parent_id、节点类型、模型、prompt_tokens、completion_tokens、total_tokens、elapsed_ms、usage_source、备注。下面是一个示例表数据仅用于说明结构实际以你的审计结果为准event_idparent_id节点类型模型prompt_tokenscompletion_tokenstotal_tokenselapsed_msusage_source备注root-QUERY-00015.2none查询入口e1rootRETRIEVE-00042.8none本地向量检索e2rootLLMmimo-v2.6-flash182042022401350.4api生成计划e3e2FUNCTION_CALLmimo-v2.6-flash64088728410.7api工具调用e4e2LLMmimo-v2.6-pro4120102451443820.1api长上下文首次尝试e5e2LLMmimo-v2.6-pro408098050603600.5api失败重试e6rootSYNTHESIZEmimo-v2.6-pro236064030001820.9api汇总输出total---1302031521617211060.6-重试占 LLM token 约 31%这个表能直接回答几个问题。第一如果RETRIEVE节点 token 为 0但后续LLM的 prompt_tokens 很高说明成本主要来自上下文拼接而不是检索本身。第二如果同一个parent_id下出现多次相似LLM事件且retry标记增加说明失败重试在放大成本。第三如果elapsed_ms很高但 token 很低可能是排队、网络或工具等待如果 token 很高但elapsed_ms不高通常是输入太长。第四如果 pro 和 flash 混用逐节点表能看出 pro 被用在了哪些节点是否可以用 flash 替换。当 usage 缺失时可以本地估算。以下函数只做粗略兜底不能替代 API usage但能帮助发现异常def estimate_tokens(text: str) - int: if not text: return 0 # 粗略估算英文约 4 字符/token中文按 1.5-2 字符/token 估 return max(1, int(len(text) / 3.5))也可以使用 tiktoken 做更细的估算import tiktoken def estimate_with_tiktoken(text: str, model: str cl100k_base) - int: encoding tiktoken.get_encoding(model) return len(encoding.encode(text or ))然后把估算值写回审计记录def backfill_missing_usage(records, text_getter): for row in records: if row.get(usage_source) missing and row.get(event_type) CBEventType.LLM: text text_getter(row) row[prompt_tokens] estimate_tokens(text) row[completion_tokens] 0 row[total_tokens] row[prompt_tokens] row[usage_source] estimated return records注意审计日志建议落在本地 CSV、SQLite 或表格文件中由读者本地执行分析。不要从 Agent 直接连接生产库也不要把 Key 写进审计日志。你可以在本地用 Python 或 SQL 查询 CSV但 SQL 命令必须由你在本地环境执行。5. Claude Code / Codex / CC Switch 三件套配置不要混用环境变量推理链路审计不只发生在 LlamaIndex。如果你还用 Claude Code、Codex 或 CC Switch 做编码任务同样需要把供应商入口统一到 TaoToken。这里最容易犯的错误是环境变量串用Claude Code 用ANTHROPIC_*Codex 用config.tomlCC Switch 则填 Base URL、API Key、Model 三件套。不要把ANTHROPIC_*写进 Codex 的config.toml也不要把 Codex 的env_key套到 Claude Code。Claude Code 可以在~/.claude/settings.json中配置{ env: { ANTHROPIC_BASE_URL: https://taotoken.net/api, ANTHROPIC_AUTH_TOKEN: YOUR_API_KEY, ANTHROPIC_MODEL: 控制台可用模型名, ANTHROPIC_SMALL_FAST_MODEL: 控制台可用轻量模型名 } }如果你的 Claude Code 版本使用ANTHROPIC_API_KEY也可以按版本文档调整但核心仍然是 Base URL 指向https://taotoken.net/apiKey 使用YOUR_API_KEY。修改后重启终端或重新加载配置再用一个简单任务验证是否走到了 TaoToken。Claude Code 文档入口会在文末给出。Codex 使用~/.codex/config.toml配置方式与 Claude Code 不同model_provider taotoken model 控制台可用模型名 model_reasoning_effort medium [model_providers.taotoken] name TaoToken base_url https://taotoken.net/api env_key TAOTOKEN_API_KEY wire_api chat然后在本地环境变量中设置export TAOTOKEN_API_KEYYOUR_API_KEY这里env_key写的是环境变量名不是 Key 本身。Codex 会读取TAOTOKEN_API_KEY的值。不要把ANTHROPIC_AUTH_TOKEN填进env_key否则会出现找不到 Key 或 401。CC Switch 可以理解成供应商切换器配置时重点检查三件套配置项填什么说明Base URLhttps://taotoken.net/api不加 UTM不加多余斜杠API KeyYOUR_API_KEY从 TaoToken 控制台创建不要复用旧 KeyModel控制台模型名不要手写不存在的别名如果你在 CC Switch 里同时维护多个供应商建议给 TaoToken 单独命名例如taotoken-pro、taotoken-flash这样切换后能从模型名反推审计记录。更多入口和配置说明可以在 TaoToken 官网 查看工具接入说明 获取。注意编码工具本身不一定输出逐节点 token usage所以真正的细粒度审计仍建议放在 LlamaIndex 或你自己的请求网关层。6. 排障Base URL、401、404、流式响应无 usage 的审计兜底把入口统一到 TaoToken 后常见问题集中在四类401、404、429 和流式响应无 usage。下面按症状给出排查顺序。第一401 未授权。先确认 Key 是否完整再确认请求头格式是否为Authorization: Bearer YOUR_API_KEY。如果你用的是环境变量执行env | grep TAOTOKEN或env | grep ANTHROPIC检查当前 shell 是否真的加载。Claude Code 的settings.json修改后需要重启工具Codex 的TAOTOKEN_API_KEY需要在启动前导出。第二404 路径错误。最常见原因是 Base URL 被客户端二次拼接。TaoToken 的 Base URL 是https://taotoken.net/api。某些 OpenAI 兼容客户端会自己追加/v1这时你应以客户端实际请求路径为准但不要同时保留两个不同的域名。排查时可以先用最小 curl 请求确认通道。第三429 限流。Agentic RL 或批量审计场景容易并发过高触发限制后如果代码自动重试会产生额外 token。审计表里应记录重试次数和错误信息def mark_retry(record: dict, error: str) - None: record[retry] int(record.get(retry, 0)) 1 record[error] error[:200]然后在导出表里把retry 0的行标红或单独汇总。重试成本经常被低估尤其是长上下文请求失败后重试prompt_tokens 会重复计入。第四流式响应无 usage。某些兼容接口只在最后一个 chunk 返回 usage或者完全不返回。可以在回调里累积 chunkclass StreamingUsageMerger: def __init__(self): self.prompt_tokens 0 self.completion_tokens 0 self.total_tokens 0 self.text_parts [] def feed(self, chunk): if chunk is None: return if isinstance(chunk, dict): usage chunk.get(usage) delta chunk.get(choices, [{}])[0].get(delta, {}) content delta.get(content) else: usage getattr(chunk, usage, None) content getattr(chunk, delta, None) or getattr(chunk, text, None) if content: self.text_parts.append(str(content)) if usage: if isinstance(usage, dict): self.prompt_tokens usage.get(prompt_tokens, self.prompt_tokens) self.completion_tokens usage.get(completion_tokens, self.completion_tokens) self.total_tokens usage.get(total_tokens, self.total_tokens) else: self.prompt_tokens getattr(usage, prompt_tokens, self.prompt_tokens) self.completion_tokens getattr(usage, completion_tokens, self.completion_tokens) self.total_tokens getattr(usage, total_tokens, self.total_tokens) def fallback(self): if self.total_tokens 0: self.prompt_tokens 0 self.completion_tokens estimate_tokens(.join(self.text_parts)) self.total_tokens self.prompt_tokens self.completion_tokens return { prompt_tokens: self.prompt_tokens, completion_tokens: self.completion_tokens, total_tokens: self.total_tokens, usage_source: stream-merged if self.total_tokens else missing, }把流式合并结果写回审计记录再导出表格。这样即使接口不返回完整 usage你也能看到输出侧的 token 量级。不过要明确估算只用于发现异常和排障不能替代账单数据。除了错误排查还要看时间线。elapsed_ms和 token 数交叉分析很有用现象可能原因处理token 高、耗时低上下文太长模型快速消费压缩检索结果限制历史消息token 低、耗时长排队、网络、工具等待检查并发和超时同一 parent_id 多次 LLM重试或递归调用增加重试上限记录错误pro 模型集中在某个节点模型路由过重评估 flash 替换usage_source 多为 missing流式未合并或接口不返回启用合并与估算兜底最后审计表要能按trace_id聚合。虽然上面的 LlamaIndex 回调记录了event_id和parent_id但你可以在业务层给每次查询或 rollout 加一个外部 trace_id写进 payload 或日志。这样在 Agentic RL 后训练场景里可以把一条轨迹下的所有模型调用串起来看看究竟是规划阶段、工具阶段还是汇总阶段消耗最多。7. 文末 CTA模型对话、Coding Plan、创建 Key、Claude Code 文档如果你已经准备把推理链路审计落到可复现的流程里可以按下面的顺序操作。先验证模型对话确认 Base URL 和 Key 可用再决定是否把编码工具接入 Coding Plan然后创建独立 API Key避免和旧 Key 混用最后按 Claude Code 文档完成settings.json配置。以下入口都带 UTM便于区分来源模型对话先用一个短请求确认https://taotoken.net/api和YOUR_API_KEY可用。Coding Plan如果要把 Claude Code、Codex、CC Switch 一起接入先确认编码场景的模型与额度策略。创建 API Key为审计环境单独创建一个 Key不要和测试 Key 混用。Claude Code 文档按文档配置settings.json中的ANTHROPIC_*不要套用到 Codex。最后给一个本地行动清单。第一步在 TaoToken 官网 获取 Key 并确认可用模型名。第二步把 LlamaIndex 的api_base设为https://taotoken.net/apiKey 使用YOUR_API_KEY。第三步注册TokenAuditHandler运行一次带检索和工具调用的 query。第四步导出token_audit.csv整理成逐节点 Token 输出表重点看重试和长上下文节点。第五步如果还要接 Claude Code 或 Codex分别按settings.json和config.toml配置不要混用环境变量。做到这五步再讨论“谁在消耗 Token”你手里就有账本而不是只有一个总价。
