AI 服务可观测性设计

AI 服务特有的可观测性需求：LLM 调用的结构化日志、token 用量追踪、延迟指标、SSE 管道的 trace ID 串联、告警规则设计。

#type / concept #status / growing #tech / ai #tech / ops

[!info] related notes

所属 MOC: BodySense MOC

平台级可观测: Go 可观测性

链路追踪: 多服务分布式追踪

质量门禁: [[ai-quality-gates|AI 质量门禁]]

降级策略: [[ai-service-graceful-degradation|AI 服务优雅降级]]

AI 服务可观测性设计

这篇解决什么问题

传统服务的可观测性关注”请求是否成功、延迟多少”。AI 服务在此基础上还需要回答：

LLM 花了多少钱 — 每次调用消耗了多少 token？
LLM 回答得好不好 — faithfulness 检查通过了吗？red flag 触发了吗？
流式管道断在哪 — SSE 事件流在哪个环节中断了？
RAG 检索质量如何 — 检索到的知识和用户问题相关吗？

通用可观测性（如 go-observability）解决的是”服务挂了怎么排查”，这里解决的是”AI 回答错了怎么排查”。

LLM 调用的结构化日志

每次 LLM 调用必须记录以下字段：

logger.info("llm_call_completed",
    # 请求标识
    request_id=req_id,
    session_id=session_id,
    turn_index=turn_index,

    # 模型信息
    model=model_name,
    provider=provider_name,

    # Token 用量
    prompt_tokens=usage.prompt_tokens,
    completion_tokens=usage.completion_tokens,
    total_tokens=usage.total_tokens,

    # 延迟
    latency_ms=latency_ms,
    time_to_first_token_ms=ttft_ms,

    # 质量指标
    faithfulness_passed=faithfulness_result.faithful,
    red_flag_triggered=len(red_flags) > 0,
    red_flag_categories=[f.category for f in red_flags],

    # 工具调用
    tool_calls=[tc.name for tc in tool_calls],
    tool_call_count=len(tool_calls),
)

关键设计：

request_id 贯穿整个请求链路（前端 → Go → Python），用于跨服务关联
session_id + turn_index 用于重建完整对话上下文
token 用量和延迟必须记录——这是成本和性能分析的基础

Token 用量追踪

# 按维度聚合 token 用量
class TokenTracker:
    def record(self, session_id: str, model: str, usage: TokenUsage):
        # 写入时序数据库或 Prometheus histogram
        TOKEN_COUNTER.labels(
            model=model,
            direction="prompt"
        ).inc(usage.prompt_tokens)
        TOKEN_COUNTER.labels(
            model=model,
            direction="completion"
        ).inc(usage.completion_tokens)

需要监控的维度：

维度	用途	告警阈值示例
每次调用 token 数	发现异常大请求	> 8000 tokens
每会话总 token 数	成本控制	> 50k tokens/session
每日总 token 数	预算告警	> 1M tokens/day
prompt vs completion 比例	发现生成过长	completion > 3x prompt

LLM 延迟指标

# 两个关键延迟
LLM_LATENCY.labels(model=model).observe(latency_ms)           # 总延迟
LLM_TTFT.labels(model=model).observe(time_to_first_token_ms)  # 首 token 延迟

指标	含义	用户感知
TTFT（首 token 延迟）	从请求到第一个 token 到达	”AI 在思考”
总延迟	从请求到完整回复	”AI 回答完了”
流式吞吐	每秒 token 数	”打字速度”

SSE 场景特殊处理：流式响应的延迟不是”请求-响应”的单点，而是一个流。需要记录：

流开始时间（首 token）
流结束时间
流是否被中断

SSE 管道的 Trace ID 串联

多服务架构下，SSE 管道的 trace 串联是最容易断裂的地方：

React (fetch) → Go (proxy) → Python (LLM stream)
   ↓               ↓              ↓
   trace_id        trace_id       trace_id
   (前端生成)      (透传)         (透传+创建 span)

// Go SSE 代理：把 trace_id 注入到每个 SSE 事件中
func proxySSE(c *gin.Context, aiURL string) {
    ctx := c.Request.Context()
    span := trace.SpanFromContext(ctx)

    c.Header("Content-Type", "text/event-stream")
    c.Header("X-Trace-ID", span.SpanContext().TraceID().String())

    // 每个 SSE 事件都带上 trace_id
    scanner := bufio.NewScanner(resp.Body)
    for scanner.Scan() {
        line := scanner.Text()
        if strings.HasPrefix(line, "data:") {
            // 注入 trace_id 到事件数据
            enriched := enrichWithTraceID(line, span.SpanContext().TraceID())
            fmt.Fprintf(c.Writer, "%s\n\n", enriched)
        }
    }
}

# Python AI 服务：从请求头提取 trace_id
@app.post("/chat/stream")
async def chat_stream(request: Request):
    trace_id = request.headers.get("X-Trace-ID", generate_trace_id())
    # 所有后续日志都带上这个 trace_id
    logger = logger.bind(trace_id=trace_id)

AI 特有的告警规则

告警	条件	严重程度	含义
LLM 延迟飙升	p95 > 10s 持续 5min	warning	模型服务变慢
LLM 错误率	> 5% 持续 3min	critical	模型服务不可用
Faithfulness 失败率	> 20% 持续 10min	warning	知识库或 prompt 有问题
Red Flag 触发率突增	> 2x 基线	info	可能有新类型的危险症状
Token 用量异常	> 2x 日均	warning	可能有循环调用或滥用
SSE 流中断率	> 10% 持续 5min	critical	管道有问题

与通用可观测性的关系

维度	通用可观测性	AI 特有
日志	请求级	需要 LLM 调用级（token、model、工具调用）
指标	QPS、延迟、错误率	+ token 用量、faithfulness 率、red flag 率
追踪	服务间调用链	+ LLM 内部的 tool call 链、RAG 检索链
告警	服务健康	+ 质量退化、成本异常

通用可观测性是基础，AI 可观测性是在此之上增加 AI 特有的信号。两者共用同一套 trace ID 体系。

设计要点

trace_id 必须贯穿前端到 AI 服务 — 否则无法还原”用户看到的回复”对应的完整链路
token 用量是第一优先级的指标 — 没有 token 监控就无法做成本控制
faithfulness 和 red flag 结果必须入日志 — 这是事后审计”AI 为什么给了错误回答”的关键证据
SSE 流的开始/结束/中断都要记录 — 流式场景的调试比请求-响应模式难得多
日志要结构化 — JSON 格式，不要字符串拼接，否则无法聚合分析

常见错误

只记录成功调用

# ❌ 只在成功时记录
logger.info("llm_call", tokens=usage.total_tokens)

# ✅ 成功和失败都记录，失败时带上错误信息
logger.info("llm_call",
    success=True,
    tokens=usage.total_tokens,
    error=None,
)
logger.error("llm_call",
    success=False,
    error=str(e),
    tokens=None,
)

Trace ID 在 SSE 流中断裂

# ❌ SSE 事件不带 trace_id
yield f"data: {json.dumps({'text': delta})}\n\n"

# ✅ 每个事件都带 trace_id
yield f"data: {json.dumps({'text': delta, 'trace_id': trace_id})}\n\n"

Token 用量只记总数不分维度

# ❌ 只记总数，无法分析哪个模型/哪个功能消耗最多
TOKEN.inc(usage.total_tokens)

# ✅ 按模型、功能、会话多维度记录
TOKEN.labels(model=model, feature="diagnosis").inc(usage.total_tokens)