Langfuse 深度解析

梯形图转SCL | 博途AI辅助编程文档 | AI模型价格对比 | AI工具导航 | ONNX模型库 | Vibe Coding教程 | PLC在线仿真器 | Tripo 3D | Meshy AI | ElevenLabs | KlingAI | ArtSpace | Phot.AI | InVideo

你发布了一个 LLM 智能体。然后呢？

你熬夜把 GPT-4o、向量数据库和复杂的路由提示词组装在一起。你部署了新的 RAG 应用。用户真的在调用 API。然后……一片寂静。

你完全不知道它是否真的表现良好。你的标准错误日志显示 HTTP 状态码，但它们完全没有告诉你为什么你的智能体给了一个奇怪的、幻觉化的答案——用户刚刚截图并在 Twitter 上艾特了你的团队。

这就是超越简单聊天提示词的现实：生产环境是一个黑盒。

你无法用 console.log() 来调试一个 15 步的 RAG 管道。你无法为非确定性的模型行为编写标准单元测试。你肯定无法向你的 CTO 解释为什么你的客服智能体告诉用户他们的订单"可能存在于宇宙的某个地方"。

1、缺失的可观测性层

这就是 Langfuse 的用武之地。它提供了复杂的 RAG 系统和自主智能体真正需要的深度追踪和遥测。

你不再需要猜测用户会话期间发生了什么，而是获得：

• 完整追踪： 可视化智能体执行的整个生命周期——从初始查询重写和向量数据库检索步骤到最终的 LLM 综合。
• 结构化评估： 运行自动化评分管道来测试幻觉、相关性和毒性。
• 提示词管理： 独立于应用代码部署来对提示词进行版本控制。
• 回归测试： 从真实用户交互中构建"黄金数据集"，确保模型更新不会破坏现有行为。

所有这些都打包在一个开源平台中。

2、我们要构建什么

本指南是一个实用蓝图，用于将你的 RAG 和智能体工作流从脆弱的原型转变为生产级部署。我们将分解核心架构概念，为 AI 客服智能体提供一个动手实现的示例，并诚实地比较 Langfuse 与替代工具，以便你为团队选择正确的技术栈。

让我们打开这个黑盒。

3、什么是 Langfuse，为什么它存在？

在 Langfuse 之前，构建 LLM 应用的工程团队遇到了一个令人沮丧的事实：当你的核心应用逻辑由非确定性统计模型处理时，传统的调试工具就失效了。

你无法在神经网络中设置断点。你无法保证相同输入产生相同输出。当智能体失败时，bug 通常不是语法错误，而是格式不良的提示词、向量数据库中糟糕的文档分块策略，或者基础模型本身的意外行为。

Langfuse 由 Max Langkamp、Marc Klingen 和 Clemens Rawert 于 2023 年创立，当时他们在构建自己的 AI 应用时遇到了这些确切的障碍。他们在 Apache 2.0 许可证下开源了该平台，它很快成为 GitHub 上最突出的 LLM 工程工具库之一。

它解决什么问题？

当你从简单的 API 调用转向多步骤 RAG 管道和自主智能体时，操作挑战完全转变：

4、核心概念：追踪、跨度、生成和评分

在编写第一行代码之前，你需要理解 Langfuse 的四个基础原语。如果你曾经使用过 OpenTelemetry，这些概念会立即让你感到熟悉，但它们专门为 RAG 管道和自主智能体量身定制。

它们决定了数据在平台内部的结构：

1. 追踪（Trace）

追踪代表你的应用程序中一个完整的端到端操作——通常是一个单独的用户请求。它作为整个数据树的根，捕获关键的顶级元数据，如唯一 ID、可选名称、用户或会话 ID 以及总执行时间。把它想象成分布式系统追踪中的请求追踪，只是它包含了你整个 LLM 推理链。

2. 跨度（Span）

跨度是追踪中任何不是显式 LLM 调用的工作单元。跨度包装了围绕模型的工程逻辑。数据库查询、向量相似性搜索、文档重排序步骤或外部 API 调用都变成跨度。由于你可以任意深度嵌套跨度，它们允许你将复杂的多步骤智能体行为映射成清晰的可视化调用树。

3. 生成（Generation）

生成是专门为 LLM 调用保留的特殊类型的跨度。Langfuse 独立跟踪这些，因为它们需要独特的跟踪参数。生成会自动捕获确切的输入消息、模型输出、提示词名称和 token 计数。你永远不需要手动计算成本；Langfuse 引用自己的托管定价数据库来实时计算 token 支出。

4. 评分（Score）

评分是直接附加到追踪、跨度或生成的质量信号。评分是将 Langfuse 从被动日志工具转变为真正评估平台的机制。它们允许你基于三个来源附加量化指标：

• 人工审核员： 直接在 Langfuse UI 中注释生产输出。
• LLM 裁判： 大规模运行自动化评估以检查相关性或语气。
• 你自己的代码： 以编程方式执行基于规则的检查（例如，验证输出是否包含必需的免责声明或是否是有效的 JSON）。

5、五分钟完成设置

步骤 1：创建 Langfuse 帐户

访问 cloud.langfuse.com，创建一个项目，然后从 Settings → API Keys 获取 API 密钥。

步骤 2：安装 SDK

# Python
pip install langfuse openai anthropic

# Node.js
npm install langfuse openai

步骤 3：设置环境变量

# .env
LANGFUSE_PUBLIC_KEY=pk-lf-xxxxxxxxxxxxxxxx
LANGFUSE_SECRET_KEY=sk-lf-xxxxxxxxxxxxxxxx
LANGFUSE_HOST=https://cloud.langfuse.com  # 或你的自托管 URL

步骤 4：你的第一个追踪（直接集成）

这就是 Langfuse 获得"零配置"声誉的地方。如果你使用 OpenAI 或 Anthropic，只需替换你的导入，然后就完成了：

import os, base64
from dotenv import load_dotenv
load_dotenv()

# 一次性 OTEL 设置 → 将追踪发送到 Langfuse
# (langfuse.anthropic 在 v4 中已移除；这是 v4 的等效版本)
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry import trace
from opentelemetry.instrumentation.anthropic import AnthropicInstrumentor

auth = base64.b64encode(
    f"{os.environ['LANGFUSE_PUBLIC_KEY']}:{os.environ['LANGFUSE_SECRET_KEY']}".encode()
).decode()
provider = TracerProvider()
provider.add_span_processor(BatchSpanProcessor(
    OTLPSpanExporter(
        endpoint=f"{os.environ['LANGFUSE_BASE_URL']}/api/public/otel/v1/traces",
        headers={"Authorization": f"Basic {auth}"}
    )
))
trace.set_tracer_provider(provider)
AnthropicInstrumentor().instrument()

# 之前
# from anthropic import Anthropic

# 之后——没有其他变化
from anthropic import Anthropic

client = Anthropic()

message = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "What is LLM observability?"}]
)

print(message.content[0].text)
# ✅ 现在你的 Langfuse 仪表板中会出现一个追踪

就这样。没有配置、没有包装器、没有额外参数。每次调用都会自动追踪，包含完整的输入、输出、模型、token 和成本。

6、你将在仪表板中看到什么

运行该代码后几秒钟内，你的 Langfuse 仪表板将显示：

• 发送给模型的确切消息
• 模型的完整响应
• 延迟（首 token 时间 + 总时间）
• token 计数（提示词、补全、总计）
• 以美元计的成本（自动计算）
• 所有跨度的时间线视图

7、使用 @observe() 构建结构化追踪

直接集成对简单脚本很有用。对于真实应用——RAG 管道、智能体、多步骤链——你需要结构化追踪来准确显示每个步骤发生了什么。

Langfuse 的 @observe() 装饰器是为此的主要工具。它包装任何 Python 函数并在追踪树中创建一个跨度。

基本装饰器用法

from dotenv import load_dotenv
load_dotenv()

from langfuse.decorators import observe, langfuse_context
from anthropic import Anthropic

client = Anthropic()

@observe()
def retrieve_documents(query: str, top_k: int = 5):
    docs = vector_db.similarity_search(query, k=top_k)
    langfuse_context.update_current_observation(
        metadata={
            "num_docs_retrieved": len(docs),
            "top_score": docs[0].score if docs else None
        }
    )
    return [doc.page_content for doc in docs]

@observe()
def rerank_documents(query: str, docs: list):
    reranked = cross_encoder.rank(query, docs)
    return reranked[:3]

@observe()
def answer_question(user_query: str, user_id: str):
    langfuse_context.update_current_trace(
        user_id=user_id,
        tags=["rag", "production"],
        session_id=f"session_{user_id}"
    )
    docs = retrieve_documents(user_query)
    top_docs = rerank_documents(user_query, docs)
    context = "\n\n".join(top_docs)
    message = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[
            {
                "role": "user",
                "content": f"Answer based on this context:\n{context}\n\nQuestion: {user_query}"
            }
        ]
    )
    return message.content[0].text

answer = answer_question("What is our return policy?", user_id="user_42")
print(answer)
# ✅ 根追踪 + 子跨度出现在你的 Langfuse 仪表板中

这种可见性是在黑暗中调试和完全了解管道全貌之间的区别。

异步支持

Langfuse 的装饰器在异步应用程序（FastAPI 等）中无缝工作：

import asyncio
from langfuse.decorators import observe, langfuse_context
from anthropic import AsyncAnthropic

client = AsyncAnthropic()

@observe()
async def async_retrieve(query: str, top_k: int = 5):
    await asyncio.sleep(0)  # 模拟异步数据库调用
    docs = DOCS[:top_k]
    langfuse_context.update_current_observation(
        metadata={"num_docs_retrieved": len(docs)}
    )
    return [doc.page_content for doc in docs]

@observe()
async def async_rag_pipeline(query: str, user_id: str):
    langfuse_context.update_current_trace(user_id=user_id)
    docs = await async_retrieve(query)
    response = await async_llm_call(query, docs)
    return response

8、评估和评分：了解你的应用是否真正良好

追踪告诉你发生了什么。评分告诉你发生得有多好。这是 Langfuse 将认真团队与只是在记录日志的团队区分开来的部分。

三种类型的评分：

8.1 基于规则的评分（从这里开始）

基于规则的评分是即时的、免费的，不需要 LLM 调用。将它们内联添加到你的管道中：

@observe()
def run_support_pipeline(query: str):
    langfuse_context.update_current_trace(tags=["rule-based-scores"])
    response = call_llm(query)
    
    # 评分：响应长度
    word_count = len(response.split())
    langfuse_context.score_current_trace(name="response_length", value=word_count)
    
    # 评分：模型是否拒绝？
    refused = any(phrase in response.lower() for phrase in
                  ["i cannot", "i'm unable", "i don't have access"])
    langfuse_context.score_current_trace(name="refusal", value=1 if refused else 0)
    
    # 评分：响应是否为有效 JSON？
    try:
        json.loads(response)
        langfuse_context.score_current_trace(name="valid_json", value=1)
    except Exception:
        langfuse_context.score_current_trace(name="valid_json", value=0)
    
    return response

8.2 LLM 作为裁判（用于细微质量评估）

对于有用性、语气、事实准确性或幻觉检测等，你需要一个 LLM 来评估：

def evaluate_hallucination(trace_id: str, output: str, context: str):
    """LLM 裁判：输出是否包含超出所提供上下文的幻觉？"""
    eval_prompt = f"""你是一个事实准确性评估器。
    
给定上下文和 AI 的输出，确定输出是否包含任何
上下文中不支持的事实声明（幻觉）。

上下文：{context}
AI 输出：{output}

仅以有效的 JSON 格式回复：
{{"score": 0.95, "reasoning": "...", "hallucination_detected": false}}

其中 score 1.0 = 完全基于事实，0.0 = 完全幻觉。"""

    result = eval_client.messages.create(
        model="claude-haiku-4-5-20251001",  # 用于评估的廉价模型
        max_tokens=512,
        messages=[{"role": "user", "content": eval_prompt}]
    )
    
    # 解析结果并附加评分
    lf.score(trace_id=trace_id, name="groundedness", value=data["score"])

8.3 人工标注（你的基本事实）

在 Langfuse UI 中配置标注队列（Settings → Scores → Score Configs）来定义自定义评分标准：

• 有用性 — 1 到 5 的等级
• 语气 — 分类：专业 / 中立 / 不当
• 包含 PII — 布尔值
• 客户是否满意 — 布尔值

然后创建一个标注队列，自动将随机样本（例如 5% 的生产流量、100% 质量评分 < 0.6 的追踪）路由给你的质量团队进行审核。

人工标签有两个关键目的：

1. 它们是测量其他一切的基本事实
2. 它们让你校准 LLM 裁判 — 将裁判评分与相同追踪上的人工评分进行比较以验证你的评估管道

9、提示词管理：无需部署代码即可发布提示词更改

这是 Langfuse 最被低估的功能之一。在大多数团队中，更改提示词意味着编辑代码中的字符串、打开拉取请求、等待审查——这个周期可能需要数小时或数天。Langfuse 完全打破了这种耦合。

9.1 工作原理

from langfuse import Langfuse

lf = Langfuse()

# 1. 在 Langfuse 中创建提示词（只需一次）
lf.create_prompt(
    name="customer-support-agent",
    type="text",
    prompt=(
        "你是 AcmeCorp 的有用客服智能体。\n\n"
        "客户姓名：{{customer_name}}\n"
        "订阅计划：{{subscription_plan}}\n"
        "当前日期：{{current_date}}\n\n"
        "帮助客户解决问题。要有礼貌、简洁和准确。"
    ),
    labels=["production"]
)

# 2. 在运行时获取并编译
prompt = lf.get_prompt("customer-support-agent")  # 获取生产版本
system_message = prompt.compile(                    # 填充变量
    customer_name=customer_name,
    subscription_plan=plan,
    current_date=...
)

# 3. 将生成链接到提示词版本
langfuse_context.update_current_observation(prompt=prompt)

Langfuse 缓存提示词并在后台刷新——因此每个请求都没有延迟。

9.2 在生产中对提示词进行 A/B 测试

def get_prompt_for_user(user_id: str):
    """基于 user_id 的确定性 50/50 分割。"""
    use_variant_b = hash(user_id) % 100 < 50
    if use_variant_b:
        return lf.get_prompt("support-agent-v2")   # 挑战者
    else:
        return lf.get_prompt("support-agent")       # 控制组

@observe()
def handle_query(query: str, user_id: str):
    prompt = get_prompt_for_user(user_id)
    langfuse_context.update_current_trace(
        user_id=user_id,
        metadata={"prompt_variant": prompt.name}
    )
    # ... 使用提示词 ...

在 Langfuse 中按 metadata.prompt_variant 过滤追踪以比较平均质量评分、延迟和成本。不需要统计学博士学位。

10、数据集和实验运行：在用户发现之前捕获回归

数据集是你将"我觉得它变差了"转变为"它在问答上变差了 12%，在摘要上变好了 3%"的方式。

# 1. 构建黄金数据集
lf.create_dataset(name="customer-support-golden-set", description="用于回归测试的手工策划的 QA 对")

# 2. 添加测试用例
test_cases = [
    {"input": {"question": "60 天后我能退款吗？"}, "expected": {"must_contain": "30"}},
    {"input": {"question": "如何导出我的数据？"}, "expected": {"must_contain": "Settings"}},
]
for case in test_cases:
    lf.create_dataset_item(dataset_name="customer-support-golden-set", input=case["input"])

# 3. 运行实验
for item in dataset.items:
    with item.observe(run_name="sonnet-4-6-baseline") as trace_id:
        output = support_pipeline(query=item.input["question"], model="claude-sonnet-4-6")
        score = evaluate_against_expected(output, item.expected_output)
        lf.score(trace_id=trace_id, name="meets_criteria", value=score)

在 Langfuse 中转到 Datasets > customer-support-golden-set 查看并排运行比较。

11、实战用例：构建生产客服智能体（提供代码）

让我们用一个完整、真实的示例将所有内容整合在一起。我们将为 SaaS 产品构建一个客服助手，它：

1. 检索相关的知识库文章
2. 生成有帮助的响应
3. 自动评估质量
4. 按客户计划跟踪成本

11.1 项目结构

real_project/
├── config.py          ← 客户端 + 提示词引导
├── retrieval.py       ← 步骤 1：查找相关 KB 文章
├── generation.py      ← 步骤 2：构建提示词 + 调用 Claude
├── pipeline.py        ← 编排器：连接步骤 1+2，添加评分
├── api.py             ← FastAPI 服务器：/support/ask + /support/feedback
├── requirements.txt
├── tests/
│   ├── build_dataset.py  ← 在 Langfuse 中创建黄金测试数据集（运行一次）
│   └── ci_eval.py        ← 回归门：如果质量低于阈值则失败
└── .github/
    └── workflows/
        └── eval.yml      ← GitHub Actions：在每个 PR 上运行 ci_eval.py

11.2 config.py：客户端和提示词引导

from langfuse import Langfuse
from anthropic import Anthropic

lf = Langfuse()
prod_client = Anthropic()   # 被追踪——所有生产调用都通过 @observe 装饰器
eval_client = Anthropic()   # 未被追踪——评估调用不会污染你的生产追踪

两个单独的 Anthropic() 实例，即使它们是相同的对象。这是一个约定：prod_client 调用被捕获在 Langfuse 追踪中，因为它们在 @observe 装饰的函数内调用。eval_client 调用发生在任何追踪上下文之外，因此它们永远不会出现为跨度——你的评估运行不会污染你的生产仪表板。

11.3 pipeline.py — 编排器

这是连接所有内容并拥有根追踪的主函数。

@observe(name="support_pipeline")
def handle_support_query(query, customer_id, customer_plan, session_id):
    # 1. 将客户上下文附加到根追踪
    langfuse_context.update_current_trace(
        user_id=customer_id,
        tags=[f"plan:{customer_plan}", "channel:support-widget"]
    )
    
    # 2. 检索 + 生成
    docs = retrieve_knowledge_base(query)
    response = generate_support_response(query, docs, customer_plan)
    
    # 3. 基于规则的评分（同步，零成本）
    langfuse_context.score_current_trace(name="response_length", value=len(response.split()))
    
    # 4. 异步 LLM 裁判（非阻塞）
    trace_id = langfuse_context.get_current_trace_id()
    threading.Thread(target=_run_async_eval, args=(trace_id, ...), daemon=True).start()
    
    return response

11.4 api.py — FastAPI 服务器

两个端点：

• POST /support/ask — 调用 handle_support_query 并返回响应和 trace_id
• POST /support/feedback — 接收 trace_id 和点赞/点踩

trace_id 被返回给前端——这样 UI 可以将用户反馈附加到产生此响应的确切追踪。

11.5 tests/ci_eval.py 回归门

这是在用户发现之前捕获回归的关键。它将黄金数据集中的每个项目运行通过实时管道，等待 LLM 裁判评分，并检查它们是否低于阈值：

THRESHOLD = 0.75   # 如果任何指标平均值低于此值则失败
for item in dataset.items:
    with item.observe(run_name=RUN_NAME) as trace_id:
        result = handle_support_query(query=item.input["query"], ...)

如果任何指标失败，脚本以代码 1 退出——这会使 GitHub Actions 作业失败并阻止 PR 合并。

12、运行

启动服务器：

uvicorn api:app --reload --port 8000

发送测试查询：

curl -X POST http://localhost:8000/support/ask \
  -H "Content-Type: application/json" \
  -d '{"query": "如何导出我的数据？", "customer_id": "user_42", "customer_plan": "pro"}'

运行回归套件：

python tests/ci_eval.py

原文链接: Production-Grade agentic observability: a complete Langfuse Deep Dive

汇智网翻译整理，转载请标明出处