开篇

LLM 应用出问题，你大概率不知道问题出在哪里。

这不是夸张，是现实。传统的 APM（Application Performance Monitoring）在 API 响应层面可以告诉你"200ms 延迟，99 分位正常"，但它无法回答以下问题：

• 这一次 LLM 调用的 prompt 里注入了什么 context，模型为什么输出了这个结果？
• 用户连续交互 10 轮之后，context window 消耗到了什么程度？
• 模型"一本正经地胡说八道"，你有没有手段在生产环境里发现它？

传统微服务的可观测性建立在确定性之上：输入固定、逻辑固定、输出可预期。LLM 应用打破了所有这三个假设。当你的"后端逻辑"是一段由概率驱动的模型推理时，传统的日志 + 指标 + 链路追踪已经不够用了。

本文不介绍概念，直接讲怎么落地。代码可运行，工具经过生产验证，踩过的坑明明白白写出来。

---

一、什么是 AI Agent 可观测性

1.1 传统可观测性三支柱

业界通用的可观测性体系包含三大支柱：

• Trace（链路追踪）：请求在分布式系统中的完整调用路径，每一层的耗时与依赖关系。
• Metric（指标）：聚合后的数值型数据，例如 QPS、延迟分位、错误率、CPU 使用率。
• Log（日志）：离散的事件记录，包含时间戳、上下文和具体内容。

这三者在微服务架构下配合良好，OpenTelemetry 已经将标准统一。但 LLM 应用引入了新的维度。

1.2 LLM 应用的特殊性

LLM 应用与微服务有三个根本性差异：

非确定性输出：相同输入可能产生不同输出，“重试"不等于"修复”。传统可观测性假设可以稳定复现问题，LLM 场景下这个假设失效。

Token 维度：输入输出以 token 计量，而 token 与成本直接挂钩。传统指标体系里没有 token 这个概念，但它是 LLM 应用的核心资源。

多层嵌套调用：一个 Agent 内部可能包含：规划（Planning LLM）→ 工具选择（Tool Selection）→ 外部 API 调用（Tool Execution）→ 结果总结（Summary LLM）。这不是一条直线，是一棵动态决策树。

1.3 AI Agent 可观测性的定义

本文给出的定义：在 LLM 应用中，能够回答"模型在做什么、为什么这样做、这样做的代价是什么、结果对不对"这四个问题的完整能力体系。

对应到技术实现：

维度	回答的问题	传统等效物
Trace	模型在调用谁，调用顺序是什么	分布式链路追踪
Metric	Token 消耗、延迟、成本、错误率	系统指标
Log	Prompt、完整输出、中间状态	应用日志
额外维度：Eval	输出质量是否合格	无等效物

Eval（评估）是 LLM 应用独有的维度。传统软件里，输出格式不对是 bug，LLM 里"格式对但内容错"是更常见的问题，而这需要单独的检测手段。

---

二、三大支柱的 LLM 化实现

2.1 Trace：追踪 Agent 的每一步决策

在传统微服务里，OpenTelemetry 的 span 代表一次函数调用。在 LLM 应用里，span 的语义需要扩展：

[Agent Root Span]
  ├── [Planner LLM Span]     # 调用规划模型
  │     └── [Prompt Render Span]
  ├── [Tool Selection Span]   # 工具选择决策
  ├── [Tool Execution Span]   # 外部 API 调用
  │     └── [HTTP Client Span]
  └── [Response LLM Span]     # 最终响应生成

下面是一个用 OpenTelemetry + LLM SDK 实现的带 trace 的简单 Agent：

import os
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter
from opentelemetry.sdk.resources import Resource
from opentelemetry.semconv.resource import ResourceAttributes
from opentelemetry.trace import Status, StatusCode

# 初始化 tracer provider
provider = TracerProvider(
    resource=Resource.create({
        ResourceAttributes.SERVICE_NAME: "ai-agent-service",
        ResourceAttributes.SERVICE_VERSION: "1.0.0",
    })
)
provider.add_span_processor(BatchSpanProcessor(ConsoleSpanExporter()))
trace.set_tracer_provider(provider)
tracer = trace.get_tracer(__name__)

import openai
from openai import OpenAI

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))


class TracedAgent:
    """带完整链路追踪的简单 Agent"""

    def __init__(self, model: str = "gpt-4o"):
        self.model = model
        self.client = client
        self.tracer = tracer
        self.max_turns = 5

    def run(self, user_message: str, system_prompt: str) -> str:
        """执行 Agent 对话，带完整 trace"""

        with self.tracer.start_as_current_span(
            "agent.run",
            attributes={
                "agent.model": self.model,
                "agent.max_turns": self.max_turns,
            }
        ) as root_span:
            messages = [
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_message},
            ]

            final_response = None
            turn = 0

            while turn < self.max_turns:
                turn += 1

                # 追踪 LLM 调用
                with self.tracer.start_as_current_span(
                    f"llm.call.turn_{turn}",
                    attributes={
                        "llm.model": self.model,
                        "llm.messages_count": len(messages),
                    }
                ) as llm_span:
                    response = self.client.chat.completions.create(
                        model=self.model,
                        messages=messages,
                        temperature=0.7,
                    )
                    assistant_message = response.choices[0].message
                    assistant_content = assistant_message.content

                    # 记录 token 使用量
                    usage = response.usage
                    llm_span.set_attribute("llm.usage.prompt_tokens", usage.prompt_tokens)
                    llm_span.set_attribute("llm.usage.completion_tokens", usage.completion_tokens)
                    llm_span.set_attribute("llm.usage.total_tokens", usage.total_tokens)
                    llm_span.set_attribute("llm.finish_reason", response.choices[0].finish_reason)

                    messages.append({"role": "assistant", "content": assistant_content})

                    # 检查是否需要继续循环（假设格式为 [ACT:tool_name] 或 [DONE]）
                    if "[DONE]" in assistant_content or not assistant_content.startswith("["):
                        final_response = assistant_content
                        llm_span.set_status(Status(StatusCode.OK))
                        break

                    # 解析动作并执行
                    action_result = self._execute_action(assistant_content)
                    messages.append({"role": "user", "content": f"Action result: {action_result}"})

            root_span.set_attribute("agent.total_turns", turn)
            root_span.set_attribute("agent.total_messages", len(messages))
            return final_response or "Max turns reached"

    def _execute_action(self, action_text: str):
        """执行 Agent 决策的动作"""
        with self.tracer.start_as_current_span("agent.execute_action") as span:
            span.set_attribute("agent.action_text", action_text[:200])

            # 解析动作（简化示例）
            if "[ACT:search]" in action_text:
                return self._mock_search()
            elif "[ACT:calculate]" in action_text:
                return self._mock_calculate()
            else:
                return "unknown_action"

    def _mock_search(self):
        with self.tracer.start_as_current_span("tool.search"):
            return '[{"title": "result 1"}, {"title": "result 2"}]'

    def _mock_calculate(self):
        with self.tracer.start_as_current_span("tool.calculate"):
            return "42"


# 使用示例
if __name__ == "__main__":
    agent = TracedAgent(model="gpt-4o")
    result = agent.run(
        user_message="帮我分析一下 2024 年 Q3 的销售数据",
        system_prompt="你是一个数据分析助手。如果需要数据，用 [ACT:search] 获取；如果需要计算，用 [ACT:calculate]。完成后用 [DONE] 标记结束。"
    )
    print(result)

这段代码的 trace 输出会包含每一轮 LLM 调用的 token 消耗、完整的消息历史引用关系，以及工具执行的耗时。配合 Jaeger 或 Zipkin 渲染，可以清楚看到 Agent 的决策树。

关键点：不要只在最外层打一个 span。Agent 的价值在于中间过程，trace 必须下沉到 turn 级别，否则出问题只能看到"调用了 GPT-4"这个事实，不知道内部发生了什么。

2.2 Metric：采集 LLM 应用的核心指标

Token 消耗和延迟是 LLM 应用独有的指标，传统 APM 不会采集这些。下面的代码展示如何用 Prometheus 格式暴露这些指标：

from prometheus_client import Counter, Histogram, Gauge, generate_latest, CONTENT_TYPE_LATEST
import time
from functools import wraps

# === 核心指标定义 ===

# Token 相关
llm_token_total = Counter(
    "llm_tokens_total",
    "Total tokens consumed",
    ["model", "token_type"]  # token_type: prompt / completion
)

llm_cost_usd = Counter(
    "llm_cost_usd_total",
    "Total estimated cost in USD",
    ["model", "provider"]
)

# 延迟指标
llm_latency_seconds = Histogram(
    "llm_request_latency_seconds",
    "LLM request latency",
    ["model", "status"],  # status: success / error
    buckets=[0.5, 1.0, 2.0, 5.0, 10.0, 30.0]
)

# Agent 行为指标
agent_turns = Histogram(
    "agent_turns_total",
    "Number of turns per agent run",
    ["agent_name"],
    buckets=[1, 2, 3, 5, 10, 20]
)

agent_context_length = Gauge(
    "agent_context_tokens_current",
    "Current context token count",
    ["agent_name", "session_id"]
)

# 错误指标
llm_errors = Counter(
    "llm_errors_total",
    "Total LLM errors",
    ["model", "error_type"]  # error_type: rate_limit / timeout / invalid_request
)

# Token 单价映射（单位：USD / 1M tokens）
TOKEN_PRICING = {
    "gpt-4o": {"prompt": 5.0, "completion": 15.0},      # $5/$15 per 1M tokens
    "gpt-4o-mini": {"prompt": 0.15, "completion": 0.6},
    "gpt-4-turbo": {"prompt": 10.0, "completion": 30.0},
}


def track_llm_call(model: str, provider: str = "openai"):
    """装饰器：自动采集 LLM 调用的指标"""
    pricing = TOKEN_PRICING.get(model, {"prompt": 0.0, "completion": 0.0})

    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            start = time.monotonic()
            error_type = None
            try:
                result = func(*args, **kwargs)
                return result
            except Exception as e:
                error_type = _classify_error(e)
                llm_errors.labels(model=model, error_type=error_type).inc()
                raise
            finally:
                elapsed = time.monotonic() - start
                status = "error" if error_type else "success"
                llm_latency_seconds.labels(model=model, status=status).observe(elapsed)

        return wrapper
    return decorator


def record_token_usage(model: str, provider: str, usage):
    """记录 token 使用量和成本"""
    llm_token_total.labels(model=model, token_type="prompt").inc(usage.prompt_tokens)
    llm_token_total.labels(model=model, token_type="completion").inc(usage.completion_tokens)

    pricing = TOKEN_PRICING.get(model, {"prompt": 0.0, "completion": 0.0})
    prompt_cost = (usage.prompt_tokens / 1_000_000) * pricing["prompt"]
    completion_cost = (usage.completion_tokens / 1_000_000) * pricing["completion"]
    total_cost = prompt_cost + completion_cost

    llm_cost_usd.labels(model=model, provider=provider).inc(total_cost)
    return total_cost


def _classify_error(exception: Exception) -> str:
    """对 LLM API 错误进行分类"""
    error_msg = str(exception).lower()
    if "rate" in error_msg or "429" in error_msg:
        return "rate_limit"
    elif "timeout" in error_msg or "timed out" in error_msg:
        return "timeout"
    elif "invalid" in error_msg or "400" in error_msg:
        return "invalid_request"
    elif "401" in error_msg or "authentication" in error_msg:
        return "auth"
    else:
        return "other"


# === HTTP 端点：暴露 Prometheus 指标 ===

from flask import Flask, Response
app = Flask(__name__)

@app.route("/metrics")
def metrics():
    return Response(generate_latest(), mimetype=CONTENT_TYPE_LATEST)

这套指标体系解决了三个核心问题：

1. Token 成本可视化：每个模型、每天、每个用户消耗了多少 token，成本一目了然。
2. 延迟分布：用 Histogram 而不是平均值的原因是 LLM 延迟分布高度偏斜，99 分位可能比中位数大 10 倍，平均值会掩盖问题。
3. 错误分类：不是所有 LLM 错误都一样，rate limit 需要扩容，timeout 需要优化 prompt 长度，invalid request 可能是代码 bug——分类后才能对症下药。

2.3 Log：结构化日志的 LLM 版本

传统日志是给人看的，LLM 日志必须同时给机器看。原因：你的日志里会包含 prompt 和 completion，这些文本可能长达几千 token，你没法用 grep 找出问题在哪里。

结构化日志是必选项，不是可选项：

import json
import logging
from datetime import datetime, timezone
from typing import Any, Optional
import structlog

# 配置 structlog（结构化日志库）
structlog.configure(
    processors=[
        structlog.stdlib.filter_by_level,
        structlog.stdlib.add_logger_name,
        structlog.stdlib.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
        structlog.processors.JSONRenderer()  # 输出 JSON 格式
    ],
    wrapper_class=structlog.stdlib.BoundLogger,
    context_class=dict,
    logger_factory=structlog.stdlib.LoggerFactory(),
    cache_logger_on_first_use=True,
)

logger = structlog.get_logger()


class LLMCallLogger:
    """LLM 调用的专用日志记录器"""

    def log_llm_request(
        self,
        trace_id: str,
        span_id: str,
        model: str,
        prompt_tokens: int,
        temperature: float,
        system_prompt_hash: str,  # prompt 的哈希，用于去重分析
        request_id: Optional[str] = None,
    ):
        """记录 LLM 请求的元数据"""
        logger.info(
            "llm_request",
            event_type="llm_request",
            trace_id=trace_id,
            span_id=span_id,
            model=model,
            prompt_tokens=prompt_tokens,
            temperature=temperature,
            system_prompt_hash=system_prompt_hash,
            request_id=request_id,
        )

    def log_llm_response(
        self,
        trace_id: str,
        span_id: str,
        model: str,
        completion_tokens: int,
        total_tokens: int,
        latency_ms: float,
        finish_reason: str,
        response_text_hash: str,  # 响应文本的哈希，用于检测重复
        error: Optional[str] = None,
    ):
        """记录 LLM 响应的完整信息"""
        log_data = {
            "event_type": "llm_response",
            "trace_id": trace_id,
            "span_id": span_id,
            "model": model,
            "completion_tokens": completion_tokens,
            "total_tokens": total_tokens,
            "latency_ms": latency_ms,
            "finish_reason": finish_reason,
            "response_text_hash": response_text_hash,
        }
        if error:
            log_data["error"] = error
            logger.error("llm_response", **log_data)
        else:
            logger.info("llm_response", **log_data)

    def log_agent_turn(
        self,
        trace_id: str,
        agent_name: str,
        turn: int,
        action_taken: str,
        context_tokens_before: int,
        context_tokens_after: int,
        evaluation_score: Optional[float] = None,  # 质量评分
    ):
        """记录 Agent 单轮执行"""
        logger.info(
            "agent_turn",
            event_type="agent_turn",
            trace_id=trace_id,
            agent_name=agent_name,
            turn=turn,
            action_taken=action_taken,
            context_tokens_before=context_tokens_before,
            context_tokens_after=context_tokens_after,
            context_growth=context_tokens_after - context_tokens_before,
            evaluation_score=evaluation_score,
        )

prompt 和 completion 的实际内容不适合直接写入日志（太长、可能含敏感信息）。哈希方案是更好的实践：保留 system_prompt_hash 和 response_text_hash，在需要调试时通过 trace_id 关联到具体的完整记录，单独查询获取。

---

三、工具链实战

3.1 Langfuse：LLM 应用的一站式观测平台

Langfuse 是目前开源社区最成熟的 LLM 应用可观测性平台，直接集成了 Trace、Metric、Eval 三个维度，部署简单，数据模型设计合理。

安装与基础集成：

# pip install langfuse langchain langchain-openai

from langfuse import Langfuse
import os

langfuse = Langfuse(
    public_key=os.getenv("LANGFUSE_PUBLIC_KEY"),
    secret_key=os.getenv("LANGFUSE_SECRET_KEY"),
    host=os.getenv("LANGFUSE_HOST", "https://cloud.langfuse.com"),  # 自托管改这里
)

from langchain_openai import ChatOpenAI
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate

# Langfuse 自动为 LangChain chain 注入 trace
llm = ChatOpenAI(model="gpt-4o", temperature=0.7)

prompt = PromptTemplate.from_template(
    "你是 {role}，回答用户问题：{question}"
)

chain = LLMChain(llm=llm, prompt=prompt)

# LangChain 的 LLMChain 会自动生成 trace
response = chain.invoke({
    "role": "技术顾问",
    "question": "解释一下什么是微服务架构"
})

Langfuse 后台会自动展示：每一次 LLM 调用的 prompt、完整 completion、token 消耗、延迟、model 版本，以及调用链路。

手动添加 span（对于非 LangChain 代码）：

from langfuse.decorators import observe, langfuse_context

@observe()
def my_agent_step(query: str, context: list):
    """被 @observe() 装饰的函数自动创建 span"""

    # 在当前 span 上添加属性
    langfuse_context.update_current_span(
        input=query,
        metadata={"context_length": len(context)}
    )

    result = call_llm(query, context)

    langfuse_context.update_current_span(
        output=result,
        metadata={"result_length": len(result)}
    )

    return result

# 在 Langfuse 后台可以看到完整的调用树

用户反馈与质量评估集成：

def submit_user_feedback(trace_id: str, score: int, comment: str = ""):
    """将用户评分关联到 trace，用于后续质量分析"""
    langfuse.score(
        trace_id=trace_id,
        name="user_rating",
        value=score,  # 0-1 或 0-100
        comment=comment,
    )

Langfuse 的核心价值在于：开箱即用的 trace UI、支持 Python 和 JS、评估（Eval）功能可以基于标注数据持续优化 prompt、生产环境可用。

3.2 OpenTelemetry：统一埋点标准

Langfuse 解决了 LLM 层的观测，但生产环境不会只有 LLM。OpenTelemetry 负责统一所有层的 trace——从 API 网关到数据库到 LLM 调用。

关键配置：

from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

# 配置 OTLP exporter，将 trace 发送到 Jaeger / Tempo / Grafana
otlp_exporter = OTLPSpanExporter(
    endpoint="http://tempo:4317",  # Grafana Tempo 地址
    insecure=True,
)
provider.add_span_processor(BatchSpanProcessor(otlp_exporter))

在 Kubernetes 环境中，配合 OpenTelemetry Operator，可以实现自动 instrumentation，无需在代码中手动埋点。JavaScript/TypeScript 侧的 Agent 代码同理：

// Node.js 侧使用 @langfuse/frameworks instrumentation
import { NodeSDK } from '@opentelemetry/sdk-node';
import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-grpc';

const sdk = new NodeSDK({
  traceExporter: new OTLPTraceExporter({
    url: 'http://otel-collector:4317',
  }),
});
sdk.start();

OpenTelemetry 解决的问题是：Langfuse 负责 LLM 层，其他一切由 OTel 负责，两者在 trace_id 层面打通，最终在 Grafana 或 Jaeger 里看到的是统一的视图。

3.3 Evidently AI：数据漂移与质量监控

Langfuse 记录了"发生了什么"，Evidently AI 回答"质量是否在退化"。

Evidently 的核心能力是漂移检测（Drift Detection）。当模型输出的语义分布发生变化时（即使输出格式没变），这往往意味着模型行为出现了异常——可能是 prompt 漂移、输入数据分布变化、或模型本身退化。

from evidently.dashboard import Dashboard
from evidently.tabs import DataDriftTab, TextOverviewTab
import pandas as pd

# 假设 reference_data 是"好的"时期的输出分布
# current_data 是当前生产环境的输出

reference_data = pd.read_csv("production_data_q1.csv")  # Q1 基线数据
current_data = pd.read_csv("production_data_q2.csv")   # Q2 监控数据

# 检测数值型指标的漂移（如 token 消耗、延迟）
drift_dashboard = Dashboard(tabs=[
    DataDriftTab(verbose_level=0),
    TextOverviewTab(verbose_level=0),
])

drift_dashboard.calculate(
    reference_data=reference_data,
    current_data=current_data,
    column_mapping={
        "target": "user_satisfaction_score",
        "numerical_features": ["latency_ms", "prompt_tokens", "completion_tokens"],
    }
)

drift_dashboard.save("drift_report.html")

Evidently 的 Text Overview 标签页可以分析 LLM 输出的文本特征分布——输出长度、词汇多样性、关键词频率等。这是一个被低估的能力：大多数 LLM 质量退化不会体现在 HTTP 状态码上，而是体现在输出的"风格"变化上。

---

四、常见踩坑点

踩坑一：幻觉检测是事后诸葛

幻觉（Hallucination）没有银弹。模型说了一个不存在的事实，这不会触发任何 HTTP 错误，也不会让你的 trace 变红。

实际可行的做法是辅助检索（RAG）和结构化输出约束的组合：

from pydantic import BaseModel, Field

class FactCheckResponse(BaseModel):
    statement: str
    is_supported: bool = Field(description="该陈述是否被提供的事实依据支持")
    confidence: float = Field(description="置信度 0-1")
    cited_source: str = Field(description="引用的信息来源，没有则填 'none'")

# 强制模型输出结构化 JSON，减少自由发挥空间
structured_llm = client.beta.chat.completions.parse(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "你是一个事实核查助手。只输出 JSON，不要添加解释。"},
        {"role": "user", "content": f"基于以下事实：{retrieved_context}\n\n核查：{user_statement}"}
    ],
    response_format=FactCheckResponse,
)

result = structured_llm.choices[0].message.parsed
if not result.is_supported and result.confidence < 0.7:
    # 触发人工复核流程
    escalate_for_review(user_statement, result)

同时，在 trace 中记录 is_supported=False 的比例，作为质量指标持续监控。趋势比单次事件更重要——如果某天这个比例突然上升，说明上游数据源出了问题或模型出现了系统性偏差。

踩坑二：Token 爆炸与 Context 溢出

多轮对话的 Agent 有一个隐性陷阱：context 里的历史消息会持续累积。GPT-4o 的 context window 是 128k tokens，看起来很大，但一个每天处理 1000 个用户的客服 Agent，如果每轮对话平均 10 条消息，context 的 token 消耗会线性增长，很快逼近 limit。

实测经验数据：

• 对话前 5 轮：context 增长可预测，约每轮 +300-800 tokens（取决于消息长度）
• 对话 10-15 轮：开始出现重复内容积累（模型倾向于重复自己见过的模式）
• 对话 20 轮以上：context 利用效率急剧下降，模型开始"遗忘"早期关键信息

解决方案不是加大 context window，是主动压缩：

def compress_context(messages: list, max_tokens: int = 4000) -> list:
    """保留 system prompt + 最近 N 条消息，超出部分截断"""
    system_msg = messages[0]  # system prompt 必须保留
    history = messages[1:]

    # 从后往前保留，直到 token 数量达标
    compressed_history = []
    current_tokens = 0

    for msg in reversed(history):
        msg_tokens = estimate_tokens(msg["content"])
        if current_tokens + msg_tokens <= max_tokens:
            compressed_history.insert(0, msg)
            current_tokens += msg_tokens
        else:
            break

    return [system_msg] + compressed_history


def estimate_tokens(text: str) -> int:
    """粗略估算 token 数量：中文约 1.5 字/token，英文约 4 字符/token"""
    chinese_chars = sum(1 for c in text if '\u4e00' <= c <= '\u9fff')
    other_chars = len(text) - chinese_chars
    return int(chinese_chars / 1.5 + other_chars / 4)

更精确的压缩需要引入 semantic summarization——让模型对历史对话做摘要，而不是简单截断。但这不是免费的：每次压缩本身就是一次 LLM 调用，有成本。权衡利弊决定使用哪种策略。

踩坑三：延迟陷阱——首 token 延迟与总延迟

LLM API 的延迟不只是"模型响应总时间"，还包含一个关键指标：首 token 延迟（Time to First Token, TTFT）。

import time

# 测量 TTFT 和总延迟
start = time.monotonic()
stream = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "写一篇 1000 字的文章"}],
    stream=True,
)

first_token_time = None
total_content = ""

for chunk in stream:
    if first_token_time is None and chunk.choices[0].delta.content:
        first_token_time = time.monotonic() - start
    if chunk.choices[0].delta.content:
        total_content += chunk.choices[0].delta.content

total_latency = time.monotonic() - start

print(f"TTFT: {first_token_time:.2f}s, Total: {total_latency:.2f}s")

为什么 TTFT 重要？因为 streaming UI 可以用 TTFT 立即给用户反馈——渲染"正在思考…"——即使完整响应还需要 30 秒。如果你的监控只记录 total latency，就会错过这个优化窗口。

典型延迟分解（GPT-4o，处理一个中等复杂度的请求）：

阶段	典型耗时	优化空间
网络 + API 路由	50-200ms	CDN / 换 region
TTFT（模型开始生成）	1-3s	减少 prompt 长度
逐 token 生成	10-40ms/token	限制 max_tokens
完整响应完成	5-30s	控制输出长度

prompt 越长，TTFT 越长，因为模型需要处理更多 context。每减少 1000 个 prompt tokens，TTFT 通常可以降低 0.5-1 秒。

---

五、作者观点与工具选型建议

观点一：可观测性是 Agent 开发的第一优先级

大多数团队在 Agent 开发初期追求的是"能不能跑起来"，把可观测性留到"后面再做"。这个顺序是错的。LLM 应用的问题往往不是"不工作"，而是"以你不知道的方式工作"。没有 trace，你连 Agent 在哪个节点出错都不知道。

建议：第一个 feature 交付之前，可观测性基础设施必须到位。这不是过度工程，是基本保障。

观点二：不要试图用一个工具解决所有问题

Langfuse 很好，但它不是银弹。Langfuse 解决 LLM 层的观测，但：

• 外部 API 调用的追踪交给 OpenTelemetry
• 业务指标的采集交给 Prometheus/Grafana
• 数据漂移的检测交给 Evidently AI
• 告警规则和 on-call 交给 Alertmanager 或 PagerDuty

工具链长才是正常的，每个工具有自己擅长的事情。不要为了统一性牺牲专业性。

工具选型建议

小型团队（<5人）或 PoC 阶段：
Langfuse Cloud（免费 tier）+ Python structlog 够了。先跑起来，不要在基础设施上过度投入。

中型团队（5-20人）或生产初期：
Langfuse 自托管（Postgres 后端）+ Grafana + Tempo。保留所有 trace 数据，支持内部审计和回放。

大规模生产环境：
OpenTelemetry 全家桶（Grafana Tempo 做存储，Loki 做日志，Mimir 做 metrics）+ Langfuse 自托管（作为 LangChain/LLAM 应用的专用观测层）+ Evidently AI（定期跑质量评估任务）。

选型时最重要的一个指标： 你的团队能否在生产故障时，在 5 分钟内定位到问题链路。如果不能，工具链还不到位。

---

监控 LLM 应用不是技术选型问题，是工程成熟度问题。当你的 Agent 进入生产环境、开始服务真实用户时，每一个无法回答的"为什么模型这样回答"都是一个潜在的事故。传统 APM 不知道 LLM 里面发生了什么，你得自己建这套感知能力。

从 trace 开始，成本最低，收益最高。一个带完整 span 的 trace，顶得上 100 行调试日志。