使用 Litefuse Trace OpenAI Agents SDK
本 notebook 演示如何将 Litefuse 集成 到你的 OpenAI Agents 工作流中，以监控、调试并评估你的 AI Agent。
什么是 OpenAI Agents SDK？：OpenAI Agents SDK 是一个轻量级开源框架，让开发者可以构建 AI Agent 并编排多 Agent 工作流。它提供了构建块 —— 例如工具、handoff 和 guardrail —— 用于为大语言模型配置自定义指令与集成工具。其 Python 优先的设计支持动态指令与函数工具，便于快速原型和与外部系统集成。
什么是 Litefuse？：Litefuse 是一个面向 AI Agent 的开源可观测性平台。它帮助你可视化并监控 LLM 调用、工具使用、成本、延迟等。
1. 安装依赖
下面我们安装 openai-agents 库（OpenAI Agents SDK），以及 OpenInference OpenAI Agents instrumentation 库。
%pip install openai-agents langfuse nest_asyncio openinference-instrumentation-openai-agents2. 配置环境与 Litefuse 凭证
接下来设置你的 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-..."3. 对 Agent 进行 Instrumentation
import nest_asyncio
nest_asyncio.apply()现在我们初始化 OpenInference OpenAI Agents instrumentation。这个第三方 instrumentation 会自动捕获 OpenAI Agents 的操作并将 OpenTelemetry (OTel) span 导出到 Litefuse。
from openinference.instrumentation.openai_agents import OpenAIAgentsInstrumentor
OpenAIAgentsInstrumentor().instrument()接下来初始化 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.")4. Hello World 示例
下面我们创建一个总是以 俳句 形式回答的 OpenAI Agent。我们使用 Runner.run 运行它并打印最终输出。
import asyncio
from agents import Agent, Runner
async def main():
agent = Agent(
name="Assistant",
instructions="You only respond in haikus.",
)
result = await Runner.run(agent, "Tell me about recursion in programming.")
print(result.final_output)
loop = asyncio.get_running_loop()
await loop.create_task(main())
点击上面的链接（或你自己的项目链接）即可查看所有子 span、token 使用、延迟等信息，便于调试或优化。
5. 多 Agent Handoff 示例
这里我们创建：
- 一个仅用西班牙语回答的 Spanish Agent。
- 一个仅用英语回答的 English Agent。
- 一个根据输入语言把请求路由到合适 Agent 的 Triage Agent。
任何调用或 handoff 都会被作为 trace 的一部分捕获。这样你就能看到使用了哪个子 Agent 或工具以及最终结果。
from agents import Agent, Runner
import asyncio
spanish_agent = Agent(
name="Spanish agent",
instructions="You only speak Spanish.",
)
english_agent = Agent(
name="English agent",
instructions="You only speak English",
)
triage_agent = Agent(
name="Triage agent",
instructions="Handoff to the appropriate agent based on the language of the request.",
handoffs=[spanish_agent, english_agent],
)
result = await Runner.run(triage_agent, input="Hola, ¿cómo estás?")
print(result.final_output)
6. 函数示例
OpenAI Agents SDK 允许 Agent 调用 Python 函数。通过 Litefuse instrumentation，你可以看到调用了哪些 函数、参数以及返回值。这里我们定义一个简单的函数 get_weather(city: str) 并将其作为工具添加。
import asyncio
from agents import Agent, Runner, function_tool
# Example function tool.
@function_tool
def get_weather(city: str) -> str:
return f"The weather in {city} is sunny."
agent = Agent(
name="Hello world",
instructions="You are a helpful agent.",
tools=[get_weather],
)
async def main():
result = await Runner.run(agent, input="What's the weather in Tokyo?")
print(result.final_output)
loop = asyncio.get_running_loop()
await loop.create_task(main())
查看 trace 时，你会看到一个 span 捕获了 get_weather 函数调用以及所传的参数。
7. 对 Agent 运行进行分组
在某些工作流中，你想把多次调用归并到同一条 trace 中 —— 例如构建一系列与同一用户请求相关的 prompt chain。你可以使用 trace(...) 上下文管理器，把多次调用嵌套在一个顶层 trace 下。
from agents import Agent, Runner, trace
import asyncio
async def main():
agent = Agent(name="Joke generator", instructions="Tell funny jokes.")
with trace("Joke workflow"):
first_result = await Runner.run(agent, "Tell me a joke")
second_result = await Runner.run(agent, f"Rate this joke: {first_result.final_output}")
print(f"Joke: {first_result.final_output}")
print(f"Rating: {second_result.final_output}")
loop = asyncio.get_running_loop()
await loop.create_task(main())
每个子调用都表示为顶层 Joke workflow span 下的一个子 span，让你能轻松看到整个对话或调用序列。
关联 Litefuse Prompt
如果你使用 Litefuse Prompt Management 管理 prompt，你可以通过设置一个 OTel Span processor 将所使用的 prompt 关联到 trace。
限制： 此方法会将 Litefuse Prompt 关联到 trace 中所有以指定字符串开头的 generation span（见下一段代码）。
from contextvars import ContextVar
from typing import Optional
from opentelemetry import context as context_api, trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import Span, SpanProcessor
prompt_info_var = ContextVar("prompt_info", default=None)
# Make sure to set the name of the generation spans in your trace
class LangfuseProcessor(SpanProcessor):
def on_start(self, span: "Span", parent_context: Optional[context_api.Context] = None) -> None:
if span.name.startswith("response"): # The name of the generation spans in your trace
prompt_info = prompt_info_var.get()
if prompt_info:
span.set_attribute("langfuse.prompt.name", prompt_info.get("name"))
span.set_attribute("langfuse.prompt.version", prompt_info.get("version"))
from langfuse import get_client
langfuse = get_client()
trace.get_tracer_provider().add_span_processor(LangfuseProcessor())from openinference.instrumentation.openai_agents import OpenAIAgentsInstrumentor
OpenAIAgentsInstrumentor().instrument()# Fetch the prompt from Langfuse Prompt Management
langfuse_prompt = langfuse.get_prompt("movie-critic")
system_prompt = langfuse_prompt.compile(criticlevel = "expert", movie = "Dune 2")
# Pass the prompt to the SpanProcessor
prompt_info_var.set({
"name": langfuse_prompt.name,
"version": langfuse_prompt.version,
})
# Run the agent ...资源
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:
- Call
langfuse.flush()at the end of your application to ensure all observations are exported. - Verify that you are using the correct API keys and base URL.
- Call
- 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: