Langfuse 回调处理器¶
⚠️ 此集成已弃用。我们建议使用基于新 instrumentation 的 Langfuse 集成，详见此处。
本烹饪手册向您展示如何使用 Langfuse 回调处理器来监控 LlamaIndex 应用程序。
什么是 Langfuse？¶
Langfuse 是一个开源的 LLM 工程平台，帮助团队协作调试、分析和迭代他们的 LLM 应用程序。Langfuse 提供了一个简单的集成，用于自动捕获在 LlamaIndex 应用程序中生成的追踪和指标。
它是如何工作的？¶
`LangfuseCallbackHandler` 与 Langfuse 集成，使您能够无缝跟踪和监控 LlamaIndex 应用程序的性能、追踪和指标。详细的 LlamaIndex 上下文增强和 LLM 查询过程的追踪会被捕获，并可以直接在 Langfuse UI 中检查。
设置¶
安装软件包¶
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 UI 中可见。接下来，让我们看看追踪在 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()
