Langfuse 开源 LLM 可观测性：别再盲目调试 RAG 应用

RAG 应用在生产环境返回了一个看起来非常自信、但引用错误政策文档的答案。如果没有 trace，你只能在一个 Slack 线程里猜测："是模型问题？" "是 prompt 漂移？" "是 embeddings？" "换个 GPT 试试？" 这不是调试，这是拿着 API Key 在猜。

Langfuse 开源 LLM 可观测性工具

Langfuse 是一个开源的 LLM 可观测性平台。它提供 traces（一次用户请求）、嵌套 spans（每个流水线步骤）、generations（包含 token、成本、延迟的 LLM 调用），以及由人工或自动评估附加的 scores。目标很简单：打开那个糟糕的请求，看到底是哪一步出了问题。

你真正要解决的问题

一个典型的 RAG 路径是这样的：

检索 top-k 文本块
用这些文本块构建 prompt
调用模型
返回答案

没有插桩时，生产环境只能看到第 4 步。Langfuse 让第 1–3 步在同一棵调用树中可见，并附带每次 LLM 调用的耗时和成本。

问题从来不是"模型不行"，而是"检索层悄悄地把错误文档喂给了模型"。可观测性的价值就在于把这种静默失败暴露出来。

LLM 可观测性（LLM Observability） 是把传统软件可观测性的三大支柱——日志（logs）、指标（metrics）、链路追踪（traces）——适配到以 LLM 为核心的流水线上。传统 APM 工具通常只能看到一次 HTTP 请求的耗时和状态码，无法回答“prompt 里到底拼进了哪些文本块”“这次调用花了多少 token 和美元”“检索回来的相似度分布如何”这类问题。LLM 可观测性因此引入 span（跨度） 作为基本单位：一个 span 代表流水线中的一个步骤，可携带输入、输出、耗时、成本等结构化字段；多个嵌套 span 组成一棵 trace（追踪），对应一次端到端用户请求。在这样的模型下，生成（generation）只是若干 span 类型中的一种，专门记录 LLM 调用的模型名、token 用量与费用。

安装与凭证配置

pip install langfuse openai

设置环境变量（Cloud 或自托管使用同一套 SDK）：

export LANGFUSE_PUBLIC_KEY="pk-lf-..."
export LANGFUSE_SECRET_KEY="sk-lf-..."
export LANGFUSE_BASE_URL="https://cloud.langfuse.com"   # EU
# US: https://us.cloud.langfuse.com
# JP: https://jp.cloud.langfuse.com

可选的连通性检查（不要用于热路径）：

from langfuse import get_client
get_client().auth_check()

Langfuse 支持多区域部署（EU / US / JP / HIPAA / 自托管），LANGFUSE_BASE_URL 必须与创建项目时的区域一致。

Drop-in OpenAI：只改一行导入

如果你已经在使用 OpenAI Python SDK，Langfuse 的 drop-in 方案只需更改 import 即可自动记录 prompts、completions、用量、成本、延迟和 API 错误：

# before: import openai
from langfuse.openai import openai
# or: from langfuse.openai import OpenAI, AsyncOpenAI

completion = openai.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {"role": "system", "content": "Answer only from the context."},
        {"role": "user", "content": "What is the refund window?"},
    ],
    name="support-answer",  # shows up as the generation name in Langfuse
)
print(completion.choices[0].message.content)

你还可以在同一个调用中通过 metadata 传递 Langfuse 属性（例如 langfuse_session_id、langfuse_user_id、langfuse_tags）。

用 @observe 嵌套整条流水线

单次 OpenAI 调用只是一个 generation，而 RAG 流水线需要一棵trace 树。给每个步骤加装饰器：

from langfuse import observe, get_client, propagate_attributes
from langfuse.openai import openai

@observe(name="retrieve")
def retrieve(query: str, k: int = 4) -> list[dict]:
    return search_chunks(query, k=k)

@observe(name="build_prompt")
def build_prompt(query: str, chunks: list[dict]) -> list[dict]:
    context = "\n\n".join(c["text"] for c in chunks)
    return [
        {"role": "system", "content": "Answer only from the context. If missing, say you don't know."},
        {"role": "user", "content": f"Context:\n{context}\n\nQuestion: {query}"},
    ]

@observe(name="rag_answer")
def rag_answer(query: str, user_id: str, session_id: str) -> str:
    with propagate_attributes(
        user_id=user_id,
        session_id=session_id,
        tags=["support-rag"],
        metadata={"pipeline": "support-rag-v2"},
    ):
        chunks = retrieve(query)
        messages = build_prompt(query, chunks)
        completion = openai.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
            name="generate",
        )
        return completion.choices[0].message.content

嵌套的 @observe 调用会自动形成父子 spans，OpenAI drop-in 的 generation 会嵌套在当前活动观测之下。

关键工作流：调试错误的 RAG 答案

这才是你真正要交付的部分。

第一步：找到出错的请求

通过 user_id / session_id / 时间窗口找到对应的 trace，你应该看到类似这样的结构：

rag_answer（根 span）
- retrieve — 输入查询、输出 chunks（ID、分数、文本片段）
- build_prompt — 组装好的 messages
- generate — 模型、tokens、成本（USD）、延迟、completion 文本

第二步：从答案自底向上排查

如果 generate 的输出流畅但错误，不要先去换模型。打开 retrieve 检查：

top chunks 是否来自正确的文档 / 产品 / 地区？
相似度分数是否异常低，或聚集在样板文本上？
是否某个过滤器把预期的政策段落丢弃了？

在"错误政策"这个案例中，generate span 是无辜的。retrieve span 显示的是来自旧版退款 FAQ 的高置信度 chunks。根因是索引过期或缺少 metadata 过滤，而不是"LLM 幻觉"。

第三步：附加 score，让失败可检索

在当前观测内（或事后通过带 trace_id 的 create_score）：

from langfuse import get_client
langfuse = get_client()

langfuse.score_current_span(
    name="retrieval_relevance",
    value=0.2,
    data_type="NUMERIC",
    comment="Top chunks from outdated refund FAQ; annual-plan policy missing",
)
langfuse.score_current_trace(
    name="answer_correctness",
    value=0,
    data_type="BOOLEAN",
    comment="Wrong refund window cited",
)

Score 类型支持 NUMERIC、CATEGORICAL、BOOLEAN 和 TEXT。

第四步：修复出错的 span，重跑同一查询

新的 trace 应该显示更好的 retrieve 输出和通过的 score。这一对"修复前/修复后"的 trace 加上 scores，就是下一次 prompt 或索引变更时的回归证据。

“自底向上排查” 指的是沿 trace 树的调用链从叶子节点往根节点回溯，而不是从最终答案出发猜原因。在一次 RAG 请求中，根 span 是整条流水线，叶子是具体的 LLM 调用和检索函数。调试时先看最靠近输出、同时最依赖上游输入的步骤（通常是 retrieve），检查它返回的 chunks 是否包含正确文档、相似度分数是否合理、过滤条件是否误删了关键段落。只有当证据显示检索层正常、prompt 组装无误、模型参数也正确时，才把怀疑对象升级为模型本身。这样做的原因是 LLM 的错误输出往往是上游错误的放大而非独立故障：索引过期、metadata 过滤失效、chunk 切分不合理都会让模型在“忠实于给定上下文”的前提下给出错误答案，从外部看却像是幻觉。

容易踩的坑

短生命周期进程要 flush。 SDK 在后台批量上传，脚本和 serverless handler 退出前应调用 get_client().flush()（或 shutdown()），否则最后一批事件会丢失。

@observe 处理大 I/O。 巨大的 chunk dump 会增加开销。可以按装饰器用 capture_input=False / capture_output=False 禁用，或通过 LANGFUSE_OBSERVE_DECORATOR_IO_CAPTURE_ENABLED 全局控制。

区域必须匹配。 LANGFUSE_BASE_URL 必须与创建项目时的区域一致，错误的 base URL 表现就像"什么都没显示"。

Assistants API 覆盖有限。 OpenAI Assistants 路径（服务端状态）未被 drop-in 完全覆盖。建议在自有编排外用 @observe 包一层。

采样。 高频应用可以设置 LANGFUSE_SAMPLE_RATE（或 Langfuse(sample_rate=...)），但要保证正在调试的路径保持 100% 采样。

上手清单

pip install langfuse openai，设置 LANGFUSE_PUBLIC_KEY / SECRET_KEY / BASE_URL
切换到 from langfuse.openai import openai，自动记录 generation
在 retrieve / prompt / answer（及任何工具）上加 @observe(name=...)
尽早用 propagate_attributes 传播 user_id、session_id 和 tags
遇到坏答案，打开 trace → 先看 retrieve span，再怪模型
用 score_current_span / score_current_trace 让失败可被过滤
脚本和 serverless 中记得 flush()

当生产环境的 AI 答案出错时，你是打开一条 trace，还是开启一轮 Slack 猜测？