LLM 应用的可观测性困境

当你的 LLM 应用从 Demo 走向生产,一系列问题随之而来:某个请求为什么返回了错误结果?Prompt 改动后效果到底是变好还是变差?不同模型的延迟和成本如何对比?传统 APM 工具(如 Datadog)对 LLM 场景缺乏原生支持。Langfuse 作为开源的 LLM 可观测性平台,正填补了这一空白。

Langfuse 核心功能

功能矩阵

功能模块说明
Tracing请求级全链路追踪,含 LLM 调用、工具调用、检索流程
Prompt ManagementPrompt 版本管理、AB 测试、在线编辑
Analytics延迟、Token 用量、成本统计
Evaluation人工标注 + 自动评估(基于规则或模型裁判)
User Feedback用户点赞/点踩采集与关联
Model Usage多模型/多项目用量看板

Tracing:全链路追踪

集成方式

Python SDK(推荐)

from langfuse import Langfuse
from langfuse.decorators import observe

langfuse = Langfuse()

@observe()  # 自动创建 trace
def chat_bot(user_message: str):
    # 嵌套的 @observe 会自动建立父子关系
    retrieved_docs = retrieve(user_message)  # 被 trace
    answer = generate_response(user_message, retrieved_docs)  # 被 trace
    return answer

@observe(as_type="generation")
def generate_response(question, docs):
    response = openai.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": f"基于以下内容回答:{docs}\n问题:{question}"}],
    )
    return response.choices[0].message.content

@observe(as_type="span")
def retrieve(query):
    # 检索逻辑
    return vector_db.search(query)

OpenAI SDK 集成(Drop-in)

from langfuse.openai import openai

# 只需替换 import,所有 OpenAI 调用自动被 trace
response = openai.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}],
    metadata={"user_id": "user_123"},  # 附加用户信息
)

LangChain 集成

from langfuse.callback import CallbackHandler

handler = CallbackHandler()
chain.invoke(
    {"input": "什么是量子计算?"},
    config={"callbacks": [handler]}
)

Trace 结构

Trace (chat_bot 调用)
├── Span (retrieve - 向量检索)
│   └── Generation (embedding 调用)
├── Generation (generate_response - LLM 调用)
│   ├── input: "基于以下内容..."
│   ├── output: "量子计算是..."
│   ├── model: gpt-4o
│   ├── tokens: {input: 350, output: 280}
│   └── cost: $0.0091
└── Span (后处理)

自定义 Span

from langfuse import Langfuse

langfuse = Langfuse()

trace = langfuse.trace(name="rag-pipeline", user_id="user_123")

span = trace.span(name="document-retrieval", input={"query": "量子计算"})
# ... 执行检索 ...
span.end(output={"docs": [...]})

gen = trace.generation(
    name="answer-generation",
    model="gpt-4o",
    input={"messages": [...]},
    output={"content": "..."},
    usage={"prompt_tokens": 350, "completion_tokens": 280},
)

Prompt 管理

版本化 Prompt

from langfuse import Langfuse

langfuse = Langfuse()

# 获取生产环境的 Prompt(自动缓存)
prompt = langfuse.get_prompt(
    name="summarizer",
    version=2,  # 指定版本,或省略获取最新
)

# 编译 Prompt(变量替换)
compiled = prompt.compile(
    text="一段需要总结的长文本...",
    max_length="200"
)

# 直接用于 LLM 调用
response = openai.chat.completions.create(
    model=prompt.config["model"],
    messages=[{"role": "user", "content": compiled}],
)

Prompt AB 测试

import random

# 50% 用户用 v1,50% 用 v2
version = random.choice(["v1", "v2"])
prompt = langfuse.get_prompt(name="customer-support", version=version)

# Langfuse 自动记录每个请求使用了哪个版本
trace = langfuse.trace(name="support-chat")
trace.update(metadata={"prompt_version": version})

在线编辑

Langfuse Web UI 提供 Prompt 在线编辑器:

  • 修改变量、修改模板
  • 测试不同输入下的渲染效果
  • 发布新版本(灰度比例可调)
  • 回滚到任意历史版本

自部署指南

Docker Compose 部署

# docker-compose.yml
version: '3.8'

services:
  langfuse:
    image: langfuse/langfuse:latest
    ports:
      - "3000:3000"
    environment:
      - DATABASE_URL=postgresql://langfuse:langfuse@db:5432/langfuse
      - NEXTAUTH_SECRET=your-secret-here
      - SALT=your-salt-here
      - NEXTAUTH_URL=http://localhost:3000
      - TELEMETRY_ENABLED=false
    depends_on:
      db:
        condition: service_healthy

  db:
    image: postgres:15
    environment:
      - POSTGRES_USER=langfuse
      - POSTGRES_PASSWORD=langfuse
      - POSTGRES_DB=langfuse
    volumes:
      - langfuse-db:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U langfuse"]
      interval: 3s

volumes:
  langfuse-db:
docker compose up -d
# 访问 http://localhost:3000 创建管理员账户

数据持久化与备份

# 备份
docker exec langfuse-db-1 pg_dump -U langfuse langfuse > backup_$(date +%Y%m%d).sql

# 恢复
cat backup_20260624.sql | docker exec -i langfuse-db-1 psql -U langfuse langfuse

评估与标注

人工标注

# 在 Web UI 中配置标注队列
# 用户可以对每条 trace 进行:
# - 质量评分(1-5 星)
# - 标签标注(如 "hallucination", "good", "off-topic")
# - 文字备注

# 通过 API 查询标注结果
scores = langfuse.get_scores(
    trace_id="trace_xxx",
)

自动评估

# 基于规则的自动评估
def evaluate_response(trace_id, output, expected_keywords):
    score = 1.0 if any(kw in output for kw in expected_keywords) else 0.0
    langfuse.score(
        trace_id=trace_id,
        name="keyword-coverage",
        value=score,
        comment="关键词覆盖率检查"
    )

# 基于模型裁判的自动评估
def model_graded_score(trace_id, question, answer):
    judge_prompt = f"问题:{question}\n回答:{answer}\n请评分1-5"
    score = int(openai.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": judge_prompt}]
    ).choices[0].message.content)
    
    langfuse.score(
        trace_id=trace_id,
        name="quality",
        value=score,
    )

Langfuse vs LangSmith

维度LangfuseLangSmith
开源✅ MIT❌ 闭源
自部署✅ Docker❌ 仅 SaaS
价格免费(自部署)按量付费
Tracing
Prompt 管理
评估人工 + 自动人工 + 自动
LangChain 集成✅(原生)
非 LangChain 集成✅(SDK 通用)较弱
社区活跃增长LangChain 生态绑定
数据隐私完全自控数据在 LangChain 服务器

选型建议

  • 数据敏感、需要自部署 → Langfuse
  • 深度使用 LangChain 生态 → LangSmith
  • 需要开源可定制 → Langfuse

性能与成本监控

Token 用量看板

Langfuse 自动统计每次 LLM 调用的 token 消耗和成本:

# 在 trace 中设置 session 和 user
trace = langfuse.trace(
    name="chat",
    user_id="user_123",
    session_id="session_456",
)

# Langfuse 自动从 OpenAI 响应中提取 usage 信息
# 并按模型定价计算成本

告警配置

# 设置成本告警(在 Web UI 中配置)
# 当单日成本超过 $100 时发送通知
# 当平均延迟超过 5s 时触发告警

总结

Langfuse 填补了 LLM 应用可观测性的关键空白。作为开源方案,它提供了与商业产品 LangSmith 可比的功能,同时赋予团队数据主权。其 Tracing 能力让复杂的 RAG/Agent 链路变得可调试,Prompt 管理让 Prompt 迭代有迹可循。对于任何将 LLM 应用投入生产的团队,Langfuse 是 2026 年值得优先考虑的可观测性方案。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞,认知差异创造无限可能。