1. 为什么你的 LLM 调用需要一本“流水账”上线一个 LLM 应用之后最让人心里没底的不是模型答得不好而是你根本不知道它每次到底干了什么。产品经理跑来问“昨天那条回复为什么怪怪的”你只能翻应用日志、翻监控面板、翻代码里的 print最后发现是某次“小幅优化”把 System Prompt 改了一行或者上下文拼接多塞了一段历史又或者模型侧悄悄换了版本。这些问题的共同根源是每次 Prompt 调用对开发者来说是个黑盒。Prompt 可观测性要解决的就是这件事。它不是简单监控几个 QPS 指标而是把每一次 LLM 调用的输入、输出、延迟、Token 用量完整记录下来让调用链路可追溯、可量化、可归因。适合谁适合已经把 LLM 应用跑起来、开始关心成本归因和效果审计的开发者尤其是做 Agent、RAG、批量内容生成这类调用量大、Prompt 结构复杂的场景。我试过在业务代码里到处埋 print 和计时器结果是日志格式五花八门Token 数只能靠估算延迟统计口径不一致排查一次问题要拼半天。后来把观测逻辑收敛到一个统一的 API 通道上用同一把 Key 走所有调用日志和指标自然就对齐了。这篇就按这个思路给你一套可复制的埋点骨架从统一 Key 接入到一次调用链路的核对全部走一遍。2. 用 TaoToken 统一 Key 作为观测入口观测要落地第一步是让所有 LLM 调用走同一条通道。如果每个模型、每个服务各用各的 Key、各连各的地址日志就会散落在不同地方trace_id 也对不齐。TaoToken 在这里的作用是提供一个统一的 API 入口和 Key 管理你可以在一个控制台里看到所有模型的调用接入方式兼容常见的 OpenAI 风格接口改一行 base_url 就能接上。具体来说你需要先拿到一把 API Key。打开控制台https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite在 API Keys 页面创建一个新 Key建议按环境区分命名比如prod-agent、staging-rag这样后面做成本归因时能直接按 Key 维度拆分。创建完成后复制保存页面只显示一次。拿到 Key 之后接入地址用https://taotoken.net/api不要带任何多余路径。模型名按你实际要用的填比如gpt-4o、claude-3-5-sonnet这类。如果你用的是 OpenAI SDK只需要改两个地方from openai import OpenAI client OpenAI( api_keysk-你的TaoTokenKey, base_urlhttps://taotoken.net/api, ) resp client.chat.completions.create( modelgpt-4o, messages[{role: user, content: 用一句话解释什么是可观测性}], ) print(resp.choices[0].message.content)这一步的意义不只是“能调通”而是你后续所有的埋点、日志、指标都挂在这条统一通道上。Key 是观测的锚点trace 是观测的骨架两者配合才能把一次调用完整串起来。如果你还没决定用哪些模型可以先去模型对话页面https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite试几条 Prompt确认模型行为符合预期再写进生产代码。3. 可复制的日志埋点配置骨架观测的核心是“包装”。不要在每个业务函数里手写计时和日志而是写一个中间件把 LLM 调用包起来自动记录输入快照、首 Token 时间、总延迟、Token 用量和输出文本。下面这份骨架可以直接复制到你的项目里按需改字段。先定义一次调用的数据结构用 dataclass 保证字段清晰import time import uuid import json from dataclasses import dataclass, field, asdict from typing import Optional dataclass class PromptTrace: trace_id: str field(default_factorylambda: str(uuid.uuid4())) # 输入快照 system_prompt: str user_input: str context_injected: str # 时间线 start_time: float 0.0 first_token_time: float 0.0 end_time: float 0.0 ttft_ms: float 0.0 total_ms: float 0.0 # 输出 output_text: str stop_reason: str # Token 用量 prompt_tokens: int 0 completion_tokens: int 0 total_tokens: int 0 # 元数据 model_name: str request_id: str user_id: str session_id: str # 错误 error: Optional[str] None error_type: Optional[str] None def to_log_line(self) - str: return json.dumps(asdict(self), ensure_asciiFalse, defaultstr)然后是观测中间件负责包装调用、计时、抓取 Token 用量。注意 Token 用量优先从 API 返回的usage字段读取读不到再用估算兜底import asyncio class PromptObserver: def __init__(self, log_writer, sample_rate: float 1.0, max_snapshot_chars: int 5000): self._log log_writer self._sample_rate sample_rate self._max_snapshot max_snapshot_chars async def observe(self, client, model: str, messages: list, metadata: dict | None None): trace PromptTrace(model_namemodel, **(metadata or {})) # 拆解输入快照 for m in messages: if m[role] system: trace.system_prompt m[content][:self._max_snapshot] elif m[role] user: trace.user_input m[content][:self._max_snapshot] trace.start_time time.monotonic() try: resp await client.chat.completions.create( modelmodel, messagesmessages, ) trace.end_time time.monotonic() trace.total_ms (trace.end_time - trace.start_time) * 1000 trace.output_text resp.choices[0].message.content trace.stop_reason resp.choices[0].finish_reason trace.request_id getattr(resp, id, ) if resp.usage: trace.prompt_tokens resp.usage.prompt_tokens trace.completion_tokens resp.usage.completion_tokens trace.total_tokens resp.usage.total_tokens return resp except Exception as exc: trace.error str(exc) trace.error_type type(exc).__name__ trace.end_time time.monotonic() trace.total_ms (trace.end_time - trace.start_time) * 1000 raise finally: asyncio.create_task(self._log.write(trace.to_log_line()))日志写入器用异步批量刷盘避免每条调用都同步写文件拖慢主流程class AsyncLogWriter: def __init__(self, file_path: str, flush_interval: float 5.0, max_buffer_size: int 1000): self._path file_path self._interval flush_interval self._max_buffer max_buffer_size self._buffer: list[str] [] self._lock asyncio.Lock() async def write(self, line: str): async with self._lock: self._buffer.append(line) if len(self._buffer) self._max_buffer: await self._flush() async def _flush(self): if not self._buffer: return lines self._buffer.copy() self._buffer.clear() await asyncio.to_thread(self._write_to_file, lines) def _write_to_file(self, lines: list[str]): with open(self._path, a, encodingutf-8) as f: for line in lines: f.write(line \n)这套骨架的关键设计点有三个。第一trace_id在调用发起时生成贯穿整条链路后面查日志就靠它。第二Token 用量以 API 返回的usage为准估算只作为兜底避免成本归因失真。第三日志写入放在finally里异步执行即使调用抛异常也能记录不会因为观测逻辑本身影响主流程。4. 验证一次调用链路是否真的可追溯埋点写完不算完得验证数据真的落下来了、字段真的对得上。下面走一遍完整的核对步骤。第一步发起一次真实调用把 trace_id 打出来import asyncio async def main(): writer AsyncLogWriter(prompt_trace.jsonl) observer PromptObserver(writer) client AsyncOpenAI( api_keysk-你的TaoTokenKey, base_urlhttps://taotoken.net/api, ) resp await observer.observe( clientclient, modelgpt-4o, messages[ {role: system, content: 你是一个简洁的助手}, {role: user, content: 用一句话解释 Token 是什么}, ], metadata{user_id: u_1001, session_id: s_abc}, ) print(输出:, resp.choices[0].message.content) print(Token 用量:, resp.usage.total_tokens) asyncio.run(main())第二步等几秒让异步刷盘完成然后查看日志文件tail -n 1 prompt_trace.jsonl | python -m json.tool你应该能看到类似这样的结构{ trace_id: 3f2a..., system_prompt: 你是一个简洁的助手, user_input: 用一句话解释 Token 是什么, total_ms: 842.3, output_text: Token 是模型处理文本的最小单位..., prompt_tokens: 28, completion_tokens: 35, total_tokens: 63, model_name: gpt-4o, user_id: u_1001, session_id: s_abc }第三步核对三个关键点。延迟方面total_ms应该和你手动计时的结果接近误差在几十毫秒内算正常。Token 方面total_tokens应该等于prompt_tokens completion_tokens且和 API 返回的 usage 一致。输入输出方面system_prompt和user_input应该完整还原你传入的内容没有被截断除非超过max_snapshot_chars。第四步做一次成本归因的快速验证。按 Key 维度统计当天所有调用的 Token 总量乘以对应模型的单价和账单对一下。如果差异超过 5%优先检查是不是有调用没走观测中间件或者 Token 估算兜底被误用了。如果你在验证过程中发现某些字段缺失比如request_id为空先确认你用的 SDK 版本是否返回了该字段如果usage为 None检查是不是用了流式模式但没开stream_options{include_usage: True}。这些细节在接入文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite里有对应说明。5. 本篇常见错排查报错一401 Unauthorized提示 Key 无效。最常见的原因是 Key 复制时带了空格或者用了已经删除的 Key。去 API Keys 页面重新生成一把注意创建后只显示一次。另外确认base_url写的是https://taotoken.net/api不要多加/v1之类的路径。报错二日志文件里total_tokens一直是 0。说明resp.usage没取到。如果你用的是流式调用需要在请求参数里加stream_options{include_usage: True}否则最后一个 chunk 不会带 usage。非流式调用一般都有检查一下 SDK 版本是否过旧。报错三ttft_ms始终为 0。这份骨架里 TTFT 需要流式模式才能测非流式调用只有总延迟。如果你需要首 Token 延迟指标把调用改成流式在收到第一个 chunk 时记录时间戳。注意流式模式下output_text需要自己拼接所有 chunk。报错四日志写入延迟高主流程被拖慢。检查max_buffer_size是不是设得太小导致频繁刷盘。一般设 500 到 1000 比较合适。另外确认_write_to_file走的是asyncio.to_thread没有阻塞事件循环。报错五采样率设了 0.1但错误调用没被记录。采样逻辑要区分正常和异常错误调用建议全量记录。在observe的finally里判断trace.error是否为空有错误就跳过采样直接写。报错六多进程部署时日志文件互相覆盖。每个进程写同一个文件会出问题。按进程 ID 分文件比如prompt_trace_{pid}.jsonl或者直接写到 stdout 由日志采集器统一收集。6. 把观测数据用起来数据落下来只是第一步真正产生价值的是拿它回答问题。比如“上周 Prompt 效果为什么变差”你可以按session_id拉出那段时间的所有 trace对比system_prompt字段有没有变化再看output_text的分布是否偏移。“哪个环节在拖慢延迟”按model_name分组统计total_ms的 P50 和 P99如果某个模型 P99 突然飙高基本能定位到模型侧或网络侧。“Token 成本花在哪里”按user_id或session_id聚合total_tokens找出消耗大户。如果你做的是长期编码或 Agent 场景调用量大、Prompt 结构复杂建议把观测数据和 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite结合使用按项目维度拆分 Key这样成本归因能直接落到具体项目上不用事后猜。接入过程中遇到字段对不上或者 SDK 兼容问题优先查接入文档大部分坑里面都有记录。 Prompt 可观测性实战：用 TaoToken 统一 Key 记录每次 LLM 调用的输入输出、延迟与 Token
1. 为什么你的 LLM 调用需要一本“流水账”上线一个 LLM 应用之后最让人心里没底的不是模型答得不好而是你根本不知道它每次到底干了什么。产品经理跑来问“昨天那条回复为什么怪怪的”你只能翻应用日志、翻监控面板、翻代码里的 print最后发现是某次“小幅优化”把 System Prompt 改了一行或者上下文拼接多塞了一段历史又或者模型侧悄悄换了版本。这些问题的共同根源是每次 Prompt 调用对开发者来说是个黑盒。Prompt 可观测性要解决的就是这件事。它不是简单监控几个 QPS 指标而是把每一次 LLM 调用的输入、输出、延迟、Token 用量完整记录下来让调用链路可追溯、可量化、可归因。适合谁适合已经把 LLM 应用跑起来、开始关心成本归因和效果审计的开发者尤其是做 Agent、RAG、批量内容生成这类调用量大、Prompt 结构复杂的场景。我试过在业务代码里到处埋 print 和计时器结果是日志格式五花八门Token 数只能靠估算延迟统计口径不一致排查一次问题要拼半天。后来把观测逻辑收敛到一个统一的 API 通道上用同一把 Key 走所有调用日志和指标自然就对齐了。这篇就按这个思路给你一套可复制的埋点骨架从统一 Key 接入到一次调用链路的核对全部走一遍。2. 用 TaoToken 统一 Key 作为观测入口观测要落地第一步是让所有 LLM 调用走同一条通道。如果每个模型、每个服务各用各的 Key、各连各的地址日志就会散落在不同地方trace_id 也对不齐。TaoToken 在这里的作用是提供一个统一的 API 入口和 Key 管理你可以在一个控制台里看到所有模型的调用接入方式兼容常见的 OpenAI 风格接口改一行 base_url 就能接上。具体来说你需要先拿到一把 API Key。打开控制台https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite在 API Keys 页面创建一个新 Key建议按环境区分命名比如prod-agent、staging-rag这样后面做成本归因时能直接按 Key 维度拆分。创建完成后复制保存页面只显示一次。拿到 Key 之后接入地址用https://taotoken.net/api不要带任何多余路径。模型名按你实际要用的填比如gpt-4o、claude-3-5-sonnet这类。如果你用的是 OpenAI SDK只需要改两个地方from openai import OpenAI client OpenAI( api_keysk-你的TaoTokenKey, base_urlhttps://taotoken.net/api, ) resp client.chat.completions.create( modelgpt-4o, messages[{role: user, content: 用一句话解释什么是可观测性}], ) print(resp.choices[0].message.content)这一步的意义不只是“能调通”而是你后续所有的埋点、日志、指标都挂在这条统一通道上。Key 是观测的锚点trace 是观测的骨架两者配合才能把一次调用完整串起来。如果你还没决定用哪些模型可以先去模型对话页面https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite试几条 Prompt确认模型行为符合预期再写进生产代码。3. 可复制的日志埋点配置骨架观测的核心是“包装”。不要在每个业务函数里手写计时和日志而是写一个中间件把 LLM 调用包起来自动记录输入快照、首 Token 时间、总延迟、Token 用量和输出文本。下面这份骨架可以直接复制到你的项目里按需改字段。先定义一次调用的数据结构用 dataclass 保证字段清晰import time import uuid import json from dataclasses import dataclass, field, asdict from typing import Optional dataclass class PromptTrace: trace_id: str field(default_factorylambda: str(uuid.uuid4())) # 输入快照 system_prompt: str user_input: str context_injected: str # 时间线 start_time: float 0.0 first_token_time: float 0.0 end_time: float 0.0 ttft_ms: float 0.0 total_ms: float 0.0 # 输出 output_text: str stop_reason: str # Token 用量 prompt_tokens: int 0 completion_tokens: int 0 total_tokens: int 0 # 元数据 model_name: str request_id: str user_id: str session_id: str # 错误 error: Optional[str] None error_type: Optional[str] None def to_log_line(self) - str: return json.dumps(asdict(self), ensure_asciiFalse, defaultstr)然后是观测中间件负责包装调用、计时、抓取 Token 用量。注意 Token 用量优先从 API 返回的usage字段读取读不到再用估算兜底import asyncio class PromptObserver: def __init__(self, log_writer, sample_rate: float 1.0, max_snapshot_chars: int 5000): self._log log_writer self._sample_rate sample_rate self._max_snapshot max_snapshot_chars async def observe(self, client, model: str, messages: list, metadata: dict | None None): trace PromptTrace(model_namemodel, **(metadata or {})) # 拆解输入快照 for m in messages: if m[role] system: trace.system_prompt m[content][:self._max_snapshot] elif m[role] user: trace.user_input m[content][:self._max_snapshot] trace.start_time time.monotonic() try: resp await client.chat.completions.create( modelmodel, messagesmessages, ) trace.end_time time.monotonic() trace.total_ms (trace.end_time - trace.start_time) * 1000 trace.output_text resp.choices[0].message.content trace.stop_reason resp.choices[0].finish_reason trace.request_id getattr(resp, id, ) if resp.usage: trace.prompt_tokens resp.usage.prompt_tokens trace.completion_tokens resp.usage.completion_tokens trace.total_tokens resp.usage.total_tokens return resp except Exception as exc: trace.error str(exc) trace.error_type type(exc).__name__ trace.end_time time.monotonic() trace.total_ms (trace.end_time - trace.start_time) * 1000 raise finally: asyncio.create_task(self._log.write(trace.to_log_line()))日志写入器用异步批量刷盘避免每条调用都同步写文件拖慢主流程class AsyncLogWriter: def __init__(self, file_path: str, flush_interval: float 5.0, max_buffer_size: int 1000): self._path file_path self._interval flush_interval self._max_buffer max_buffer_size self._buffer: list[str] [] self._lock asyncio.Lock() async def write(self, line: str): async with self._lock: self._buffer.append(line) if len(self._buffer) self._max_buffer: await self._flush() async def _flush(self): if not self._buffer: return lines self._buffer.copy() self._buffer.clear() await asyncio.to_thread(self._write_to_file, lines) def _write_to_file(self, lines: list[str]): with open(self._path, a, encodingutf-8) as f: for line in lines: f.write(line \n)这套骨架的关键设计点有三个。第一trace_id在调用发起时生成贯穿整条链路后面查日志就靠它。第二Token 用量以 API 返回的usage为准估算只作为兜底避免成本归因失真。第三日志写入放在finally里异步执行即使调用抛异常也能记录不会因为观测逻辑本身影响主流程。4. 验证一次调用链路是否真的可追溯埋点写完不算完得验证数据真的落下来了、字段真的对得上。下面走一遍完整的核对步骤。第一步发起一次真实调用把 trace_id 打出来import asyncio async def main(): writer AsyncLogWriter(prompt_trace.jsonl) observer PromptObserver(writer) client AsyncOpenAI( api_keysk-你的TaoTokenKey, base_urlhttps://taotoken.net/api, ) resp await observer.observe( clientclient, modelgpt-4o, messages[ {role: system, content: 你是一个简洁的助手}, {role: user, content: 用一句话解释 Token 是什么}, ], metadata{user_id: u_1001, session_id: s_abc}, ) print(输出:, resp.choices[0].message.content) print(Token 用量:, resp.usage.total_tokens) asyncio.run(main())第二步等几秒让异步刷盘完成然后查看日志文件tail -n 1 prompt_trace.jsonl | python -m json.tool你应该能看到类似这样的结构{ trace_id: 3f2a..., system_prompt: 你是一个简洁的助手, user_input: 用一句话解释 Token 是什么, total_ms: 842.3, output_text: Token 是模型处理文本的最小单位..., prompt_tokens: 28, completion_tokens: 35, total_tokens: 63, model_name: gpt-4o, user_id: u_1001, session_id: s_abc }第三步核对三个关键点。延迟方面total_ms应该和你手动计时的结果接近误差在几十毫秒内算正常。Token 方面total_tokens应该等于prompt_tokens completion_tokens且和 API 返回的 usage 一致。输入输出方面system_prompt和user_input应该完整还原你传入的内容没有被截断除非超过max_snapshot_chars。第四步做一次成本归因的快速验证。按 Key 维度统计当天所有调用的 Token 总量乘以对应模型的单价和账单对一下。如果差异超过 5%优先检查是不是有调用没走观测中间件或者 Token 估算兜底被误用了。如果你在验证过程中发现某些字段缺失比如request_id为空先确认你用的 SDK 版本是否返回了该字段如果usage为 None检查是不是用了流式模式但没开stream_options{include_usage: True}。这些细节在接入文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite里有对应说明。5. 本篇常见错排查报错一401 Unauthorized提示 Key 无效。最常见的原因是 Key 复制时带了空格或者用了已经删除的 Key。去 API Keys 页面重新生成一把注意创建后只显示一次。另外确认base_url写的是https://taotoken.net/api不要多加/v1之类的路径。报错二日志文件里total_tokens一直是 0。说明resp.usage没取到。如果你用的是流式调用需要在请求参数里加stream_options{include_usage: True}否则最后一个 chunk 不会带 usage。非流式调用一般都有检查一下 SDK 版本是否过旧。报错三ttft_ms始终为 0。这份骨架里 TTFT 需要流式模式才能测非流式调用只有总延迟。如果你需要首 Token 延迟指标把调用改成流式在收到第一个 chunk 时记录时间戳。注意流式模式下output_text需要自己拼接所有 chunk。报错四日志写入延迟高主流程被拖慢。检查max_buffer_size是不是设得太小导致频繁刷盘。一般设 500 到 1000 比较合适。另外确认_write_to_file走的是asyncio.to_thread没有阻塞事件循环。报错五采样率设了 0.1但错误调用没被记录。采样逻辑要区分正常和异常错误调用建议全量记录。在observe的finally里判断trace.error是否为空有错误就跳过采样直接写。报错六多进程部署时日志文件互相覆盖。每个进程写同一个文件会出问题。按进程 ID 分文件比如prompt_trace_{pid}.jsonl或者直接写到 stdout 由日志采集器统一收集。6. 把观测数据用起来数据落下来只是第一步真正产生价值的是拿它回答问题。比如“上周 Prompt 效果为什么变差”你可以按session_id拉出那段时间的所有 trace对比system_prompt字段有没有变化再看output_text的分布是否偏移。“哪个环节在拖慢延迟”按model_name分组统计total_ms的 P50 和 P99如果某个模型 P99 突然飙高基本能定位到模型侧或网络侧。“Token 成本花在哪里”按user_id或session_id聚合total_tokens找出消耗大户。如果你做的是长期编码或 Agent 场景调用量大、Prompt 结构复杂建议把观测数据和 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite结合使用按项目维度拆分 Key这样成本归因能直接落到具体项目上不用事后猜。接入过程中遇到字段对不上或者 SDK 兼容问题优先查接入文档大部分坑里面都有记录。