AI 应用的可观测性：LLM 调用的 Tracing、Token 消耗监控、延迟分析、Prompt 版本管理
问题
AI 应用上线后，传统监控告警（CPU、内存、QPS、5xx 错误率）完全不够用。LLM 调用比普通 API 调用多了几个维度：模型会输出不确定的答案，Token 消耗直接决定成本，Prompt 一变输出就全变。线上出了幻觉回答、Token 成本翻 10 倍、延迟飙到 5s，你拿什么复盘？
我见过不止一个团队在 AI 应用上线后，传统监控面板一片绿，但用户反馈"越来越蠢了"。你点开 Grafana 看到 CPU 20%、内存 50%、QPS 稳定，但问题的根因是：系统 Prompt 上周被产品改了一版没有通知任何人，导致 Agent 行为完全变了。传统监控对这种情况毫无感知。
更具体的坑： 某团队在 2025 年初上线了一个 AI 客服 Agent，上线前做了 2 周压力测试，CPU 和内存表现正常。上线第 3 天用户反馈"答非所问"，运维查了所有指标都正常。最后翻代码才发现，产品经理在 Prompt 里加了一句"如果用户情绪激动，先安抚再回答"，结果 LLM 把"情绪激动"解释得过于宽泛，60% 的请求都走了安抚流程，回答变成了"我理解您的心情...但无法回答您的问题"。传统监控抓不到这种"语义级"的故障。
传统监控 vs AI 可观测性的核心差异
| 维度 | 传统后端 API | LLM 调用 |
|---|---|---|
| 输入输出确定性 | 同一入参=同一出参 | 同一 Prompt 可能输出不同 |
| 错误类型 | 4xx/5xx/超时 | 格式错误/幻觉/拒绝回答/安全违规 |
| 性能指标 | p50/p99 延迟 | TTFT + TPOT + 总延迟 + 重试放大 |
| 成本单位 | 服务器资源 | Token 计价（随模型和长度非线性增长） |
| 变更影响面 | 代码发布 | Prompt 修改一版、模型切换、温度参数变化 |
| 排查手段 | 日志/链路追踪 | 日志 + 链路 + Prompt 快照 + 评估结果 |
结论： 传统监控能告诉你"系统挂了"，但说不清"为什么回答质量下降了"。AI 可观测性必须覆盖四个独立维度：Tracing、Metrics、延迟分析、质量评估。
一、Tracing —— 每次 LLM 调用的完整链路
每次 LLM 调用记录以下字段，缺一不可：
- 请求层： Prompt（完整内容，至少前 1000 token）、模型名称、temperature/top_p 参数、用户 ID、应用 ID
- 响应层： Response 内容、Token 用量（input/output/total）、Finish reason（stop / length / content_filter）
- 性能层： TTFT（首 Token 延迟）、TPOT（每 Token 生成时间）、总延迟、重试次数
- 错误层： 异常类型、重试历史、降级模型名称
为什么传统 API 的 Tracing 方案不够用
传统 OpenTelemetry 的 Span 模型假设：一个请求 = 一个已知的固定端点。但 LLM 调用有四个特殊问题：
- Prompt 不是固定参数：同一个端点（GPT-4o chat）每次 payload 不同，Span 的 attribute 没法预设 schema。
- 流式响应切片：传统 HTTP Span 在 response 结束后才 close，但流式场景下 TTFT 需要在第一个 chunk 到达时就记录。
- 重试语义：传统方案重试=新 Span，但 LLM 重试应视为同一 Span 的多次尝试，否则 trace 图会被重试污染到不可读。
- Token 成本需要聚合：传统 Span 不关心"成本"，但 LLM 场景下这是核心指标。
真实案例： 某团队用 Spring Cloud Sleuth 做了 LLM Tracing，结果每次重试生成了 3 条独立 Span，Jaeger 里一个请求的 trace 图被 3 条 LLM Span 和 3 条 HTTP Span 填满，一个 2 秒的请求画了 12 个 Span 节点，排查问题是先看"哪条 Span 失败了"，而不是"总共失败了几次"。后来改成自定义 Span 模型，把重试次数作为 Span attribute，trace 图才从 12 个节点压缩到 4 个。
代码实现：基于 OpenTelemetry 的 LLM Tracing 中间件
下面是一个生产可用的 LLM 调用追踪中间件，修复了常见实现中 TTFT 测量不准的问题：
import time
import json
from opentelemetry import trace
from opentelemetry.metrics import get_meter
from opentelemetry.sdk.trace import TracerProvider
from dataclasses import dataclass, field, asdict
from typing import Optional, Generator
tracer = trace.get_tracer("llm.observability")
meter = get_meter("llm.observability")
# 指标定义
token_counter = meter.create_counter("llm.token.total", description="Total tokens consumed")
token_input_counter = meter.create_counter("llm.token.input", description="Input tokens")
token_output_counter = meter.create_counter("llm.token.output", description="Output tokens")
latency_histogram = meter.create_histogram("llm.latency.ms", description="LLM call latency in ms")
ttft_histogram = meter.create_histogram("llm.ttft.ms", description="Time to first token in ms")
cost_counter = meter.create_counter("llm.cost.usd", description="Cost in USD", unit="usd")
@dataclass
class LLMCallSpan:
"""单次 LLM 调用的追踪数据"""
trace_id: str
span_id: str
model: str
prompt_preview: str
response_preview: str
input_tokens: int
output_tokens: int
total_tokens: int
latency_ms: float
ttft_ms: float
retry_count: int = 0
error: Optional[str] = None
user_id: Optional[str] = None
app_id: Optional[str] = None
prompt_version: Optional[str] = None
finish_reason: Optional[str] = None
class LLMTracingMiddleware:
"""
LLM 调用追踪中间件，包装任意 LLM 客户端。
关键设计：
1. 流式模式下精确测量 TTFT —— 从请求发出到收到第一个 chunk 的耗时
2. 非流式模式下 TTFT ≈ 总延迟（模型一次性返回）
3. 重试策略计入 Span 属性，不会产生多条 Span
4. 成本估算按模型官方定价，可配置化
"""
MODEL_PRICING = {
"gpt-4o": {"input": 2.50, "output": 10.00},
"gpt-4o-mini": {"input": 0.15, "output": 0.60},
"claude-3.5-sonnet": {"input": 3.00, "output": 15.00},
"deepseek-v2": {"input": 0.14, "output": 0.28},
}
def __init__(self, llm_client, model_name: str = "gpt-4o", pricing: dict = None):
self._client = llm_client
self._model = model_name
self._pricing = pricing or self.MODEL_PRICING.get(model_name, {"input": 2.00, "output": 8.00})
self._max_retries = 3
self._retry_delay = 1.0 # 秒
def chat_completion(self, messages: list, stream: bool = False, **kwargs):
prompt_json = json.dumps(messages, ensure_ascii=False)
start = time.monotonic()
retry_count = 0
last_error = None
ttft_ms = 0.0
response_text = ""
finish_reason = None
with tracer.start_as_current_span("llm.chat_completion") as span:
span.set_attribute("llm.model", self._model)
span.set_attribute("llm.prompt.length", len(prompt_json))
span.set_attribute("llm.prompt.preview", prompt_json[:500])
span.set_attribute("llm.stream", str(stream))
for attempt in range(self._max_retries):
try:
# 精确测量 TTFT —— 对非流式请求，TTFT 就是首次响应到达时间
response_start = time.monotonic()
if stream:
# 流式模式：第一个 chunk 到达即记录 TTFT
collected_chunks = []
for chunk in self._client.chat(messages, stream=True, **kwargs):
if not collected_chunks: # 第一个 chunk
ttft_ms = (time.monotonic() - response_start) * 1000
collected_chunks.append(chunk)
# 拼装完整响应
response_text = "".join(
c.get("choices", [{}])[0].get("delta", {}).get("content", "")
for c in collected_chunks
)
# 取最后一个 chunk 的 finish_reason
last = collected_chunks[-1] if collected_chunks else {}
finish_reason = last.get("choices", [{}])[0].get("finish_reason", "stop")
else:
# 非流式：直接返回完整响应，TTFT ≈ 总延迟
response = self._client.chat(messages, **kwargs)
ttft_ms = (time.monotonic() - response_start) * 1000
response_text = response.get("choices", [{}])[0].get("message", {}).get("content", "")
finish_reason = response.get("choices", [{}])[0].get("finish_reason", "stop")
end = time.monotonic()
latency_ms = (end - start) * 1000
# 解析 Token 用量
input_tokens = response.get("usage", {}).get("prompt_tokens", 0)
output_tokens = response.get("usage", {}).get("completion_tokens", 0)
total_tokens = input_tokens + output_tokens
# 构建 Span 数据
call = LLMCallSpan(
trace_id=format(span.get_span_context().trace_id, "032x"),
span_id=format(span.get_span_context().span_id, "016x"),
model=self._model,
prompt_preview=prompt_json[:1000],
response_preview=response_text[:1000],
input_tokens=input_tokens,
output_tokens=output_tokens,
total_tokens=total_tokens,
latency_ms=latency_ms,
ttft_ms=ttft_ms,
retry_count=retry_count,
error=None,
finish_reason=finish_reason,
)
# 填写 Span 属性
span.set_attribute("llm.token.input", input_tokens)
span.set_attribute("llm.token.output", output_tokens)
span.set_attribute("llm.token.total", total_tokens)
span.set_attribute("llm.latency.ms", latency_ms)
span.set_attribute("llm.ttft.ms", ttft_ms)
if finish_reason:
span.set_attribute("llm.finish_reason", finish_reason)
# 记录指标
token_counter.add(total_tokens, {"model": self._model})
token_input_counter.add(input_tokens, {"model": self._model})
token_output_counter.add(output_tokens, {"model": self._model})
latency_histogram.record(latency_ms, {"model": self._model})
ttft_histogram.record(ttft_ms, {"model": self._model})
# 成本估算
cost = input_tokens / 1_000_000 * self._pricing["input"] \
+ output_tokens / 1_000_000 * self._pricing["output"]
cost_counter.add(cost, {"model": self._model})
# 持久化
self._persist_span(call)
return response if not stream else response_text, call
except Exception as e:
retry_count += 1
last_error = str(e)
if attempt < self._max_retries - 1:
time.sleep(self._retry_delay * (2 ** attempt)) # 指数退避: 1s, 2s, 4s
continue
# 所有重试失败
span.set_attribute("llm.error", last_error)
span.set_attribute("llm.retry_count", retry_count)
span.set_status(trace.Status(trace.StatusCode.ERROR, last_error))
end = time.monotonic()
failed_call = LLMCallSpan(
trace_id=format(span.get_span_context().trace_id, "032x"),
span_id=format(span.get_span_context().span_id, "016x"),
model=self._model,
prompt_preview=prompt_json[:1000],
response_preview="",
input_tokens=0,
output_tokens=0,
total_tokens=0,
latency_ms=(end - start) * 1000,
ttft_ms=0,
retry_count=retry_count,
error=last_error,
)
self._persist_span(failed_call)
raise
def _persist_span(self, call: LLMCallSpan):
"""持久化 Span 数据 → 写入 Kafka / ES / LangFuse API"""
record = asdict(call)
record["timestamp"] = int(time.time() * 1000)
record["model_pricing"] = self._pricing
# 生产环境：写入 Kafka topic "llm-traces" 或直接调用 LangFuse API
print(json.dumps(record, ensure_ascii=False))这段代码的重点设计
- TTFT 精确测量： 流式模式下，在第一个
for chunk迭代时记录 TTFT；非流式模式 TTFT ≈ 总延迟（因为模型一次性返回）。之前很多实现把 TTFT 和总延迟混为一谈，或者直接写死 0。 - 重试计入 Span 而非多条 Span： 重试是同一请求的多次尝试，应该在同一 Span 内体现（
retry_count属性），而不是创建多条独立 Span。后者会污染链路图和聚合统计。 - 成本模型可配置： 不同模型定价不同，hardcode 会被审计骂。
MODEL_PRICING字典支持按需扩展。
面试追问：Tracing 在 LLM 场景下的特殊问题
面试官可能会追问：
Q1：为什么不用 Spring Cloud Sleuth 或者 Micrometer 直接做 LLM Tracing？ A：Sleuth 的 Span 模型是为 HTTP RPC 设计的，没法处理 LLM 的流式响应。你没法在 Sleuth 的 Span 里插一个"到达第一个 chunk 时记录 TTFT"的回调。Micrometer 可以记录指标，但没法做 Prompt 快照和历史比对。所以 LLM Tracing 需要自建一层，OpenTelemetry 只做底层传输，上层需要定制 LLM 语义。
Q2：TTFT 和总延迟的区别在面试中怎么答？ A：TTFT（Time to First Token）决定用户首次感知，总延迟决定用户等待完毕。流式场景下，TTFT 300ms 总延迟 8s 的用户体验，远好于 TTFT 2s 总延迟 3s 的用户体验。因为用户看到文字在出，心理上会"觉得没那么慢"。所以优化 TTFT 比优化 TPOT 优先级更高。
Q3：如果让你设计一个 LLM Tracing 的 Prometheus 指标，你选哪些？ A：必选五个：llm_ttft_ms（Histogram，分桶 100/300/500/1000/2000/5000）、llm_tpot_ms（Histogram，分桶 10/20/50/100/200）、llm_latency_ms（Histogram）、llm_token_total（Counter，label 分 input/output）、llm_cost_usd（Counter）。另外加一个 Gauge llm_error_rate，按 errors / total 在应用层算好再推，避免 Prometheus 的 rate() 函数在低 QPS 下不准。坑： Prometheus Histogram 的桶边界要按模型类型调。GPT-4o 的 TTFT 集中在 200-800ms，GPT-4o-mini 的 TTFT 集中在 50-200ms，用一个桶集会导致大部分数据落在同一个桶里，p99 估算偏差大。
Spring Boot 集成 LLM Tracing 示例
如果你是 Java 后端，在 Spring Boot 里集成 LLM Tracing 的典型做法：
@Component
public class LLMTracingAspect {
private static final Tracer tracer = GlobalOpenTelemetry.getTracer("llm.observability");
private static final Meter meter = GlobalOpenTelemetry.getMeter("llm.observability");
private final LongCounter tokenCounter = meter.counterBuilder("llm.token.total").build();
private final DoubleHistogram latencyHistogram = meter.histogramBuilder("llm.latency.ms").build();
@Around("@annotation(LLMTraced)")
public Object traceLLMCall(ProceedingJoinPoint pjp) throws Throwable {
Span span = tracer.spanBuilder("llm.chat").startSpan();
long start = System.nanoTime();
try (Scope scope = span.makeCurrent()) {
Object result = pjp.proceed();
long latencyMs = (System.nanoTime() - start) / 1_000_000;
// 从 result 中提取 token 用量（假设返回对象有 getUsage()）
if (result instanceof LLMResponse) {
LLMResponse resp = (LLMResponse) result;
span.setAttribute("llm.token.total", resp.getUsage().getTotalTokens());
tokenCounter.add(resp.getUsage().getTotalTokens());
latencyHistogram.record(latencyMs);
}
return result;
} catch (Exception e) {
span.setStatus(StatusCode.ERROR, e.getMessage());
throw e;
} finally {
span.end();
}
}
}坑： Spring AOP 默认用 JDK 代理，如果 LLM 客户端是 final 类或者没有实现接口，要用 CGLIB 或手动织入 AspectJ。Spring Boot 3.x 默认用 CGLIB，但如果你自己 new 了 LLM 客户端，AOP 切不进去，必须通过 Spring 容器获取 Bean。具体来说，new OpenAiClient() 不会被 AOP 拦截，必须 @Autowired private OpenAiClient client。
二、延迟分析 —— 为什么 LLM 延迟的 p50 和 p99 差距这么大？
实测数据（生产环境，GPT-4o，1K 上下文，2025 年 3 月线上采集）：
| 指标 | 值 | 说明 |
|---|---|---|
| p50 总延迟 | 820ms | 一半请求在 1s 内返回 |
| p95 总延迟 | 2.3s | 5% 的请求超过 2.3s |
| p99 总延迟 | 3.8s | 1% 的请求接近 4s |
| TTFT p50 | 320ms | 模型首 token 生成较快 |
| TTFT p99 | 1.5s | GPU 排队时首 token 延迟飙升 |
| TPOT | 35ms/token | 200 token 输出约 7s |
差距根本原因： 三个因素叠加：
- Prompt 长度： 输入越长，prefill 阶段越慢。1K token 输入 vs 8K token 输入，TTFT 差 3-5 倍。
- 模型共享负载： 共享 GPU 上，同一时间段的请求排队。夜里 2 点 vs 晚上 8 点，p99 延迟差 2 倍以上。
- 网络抖动： 跨 region 调用（如国内调 OpenAI API）加 100-300ms 网络延迟。
我踩过的坑： 之前用平均值设了 2s 超时，结果 p99 5% 的请求 3.8s 全被熔断，用户体验是"每 20 次请求就有 1 次报错"。后来改成 p99 + 2σ 设 5s 超时，配合熔断器（详见第 14 篇《LLM API 工程的兼容性、重试与熔断器》），才解决。
延伸案例： 某电商 AI 客服上线后，发现晚间 8-10 点 TTFT p99 从 300ms 飙到 3.2s，排查发现共享 GPU 上同时跑了 3 个团队的模型推理任务。运维查 GPU 利用率才 40%，但 vLLM 的调度队列显示排队长度高峰期到 80 个请求。核心问题不是 GPU 满载，而是 vLLM 的 Continuous Batching 策略在队列深度超过 50 时，Prefill 阶段的 batch 等待时间指数级增长。 最终解法是给客服模型单独分配一个 GPU 实例，TTFT p99 降到 400ms。
延迟拆解流程图
用户请求 → [网络延迟] → [GPU 排队] → [Prefill 阶段] → [Decode 阶段] → [网络回传]
↑ ↑ ↑
等待其他请求 处理 Prompt 逐 token 生成
非固定时长 O(n²) 计算 O(n) 计算
影响因素： n=Prompt token m=输出 token
并发数、GPU 型号- Prefill（TTFT）： 和 Prompt 长度成正比。一个 8K prompt 的 prefill 耗时约是 1K prompt 的 5x（因为 attention 计算是 O(n²)）。
- Decode（TPOT）： 和输出长度成正比。输出 500 token 比输出 100 token 多花 5 倍时间。
- 总延迟公式：
Latency = TTFT + TPOT × output_tokens + 网络延迟
面试追问：为什么 Prefill 是 O(n²) 而 Decode 是 O(n)？
Attention 的计算公式：Attention(Q,K,V) = softmax(QK^T / √d)V。Prefill 阶段需要计算 Prompt 所有 token 之间的注意力，矩阵乘法维度是 n × d 乘 d × n，得到 n × n 的注意力矩阵，复杂度 O(n²)。Decode 阶段每次只生成一个 token，新 token 与已有 KV-Cache 做注意力，复杂度 O(n)。所以长 Prompt 的 TTFT 会暴涨，而输出长文本的 TPOT 是线性的。
面试加分回答： vLLM 等推理引擎通过 PagedAttention 和 Continuous Batching 把 Prefill 和 Decode 混合调度，减少了 GPU 空闲时间，但 Prefill 的 O(n²) 本质没变。所以极端场景（如 128K 上下文）下，Prefill 耗时可能占 80% 以上。这也是为什么 FlashAttention 对长上下文场景至关重要——它把空间复杂度从 O(n²) 降到了 O(n)，但时间复杂度还是 O(n²)。面试官如果问"怎么优化 Prefill"，回答顺序：FlashAttention → PagedAttention → 减少输入长度（RAG 截断、Prompt 压缩）。
更精细的延迟拆解
| 阶段 | 耗时占比 | 影响因素 |
|---|---|---|
| 网络延迟(TX) | 5-10% | 机房距离、代理转发 |
| GPU 排队 | 10-30% | 并发数、GPU 型号、vLLM 批处理策略 |
| Prefill(TTFT) | 30-50% | Prompt 长度、模型参数量 |
| Decode(TPOT) | 20-40% | 输出长度 × 每 token 生成时间 |
| 网络延迟(RX) | 5-10% | 同上 |
三、Token 消耗监控 —— 成本失控的五个坑
坑 1：上线前 Token 用量预估偏差 10 倍
很多团队上线前用"预估用户数 × 平均对话轮次 × 平均 Prompt 长度"算，结果实际用量差 10 倍。原因：
- Prompt 实际长度远超预估： RAG 的 Top-K 设大了，每次检索带回 5 篇文档，每篇 2000 token，Prompt 里塞了 10K+ token。
- 系统 Prompt 加塞： 产品经理每周加一段"提示词优化"，三个月后系统 Prompt 从 200 token 膨胀到 2000 token。
- 重试和 fallback 放大： 一次失败重试 3 次，用更大模型降级，成本翻 4 倍。
解法： 线上先放 10% 流量，观察三天 Token 消耗曲线，用回归模型预测全量成本，再逐步放量。
坑 2：Token 消耗和用户数不是线性关系
用户数涨 10 倍，Token 消耗可能涨 30 倍。因为：
- 新用户使用模式不同（更爱问长问题）
- Prompt 缓存命中率下降
- 并发升高导致模型降级（用更贵模型兜底）
真实案例： 某 AI 客服上线第 1 周日均 1000 用户，Token 消耗 200 万/天；第 2 周 3000 用户，Token 消耗飙到 800 万/天。不是 3 倍而是 4 倍，原因是系统 Prompt 里加了用户画像，每次请求都要带 500 token 的额外上下文。
坑 3：输出 Token 比输入 Token 贵得多
以 GPT-4o 为例：输入 $2.50/1M tokens，输出 $10.00/1M tokens。输出价格是输入的 4 倍。如果输出长回答（2000 token），一个请求的成本 = 主要来自输出端。
优化方向：
- 能短回答的不要长回答，设置
max_tokens上限 - 使用
gpt-4o-mini做简单问答，gpt-4o仅做复杂推理 - 缓存常见问题回答（见下文 Redis 缓存策略）
坑 4：上下文窗口超限导致隐形成本上涨
当 Prompt 超过模型上下文窗口（如 GPT-4o 的 128K），有些 SDK 会静默截断，有些会报错重试。截断的实际效果是：Token 没少花，但回答质量下降，用户反复追问，总成本反而更高。
真实案例： 某 RAG 应用，每次检索 5 篇文档，最长的文档 30K token，加起来超了 128K 窗口。SDK 自动截断到 128K，但截断的是文档中间部分，结果回答总是缺少关键信息，用户平均追问 2.3 次才得到满意答案。追问题平均消耗 600 token 输出 + 200 token 输入，每用户成本翻倍。解法： 在应用层做 Token 计数 + 截断策略，不要让 SDK 静默处理。
坑 5：Prompt Cache 命中率低等于白建
vLLM 等框架支持 Prefix Cache：相同 Prompt 前缀的请求可以复用 KV-Cache，节省 Prefill 计算。但很多团队的系统 Prompt 天天改，Cache 命中率不到 10%。Cache 命中率 = 60% 时，TTFT 降低 40%；Cache 命中率降到 10%，基本等于没 Cache。
具体数据： 某团队用 vLLM 部署了 GPT-4o 的替代模型，上线第一个月 Prefix Cache 命中率 65%，TTFT p50 180ms。第二个月产品经理频繁调整系统 Prompt，Cache 命中率降到 8%，TTFT p50 升到 420ms。根因： 每次修改了 Prompt 前缀的 50 个字符，整个 KV-Cache 全部失效。解法： 把系统 Prompt 中的动态部分（如用户信息、时间戳）移到 Prompt 末尾，固定前缀部分保持不变，Cache 命中率回升到 55%。
成本预估速算表
| 场景 | 日均用户 | 日均请求 | 平均 Prompt | 平均输出 | 单次成本 | 日成本 |
|---|---|---|---|---|---|---|
| 简单问答(gpt-4o-mini) | 10K | 50K | 500 token | 200 token | $0.0002 | ~$10 |
| 复杂推理(gpt-4o) | 1K | 10K | 2K token | 800 token | $0.013 | ~$130 |
| RAG 问答(gpt-4o) | 5K | 25K | 5K token | 500 token | $0.0175 | ~$437.5 |
| 代码生成(gpt-4o) | 500 | 5K | 3K token | 2K token | $0.028 | ~$140 |
注意： 上表是理想情况，实际因为重试、上下文窗口超限、降级模型，日成本容易翻 2-3 倍。上线前要按 3 倍 reserve 预算。
四、Prompt 版本管理 —— 解决"谁改了什么"的问题
核心需求
- 每次 Prompt 变更都有版本记录，可追溯
- 线上可以指定使用哪个版本
- 支持 A/B 测试，对比新老版本效果
实现方案
import hashlib
import random
from datetime import datetime, timezone
from typing import Optional
class PromptRegistry:
"""Prompt 模板的版本管理与 A/B 测试，支持 Git 式版本追溯"""
def __init__(self):
# name -> [version_records]，按时间升序排列
self._templates: dict[str, list[dict]] = {}
def register(self, name: str, template: str, tags: dict = None,
author: str = "") -> str:
"""
注册/更新一个 Prompt 模板。
版本号用 SHA256 摘要前 12 位，保证内容相同版本相同。
"""
version = hashlib.sha256(template.encode()).hexdigest()[:12]
record = {
"name": name,
"version": version,
"template": template,
"created_at": datetime.now(timezone.utc).isoformat(),
"author": author,
"tags": tags or {},
}
self._templates.setdefault(name, []).append(record)
print(f"[PromptRegistry] Registered {name}@{version} by {author}")
return version
def get_latest(self, name: str) -> str:
"""获取最新版本"""
versions = self._templates.get(name)
if not versions:
raise ValueError(f"Prompt {name} not found")
return versions[-1]["template"]
def get_version(self, name: str, version: str) -> str:
"""获取指定版本"""
for v in self._templates.get(name, []):
if v["version"] == version:
return v["template"]
raise ValueError(f"Version {version} not found for {name}")
def list_versions(self, name: str) -> list[dict]:
"""列出所有版本历史"""
return self._templates.get(name, [])
def ab_test(self, name: str, variant_a: str, variant_b: str,
ratio: float = 0.5, author: str = "ab-test") -> tuple[str, str]:
"""
A/B 测试分组。
返回 (version, template)，version 用于日志记录，后续分析效果。
"""
va = self.register(f"{name}_variant_a", variant_a,
{"variant": "a", "ab_test": name}, author)
vb = self.register(f"{name}_variant_b", variant_b,
{"variant": "b", "ab_test": name}, author)
if random.random() < ratio:
return va, variant_a
return vb, variant_b
# 使用示例
registry = PromptRegistry()
v1 = registry.register("qa_system",
"You are a {role}. Answer the question: {question}",
author="alice")
v2 = registry.register("qa_system",
"You are a {role}. Answer concisely: {question}",
author="bob")
print(f"Active version: {registry.get_latest('qa_system')}")面试追问：为什么用 SHA256 摘要做版本号而不是自增 ID？
两个原因：
- 内容即标识： 相同的 Prompt 内容产生相同的版本号，如果 A 和 B 同时改了同一个 Prompt，Git merge 冲突后 SHA 不变，不会出现"两个版本号但内容一样"的脏数据。
- 可追溯： 只要你存了原始 Prompt 文件，就能重新算出版本号，不会出现"版本号 3 是什么内容来着我忘了"的情况。
生产实践：Prompt 变更流程
开发 → 编辑 Prompt → 注册新版本(dev) → 跑回归测试集(500条) →
→ 对比语义相似度(≥0.85) → A/B 测试(5%流量) → 观察 2h →
→ 全量上线 → 注册新版本(prod)回归测试集要求： 500-1000 条典型用户问题 + 对应期望回答。评估用 Embedding Cosine 相似度（阈值 0.85）替代字符串匹配——因为 LLM 输出天然不固定，字符串匹配误报率极高。
我踩过的坑： 一开始用 BLEU 分数做回归测试，结果经常低于 0.3，因为 LLM 回答的措辞和参考答案不一样。后来换成 Embedding 相似度，调了 3 个 Embedding 模型（text-embedding-3-small、bge-large-zh、jina-embeddings-v2），发现 text-embedding-3-small 对中文长文本最稳定，阈值 0.85 能覆盖 95% 的正常变更，漏报率 0.5%。
更进一步的坑： 阈值 0.85 在简单问答场景下效果很好，但在代码生成场景下需要降到 0.70。因为代码生成的回答结构差异大，即使语义正确，Cosine 相似度也可能只有 0.6。我们后来按场景分了三套阈值：简单问答 0.85、复杂推理 0.78、代码生成 0.70。
五、工具链选型对比
| 工具 | 开源/付费 | 部署方式 | Tracing | 评估 | 成本管理 | 适合场景 |
|---|---|---|---|---|---|---|
| LangFuse | 开源 + 云版 | 自托管 / SaaS | ✅ | ✅ | ✅ | 小团队首选，自托管无费用 |
| LangSmith | 付费 | SaaS | ✅ | ✅ | ✅ | LangChain 深度集成，贵 |
| Arize AI | 开源 + 云版 | 自托管 / SaaS | ✅ | ✅ | 部分 | Embedding 漂移检测强 |
| 自建 OTel 栈 | 全开源 | 自托管 | ✅ | 需自建 | 需自建 | 大团队，灵活性最高 |
选型建议： 小团队（<10 人）直接上 LangFuse 自托管（免费，API 兼容 OpenTelemetry），配套 Grafana 面板看成本趋势。大团队（>50 人）自建 OpenTelemetry 扩展 + Prometheus + Grafana + Elasticsearch 四件套，但需要 1 个 SRE 专职维护。追求极致集成度用 LangSmith，但每百万 token 要额外付 $0.5 的追踪费，日均 100 万 token 就多 $50/天，一年 $18,000，这笔钱够请半个 SRE了。
LangFuse 自托管踩坑： 我们用 Docker Compose 部署 LangFuse，默认配置下 PostgreSQL 的 max_connections 只有 100，QPS 超过 500 时连接池打满，Tracing 数据丢失。解法： 在 LangFuse 的 docker-compose.yml 里改 DATABASE_URL 连接字符串参数：postgresql://user:pass@db:5432/langfuse?connection_limit=300，并用 PgBouncer 做连接池代理。直接设 POSTGRES_MAX_CONNECTIONS 环境变量 LangFuse 不认，只有 PostgreSQL 原生镜像才认这个变量，这是我们在排错时才发现的事。另外 LangFuse 的 Python SDK 默认是同步阻塞的，高并发场景下会阻塞主线程，必须用 langfuse.task_manager.set_max_retries(3) + 异步模式。
六、生产落地的告警阈值
以下是我在线上跑过的阈值，供参考：
| 指标 | 告警阈值 | 严重级别 | 说明 |
|---|---|---|---|
| p99 延迟 | > 3s 持续 5min | P1 | 用户体验明显受损 |
| Token 日消耗 | > 预估 150% | P2 | 成本可能失控 |
| 错误率 | > 5% 持续 3min | P1 | 模型/网络异常 |
| 输出长度 | 平均 > 2000 token | P2 | 模型可能跑偏，输出过长 |
| Prompt 变更 | 无版本号的调用 | P3 | 遗漏了 Tracing |
| Cache 命中率 | < 30% 持续 1h | P3 | Prefix Cache 配置失效 |
告警去重实践： 如果直接用 Prometheus 的 rate() 算错误率，在低 QPS（<10 req/s）时会剧烈抖动。我们改成滑动窗口 5 分钟计数，errors_in_5min / total_in_5min，阈值 5%，避免了因为 1 个请求失败导致 20% 误报。
七、面试高频问题整理
Q1：线上 LLM 调用的 p99 延迟突然从 1s 涨到 5s，你怎么排查？
排查路线：
- 先看 TTFT 是否飙升 → 看 Grafana 的
llm.ttft.ms指标。如果 TTFT 从 300ms 涨到 2s，说明 GPU 排队了（并发高了）或 Prompt 变长了（检查 Prompt 长度分布）。 - 如果 TTFT 正常（300ms），看 TPOT 是否变长 → 说明输出变长了。检查
finish_reason分布：length占比是否上升？如果是，说明max_tokens被改了或模型生成变长了。 - 如果 TTFT 和 TPOT 都正常，看网络延迟 → 跨 region 调用的网络抖动。
- 最后看模型降级策略：是否被触发换到了更慢的模型？是否触发了 3 次重试把延迟叠加了？
实战案例： 某次线上 p99 延迟从 1.2s 涨到 4.5s，排查步骤 1 发现 TTFT 正常（280ms），步骤 2 发现 TPOT 从 35ms/token 涨到 120ms/token。继续深挖，发现 finish_reason 中 length 占比从 5% 涨到 40%。根因是前一天晚上上线了一个新版本，把 max_tokens 从 500 改成了 2000，导致平均输出长度从 200 token 涨到 800 token。复盘： 配置变更没有走灰度，直接全量上线了。
Q2：AI 应用的可观测性系统和传统微服务观测系统怎么融合？
两个层面的融合：
- 基础设施层： LLM 调用的 Tracing 数据写入同一个 OpenTelemetry Collector，和普通 HTTP 请求的 Trace 在同一个链路里。比如一个 HTTP 请求 → 调用了 3 次 LLM → 1 次 Redis 查询，在 Jaeger 里能看到完整的调用链。
- 告警层： 传统告警（QPS、5xx 错误率）和 AI 告警（Token 消耗、TTFT、Finish Reason 分布）放在同一套告警引擎里（Prometheus AlertManager），统一通知飞书/钉钉。
具体实现： 我们的方案是自建一个 Spring Boot 应用，通过 OpenTelemetry Java Agent 自动注入，然后在代码里用 @WithSpan 注解标记 LLM 调用方法。传统 HTTP 请求的 Span 由 Agent 自动生成，LLM 调用的 Span 由 @WithSpan 生成，最终在 Jaeger 里看到的是完整链路。但有一个坑： OpenTelemetry Java Agent 默认把 LLM 响应的 body 内容也注入到了 Span attribute 中，Prometheus 的指标采集会把这些大文本全部加载到内存，导致 OOM。解法： 在 Agent 配置里设置 otel.instrumentation.micrometer.enabled=false，只保留自定义的 Histogram 和 Counter。
Q3：Prompt 变了之后，怎么量化"质量有没有下降"？
用 Embedding 相似度做回归测试。具体做法：
- 维护 500-1000 条标注好的典型问题 + 期望回答
- 新 Prompt 跑一轮这批问题，得到新回答
- 计算新回答和期望回答的 Cosine 相似度
- 平均相似度低于 0.85 就报警，人工审核
注意： 不要用字符串匹配（BLEU/ROUGE），LLM 回答的措辞多变，字符串匹配误报率极高。我见过一个团队用 BLEU 做回归，平均分 0.2，每次改 Prompt 都报警，两周后开发团队直接把报警规则关了——狼来了效应。回归测试一定要用语义相似度，而且要有专人维护测试集，每个月至少补充 50 条新问题覆盖新场景。
进阶做法： 除了 Embedding 相似度，我们还加了一个"语义一致性"检查：对同一个问题，用 3 个不同的 Embedding 模型分别计算相似度，取中位数。如果某个模型给出的相似度比其他两个低 20% 以上，说明该模型对这次变更不够敏感，需要更新 Embedding 模型。实操发现： bge-large-zh 对中文代码生成的相似度打分比 text-embedding-3-small 低了 15%，但 text-embedding-3-small 对中英文混合内容的稳定性更好。最终我们选 text-embedding-3-small 做主力，bge-large-zh 做交叉验证。
八、从 Java 后端视角看 AI 可观测性
如果你是从 Java 后端转过来做 Agent 工程，下面这些映射关系能帮你快速上手：
| 传统后端概念 | AI 应用对应概念 | 关键差异 |
|---|---|---|
| SLF4J 日志 | LLM Span + Prompt 快照 | 日志是单向的，Span 需要关联请求上下文 |
| Micrometer 指标 | Token 计数器 + 延迟 Histogram | 指标维度多了 model/token_type |
| Spring Cloud Sleuth | OpenTelemetry + 自定义 LLM Span | 流式请求需要自定义 Span 结束时机 |
| Prometheus + Grafana | 同上 + 成本面板 + 评估面板 | 多了成本 Curve 和 Prompt 版本对比 |
| ELK 日志聚合 | 同上 + Prompt 搜索 + 版本回溯 | 搜索维度多了 model 和 prompt_hash |
转型建议： 你熟悉的 OpenTelemetry、Prometheus、Grafana 这套体系可以直接复用。额外需要掌握的是：LLM 调用的语义级追踪（不是追踪 HTTP 请求，而是追踪模型的输入输出）、Token 成本模型（按模型定价做实时计数）、以及 Prompt 版本管理（Git 式版本化 + 回归测试）。这三个是传统后端没有的，也是最容易在面试中被问到的。
总结
- Tracing 是 AI 可观测性的基石，每次 LLM 调用必须记录：Prompt 摘要、Response 摘要、Token 用量、TTFT、总延迟、finish_reason。缺任何一项，出问题都查不了。
- TTFT 和 TPOT 要分开看，流式请求下 TTFT 决定用户"第一感知"，TPOT × 输出长度决定总等待时间。用
p99 + 2σ设超时，别用平均值。 - Token 消耗监控直接关联成本，上线前做用量预估 + 渐进式放量，监控日消耗变化率，超过预估 150% 自动告警。
- Prompt 版本管理防止"谁改了什么"的扯皮，配合 A/B 测试和回归测试（Embedding 相似度 ≥ 0.85），才能让模型输出保持可控。
- 面试常问： "线上 LLM 调用的 p99 延迟突然从 1s 涨到 5s，你怎么排查？" 答案：先看 TTFT 是否飙升（GPU 排队），再看输出长度是否变长（Prompt 变了），再看网络延迟，最后看模型降级策略是否被触发。
— 📚 小杰