Langfuse：LLM 应用的 observability（trace + eval）

起因

LLM 应用上生产后：

用户报"AI 回答错" → 哪 prompt？哪 model call？怎么调试？
prompt 改一版 → 怎么评估质量没退步？
月 OpenAI bill $5000 → 哪些 endpoint 烧钱？

传统 web app observability 工具不适合：

输入 / 输出是大段文本
嵌套 LLM call（agent / tool use）
需要 evaluation（不只是 metric）

Langfuse：LLM 应用专用 observability + prompt management + eval。
open source + self-host 友好。

装

# self-host
docker compose up -d -f langfuse-docker-compose.yml
# UI on localhost:3000

或 SaaS：cloud.langfuse.com 免费 tier。

SDK 集成

from langfuse import Langfuse
from langfuse.decorators import observe

lf = Langfuse(public_key='pk-...', secret_key='sk-...')

@observe()
def answer_question(q: str) -> str:
    # 自动 trace 这函数 + 嵌套 LLM call
    context = retrieve(q)
    prompt = f"Context: {context}\n\nQ: {q}"
    response = openai.chat.completions.create(
        model='gpt-4',
        messages=[{'role': 'user', 'content': prompt}],
    )
    return response.choices[0].message.content

@observe() 装饰器自动 track 函数调用 + 输入 / 输出 + 子调用 trace。

OpenAI call 用 Langfuse 包装的 client → 自动捕获 prompt / response /
token / cost。

langchain / llamaindex 集成

from langfuse.callback import CallbackHandler
handler = CallbackHandler()

chain.invoke({'q': q}, config={'callbacks': [handler]})

LangChain 整 chain / agent 自动 trace（每步骤 / 每 tool call / 每 LLM call）。

UI

trace view：

trace: answer_question("How to deploy?")
├─ retrieve (5 docs found)
├─ format_prompt
├─ openai.chat.completions
│   model: gpt-4
│   tokens: 1234 in / 234 out
│   cost: $0.024
│   latency: 1.2s
└─ output: "To deploy, run..."

每 LLM call 看到：
- input / output 全文
- model / params
- token + cost
- latency
- error

嵌套 trace 直观看 agent 步骤。

sessions + users

@observe()
def chat(user_id: str, session_id: str, message: str):
    ...

lf.update_current_trace(user_id=user_id, session_id=session_id)

UI 按 user / session 聚合 → 看某用户对话历史 + cost。

prompt management

prompt 不写代码里，存 Langfuse：

prompt = lf.get_prompt('answer-question', version='production')
formatted = prompt.compile(context=ctx, question=q)

UI 改 prompt → 自动 versioned → 部署不用改 code。

evaluation

每 trace 加 score：

lf.score(
    trace_id=trace.id,
    name='accuracy',
    value=0.9,
)

或者 user feedback：

@app.post('/feedback')
async def feedback(trace_id: str, thumbs_up: bool):
    lf.score(trace_id=trace_id, name='user_feedback', value=1 if thumbs_up else 0)

UI 聚合：哪些 prompt 评分低？哪些 model 效果好？

LLM-as-judge eval：

def llm_judge(trace):
    judgment = openai.chat.completions.create(
        model='gpt-4',
        messages=[{
            'role': 'user',
            'content': f'Rate this answer 1-5: Q: {trace.input}, A: {trace.output}',
        }],
    )
    lf.score(trace_id=trace.id, name='llm_judge', value=parse(judgment))

LLM 评 LLM → 自动 quality monitoring。

dataset + experiment

# 上传 test dataset
lf.create_dataset(name='qa-test-set', items=[
    {'input': q1, 'expected': a1},
    {'input': q2, 'expected': a2},
])

# 跑实验
for item in lf.get_dataset('qa-test-set'):
    actual = answer_question(item.input)
    lf.create_dataset_item_score(item, 'exact_match', actual == item.expected)

新 prompt / model → 跑 dataset → 对比 score。
回归测试 for LLM。

cost tracking

dashboard 看：

每 model 总 cost
每 user 总 cost
每 endpoint trace count + cost
趋势

Last 7 days:
  gpt-4: $234 (1234 calls)
  gpt-4o: $89 (5678 calls)
  embedding: $12

Top user: user-42 ($45 last week)

提前发现"某 endpoint 烧钱" → 优化（用 cheaper model / cache）。

与替代品

	Langfuse	LangSmith	Helicone	Weights & Biases
开源	✅	❌（SaaS）	✅	❌
Self-host	✅	❌	✅	enterprise
trace	✅	✅	✅	✅
prompt mgmt	✅	✅	弱	❌
eval	✅	✅	弱	✅
price	free / 用量	付费	free / 用量	付费

我用 Langfuse self-host（数据敏感 + 开源 + 功能全）。

真实 case

我们一个客户 RAG 应用上线：

1000+ QPS LLM call
接入 Langfuse 一周
发现：
某 endpoint 答错率 30%（prompt 有 bug，UI 看 trace 立刻发现）
某 user 每天 $50 cost（abuse 检测）
retrieval 召回率低 → 改 chunking 策略

如果没 trace，全靠 user complain → 慢 + 漏 80%。

privacy

LLM 内容可能含 PII。
Langfuse 配 PII redaction：

lf.flush()      # 含 mask sensitive

或者在客户端 mask 后再发：

def mask(text):
    return re.sub(r'\b\d{16}\b', '[CARD]', text)

@observe()
def chat(msg):
    msg_masked = mask(msg)
    ...

部署

self-host 简单 docker compose（langfuse + postgres + clickhouse）。
clickhouse 存 trace（大量 string 数据 → columnar 高效）。

与 OTEL 集成

Langfuse 1.0+ 支持 OTLP receiver：

# OpenTelemetry trace 自动转 Langfuse

跟现有 observability stack 整合。

踩过的坑

flush 异步：默认异步 send，应用退出前要 lf.flush() 否则
trace 丢。
trace input/output 大：长文本占 DB。配 truncate。
cost 计算不准：Langfuse 内置 cost map 可能滞后于 OpenAI 价格
调整。自己 update 或者验证。
self-host clickhouse 重：单 server 几 GB RAM 起。小项目用
SaaS 简单。
prompt version 滥：每改一字一版本 → UI 难看。考虑 staging /
production tag。