一次用户提问会穿过网关、模型供应商、工具执行和重试，普通访问日志很难把首字延迟、工具失败和费用对到同一条链路上。按现有微服务的追踪习惯，把 prompt 版本、tool span、token 用量与成本写成 OpenTelemetry 属性与子跨度即可。同一套排障方式就能定位超时、贵请求和局部失败，而不必先养一套 AI 专用控制台。
访问日志对不齐的是链路，不是字段
LLM 请求的麻烦不在于少打了几行日志，而在于一次提问本来就不是一次 HTTP。网关接入、供应商 chat、工具回调、失败重试，每一段都有自己的耗时和失败模式。访问日志通常只有外层状态码和总耗时：用户侧可能是 200，里面已经超时过一次天气 API；账单上的费用要等 usage 回来才能算，和「请求结束」不在同一个时刻；流式场景里用户感知的是首字延迟，不是整段 duration_ms。
把这些事件靠 request_id 事后拼接，也能做，但你等于在维护第二套因果顺序。微服务里已经有现成答案：一条 Trace，若干 Span，属性记录稳定维度，Event 记录过程中的点。模型调用无非是一种高延迟、带计量的出站依赖，不必换成另一套观测语言。
下面的约定很克制：能用 OpenTelemetry 语义约定里已有的 gen_ai.* 就用；约定没覆盖的（prompt 版本、美元成本、第几次重试）用 app.* 自定义属性。不要把完整 prompt 默认写进属性——体积、检索噪声和合规都划不来。需要排障时再对脱敏后的片段打 Event。
一次提问：父 Span 包住模型、工具和重试
建议的拓扑很简单：
- 入口 Span：网关的
POST /v1/chat（SpanKind.SERVER），表示用户这一次提问。 - 模型 Span：
chat {model}（SpanKind.CLIENT），表示对供应商的一次或一轮对话。 - 工具 Span：
execute_tool {name}，每一次执行、每一次重试都单独开子 Span，失败就StatusCode.ERROR并record_exception。
这样在 Jaeger / Grafana Tempo / 任何 OTLP 后端里，超时点、贵请求、局部失败都落在同一条时间线上。生产环境把导出器换成 OTLP 即可，和现有微服务共用同一套 Collector。
下面是一份可直接运行的脚本，只依赖 opentelemetry-api 与 opentelemetry-sdk。模型与工具用占位实现，Span API 都是 SDK 里真实存在的写法；换成供应商 SDK 时，保留 Span 边界，把 fake_model_call 换成真实响应即可。
import time
from decimal import Decimal, ROUND_HALF_UP
from opentelemetry import trace
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import ConsoleSpanExporter, SimpleSpanProcessor
from opentelemetry.trace import SpanKind, Status, StatusCode
resource = Resource.create({
"service.name": "llm-gateway",
"service.version": "0.1.0",
})
provider = TracerProvider(resource=resource)
provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter()))
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("llm-gateway.chat", "0.1.0")
# 示例单价，务必换成供应商当前价目；单位 USD / token
PRICE_USD_PER_TOKEN = {
"gpt-4o-mini": {
"input": Decimal("0.00000015"),
"output": Decimal("0.0000006"),
},
}
def usd_cost(model: str, input_tokens: int, output_tokens: int) -> float:
price = PRICE_USD_PER_TOKEN[model]
total = price["input"] * input_tokens + price["output"] * output_tokens
return float(total.quantize(Decimal("0.000001"), rounding=ROUND_HALF_UP))
def fake_model_call() -> dict:
"""占位：真实项目替换为供应商 chat 调用，并从响应里读 usage。"""
time.sleep(0.05)
first_token_at = time.perf_counter()
time.sleep(0.08)
return {
"first_token_at": first_token_at,
"output": "今天上海多云，气温 22 度。",
"input_tokens": 96,
"output_tokens": 18,
"model": "gpt-4o-mini",
"finish_reason": "stop",
}
def execute_tool(name: str, arguments: dict, attempt: int) -> dict:
if name == "get_weather" and attempt == 1:
raise TimeoutError("upstream weather api timeout")
time.sleep(0.02)
return {"city": arguments["city"], "weather": "cloudy", "temp_c": 22}
def handle_user_question(question: str, prompt_version: str) -> str:
request_started = time.perf_counter()
with tracer.start_as_current_span("POST /v1/chat", kind=SpanKind.SERVER) as root:
root.set_attribute("http.route", "/v1/chat")
root.set_attribute("app.prompt.version", prompt_version)
root.set_attribute("app.user.question_chars", len(question))
model = "gpt-4o-mini"
input_tokens = 0
output_tokens = 0
with tracer.start_as_current_span(f"chat {model}", kind=SpanKind.CLIENT) as chat:
chat.set_attribute("gen_ai.operation.name", "chat")
chat.set_attribute("gen_ai.provider.name", "openai")
chat.set_attribute("gen_ai.request.model", model)
chat.set_attribute("app.prompt.version", prompt_version)
first = fake_model_call()
input_tokens += first["input_tokens"]
output_tokens += first["output_tokens"]
chat.add_event(
"model.first_token",
attributes={"app.llm.stage": "plan_tools"},
)
tool_name = "get_weather"
tool_args = {"city": "上海"}
tool_result = None
last_error = None
for attempt in (1, 2):
with tracer.start_as_current_span(
f"execute_tool {tool_name}",
kind=SpanKind.CLIENT,
) as tool_span:
tool_span.set_attribute("gen_ai.operation.name", "execute_tool")
tool_span.set_attribute("gen_ai.tool.name", tool_name)
tool_span.set_attribute("app.tool.attempt", attempt)
try:
tool_result = execute_tool(tool_name, tool_args, attempt)
tool_span.set_attribute("app.tool.ok", True)
except TimeoutError as exc:
last_error = exc
tool_span.set_status(Status(StatusCode.ERROR, str(exc)))
tool_span.record_exception(exc)
tool_span.set_attribute("error.type", type(exc).__name__)
tool_span.set_attribute("app.tool.ok", False)
if tool_result is None:
chat.set_status(Status(StatusCode.ERROR, str(last_error)))
raise last_error
final = fake_model_call()
input_tokens += 40 + final["input_tokens"]
output_tokens += final["output_tokens"]
chat.add_event(
"first_token",
attributes={
"app.llm.stage": "user_visible",
"app.llm.ttft_ms": int(
(final["first_token_at"] - request_started) * 1000
),
},
)
cost = usd_cost(model, input_tokens, output_tokens)
chat.set_attribute("gen_ai.usage.input_tokens", input_tokens)
chat.set_attribute("gen_ai.usage.output_tokens", output_tokens)
chat.set_attribute("gen_ai.response.finish_reasons", ("stop",))
chat.set_attribute("app.llm.cost_usd", cost)
root.set_attribute("gen_ai.usage.input_tokens", input_tokens)
root.set_attribute("gen_ai.usage.output_tokens", output_tokens)
root.set_attribute("app.llm.cost_usd", cost)
return final["output"]
if __name__ == "__main__":
print(handle_user_question("上海今天天气怎么样？", "weather-qa@2025-03-12"))跑起来后，控制台会打出同一 trace_id 下的入口 Span、chat Span，以及两次 execute_tool（第一次 ERROR，第二次 OK）。这已经足够回答「慢在模型还是慢在工具」「失败有没有被重试吃掉」。
Prompt 版本、Token、成本：属性而不是另一份日志
Token 和费用只在供应商响应返回后才有，所以写在 Span 结束前 的属性里，而不是再打一条 billing 日志。gen_ai.usage.input_tokens / gen_ai.usage.output_tokens 是语义约定里已有的名字；成本没有统一字段，用 app.llm.cost_usd 即可。单价不要写死在观测后端的推导里，让网关用当前价目表算完再上报，避免后面改价导致历史 Trace 无法复算。
几个容易对不齐的细节：
- prompt 版本用
app.prompt.version（或你们自己的模板 ID）。模型名只能说明权重，说明不了系统提示词改过哪一版。 - 缓存命中 token如果供应商单独返回，用自定义属性另记，不要合并进
input_tokens，否则费用对不上发票。 - 入口 Span 也抄一份 usage 和 cost。按费用筛「贵请求」时，不必先展开子 Span；按工具名筛失败时，再下钻
gen_ai.tool.name。 - 属性只放稳定、可过滤的标量。对话全文、工具原始 JSON 需要时打 Event，并截断、脱敏。
网关若已有 W3C traceparent，继续用 OpenTelemetry 的上下文传播即可，模型与工具会自动挂到用户那条链上，不必为 LLM 再发明一个 session id。
工具失败、超时与首字延迟
工具不要把错误吞进模型的 200 响应里再只打日志。每一次执行都是子 Span：超时、4xx、解析失败设 Status 和 error.type，app.tool.attempt 区分第几次。父 chat Span 是否标 ERROR，取决于你有没有降级成功——重试成功则父 Span 可以保持 OK，失败的子 Span 仍然可见。这和下游 HTTP 重试的排障方式一致：局部失败不必等于整单失败，但必须能在同一条 Trace 里看见。
首字延迟不要再写一行独立日志。流式响应里，读到第一个 chunk 时对当前 Span add_event("first_token")。注意两件事不是同一个点：
model.first_token：某一次供应商调用自己的首字，用来看模型或地域延迟。- 用户可感知的
first_token：往往在工具跑完之后。用app.llm.stage分开，避免把「工具排队」误判成「模型变慢」。
Event 的时间戳跟着 Span，和 token、成本、工具错误共享 trace_id。后端里按 app.llm.cost_usd 找贵请求，按 error.type=TimeoutError 找工具超时，按 first_token 的时间相对入口 Span 算 TTFT，都还是同一套 TraceQL / 检索语法。
总结
LLM 观测要对齐的是一次提问的因果链：入口、模型、工具、重试、计量。OpenTelemetry 已经提供 Span、属性、Event 和状态，缺的只是你是否把 prompt 版本、tool 尝试、token 和成本写进去。日志仍可保留审计文本，但排障主路径应回到现有追踪后端，而不是再做一个只能看对话的控制台。先把上面这条拓扑接到 Collector；等真有「跨请求的 prompt 回归」或「按租户摊销」需求，再在同一份属性上做聚合，而不是先换一套体系。
评论