Langfuse 回调处理器¶

⚠️ 该集成方案已弃用。我们推荐使用基于 instrumentation 的新版 Langfuse 集成方案，详见此处。

本指南将展示如何使用 Langfuse 回调处理器来监控 LlamaIndex 应用程序。

什么是 Langfuse？¶

Langfuse 是一个开源的 LLM 工程平台，可帮助团队协作调试、分析和迭代他们的 LLM 应用程序。Langfuse 提供简单集成方案，可自动捕获 LlamaIndex 应用程序生成的追踪数据和指标。

工作原理¶

LangfuseCallbackHandler 与 Langfuse 深度集成，使您能够无缝追踪和监控 LlamaIndex 应用程序的性能、调用链和指标。该处理器会捕获 LlamaIndex 上下文增强和 LLM 查询过程的详细调用链，您可以直接在 Langfuse 用户界面中查看这些数据。

安装¶

安装软件包¶

In [ ]:

%pip install llama-index llama-index-callbacks-langfuse

%pip install llama-index llama-index-callbacks-langfuse

配置环境¶

若尚未完成，请先在 Langfuse 上注册并从项目设置中获取您的 API 密钥。

In [ ]:

import os

# Get keys for your project from the project settings page https://cloud.langfuse.com
os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-..."
os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-..."
os.environ["LANGFUSE_HOST"] = "https://cloud.langfuse.com"  # 🇪🇺 EU region
# os.environ["LANGFUSE_HOST"] = "https://us.cloud.langfuse.com" # 🇺🇸 US region

# OpenAI
os.environ["OPENAI_API_KEY"] = "sk-..."

import os # Get keys for your project from the project settings page https://cloud.langfuse.com os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-..." os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-..." os.environ["LANGFUSE_HOST"] = "https://cloud.langfuse.com" # 🇪🇺 EU region # os.environ["LANGFUSE_HOST"] = "https://us.cloud.langfuse.com" # 🇺🇸 US region # OpenAI os.environ["OPENAI_API_KEY"] = "sk-..."

注册 Langfuse 回调处理器¶

选项1：设置全局LlamaIndex处理器¶

In [ ]:

from llama_index.core import global_handler, set_global_handler

set_global_handler("langfuse")
langfuse_callback_handler = global_handler

from llama_index.core import global_handler, set_global_handler set_global_handler("langfuse") langfuse_callback_handler = global_handler

选项 2：直接使用 Langfuse 回调¶

In [ ]:

from llama_index.core import Settings
from llama_index.core.callbacks import CallbackManager
from langfuse.llama_index import LlamaIndexCallbackHandler

langfuse_callback_handler = LlamaIndexCallbackHandler()
Settings.callback_manager = CallbackManager([langfuse_callback_handler])

from llama_index.core import Settings from llama_index.core.callbacks import CallbackManager from langfuse.llama_index import LlamaIndexCallbackHandler langfuse_callback_handler = LlamaIndexCallbackHandler() Settings.callback_manager = CallbackManager([langfuse_callback_handler])

将事件刷新至 Langfuse¶

Langfuse SDK 会在后台对事件进行队列处理和批量发送，以减少网络请求次数并提升整体性能。在退出应用程序之前，请确保所有排队事件都已刷新至 Langfuse 服务器。

In [ ]:

# ... your LlamaIndex calls here ...

langfuse_callback_handler.flush()

# ... your LlamaIndex calls here ... langfuse_callback_handler.flush()

完成！✨ 您的 LlamaIndex 应用生成的追踪数据和指标现已自动记录至 Langfuse。当您构建新索引或结合上下文文档查询 LLM 时，所有追踪数据和指标会实时显示在 Langfuse 界面中。接下来，我们将查看这些追踪数据在 Langfuse 中的呈现形式。

示例¶

获取并保存示例数据。

In [ ]:

!mkdir -p 'data/'
!wget 'https://raw.githubusercontent.com/run-llama/llama_index/main/docs/docs/examples/data/paul_graham/paul_graham_essay.txt' -O 'data/paul_graham_essay.txt'

!mkdir -p 'data/' !wget 'https://raw.githubusercontent.com/run-llama/llama_index/main/docs/docs/examples/data/paul_graham/paul_graham_essay.txt' -O 'data/paul_graham_essay.txt'

运行示例索引构建、查询和聊天功能。

In [ ]:

from llama_index.core import SimpleDirectoryReader, VectorStoreIndex

# Create index
documents = SimpleDirectoryReader("data").load_data()
index = VectorStoreIndex.from_documents(documents)

# Execute query
query_engine = index.as_query_engine()
query_response = query_engine.query("What did the author do growing up?")
print(query_response)

# Execute chat query
chat_engine = index.as_chat_engine()
chat_response = chat_engine.chat("What did the author do growing up?")
print(chat_response)

# As we want to immediately see result in Langfuse, we need to flush the callback handler
langfuse_callback_handler.flush()

from llama_index.core import SimpleDirectoryReader, VectorStoreIndex # Create index documents = SimpleDirectoryReader("data").load_data() index = VectorStoreIndex.from_documents(documents) # Execute query query_engine = index.as_query_engine() query_response = query_engine.query("What did the author do growing up?") print(query_response) # Execute chat query chat_engine = index.as_chat_engine() chat_response = chat_engine.chat("What did the author do growing up?") print(chat_response) # As we want to immediately see result in Langfuse, we need to flush the callback handler langfuse_callback_handler.flush()

完成！✨ 您现在可以在 Langfuse 项目中查看索引和查询的跟踪记录了。

示例跟踪记录（公开链接）：

1. 查询
2. 查询（聊天）
3. 会话

📚 更多详情¶

查看完整的 Langfuse 文档，了解更多关于 Langfuse 的追踪分析功能以及如何充分利用此集成的详细信息。

反馈¶

如有任何反馈或需求，请在 GitHub 上提交 Issue 或通过 Discord 与社区分享您的想法。