将 Litefuse 与 Pydantic AI 集成

本 notebook 提供一个分步指南，演示如何将 Litefuse 与 Pydantic AI 集成，从而对你的 LLM 应用实现可观测性与调试。

关于 PydanticAI： PydanticAI 是一个 Python Agent 框架，旨在简化生产级生成式 AI 应用的开发。它将 FastAPI 中的类型安全、人性化 API 设计与开发者体验带入到 GenAI 应用开发的世界。

什么是 Litefuse？ Litefuse 是一个开源的 AI Agent 可观测性与评估平台。它为 AI 应用提供 tracing 与监控能力。Litefuse 通过提供详细洞察以及通过原生集成、OpenTelemetry 和专用 SDK 与众多工具和 Pydantic AI 对接，帮助开发者调试、分析并优化其 AI 系统。

快速开始

我们通过一个实际示例演示如何使用 Pydantic AI 并将其与 Litefuse 集成，以获得全面的 tracing。

第 1 步：安装依赖

%pip install langfuse pydantic-ai -U

第 2 步：配置 Langfuse SDK

接下来设置你的 Litefuse API Key。可通过注册免费的 Litefuse Cloud 账号或自托管 Litefuse 获取这些 Key。这些环境变量是 Langfuse 客户端鉴权并向 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 步：初始化 Pydantic AI Instrumentation

现在我们初始化 Pydantic AI Instrumentation。它会自动捕获 Pydantic AI 的操作并将 OpenTelemetry (OTel) span 导出到 Litefuse。

from pydantic_ai.agent import Agent
 
# Initialize Pydantic AI instrumentation
Agent.instrument_all()

第 4 步：基础 Pydantic AI 应用

最后，运行你的 Pydantic AI Agent，生成将被发送到 Litefuse 的 trace 数据。下面示例中，Agent 在一个依赖值（中奖号码）与自然语言输入下被执行。工具函数的输出随后被打印出来。

请确保在配置 Agent 时传入 instrument=True。

from pydantic_ai import Agent, RunContext
 
roulette_agent = Agent(
    'openai:gpt-4o',
    deps_type=int,
    system_prompt=(
        'Use the `roulette_wheel` function to see if the '
        'customer has won based on the number they provide.'
    ),
    instrument=True
)
 
@roulette_agent.tool
async def roulette_wheel(ctx: RunContext[int], square: int) -> str:
    """check if the square is a winner"""
    return 'winner' if square == ctx.deps else 'loser'

# Run the agent - using await since we're in a Jupyter notebook
success_number = 18
result = await roulette_agent.run('Put my money on square eighteen', deps=success_number)
print(result.output)

第 5 步：在 Litefuse 中查看 Trace

执行应用后，进入你的 Litefuse Trace 表。你将看到应用执行的详细 trace，包括 LLM 调用、检索操作、输入、输出以及性能指标的洞察。

Litefuse 中的示例 Trace

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

Pipecat Quarkus LangChain4j

这个页面对你有帮助吗？

支持