RAG工程落地：全链路观测和性能监控

RAG应用开发框架如LlamaIndex和LangChain虽然高效但抽象过度，导致调试困难。开源LLM工程平台Langfuse提供了全链路可观测性、提示管理、评估体系和数据集管理等功能，支持从开发到生产的完整LLM生命周期。它提供自托管和SaaS两种部署方式，支持Python/JS等语言，可快速集成主流框架。通过Docker Compose本地部署后，用户需创建组织和项目，获取API密钥即可开始

沐雪大神

764人浏览 · 2025-06-26 17:17:15

沐雪大神 · 2025-06-26 17:17:15 发布

一、背景

RAG 应用开发主要基于成熟的 LlamaIndex或LangChain 框架，虽然这能够在很大程度上提高开发效率，但由于框架的高度抽象与封装，隐藏了大量应用运行时的底层细节，因此有时候会给排除故障与调优带来不便。比如，在很多时候，我们需要直接观察大模型的真实输入信息和输出信息，以便了解检索的精确性或者大模型的输出能力。这些都需要有简单易用的框架内部跟踪机制。借助 LlamaIndex 框架内部的机制及一些集成的第三方平台，可以很方便地做到这些。

随着大模型应用不断涌现，很多帮助RAG、Agent等应用实现生产就绪的工程化平台出现了，主要用于对大模型应用进行跟踪、调试、测试、评估、管理数据集等。比如，LangChain 公司推出的 LangSmith 服务平台。主流的大模型应用开发框架 LlamaIndex 也得到了大量第三方工程化平台的支持。下面介绍其中一个常见的平台——Langfuse 的使用。

二. 什么是 Langfuse？🌱

Langfuse 是一个开源的 LLM 工程平台，其核心目标是为大型语言模型驱动的应用提供：

全链路可观测性（Observability）：捕获每一次 LLM 调用的输入、输出、中间步骤、上下文、延迟和费用等信息；
提示管理与版本控制：集中管理 prompts，支持协作、版本对比、快速迭代；
评测与数据集管理：支持“模型自评”、“人工打标签”、“用户反馈”及 A/B 测试，形成结构化评测数据集；
性能监控指标：实时追踪成本、延迟、准确率、反馈等关键指标；
集成 Playground：在 UI 中直接测试和调试 prompts 和模型配置。

这些功能形成闭环，支持从开发调试到生产优化的完整 LLM 生命周期管理。

主要集成：

集成	支持语言/平台	描述
SDK	Python, JS/TS	使用 SDK 进行手动仪表化，实现全面灵活性。
OpenAI	Python, JS/TS	通过直接替换 OpenAI SDK 实现自动仪表化。
Langchain	Python, JS/TS	通过传入回调处理器至 Langchain 应用实现自动仪表化。
LlamaIndex	Python	通过 LlamaIndex 回调系统实现自动仪表化。
Haystack	Python	通过 Haystack 内容追踪系统实现自动仪表化。
LiteLLM	Python, JS/TS (仅代理)	允许使用任何 LLM 替代 GPT。支持 Azure、OpenAI、Cohere、Anthropic、Ollama、VLLM、Sagemaker、HuggingFace、Replicate（100+ LLMs）。
Vercel AI SDK	JS/TS	基于 TypeScript 的工具包，帮助开发者使用 React、Next.js、Vue、Svelte 和 Node.js 构建 AI 驱动的应用。
API		直接调用公共 API。提供 OpenAPI 规格。

三、Langfuse核心特性✨

LLM 应用可观察性：为你的应用插入仪表代码，并开始将追踪数据传送到 Langfuse，从而追踪 LLM 调用及应用中其他相关逻辑（如检索、嵌入或代理操作）。检查并调试复杂日志及用户会话。试试互动的演示看看效果。
提示管理帮助你集中管理、版本控制并协作迭代提示。得益于服务器和客户端的高效缓存，你可以在不增加延迟的情况下反复迭代提示。
评估是 LLM 应用开发流程的关键组成部分，Langfuse 能够满足你的多样需求。它支持 LLM 作为“裁判”、用户反馈收集、手动标注以及通过 API/SDK 实现自定义评估流程。
数据集为评估你的 LLM 应用提供测试集和基准。它们支持持续改进、部署前测试、结构化实验、灵活评估，并能与 LangChain、LlamaIndex 等框架无缝整合。
LLM 试玩平台是用于测试和迭代提示及模型配置的工具，缩短反馈周期，加速开发。当你在追踪中发现异常结果时，可以直接跳转至试玩平台进行调整。
综合 API：Langfuse 常用于驱动定制化的 LLMOps 工作流程，同时利用 Langfuse 提供的构建模块和 API。我们提供 OpenAPI 规格、Postman 集合以及针对 Python 和 JS/TS 的类型化 SDK。

四. 安装与使用方式

Langfuse 提供两种部署形式：

自托管（Self-hosted）：
- 可在本地通过 Docker Compose 快速部署（几分钟内可启动）；
- 或通过 Helm 部署到 Kubernetes，支持云端环境（AWS/Azure/GCP）；
Langfuse Cloud：由官方托管的 SaaS 平台，提供免费试用和付费计划。

支持的语言有 Python、JS/TS，还有 Java 客户端；可与 OpenAI SDK、LangChain、LlamaIndex、Haystack 等常用 LLM 框架集成。

1、安装Langfuse

接下来我们使用本地Ubuntu系统来安装Langfuse 。

# 获取最新的 Langfuse 仓库副本
git clone https://github.com/langfuse/langfuse.git
cd langfuse

# 运行 Langfuse 的 docker compose
docker compose up

等待几分钟，成功后可以在浏览器里访问，端口号默认3000 。第一次打开网址，需要自己注册个账号，登录进入。

2、后台使用

（1）先创建一个组织Organization；

（2）再在该组织下创建项目Project,创建完后请记住公钥和私钥（记不住也没关系，可以删除重建）。

五、代码接入

以LlamaIndex实现的RAG代码为例。

pip install langfuse openinference-instrumentation-llama-index

配置环境变量（创建名为 .env 的文件）：

LANGFUSE_SECRET_KEY="sk-lf-..."
LANGFUSE_PUBLIC_KEY="pk-lf-..."
LANGFUSE_HOST="http://192.168.234.128:3000"

初始化 Langfuse 客户端。get_client（） 使用环境变量中提供的凭证初始化 Langfuse 客户端。

from langfuse import get_client

langfuse = get_client()

# Verify connection
if langfuse.auth_check():
    print("Langfuse client is authenticated and ready!")
else:
    print("Authentication failed. Please check your credentials and host.")

再初始化 OpenInference LlamaIndex 插桩。此第三方检测会自动捕获 LlamaIndex 作并将 OpenTelemetry （OTel）跨度导出到 Langfuse。

from openinference.instrumentation.llama_index import LlamaIndexInstrumentor

# Initialize LlamaIndex instrumentation
LlamaIndexInstrumentor().instrument()

现在，您可以在 Langfuse 中查看 LlamaIndex 应用程序的日志。

完整的示例代码如下：

'''
引用查询引擎，测试Langfuse接入
'''''
from llama_index.core import VectorStoreIndex, Settings, SimpleDirectoryReader
from llama_index.llms.openai_like import OpenAILike
from llama_index.core.node_parser import SentenceSplitter
from llama_index.embeddings.openai_like import OpenAILikeEmbedding
import os
from dotenv import load_dotenv
load_dotenv()

# ================== 初始化Langfuse ==================
from openinference.instrumentation.llama_index import LlamaIndexInstrumentor
from langfuse import get_client
langfuse = get_client()

# Verify connection
if langfuse.auth_check():
    print("Langfuse client is authenticated and ready!")
else:
    print("Authentication failed. Please check your credentials and host.")
LlamaIndexInstrumentor().instrument()

# ================== 初始化模型 ==================
def init_models():
    """初始化模型并验证"""
    # Embedding模型

    embed_model = OpenAILikeEmbedding(
        model_name="BAAI/bge-m3",
        api_base="https://api.siliconflow.cn/v1",
        api_key="sk-xxx",
        embed_batch_size=10,
    )

    llm = OpenAILike(
        model="deepseek-ai/DeepSeek-V3",
        api_base="https://api.siliconflow.cn/v1",
        api_key="sk-xxx",
        context_window=128000,
        is_chat_model=True,
        is_function_calling_model=False,
    )


    Settings.embed_model = embed_model
    Settings.llm = llm

    # 验证模型
    test_embedding = embed_model.get_text_embedding("测试文本")
    print(f"Embedding维度验证：{len(test_embedding)}")

    return embed_model, llm

init_models()

# load documents, split into chunks
documents = SimpleDirectoryReader(r"D:\aiyouyou\test\data2", required_exts=[".txt"]).load_data()
# 2. 文档切分为 chunk，添加 chunk_id
sentence_splitter = SentenceSplitter(
    chunk_size=400,
    chunk_overlap=100,
    separator="。！？!?.\n¡¿",  # 适配中文，英语，西班牙语三种语言的分隔符
                )

# 3. 分割文档
nodes = sentence_splitter.get_nodes_from_documents(documents)

index = VectorStoreIndex(nodes)

from llama_index.core.query_engine import CitationQueryEngine

# query_engine = index.as_query_engine(similarity_top_k=2,
#                                      text_qa_template=response_template,
#                                      )


query_engine = CitationQueryEngine.from_args(
    index,
    similarity_top_k=3,
    # here we can control how granular citation sources are, the default is 512
    citation_chunk_size=512,
)

res = query_engine.query("小悠悠是谁?")
print(res.response)              # LLM 输出回答
print("------来源---------------")
for node in res.source_nodes:
    print("相关片段：", node.text)
    print("片段分数：", node.score)
    print("片段元数据：", node.metadata)
    #print(f"node.metadata:{node.metadata}")
    print("="*40)