可观测性：从一行日志到一次事故回放
用户说“刚才的回答不对”，你至少要能回答四个问题：这次请求走了哪条路径？调用了哪些模型和工具？哪一步变慢或失败？这次请求消耗了多少资源？
只保存最终文本，通常答不出这些问题。下面用几段短代码逐步搭出一条观测链路。示例沿用企业服务台助手：用户询问制度时查知识库，需要查工单时调用工具，创建工单前要经过权限检查。
你可以把代码复制到自己的 Python 文件中，每完成一步就打印或查询一次结果，再继续下一步。
先分清三类数据
在写代码前先画一条线。生产系统里通常同时有三类数据：
一次请求
├─ 观测数据：日志、trace、指标，用来回答“发生了什么”
├─ 业务状态：任务状态、订单状态、审批记录，用来回答“现在是什么状态”
└─ 流式事件：text_delta、tool_pending、completed，用来让客户端重连和重放观测数据可以采样、脱敏和延迟上报；业务状态和流式事件则必须满足业务一致性。不要用 trace 代替 Response Job 的事件日志，也不要把完整用户回答复制到每一条日志里。
还有一个边界：观测系统记录模型的输入摘要、输出摘要、工具参数摘要和 usage，但不记录模型的隐藏推理过程。可以记录“模型选择了查询工单工具”，不应该试图保存或展示供应商未公开的 hidden reasoning。
第 1 步：给每次运行分配 run_id
这一小步要完成什么
先让一次请求的日志可以被串起来。run_id 应该在入口生成，后续的模型调用、检索、工具和错误都复用它。
from uuid import uuid4
def new_run_id() -> str:
return f"run_{uuid4().hex[:12]}"
run_id = new_run_id()
print(run_id)如果一次请求内部还有并行任务，可以再分配 span_id；如果用户重试同一个业务任务，可以保留 request_id，并为每次真正执行分配新的 run_id。不要把用户 ID、邮箱或订单号直接当作 run_id，它们可猜、可能包含隐私，也不能区分同一用户的两次运行。
运行观察
连续运行两次，确认两个 ID 不同。然后把它传进一个普通函数：
def retrieve_policy(query: str, run_id: str) -> list[str]:
print({"run_id": run_id, "step": "retrieve", "query_length": len(query)})
return ["VPN 申请需要直属主管审批。"]
retrieve_policy("怎么申请 VPN？", run_id)这里先记录问题长度，不记录完整问题。后面会再加可控的摘要和脱敏字段。
故意改错
把 retrieve_policy 里的 run_id 删除，改成函数内部重新生成。你会发现一次请求的检索日志和模型日志无法再关联。修回去后，再想一想：异步任务应该从哪里取得同一个 ID？答案是把它放进任务消息或上下文，而不是依赖线程全局变量。
第 2 步：把 print 换成结构化事件
先定义统一事件
纯文本日志适合人眼扫一眼，不适合按字段查询。先定义一个最小事件函数，每次输出一行 JSON：
import json
from datetime import datetime, timezone
from typing import Any
def log_event(run_id: str, event: str, **fields: Any) -> None:
record = {
"ts": datetime.now(timezone.utc).isoformat(),
"run_id": run_id,
"event": event,
**fields,
}
print(json.dumps(record, ensure_ascii=False, separators=(",", ":")))调用它：
log_event(
run_id,
"retrieval.finished",
result_count=3,
latency_ms=42,
status="ok",
)运行观察
你应该得到一条可以被日志系统解析的记录：
{"ts":"2026-09-11T08:00:00+00:00","run_id":"run_x","event":"retrieval.finished","result_count":3,"latency_ms":42,"status":"ok"}事件名建议使用“对象.动作”的形式，例如 model.started、model.finished、tool.failed、run.completed。不要同时出现 call_model_done、model_success 和 llm_end 三种叫法，否则后面统计时会漏数。
故意改错
把 latency_ms 写成字符串 "42ms"，然后尝试按数字求平均。修复时保留数值字段，把展示单位放到查询或前端。日志字段一旦用于聚合，就应该稳定地使用同一种类型。
第 3 步：记录一次模型调用的开始和结束
先记录时间，不急着接 tracing 平台
模型调用至少要有开始事件、结束事件和异常事件。用一个小包装器把三类事件统一起来：
from time import perf_counter
def call_model(model_client, messages: list[dict], run_id: str):
started = perf_counter()
log_event(run_id, "model.started", message_count=len(messages))
try:
response = model_client.generate(messages)
except Exception as exc:
log_event(
run_id,
"model.failed",
error_type=type(exc).__name__,
latency_ms=round((perf_counter() - started) * 1000),
)
raise
log_event(
run_id,
"model.finished",
model=response.model,
latency_ms=round((perf_counter() - started) * 1000),
finish_reason=response.finish_reason,
)
return response这里的 model_client 只是你所选 SDK 的薄适配层。不要在日志模块里写死某一家 SDK 的响应对象。
运行观察
用一个假的客户端返回：
class FakeResponse:
model = "demo-model"
finish_reason = "stop"
class FakeClient:
def generate(self, messages):
return FakeResponse()你应该看到 model.started 和 model.finished 两条事件，且后者有耗时。如果让 generate 抛出异常，应该看到 model.failed，并且异常仍然向上抛出，不能为了写日志把失败吞掉。
小练习
给模型事件增加 attempt 和 streaming 字段。想清楚：一次流式回答断线后重新续写，应该沿用同一个 run_id 还是新建一个？实践中可以沿用业务 response_id，但为每次实际调用增加 attempt，这样既能串起一次任务，也能区分恢复轮次。
第 4 步：把日志组织成 span 和 trace 树
为什么需要父子关系
只有 run_id 时，你能搜到一堆日志，却不容易看出“工具调用属于哪一次模型调用”。现在给事件增加 trace_id、span_id 和 parent_span_id：
from contextlib import contextmanager
from time import perf_counter
from uuid import uuid4
@contextmanager
def span(run_id: str, name: str, parent_span_id: str | None = None):
span_id = f"sp_{uuid4().hex[:10]}"
started = perf_counter()
log_event(
run_id,
"span.started",
trace_id=run_id,
span_id=span_id,
parent_span_id=parent_span_id,
span_name=name,
)
try:
yield span_id
except Exception as exc:
log_event(
run_id,
"span.failed",
trace_id=run_id,
span_id=span_id,
span_name=name,
error_type=type(exc).__name__,
latency_ms=round((perf_counter() - started) * 1000),
)
raise
else:
log_event(
run_id,
"span.finished",
trace_id=run_id,
span_id=span_id,
span_name=name,
latency_ms=round((perf_counter() - started) * 1000),
)使用时把模型和工具嵌套起来：
with span(run_id, "agent.run") as root_span:
with span(run_id, "model.call", root_span) as model_span:
with span(run_id, "tool.ticket_search", model_span):
ticket = find_ticket("INC-1001")最终可以在脑中还原成：
agent.run
└─ model.call
└─ tool.ticket_search这套字段与 OpenTelemetry tracing 的基本模型相近。接入 OpenTelemetry 时，保留业务字段命名，并让 SDK 负责真正的 span 上报；不要为了接平台再复制一套互不相同的事件格式。
故意改错
把 tool.ticket_search 的 parent_span_id 改成 None。调用树会出现两个根节点。排查时优先检查异步任务是否丢失了上下文，而不是先怀疑可视化平台。
第 5 步：给异步工具链路传递上下文
从同步调用扩展到队列
同步函数可以靠参数传递 run_id，队列任务也必须显式带上关联信息：
job = {
"run_id": run_id,
"parent_span_id": model_span_id,
"tool_call_id": "call_7f2",
"tool": "ticket_search",
"args_digest": "sha256:...",
}
queue.publish("ticket.search", job)Worker 取出后创建自己的 span：
def handle_ticket_job(job: dict) -> None:
with span(
job["run_id"],
"tool.ticket_search",
parent_span_id=job["parent_span_id"],
) as span_id:
log_event(
job["run_id"],
"tool.started",
span_id=span_id,
tool=job["tool"],
tool_call_id=job["tool_call_id"],
)
# 执行工具后再记录 tool.finished 或 tool.failed运行观察
让工具故意延迟 300 毫秒。回看 trace 时，应该能把“模型决定调用工具”和“队列 Worker 真正执行工具”串成同一棵树，同时看到排队耗时和执行耗时的区别。
这里不要只记录 tool_args 原文。工单标题、邮箱和检索关键词都可能包含隐私，先记录摘要、哈希或经过字段级脱敏的版本。
故意改错
把 parent_span_id 从队列消息中删掉，让 Worker 自己生成一个新的父节点。回看调用树时，工具会脱离原来的模型调用。修复方式是把父 span 的 ID 当作任务元数据传递；不要用“最近一次模型调用”这种全局猜测来补关系。
第 6 步：在适配层统一 usage 和成本
供应商字段只在边界转换
不同 API 可能把输入 token 叫 input_tokens 或 prompt_tokens，缓存字段也可能嵌套在 details 中。业务层只接收一套内部结构：
from dataclasses import dataclass
@dataclass
class Usage:
input_tokens: int = 0
cached_tokens: int = 0
cache_write_tokens: int = 0
output_tokens: int = 0
reasoning_tokens: int = 0
def normalize_usage(raw: dict) -> Usage:
input_details = raw.get("input_tokens_details") or raw.get("prompt_tokens_details") or {}
output_details = raw.get("output_tokens_details") or raw.get("completion_tokens_details") or {}
return Usage(
input_tokens=raw.get("input_tokens", raw.get("prompt_tokens", 0)),
cached_tokens=input_details.get("cached_tokens", 0),
cache_write_tokens=input_details.get("cache_write_tokens", 0),
output_tokens=raw.get("output_tokens", raw.get("completion_tokens", 0)),
reasoning_tokens=output_details.get("reasoning_tokens", 0),
)适配器要允许字段缺失。不要因为某个供应商没有 cache_write_tokens 就让整条请求失败，也不要把 reasoning_tokens 再加一次；它通常是输出 token 的明细。只有供应商明确说明单独计费时，才把它单列计价。
用价格版本计算估算值
价格不是代码常量。把价格表作为带版本的配置输入：
def estimate_cost(usage: Usage, price: dict) -> float:
uncached = max(
usage.input_tokens - usage.cached_tokens - usage.cache_write_tokens,
0,
)
return (
uncached / 1_000_000 * price["input"]
+ usage.cached_tokens / 1_000_000 * price.get("cached_input", price["input"])
+ usage.cache_write_tokens / 1_000_000 * price.get("cache_write", price["input"])
+ usage.output_tokens / 1_000_000 * price["output"]
)事件中记录 pricing_version 和 estimated_cost，不要把某个模型今天的价格写进教程结论。估算账单还应加上 embedding、reranker、搜索和图片等外部服务成本，且明确它是运营估算，不是供应商最终账单。
运行观察
用下面的输入验证计算：
usage = Usage(input_tokens=10_000, cached_tokens=6_000, cache_write_tokens=1_000, output_tokens=1_200)
price = {"input": 0.20, "cached_input": 0.02, "cache_write": 0.25, "output": 1.20}
print(round(estimate_cost(usage, price), 6))先手算，再对照程序结果。把 cache_write_tokens 改成 0，确认公式不会出现负数输入。
故意改错
把 cached_tokens 填成大于 input_tokens 的值。程序虽然可能仍然给出一个数字，但这说明适配层缺少数据校验。实际记录前应拒绝负的普通输入，并把异常 usage 标成 usage_invalid，交给供应商字段适配层排查。
第 7 步：观察缓存，而不是盲目追求命中率
Prompt cache 是多次请求之间复用稳定前缀的机制，和一次生成内部的 KV cache 不是一回事。你至少要记录：
{
"event": "model.finished",
"input_tokens": 4200,
"cached_tokens": 3000,
"cache_write_tokens": 0,
"cache_key_group": "support:v3",
"latency_ms": 910
}批量统计时使用：
cache_read_rate = sum(cached_tokens) / sum(input_tokens)
cache_write_rate = sum(cache_write_tokens) / sum(input_tokens)命中率上升不一定等于系统变好：如果稳定前缀包含了不该共享的租户数据，可能引入权限风险；如果为了缓存把检索结果塞进固定前缀，答案质量也可能变差。实际优化时，让 system prompt、工具 schema 和固定示例保持稳定，把用户问题和本轮检索结果放在后面；缓存 key 只按安全的租户、应用和提示词版本分组。
故意改错
在稳定前缀最前面加入当前时间和随机 request ID，然后连续发送两次相同请求。cached_tokens 下降时，检查变化内容是否污染了前缀。不要为了让命中率好看而删除必要的权限边界。
运行观察
连续跑三次同一类请求，比较 cached_tokens、cache_write_tokens 和 latency_ms，不要只看一个命中率。若第二、三次没有增加缓存读取，先检查前缀是否真的完全一致，再检查供应商是否提供了缓存字段。
第 8 步：错误、状态和流式事件要分开记录
统一错误事件
错误日志要让人知道“哪一步失败、能否重试、用户看到什么”，但不要把完整异常堆栈和敏感输入直接发给前端：
log_event(
run_id,
"tool.failed",
span_id=span_id,
tool="ticket_create",
error_code="PERMISSION_DENIED",
retryable=False,
user_message="当前账号没有创建工单的权限",
)模型的 finish_reason 也应归一化为内部状态，例如 completed、truncated_by_limit、content_blocked、tool_pending、cancelled、upstream_failed。不要把所有非 stop 的情况都叫“失败”，否则恢复策略会做错。
和流式事件日志的边界
流式事件面向客户端重连，观测事件面向排障。一个 text_delta 可以合并成较大的批次写入事件日志，但观测 trace 不必记录每个 token；记录 delta_count、首 token 时间和最终长度即可。
流式事件：text_delta(seq=18, text="需要直属主管审批")
观测事件：stream.finished(first_token_ms=420, delta_count=36, output_chars=812)这样既能按 seq 重放，也不会让观测系统因为每个 token 产生大量 span。
运行观察和故意改错
模拟一次断线：事件日志写到 seq=18，但 trace 只写出 stream.interrupted。恢复后检查两边是否仍能用 run_id 关联。故意把 seq 写进指标标签，观察指标时间序列数量迅速膨胀；修复时让 seq 只存在事件日志和单次查询中。
第 9 步：脱敏、采样和保留策略
先在写日志前脱敏
脱敏应该发生在日志函数或事件适配器边界，而不是数据已经进入日志平台后才补救：
import re
def redact(text: str) -> str:
text = re.sub(r"[\w.+-]+@[\w-]+\.[\w.-]+", "[EMAIL]", text)
text = re.sub(r"sk-[A-Za-z0-9_-]+", "[API_KEY]", text)
return text[:500]更可靠的做法是按字段处理：邮箱、手机号、身份证、访问令牌和工单正文分别使用不同策略。不要把 redact 当成完备的 DLP 系统；高风险字段最好直接不写原文，只写长度、哈希或分类标签。
再决定采样和保留时间
可以使用这样的起步策略：
所有错误和高延迟请求：保留完整结构化字段，但不保留敏感正文
成功请求：按比例采样 trace，保留聚合指标
高风险工具调用：无论是否采样，都保留审计事件
流式正文：按业务需要保留，和 trace 分开设置保留期采样不能让安全审计消失。把“是否采样”作为观测策略，把“是否记录高风险操作”作为合规规则，两者不要混为一谈。
小练习
设计一张保留策略表：成功请求保留多少天，错误请求保留多少天，高风险工具调用保留多久，谁可以读取原文。先写清楚数据用途和访问角色，再决定存储时间，不要从“日志平台默认保留多久”倒推业务规则。
第 10 步：从指标到慢请求告警
先定义能行动的指标
不要只盯着“请求数”。服务台 Agent 可以先记录：
run_success_rate
run_error_rate
model_ttft_ms
run_latency_ms
tool_latency_ms
tool_retry_count
input_tokens / output_tokens
estimated_cost按模型、工具、租户和版本分组时要小心高基数。run_id 可以用于查单次请求，但不应该作为指标标签，否则指标系统会被数百万个时间序列拖垮。
写一个简单的告警判断
def should_alert(latencies_ms: list[int], error_rate: float) -> bool:
if not latencies_ms:
return False
p95 = sorted(latencies_ms)[int(len(latencies_ms) * 0.95) - 1]
return p95 > 3000 or error_rate > 0.05告警还要加入时间窗口、最小样本数、连续触发次数和恢复条件。否则一条偶发慢请求就会把值班群吵醒。
运行观察
构造 20 个延迟样本，其中几个超过 3 秒，观察告警是否触发。再把样本数改成 1，确认你会加上“样本不足不告警”的条件。
故意改错
把 run_id 加到告警指标的标签里，再生成几百次请求。你会看到每个请求都产生一条新时间序列。修复时只保留低基数标签，例如模型版本、工具名和租户等级；单个 run_id 放到日志查询链接里。
第 11 步：用 trace 做一次事故回放
假设用户反馈：“创建工单花了很久，最后还失败了。”按下面顺序查询同一个 run_id：
1. run.started / run.completed：任务总时长和最终状态
2. span 树：模型、检索、权限检查、工单工具分别耗时多久
3. tool.failed：错误码、是否重试、是否已经产生副作用
4. model.finished：finish_reason、usage、成本和上下文长度
5. 流式事件日志：客户端最后收到哪个 seq你可能得到这样的结论：模型在 420 毫秒内决定调用工具，权限检查通过；工单服务第一次超时，第二次重试成功；但浏览器在最终 completed 事件前断开。这个结论需要同时看 trace、业务状态和流式事件，单看最终文本或单看日志都不够。
故意改错
删除 tool_call_id，只保留工具名。两次并行工单查询发生时，你会发现无法判断哪个结果对应哪次模型调用。修复时为每次工具调用保留稳定的调用 ID，并把它放入 span 属性和业务审计记录。
常见误区
把完整 Prompt 和完整回答全部写入 trace
这会扩大隐私泄露面、提高存储成本，也可能让日志查询变慢。优先记录消息数量、字符数、哈希、版本号和经过脱敏的短摘要；只有在明确授权、短期调试且有访问控制时，才临时保留原文。
把 hidden reasoning 当作必须采集的调试信息
你可以记录供应商返回的 reasoning_tokens 数量（如果有），用于 usage 和成本分析；不要要求模型暴露隐藏推理，也不要把它拼进用户可见日志。调试决策时看工具调用、状态转移和输入输出证据。
用 Redis Pub/Sub 或 trace 代替可回放事件日志
Pub/Sub 适合实时通知，断线后不能补历史；trace 适合排障，也不一定按客户端需要的 seq 存储。需要重连的回答应使用独立的 Response Job 事件日志，参见 流式响应恢复教程。
只看平均延迟和总成本
平均值会掩盖 P95/P99 的慢请求，总成本也掩盖单个异常循环。至少同时看分位数、错误率、最大轮数、重试次数和单次任务成本。
到这里你应该能完成什么
完成这些步骤后，你应该能够：
- 用
run_id找回一次完整运行； - 从 span 父子关系定位慢节点和重复工具调用；
- 在适配层统一不同 API 的 usage 字段；
- 区分缓存读取、缓存写入和普通输入成本；
- 在不记录 hidden reasoning 的前提下解释一次决策；
- 让错误、业务状态、流式事件和观测 trace 各司其职；
- 在日志进入平台前做脱敏，并为成功、失败和高风险操作设置不同保留策略；
- 用一次 trace、一次业务状态查询和一段事件日志完成事故回放。
下一步可以把这些事件接入 OpenTelemetry、LangGraph 或 Agent SDK 的 tracing 适配器。接入平台后，字段边界和安全策略仍然保持不变；平台只是帮你存储、查询和展示这些数据。