跳转至

Langfuse LLM 可观测性

为什么要学

当你把 LLM 应用部署到生产环境，你需要回答这些问题：

  • 每次调用花了多少钱？哪些 prompt 最费 token？
  • 用户满意度如何？哪些对话出了问题？
  • 延迟分布如何？瓶颈在哪？
  • RAG 检索质量怎么样？Agent 执行效率如何？

Langfuse 是开源的 LLM 可观测性平台，提供：

  • Trace 追踪：记录每次 LLM 调用的完整链路
  • Cost 计费：自动计算 token 用量和费用
  • Eval 评估：人工/自动化质量评分
  • Prompt 管理：版本化管理和 A/B 测试
  • 自托管：数据完全在你的服务器

核心概念

白话解释

把 Langfuse 想象成 LLM 应用的"体检中心"： - Trace（追踪）：一次完整的用户请求，从输入到最终输出 - Span（跨度）：Trace 中的一个步骤（如检索文档、调用LLM） - Generation（生成）：一次具体的 LLM 调用（含 prompt、completion、token 数） - Score（评分）：对输出质量的评价（人工或自动）

核心概念对照表

概念说明类比
Trace一次完整的请求追踪一次用户旅程
SpanTrace中的子步骤函数调用层级
Generation一次LLM调用API请求详情
Score质量评分(0-1或自定义)用户评价/测试打分
Session同一用户的多次Trace用户会话
Prompt版本化的提示模板配置文件版本
Dataset评估用的输入/期望输出对测试用例集
ObservationSpan或Generation的统称日志条目

安装配置

方式一：使用 Langfuse Cloud（快速开始）

  1. 注册 Langfuse Cloud
  2. 创建项目
  3. 获取 API Keys（Public + Secret）

方式二：自托管 Docker

# 克隆仓库
git clone https://github.com/langfuse/langfuse.git
cd langfuse

# 使用docker-compose启动
docker compose up -d

# 访问 http://localhost:3000
# 默认创建账号登录

# docker-compose.yml 核心配置
services:
  langfuse:
    image: langfuse/langfuse:latest
    ports:
      - "3000:3000"
    environment:
      - DATABASE_URL=postgresql://postgres:postgres@db:5432/langfuse
      - NEXTAUTH_SECRET=your-secret-key
      - NEXTAUTH_URL=http://localhost:3000
      - SALT=your-salt
  db:
    image: postgres:15
    environment:
      - POSTGRES_PASSWORD=postgres
      - POSTGRES_DB=langfuse
    volumes:
      - pgdata:/var/lib/postgresql/data

Python SDK 安装

pip install langfuse
# 或
pip install langfuse[openai]  # OpenAI集成
pip install langfuse[langchain]  # LangChain集成

环境变量配置

export LANGFUSE_PUBLIC_KEY="pk-..."
export LANGFUSE_SECRET_KEY="sk-..."
export LANGFUSE_HOST="https://cloud.langfuse.com"  # 或自托管地址

快速上手

基础追踪：OpenAI 集成

from langfuse.openai import openai  # 替换标准openai导入

# 完全兼容OpenAI SDK，自动追踪
response = openai.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {"role": "system", "content": "你是一个有帮助的助手"},
        {"role": "user", "content": "什么是向量数据库?"}
    ],
    metadata={"user_id": "user-123"},  # 额外元数据
)

print(response.choices[0].message.content)
# Langfuse自动记录: model, tokens, latency, cost

手动追踪：装饰器方式

from langfuse.decorators import observe, langfuse_context
from openai import OpenAI

client = OpenAI()

@observe()  # 自动创建Trace
def answer_question(question: str) -> str:
    # 自动创建Span
    context = retrieve_documents(question)
    answer = generate_answer(question, context)
    return answer

@observe()  # 自动创建子Span
def retrieve_documents(query: str) -> list[str]:
    # 模拟检索
    return ["文档1内容...", "文档2内容..."]

@observe(as_type="generation")  # 标记为Generation
def generate_answer(question: str, context: list[str]) -> str:
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": f"基于以下上下文回答：{context}"},
            {"role": "user", "content": question}
        ]
    )
    return response.choices[0].message.content

# 使用
result = answer_question("什么是RAG?")

添加评分

from langfuse import Langfuse

langfuse = Langfuse()

# 对特定trace打分
langfuse.score(
    trace_id="trace-id-here",
    name="helpfulness",
    value=0.9,
    comment="回答准确且完整"
)

# 在装饰器中打分
@observe()
def rated_answer(question: str) -> str:
    answer = generate_answer(question, [])

    # 自动关联到当前trace
    langfuse_context.score_current_trace(
        name="relevance",
        value=0.8
    )
    return answer

进阶用法

1. LangChain 集成

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langfuse.callback import CallbackHandler

# 创建Langfuse回调
langfuse_handler = CallbackHandler(
    public_key="pk-...",
    secret_key="sk-...",
    host="https://cloud.langfuse.com"
)

# LangChain正常使用，添加callback
llm = ChatOpenAI(model="gpt-4o-mini")
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是{role}"),
    ("user", "{question}")
])

chain = prompt | llm

# 自动追踪整条链路
response = chain.invoke(
    {"role": "数据专家", "question": "解释MapReduce"},
    config={"callbacks": [langfuse_handler]}
)

2. Prompt 版本管理

from langfuse import Langfuse

langfuse = Langfuse()

# 获取线上prompt（支持版本/标签）
prompt = langfuse.get_prompt("qa-prompt", version=2)

# 使用prompt
compiled = prompt.compile(
    context="相关文档内容...",
    question="什么是Langfuse?"
)

# 在Web UI中管理:
# - 创建/编辑prompt模板
# - 标记production/staging版本
# - 查看各版本的性能指标

3. 自动化评估(Evals)

from langfuse import Langfuse

langfuse = Langfuse()

# 创建评估数据集
dataset = langfuse.create_dataset(name="qa-eval-v1")

# 添加测试用例
langfuse.create_dataset_item(
    dataset_name="qa-eval-v1",
    input={"question": "什么是向量数据库?"},
    expected_output="向量数据库是专门存储和查询高维向量的数据库..."
)

# 运行评估
@observe()
def run_qa(question):
    return generate_answer(question, retrieve_documents(question))

for item in dataset.items:
    with item.observe(run_name="qa-v2") as trace_id:
        output = run_qa(item.input["question"])

        # 自动评分（可用LLM-as-judge）
        item.link(trace_id, run_name="qa-v2")

4. 成本追踪

# Langfuse自动计算成本(基于model pricing)
# 在Dashboard中查看:
# - 按模型的成本分布
# - 按用户的成本
# - 按功能的成本
# - 日/周/月成本趋势

# 自定义成本
@observe()
def expensive_operation():
    langfuse_context.update_current_observation(
        usage={
            "input": 1500,   # input tokens
            "output": 500,   # output tokens
            "total": 2000,   # total tokens
            "unit": "TOKENS"
        },
        model="gpt-4o"
    )

5. 用户反馈收集

# 前端收集用户反馈后发送到Langfuse
from langfuse import Langfuse

langfuse = Langfuse()

# 用户点击👍
langfuse.score(
    trace_id="trace-xxx",
    name="user-feedback",
    value=1,
    data_type="BOOLEAN"
)

# 用户点击👎并留言
langfuse.score(
    trace_id="trace-xxx",
    name="user-feedback",
    value=0,
    data_type="BOOLEAN",
    comment="回答不准确，忽略了关键信息"
)

6. Session 追踪

@observe()
def chat(session_id: str, user_message: str):
    langfuse_context.update_current_trace(
        session_id=session_id,
        user_id="user-456"
    )
    # ... 生成回复
    return response

# 同一session_id的多次调用会被关联
chat("session-001", "你好")
chat("session-001", "继续上面的话题")

常见问题

Q1: 追踪数据延迟很高？

# Langfuse SDK默认异步发送，不阻塞主线程
# 如果需要确保发送完成（如脚本结束前）：
from langfuse import Langfuse
langfuse = Langfuse()
# ... 操作 ...
langfuse.flush()  # 强制发送所有pending数据

Q2: 如何过滤/搜索 Traces？

在 Web UI 中可以按以下维度过滤： - 时间范围 - User ID - Session ID - 标签(Tags) - 元数据(Metadata) - 评分范围

Q3: 自托管资源需求？

规模PostgreSQL应用服务器
开发1GB RAM1GB RAM
小团队4GB RAM2GB RAM
生产8GB+ RAM4GB+ RAM

Q4: 与其他可观测工具对比？

工具特点开源自托管
Langfuse全面/生态好✓✓
LangSmithLangChain官方✗✗
Phoenix侧重ML评估✓✓
Helicone简单代理模式✓✓

Q5: 数据量大时性能问题？

  • 使用采样：不需要追踪 100% 的请求
  • 设置数据保留策略
  • PostgreSQL 定期 VACUUM
  • 考虑使用 ClickHouse 替代（大规模场景）

参考资源