LlamaIndex Workflows 的可观测性

本 cookbook 演示如何使用 Litefuse 为你的 LlamaIndex Workflows 提供实时可观测性。

什么是 LlamaIndex Workflows？ LlamaIndex Workflows 是一个灵活的事件驱动框架，旨在构建强大的 AI Agent。在 LlamaIndex 中，工作流由多个步骤串联而成 —— 每个步骤都通过 @step 装饰器定义并校验。每个步骤处理特定的事件类型，让你能够编排复杂的流程，例如 AI Agent 协作、RAG 流程、数据抽取等。

什么是 Litefuse？ Litefuse 是开源的 AI Agent 可观测性与评估平台。它帮助团队协作管理 prompt、trace 应用、调试问题，并在生产环境中评估 LLM 系统。

快速开始

我们将通过一个简单示例演示如何使用 LlamaIndex Workflows 并将其与 Litefuse 集成。

第 1 步：安装依赖

%pip install langfuse openai llama-index-workflows llama-index-core llama-index-llms-openai openinference-instrumentation-llama_index llama-index-instrumentation

第 2 步：设置环境变量

配置你的 Litefuse API Key。可通过注册 Litefuse Cloud 或自托管 Litefuse 获取。

import os
 
# Get keys for your project from the project settings page: https://litefuse.cloud
os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-..." 
os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-..." 
os.environ["LANGFUSE_BASE_URL"] = "https://litefuse.cloud"
 
# Your openai key
os.environ["OPENAI_API_KEY"] = "sk-proj-..."

设置好环境变量后，我们就可以初始化 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.")

第 3 步：初始化 LlamaIndex Instrumentation

现在我们初始化 OpenInference LlamaIndex instrumentation。这个第三方 instrumentation 会自动捕获 LlamaIndex 的操作并将 OpenTelemetry (OTel) span 导出到 Litefuse。

from openinference.instrumentation.llama_index import LlamaIndexInstrumentor
 
# Initialize LlamaIndex instrumentation
LlamaIndexInstrumentor().instrument()

第 4 步：创建一个简单的 LlamaIndex Workflows 应用

在 LlamaIndex Workflows 中，你通过用 @step 装饰器定义步骤来构建事件驱动的 AI Agent。每个步骤处理一个事件，并在合适时发出新事件。在本示例中，我们创建一个简单的工作流，包含两个步骤：一个对传入事件进行预处理，另一个生成回复。

from llama_index.core.llms import ChatMessage
from llama_index.llms.openai import OpenAI
from typing import Annotated
 
from workflows import Workflow, step
from workflows.events import StartEvent, StopEvent
from workflows.resource import Resource
 
 
def get_llm(**kwargs):
    return OpenAI(model="gpt-4.1-mini")
 
 
class MyWorkflow(Workflow):
    @step
    async def step1(
        self, ev: StartEvent, llm: Annotated[OpenAI, Resource(get_llm)]
    ) -> StopEvent:
        msg = ChatMessage(role="user", content=ev.get("input"))
        response = await llm.achat([msg])
        return StopEvent(result=response.message.content)
 
 
w = MyWorkflow()

response = await w.run(input="Hello, what is Litefuse?")
print(response)

第 5 步：在 Litefuse 中查看 Trace

运行工作流后，登录 Litefuse 即可查看生成的 trace。你将看到每个工作流步骤的日志，以及 token 数、延迟和执行路径等指标。

Litefuse Trace 示例

Litefuse 中的公开示例 trace

ℹ️

注意： 要添加额外的 trace 属性如 tag 或元数据，或将 LlamaIndex Workflows 与其他 Litefuse 特性一起使用，请参考本指南。

Interoperability with the Python SDK

You can use this integration together with the Litefuse SDKs to add additional attributes to the observation.

The @observe() decorator provides a convenient way to automatically wrap your instrumented code and add additional attributes to the observation.

from langfuse import observe, propagate_attributes, get_client
 
langfuse = get_client()
 
@observe()
def my_llm_pipeline(input):
    # Add additional attributes (user_id, session_id, metadata, version, tags) to all spans created within this execution scope
    with propagate_attributes(
        user_id="user_123",
        session_id="session_abc",
        tags=["agent", "my-observation"],
        metadata={"email": "user@litefuse.ai"},
        version="1.0.0"
    ):
 
        # YOUR APPLICATION CODE HERE
        result = call_llm(input)
 
        return result
 
# Run the function
my_llm_pipeline("Hi")

Learn more about using the Decorator in the Langfuse SDK instrumentation docs.

Troubleshooting

No observations appearing

First, enable debug mode in the Python SDK:

export LANGFUSE_DEBUG="True"

Then run your application and check the debug logs:

OTel observations appear in the logs: Your application is instrumented correctly but observations are not reaching Litefuse. To resolve this:
1. Call langfuse.flush() at the end of your application to ensure all observations are exported.
2. Verify that you are using the correct API keys and base URL.
No OTel spans in the logs: Your application is not instrumented correctly. Make sure the instrumentation runs before your application code.

Unwanted observations in Litefuse

The Langfuse SDK is based on OpenTelemetry. Other libraries in your application may emit OTel spans that are not relevant to you. These still count toward your billable units, so you should filter them out. See Unwanted spans in Litefuse for details.

Missing attributes

Some attributes may be stored in the metadata object of the observation rather than being mapped to the Litefuse data model. If a mapping or integration does not work as expected, please raise an issue on GitHub.

Next Steps

Once you have instrumented your code, you can manage, evaluate and debug your application:

Manage prompts in Litefuse Add evaluation scores Run LLM-as-a-judge Evaluators Create datasets Create custom dashboards Test queries in the Playground

Interoperability with the Python SDK

LlamaIndex Workflows 文档

LlamaIndex Mastra

这个页面对你有帮助吗？

支持