15.2 Langfuse：开源 LLM 可观测性平台

一、Langfuse 简介

1.1 什么是 Langfuse？

Langfuse 是一个开源的 LLM 工程平台，专为开发团队提供 LLM 应用的可观测性、调试、评估和 Prompt 管理能力。由 Y Combinator W23 孵化，目前在 GitHub 上拥有超过 19,000 颗星。

核心理念："AI 本质上是非确定性的，没有可观测性工具调试你的应用更像是猜谜游戏。"

与 LangChain/LangGraph 的兼容性：虽然 Langfuse 是框架无关的，但它完全支持 LangChain 和 LangGraph。通过官方提供的 CallbackHandler，你可以无缝追踪 LangChain 链和 LangGraph 图的执行过程。这意味着：

你可以享受 LangChain/LangGraph 的开发体验
同时获得 Langfuse 开源、可自托管、成本更低的优势
不必在"框架生态"和"可观测性平台"之间做二选一

1.2 与 LangSmith 的定位对比

特性	LangSmith	Langfuse
开源协议	闭源（需企业许可）	MIT 开源（ee 文件夹除外）
自托管	需付费企业许可	免费自托管
框架依赖	深度绑定 LangChain/LangGraph	框架无关，支持任意 LLM 应用
追踪模型	Run Tree（LCEL 原生）	Span/Generation（OpenTelemetry 兼容）
目标用户	LangChain 生态用户	通用 LLM 开发者

选择建议：

如果你深度使用 LangChain 生态 → 选择 LangSmith
如果你需要框架无关、成本敏感、数据主权 → 选择 Langfuse

二、核心功能架构

Langfuse 提供三大核心模块：

Langfuse 功能概览

text

┌─────────────────────────────────────────────────────────────┐
│                    Langfuse 平台架构                         │
├─────────────────┬─────────────────┬─────────────────────────┤
│   可观测性       │   Prompt 管理    │      评估系统           │
│  (Tracing)      │  (Prompts)      │   (Evaluation)         │
├─────────────────┼─────────────────┼─────────────────────────┤
│ • Trace 追踪    │ • 版本控制       │ • LLM-as-Judge        │
│ • Span 管理     │ • 标签管理       │ • 人工标注             │
│ • Generation    │ • Playground    │ • 用户反馈             │
│ • 成本计算      │ • 协作编辑       │ • 自定义评分           │
└─────────────────┴─────────────────┴─────────────────────────┘

三、快速开始

3.1 安装与配置

bash

pip install langfuse openai

环境变量配置（.env 文件）：

bash

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

3.2 最简单的集成方式：OpenAI Drop-in 替换

Langfuse 提供了 OpenAI SDK 的直接替换，只需更改一行 import：

python

# 原来的写法
# from openai import OpenAI

# 使用 Langfuse 的写法（自动追踪所有调用）
from langfuse.openai import openai

client = openai.OpenAI()

response = client.chat.completions.create(
    model="gpt-4",
    messages=[
        {"role": "system", "content": "你是一个有帮助的助手"},
        {"role": "user", "content": "你好！"}
    ],
    # Langfuse 特有参数
    name="greeting-chat",  # 追踪名称
    metadata={"user_id": "user_123"}  # 自定义元数据
)

print(response.choices[0].message.content)

这种方式自动捕获：

所有 Prompt 和 Completion
响应延迟
Token 使用量和成本
API 错误

四、Tracing（可观测性）详解

Langfuse Trace 示例

4.1 核心概念

概念	说明	类比
Trace	一次完整的请求处理流程	一个完整的用户对话
Span	Trace 中的一个步骤（非 LLM 调用）	数据处理、检索等
Generation	一次 LLM 调用	GPT-4 API 调用

4.2 使用 @observe 装饰器

装饰器是最优雅的追踪方式：

python

from langfuse import observe, get_client
from openai import OpenAI

client = OpenAI()

@observe()
def retrieve_context(query: str) -> str:
    """模拟检索相关文档"""
    # 这个函数会被记录为一个 Span
    return f"关于 '{query}' 的相关文档内容..."

@observe(as_type="generation")
def call_llm(prompt: str, context: str) -> str:
    """调用 LLM 生成回答"""
    # 标记为 generation 类型
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": f"基于以下上下文回答问题：\n{context}"},
            {"role": "user", "content": prompt}
        ]
    )
    return response.choices[0].message.content

@observe()
def rag_pipeline(question: str) -> str:
    """RAG 流水线主函数"""
    # 自动创建父级 Trace
    context = retrieve_context(question)  # 子 Span
    answer = call_llm(question, context)   # 子 Generation
    return answer

# 执行
result = rag_pipeline("什么是 Langfuse？")
print(result)

# 重要：短生命周期应用需要刷新
get_client().flush()

4.3 装饰器高级参数

python

@observe(
    name="custom-name",           # 自定义名称
    as_type="generation",         # 类型：span 或 generation
    capture_input=True,           # 是否捕获输入
    capture_output=True,          # 是否捕获输出
)
def my_function():
    pass

# 对于大数据量，可禁用 IO 捕获提升性能
@observe(capture_input=False, capture_output=False)
def process_large_data(huge_dataset):
    pass

五、与 LangChain 集成

Langfuse 集成

5.1 回调处理器配置

python

from langfuse import get_client
from langfuse.langchain import CallbackHandler
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

# 初始化 Langfuse 回调
langfuse = get_client()
langfuse_handler = CallbackHandler()

# 创建 LangChain 组件
llm = ChatOpenAI(model="gpt-4")
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个专业的{role}"),
    ("human", "{question}")
])

chain = prompt | llm

# 调用时传入回调
response = chain.invoke(
    {"role": "程序员", "question": "解释什么是递归"},
    config={
        "callbacks": [langfuse_handler],
        "metadata": {
            "langfuse_user_id": "user-123",
            "langfuse_session_id": "session-456",
            "langfuse_tags": ["langchain", "demo"]
        }
    }
)

print(response.content)

5.2 与 LangGraph 集成

python

from langgraph.graph import StateGraph, START, END
from langfuse.langchain import CallbackHandler

langfuse_handler = CallbackHandler()

# 定义你的 LangGraph 图...
graph = StateGraph(...)
app = graph.compile()

# 执行时传入回调
result = app.invoke(
    {"messages": [{"role": "user", "content": "你好"}]},
    config={"callbacks": [langfuse_handler]}
)

六、Prompt 管理

6.1 创建 Prompt

在 Langfuse UI 中创建，或通过 SDK：

python

from langfuse import get_client

langfuse = get_client()

# 创建文本类型 Prompt
langfuse.create_prompt(
    name="movie-critic",
    prompt="作为一个{{expertise_level}}的影评人，请评价电影《{{movie_name}}》",
    labels=["production"],  # 标签：production, staging, latest
    config={
        "model": "gpt-4",
        "temperature": 0.7
    }
)

6.2 使用 Prompt

python

from langfuse import get_client
from openai import OpenAI

langfuse = get_client()
client = OpenAI()

# 获取生产环境的 Prompt
prompt = langfuse.get_prompt("movie-critic", label="production")

# 编译 Prompt（替换变量）
compiled_prompt = prompt.compile(
    expertise_level="专业",
    movie_name="盗梦空间"
)

# 使用编译后的 Prompt
response = client.chat.completions.create(
    model=prompt.config.get("model", "gpt-4"),
    temperature=prompt.config.get("temperature", 0.7),
    messages=[{"role": "user", "content": compiled_prompt}]
)

print(response.choices[0].message.content)

6.3 版本管理

python

# 获取特定版本
prompt_v1 = langfuse.get_prompt("movie-critic", version=1)

# 获取特定标签
prompt_prod = langfuse.get_prompt("movie-critic", label="production")
prompt_staging = langfuse.get_prompt("movie-critic", label="staging")

# 获取最新版本
prompt_latest = langfuse.get_prompt("movie-critic", label="latest")

七、评估系统

7.1 评分类型

Langfuse 支持灵活的评分机制：

评分方式	适用场景	自动化程度
LLM-as-Judge	自动化质量评估	全自动
人工标注	高精度评估	手动
用户反馈	生产环境监控	半自动
自定义评分	特定业务指标	可配置

7.2 通过 SDK 添加评分

python

from langfuse import get_client

langfuse = get_client()

# 为特定 Trace 添加评分
langfuse.score(
    trace_id="your-trace-id",
    name="relevance",
    value=0.9,
    comment="回答与问题高度相关"
)

# 添加分类评分
langfuse.score(
    trace_id="your-trace-id",
    name="quality",
    value="good",  # 支持字符串值
    data_type="CATEGORICAL"
)

7.3 用户反馈收集

python

# 在 @observe 装饰的函数中添加用户反馈
from langfuse import observe, get_client

@observe()
def chat_with_feedback(user_message: str) -> str:
    response = "..."  # LLM 响应
    return response

# 收集用户反馈
langfuse = get_client()
langfuse.score(
    trace_id="current-trace-id",
    name="user-feedback",
    value=1,  # 用户点赞
    comment="用户表示满意"
)

八、自托管部署

Langfuse 部署选项

8.1 Docker Compose 快速部署

yaml

# docker-compose.yml
version: "3.9"
services:
  langfuse:
    image: langfuse/langfuse:latest
    ports:
      - "3000:3000"
    environment:
      - DATABASE_URL=postgresql://user:pass@postgres:5432/langfuse
      - NEXTAUTH_SECRET=your-secret-key
      - SALT=your-salt
      - NEXTAUTH_URL=http://localhost:3000
    depends_on:
      - postgres
      - clickhouse
      - redis

  postgres:
    image: postgres:15
    environment:
      - POSTGRES_USER=user
      - POSTGRES_PASSWORD=pass
      - POSTGRES_DB=langfuse

  clickhouse:
    image: clickhouse/clickhouse-server:latest

  redis:
    image: redis:7

启动：

bash

docker-compose up -d

8.2 生产环境架构

text

┌─────────────────────────────────────────────────────────────────┐
│                     生产环境架构                                 │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  ┌─────────┐    ┌─────────┐    ┌─────────────────────────────┐  │
│  │  Web    │    │ Worker  │    │       数据存储层             │  │
│  │ Server  │    │ Service │    │  ┌─────────┐ ┌───────────┐  │  │
│  └────┬────┘    └────┬────┘    │  │Postgres │ │ClickHouse │  │  │
│       │              │         │  │(事务DB) │ │ (分析DB)  │  │  │
│       └──────┬───────┘         │  └─────────┘ └───────────┘  │  │
│              │                 │  ┌─────────┐ ┌───────────┐  │  │
│              ▼                 │  │  Redis  │ │ S3/Blob   │  │  │
│        ┌─────────┐             │  │ (缓存)  │ │ (存储)    │  │  │
│        │负载均衡  │             │  └─────────┘ └───────────┘  │  │
│        └─────────┘             └─────────────────────────────┘  │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

九、定价对比

Langfuse 定价

方案	价格	主要特性
Hobby（免费）	$0	50k units/月，30天数据保留，2用户
Core	$29/月	100k units/月，90天保留，无限用户
Pro	$199/月	无限数据保留，SOC2/ISO27001 合规
Enterprise	$2,499/月	自定义用量，专属支持，审计日志
自托管	免费	完全控制，需自行维护

与 LangSmith 定价对比

特性	Langfuse	LangSmith
免费额度	50k traces/月	5k traces/月
自托管	免费开源	需企业许可
入门付费	$29/月	$39/月
数据主权	支持自托管	仅限企业版

十、最佳实践

10.1 开发流程集成

python

import os
from langfuse import observe, get_client

# 开发环境配置
os.environ["LANGFUSE_HOST"] = "http://localhost:3000"  # 本地实例

# 生产环境配置
# os.environ["LANGFUSE_HOST"] = "https://cloud.langfuse.com"

@observe()
def my_llm_app(user_input: str) -> str:
    # 你的 LLM 应用逻辑
    pass

# 确保数据发送
get_client().flush()

10.2 性能优化建议

批量操作：Langfuse SDK 自动批量发送数据
禁用 IO 捕获：对大数据量使用 capture_input=False
异步刷新：使用 flush() 确保数据发送

10.3 与 CI/CD 集成

python

# 在测试中使用 Langfuse 评估
import pytest
from langfuse import get_client

def test_llm_quality():
    langfuse = get_client()

    # 运行 LLM 并获取 trace
    result = my_llm_app("测试输入")

    # 添加自动化评分
    langfuse.score(
        trace_id="...",
        name="test-quality",
        value=1 if "预期关键词" in result else 0
    )

    langfuse.flush()

十一、总结

Langfuse 的核心优势

开源免费：MIT 协议，可自由使用和修改
框架无关：支持任意 LLM 框架和模型
自托管选项：数据完全自主可控
成本效益：相比 LangSmith 更实惠
OpenTelemetry 兼容：标准化的可观测性

适用场景

场景	推荐方案
小团队/个人项目	Langfuse Hobby（免费）
数据敏感型企业	Langfuse 自托管
LangChain 深度用户	考虑 LangSmith
多框架混合使用	Langfuse
需要企业合规	Langfuse Pro/Enterprise

15.2 Langfuse：开源 LLM 可观测性平台 ​

一、Langfuse 简介 ​

1.1 什么是 Langfuse？ ​

1.2 与 LangSmith 的定位对比 ​

二、核心功能架构 ​

三、快速开始 ​

3.1 安装与配置 ​

3.2 最简单的集成方式：OpenAI Drop-in 替换 ​

四、Tracing（可观测性）详解 ​

4.1 核心概念 ​

4.2 使用 @observe 装饰器 ​

4.3 装饰器高级参数 ​

五、与 LangChain 集成 ​

5.1 回调处理器配置 ​

5.2 与 LangGraph 集成 ​

六、Prompt 管理 ​

6.1 创建 Prompt ​

6.2 使用 Prompt ​

6.3 版本管理 ​

七、评估系统 ​

7.1 评分类型 ​

7.2 通过 SDK 添加评分 ​

7.3 用户反馈收集 ​

八、自托管部署 ​

8.1 Docker Compose 快速部署 ​

8.2 生产环境架构 ​

九、定价对比 ​

Langfuse 定价 ​

与 LangSmith 定价对比 ​

十、最佳实践 ​

10.1 开发流程集成 ​

10.2 性能优化建议 ​

10.3 与 CI/CD 集成 ​

十一、总结 ​

Langfuse 的核心优势 ​

适用场景 ​

参考资源 ​

15.2 Langfuse：开源 LLM 可观测性平台

一、Langfuse 简介

1.1 什么是 Langfuse？

1.2 与 LangSmith 的定位对比

二、核心功能架构

三、快速开始

3.1 安装与配置

3.2 最简单的集成方式：OpenAI Drop-in 替换

四、Tracing（可观测性）详解

4.1 核心概念

4.2 使用 @observe 装饰器

4.3 装饰器高级参数

五、与 LangChain 集成

5.1 回调处理器配置

5.2 与 LangGraph 集成

六、Prompt 管理

6.1 创建 Prompt

6.2 使用 Prompt

6.3 版本管理

七、评估系统

7.1 评分类型

7.2 通过 SDK 添加评分

7.3 用户反馈收集

八、自托管部署

8.1 Docker Compose 快速部署

8.2 生产环境架构

九、定价对比

Langfuse 定价

与 LangSmith 定价对比

十、最佳实践

10.1 开发流程集成

10.2 性能优化建议

10.3 与 CI/CD 集成

十一、总结

Langfuse 的核心优势

适用场景

参考资源