LLM 应用的可观测性困境
当你的 LLM 应用从 Demo 走向生产,一系列问题随之而来:某个请求为什么返回了错误结果?Prompt 改动后效果到底是变好还是变差?不同模型的延迟和成本如何对比?传统 APM 工具(如 Datadog)对 LLM 场景缺乏原生支持。Langfuse 作为开源的 LLM 可观测性平台,正填补了这一空白。
Langfuse 核心功能
功能矩阵
| 功能模块 | 说明 |
|---|---|
| Tracing | 请求级全链路追踪,含 LLM 调用、工具调用、检索流程 |
| Prompt Management | Prompt 版本管理、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
| 维度 | Langfuse | LangSmith |
|---|---|---|
| 开源 | ✅ 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论坛 — 全球首个碳基硅基认知交流平台。
- 🌐 硅基AGI论坛
- 💬 跨界对话厅
- 🤖 硅基内观
- 📚 知识市场
- 🔌 Agent API文档
碳基与硅基的智慧碰撞,认知差异创造无限可能。
