Langfuse 评估代理
在本指南中，我们将学习如何监控 OpenAI Agent SDK 的内部步骤（跟踪）并使用 Langfuse 评估其性能。
本指南涵盖了团队用于快速可靠地将代理投入生产的在线和离线评估指标。要了解有关评估策略的更多信息，请参阅此博客文章。
为什么 AI 代理评估很重要：
- 调试任务失败或产生次优结果的问题
- 实时监控成本和性能
- 通过持续反馈提高可靠性和安全性
步骤 0：安装所需的库
下面我们安装 openai-agents 库（OpenAI Agents SDK）、pydantic-ai[logfire] OpenTelemetry 仪器、langfuse 和 Hugging Face datasets 库
%pip install openai-agents nest_asyncio "pydantic-ai[logfire]" langfuse datasets
步骤 1：仪器化您的代理
在此笔记本中，我们将使用 Langfuse 来跟踪、调试和评估我们的代理。
注意：如果您正在使用 LlamaIndex 或 LangGraph，您可以在此处和此处找到有关仪器化它们的文档。
import os
import base64
# 从项目设置页面获取您项目的密钥：https://cloud.langfuse.com
os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-..."
os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-..."
os.environ["LANGFUSE_HOST"] = "https://cloud.langfuse.com" # 🇪🇺 欧洲区域
# os.environ["LANGFUSE_HOST"] = "https://us.cloud.langfuse.com" # 🇺🇸 美国区域
# 构建基本身份验证头。
LANGFUSE_AUTH = base64.b64encode(
f"{os.environ.get('LANGFUSE_PUBLIC_KEY')}:{os.environ.get('LANGFUSE_SECRET_KEY')}".encode()
).decode()
# 配置 OpenTelemetry 端点和标头
os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = os.environ.get("LANGFUSE_HOST") + "/api/public/otel"
os.environ["OTEL_EXPORTER_OTLP_HEADERS"] = f"Authorization=Basic {LANGFUSE_AUTH}"
# 您的 openai 密钥
os.environ["OPENAI_API_KEY"] = "sk-proj-..."
设置好环境变量后，我们现在可以初始化 Langfuse 客户端。get_client() 使用环境变量中提供的凭据初始化 Langfuse 客户端。
from langfuse import get_client
langfuse = get_client()
# 验证连接
if langfuse.auth_check():
print("Langfuse 客户端已通过身份验证并准备就绪！")
else:
print("身份验证失败。请检查您的凭据和主机。")
Pydantic Logfire 为 OpenAi Agent SDK 提供了仪器化。我们使用它将跟踪发送到 Langfuse OpenTelemetry 后端。
import nest_asyncio
nest_asyncio.apply()
import logfire
# 配置 logfire 仪器。
logfire.configure(
service_name='my_agent_service',
send_to_logfire=False,
)
# 此方法自动修补 OpenAI Agents SDK，通过 OTLP 将日志发送到 Langfuse。
logfire.instrument_openai_agents()
步骤 2：测试您的仪器化
这是一个简单的问答代理。我们运行它以确认仪器化正在正常工作。如果一切设置正确，您将在可观察性仪表板中看到日志/跨度。
import asyncio
from agents import Agent, Runner
async def main():
agent = Agent(
name="Assistant",
instructions="You are a senior software engineer",
)
result = await Runner.run(agent, "Tell me why it is important to evaluate AI agents.")
print(result.final_output)
loop = asyncio.get_running_loop()
await loop.create_task(main())
langfuse.flush()
13:00:52.784 OpenAI Agents trace: Agent workflow
13:00:52.787 Agent run: 'Assistant'
13:00:52.797 Responses API with 'gpt-4o'
评估 AI 代理至关重要，原因如下：
1. **绩效评估**：它有助于确定代理是否达到了期望的目标并有效地执行任务。通过评估，我们可以评估准确性、速度和整体性能。
2. **可靠性和一致性**：定期评估可确保 AI 在不同条件下保持一致的行为，并在生产环境中保持可靠。
3. **偏见和公平性**：识别和减轻偏见对于公平和道德的 AI 至关重要。评估有助于揭示代理行为中的任何歧视性模式。
4. **安全性**：评估 AI 代理可确保它们安全运行，不会造成伤害或意外副作用，尤其是在关键应用程序中。
5. **用户信任**：适当的评估通过展示 AI 有效且符合预期来建立用户和利益相关者的信任。
6. **法规遵从性**：它确保遵守法律和道德标准，随着人工智能法规的不断发展，这一点越来越重要。
7. **持续改进**：持续评估可提供可用于随着时间推移改进代理的见解，从而优化性能并适应新的挑战。
8. **资源效率**：评估有助于确保 AI 代理有效利用资源，从而降低成本并提高可扩展性。
总之，评估对于确保 AI 代理有效、合乎道德且符合用户需求和社会规范至关重要。
请检查您的 Langfuse Traces Dashboard 以确认已记录跨度和日志。
Langfuse 中的示例跟踪：
步骤 3：观察和评估更复杂的代理
现在您已经确认了仪器化工作正常，让我们尝试一个更复杂的查询，以便可以看到高级指标（令牌使用情况、延迟、成本等）是如何跟踪的。
import asyncio
from agents import Agent, Runner, 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 Berlin?")
print(result.final_output)
loop = asyncio.get_running_loop()
await loop.create_task(main())
13:01:15.351 OpenAI Agents trace: Agent workflow
13:01:15.355 Agent run: 'Hello world'
13:01:15.364 Responses API with 'gpt-4o'
13:01:15.999 Function: get_weather
13:01:16.000 Responses API with 'gpt-4o'
柏林的天气目前是晴朗的。
跟踪结构
Langfuse 记录一个包含跨度的跟踪，跨度代表您代理逻辑的每个步骤。这里，跟踪包含整体代理运行和子跨度：
- 工具调用（get_weather）
- LLM 调用（Responses API with 'gpt-4o'）
您可以检查这些以确切了解时间花费在哪里、使用了多少令牌等：
在线评估
在线评估是指在实时、真实环境中评估代理，即在生产中的实际使用期间。这包括持续监控代理在真实用户交互中的表现并分析结果。
我们在此处写了一份关于不同评估技术的指南：https://langfuse.com/blog/2025-03-04-llm-evaluation-101-best-practices-and-challenges。
跟踪指标的常用指标
- 成本 — 仪器化捕获令牌使用情况，您可以将其通过分配每令牌价格转换为近似成本。
- 延迟 — 观察完成每个步骤或整个运行所需的时间。
- 用户反馈 — 用户可以提供直接反馈（点赞/点踩）来帮助优化或纠正代理。
- LLM 即裁判 — 使用单独的 LLM 来近乎实时地评估您的代理的输出（例如，检查毒性或正确性）。
下面我们展示了这些指标的示例。
1. 成本
下面是显示 gpt-4o 调用使用情况的屏幕截图。这有助于查看昂贵的步骤并优化您的代理。
2. 延迟
我们还可以查看完成每个步骤所需的时间。在下面的示例中，整个运行耗时 7 秒，您可以按步骤细分。这有助于您识别瓶颈并优化代理。
3. 其他属性
Langfuse 允许您将其他属性传递给您的跨度。这些可以包括 user_id、tags、session_id 和自定义 metadata。使用这些详细信息丰富跟踪对于分析、调试和监控您的应用程序在不同用户或会话中的行为非常重要。
在此示例中，我们将 user_id、session_id 和 trace_tags 传递给 Langfuse。
input_query = "Why is AI agent evaluation important?"
with langfuse.start_as_current_span(
name="OpenAI-Agent-Trace",
) as span:
# 在此处运行您的应用程序
async def main(input_query):
agent = Agent(
name = "Assistant",
instructions = "You are a helpful assistant.",
)
result = await Runner.run(agent, input_query)
print(result.final_output)
return result
result = await main(input_query)
# 将其他属性传递给跨度
span.update_trace(
input=input_query,
output=result,
user_id="user_123",
session_id="my-agent-session",
tags=["staging", "demo", "OpenAI Agent SDK"],
metadata={"email": "user@langfuse.com"},
version="1.0.0"
)
# 在短期应用程序中刷新事件
langfuse.flush()
13:02:41.552 OpenAI Agents trace: Agent workflow
13:02:41.553 Agent run: 'Assistant'
13:02:41.554 Responses API with 'gpt-4o'
AI 代理评估至关重要，原因如下：
1. **绩效指标**：它有助于确定 AI 代理执行任务的程度，确保其符合期望的标准和目标。
2. **可靠性和安全性**：评估可确保代理在不同场景下行为一致且安全，从而降低意外后果的风险。
3. **偏见检测**：通过评估 AI 代理，开发人员可以识别和减轻偏见，确保所有用户的公平和公正的结果。
4. **基准测试和比较**：评估允许比较不同的 AI 模型或版本，从而促进改进和进步。
5. **用户信任**：展示 AI 代理的有效性和可靠性可以建立用户信任，鼓励采用和使用。
6. **法规遵从性**：适当的评估有助于确保 AI 系统符合法律和法规要求，这在医疗保健或金融等敏感领域尤其重要。
7. **可扩展性和部署**：评估有助于确定 AI 代理是否可以有效扩展并在实际环境中准确运行。
总而言之，AI 代理评估是开发有效、值得信赖且合乎道德的 AI 系统的关键。
4. 用户反馈
如果您的代理嵌入到用户界面中，您可以记录直接的用户反馈（例如聊天 UI 中的点赞/点踩）。下面是一个使用 IPython.display 进行简单反馈机制的示例。
在下面的代码片段中，当用户发送聊天消息时，我们捕获 OpenTelemetry 跟踪 ID。如果用户喜欢/不喜欢最后一个答案，我们会将分数附加到跟踪中。
from agents import Agent, Runner, WebSearchTool
from opentelemetry.trace import format_trace_id
import ipywidgets as widgets
from IPython.display import display
from langfuse import get_client
langfuse = get_client()
# 使用网络搜索工具定义您的代理
agent = Agent(
name="WebSearchAgent",
instructions="You are an agent that can search the web.",
tools=[WebSearchTool()]
)
def on_feedback(button):
if button.icon == "thumbs-up":
langfuse.create_score(
value=1,
name="user-feedback",
comment="The user gave this response a thumbs up",
trace_id=trace_id
)
elif button.icon == "thumbs-down":
langfuse.create_score(
value=0,
name="user-feedback",
comment="The user gave this response a thumbs down",
trace_id=trace_id
)
print("Scored the trace in Langfuse")
user_input = input("Enter your question: ")
# 运行代理
with langfuse.start_as_current_span(
name="OpenAI-Agent-Trace",
) as span:
# 在此处运行您的应用程序
result = Runner.run_sync(agent, user_input)
print(result.final_output)
result = await main(user_input)
trace_id = langfuse.get_current_trace_id()
span.update_trace(
input=user_input,
output=result.final_output,
)
# 获取反馈
print("How did you like the agent response?")
thumbs_up = widgets.Button(description="👍", icon="thumbs-up")
thumbs_down = widgets.Button(description="👎", icon="thumbs-down")
thumbs_up.on_click(on_feedback)
thumbs_down.on_click(on_feedback)
display(widgets.HBox([thumbs_up, thumbs_down]))
# 在短期应用程序中刷新事件
langfuse.flush()
Enter your question: What is Langfuse?
13:54:41.574 OpenAI Agents trace: Agent workflow
13:54:41.575 Agent run: 'WebSearchAgent'
13:54:41.577 Responses API with 'gpt-4o'
Langfuse 是一个开源工程平台，旨在增强大型语言模型 (LLM) 应用程序的开发、监控和优化。它提供了一套工具，可提供可观察性、提示管理、评估和指标，从而有助于调试和改进基于 LLM 的解决方案。 ([toolkitly.com](https://www.toolkitly.com/langfuse?utm_source=openai))
**Langfuse 的主要特点：**
- **LLM 可观察性**：Langfuse 通过跟踪 API 调用、用户输入、提示和输出来使开发人员能够监控和分析语言模型的性能。这种可观察性有助于理解模型行为和识别改进领域。 ([toolkitly.com](https://www.toolkitly.com/langfuse?utm_source=openai))
- **提示管理**：该平台提供直接在 Langfuse 中管理、版本控制和部署提示的工具。此功能允许高效地组织和优化提示以获得最佳模型响应。 ([toolkitly.com](https://www.toolkitly.com/langfuse?utm_source=openai))
- **评估和指标**：Langfuse 提供收集和计算 LLM 完成情况分数、运行模型评估以及收集用户反馈的功能。它还跟踪关键指标，如成本、延迟和质量，通过仪表板和数据导出提供见解。 ([toolkitly.com](https://www.toolkitly.com/langfuse?utm_source=openai))
- **Playground 环境**：该平台包含一个 Playground，用户可以在其中与不同的模型和提示进行交互式实验，从而促进提示工程和测试。 ([toolkitly.com](https://www.toolkitly.com/langfuse?utm_source=openai))
- **集成功能**：Langfuse 与各种工具和框架无缝集成，包括 LlamaIndex、LangChain、OpenAI SDK、LiteLLM 等，增强了其功能并允许开发复杂的应用程序。 ([toolerific.ai](https://toolerific.ai/ai-tools/opensource/langfuse-langfuse?utm_source=openai))
- **开源和自托管**：Langfuse 是开源的，允许开发人员根据其特定需求自定义和扩展平台。它可以自托管，从而完全控制基础设施和数据。 ([vafion.com](https://www.vafion.com/blog/unlocking-power-language-models-langfuse/?utm_source=openai))
Langfuse 对于处理 LLM 的开发人员和研究人员特别有价值，它提供了一套全面的工具来提高 LLM 应用程序的性能和可靠性。其灵活性、集成功能和开源性质使其成为寻求增强 LLM 项目的开发人员的可靠选择。
How did you like the agent response?
HBox(children=(Button(description='👍', icon='thumbs-up', style=ButtonStyle()), Button(description='👎', icon='t…
Scored the trace in Langfuse
用户反馈随后在 Langfuse 中捕获：
5. LLM 即裁判
LLM 即裁判是自动评估代理输出的另一种方法。您可以设置单独的 LLM 调用来评估输出的正确性、毒性、样式或您关心的任何其他标准。
工作流程：
- 您定义一个评估模板，例如“检查文本是否具有毒性”。
- 您设置一个用作裁判模型的模型；在这种情况下是
gpt-4o-mini。 - 每次代理生成输出时，您都会将该输出与模板一起传递给您的“裁判” LLM。
- 裁判 LLM 返回一个您记录到可观察性工具中的评分或标签。
Langfuse 中的示例：
# 示例：检查代理的输出是否具有毒性。
from agents import Agent, Runner, WebSearchTool
# 使用网络搜索工具定义您的代理
agent = Agent(
name="WebSearchAgent",
instructions="You are an agent that can search the web.",
tools=[WebSearchTool()]
)
input_query = "Is eating carrots good for the eyes?"
# 运行代理
with langfuse.start_as_current_span(name="OpenAI-Agent-Trace") as span:
# 使用查询运行您的代理
result = Runner.run_sync(agent, input_query)
# 将输入和输出值添加到父跟踪
span.update_trace(
input=input_query,
output=result.final_output,
)
14:05:34.735 OpenAI Agents trace: Agent workflow
14:05:34.736 Agent run: 'WebSearchAgent'
14:05:34.738 Responses API with 'gpt-4o'
您可以看到此示例的答案被评判为“无毒性”。
6. 可观察性指标概述
所有这些指标都可以一起在仪表板中可视化。这使您可以快速了解代理在多个会话中的表现，并帮助您跟踪质量指标随时间的变化。
离线评估
在线评估对于实时反馈至关重要，但您还需要离线评估——开发期间或之前的系统检查。这有助于在将更改推送到生产环境之前维护质量和可靠性。
数据集评估
在离线评估中，您通常：
- 拥有一个基准数据集（包含提示和预期输出对）
- 在该数据集上运行您的代理
- 将输出与预期结果进行比较或使用额外的评分机制
下面，我们使用 search-dataset 来演示此方法，该数据集包含可以通过网络搜索工具回答的问题和预期的答案。
import pandas as pd
from datasets import load_dataset
# 从 Hugging Face 获取 search-dataset
dataset = load_dataset("junzhang1207/search-dataset", split = "train")
df = pd.DataFrame(dataset)
print("search-dataset 的前几行：")
print(df.head())
README.md: 0%| | 0.00/2.12k [00:00<?, ?B/s]
data-samples.json: 0%| | 0.00/2.48k [00:00<?, ?B/s]
data.jsonl: 0%| | 0.00/316k [00:00<?, ?B/s]
Generating train split: 0%| | 0/934 [00:00<?, ? examples/s]
GSM8K 数据集的几行：
id
0 20caf138-0c81-4ef9-be60-fe919e0d68d4
1 1f37d9fd-1bcc-4f79-b004-bc0e1e944033
2 76173a7f-d645-4e3e-8e0d-cca139e00ebe
3 5f5ef4ca-91fe-4610-a8a9-e15b12e3c803
4 64dbed0d-d91b-4acd-9a9c-0a7aa83115ec
question
0 steve jobs statue location budapst
1 Why is the Battle of Stalingrad considered a t...
2 In what year did 'The Birth of a Nation' surpa...
3 How many Russian soldiers surrendered to AFU i...
4 What event led to the creation of Google Images?
expected_answer category area
0 The Steve Jobs statue is located in Budapest, ... Arts Knowledge
1 The Battle of Stalingrad is considered a turni... General News News
2 This question is based on a false premise. 'Th... Entertainment News
3 About 300 Russian soldiers surrendered to the ... General News News
4 Jennifer Lopez's appearance in a green Versace... Technology News
接下来，我们在 Langfuse 中创建一个数据集实体来跟踪运行。然后，我们将数据集中的每个项目添加到系统中。
from langfuse import get_client
langfuse = get_client()
langfuse_dataset_name = "search-dataset_huggingface_openai-agent"
# 在 Langfuse 中创建数据集
langfuse.create_dataset(
name=langfuse_dataset_name,
description="search-dataset uploaded from Huggingface",
metadata={
"date": "2025-03-14",
"type": "benchmark"
}
)
Dataset(id='cm88w66t102qpad07xhgeyaej', name='search-dataset_huggingface_openai-agent', description='search-dataset uploaded from Huggingface', metadata={'date': '2025-03-14', 'type': 'benchmark'}, project_id='cloramnkj0002jz088vzn1ja4', created_at=datetime.datetime(2025, 3, 14, 14, 47, 14, 676000, tzinfo=datetime.timezone.utc), updated_at=datetime.datetime(2025, 3, 14, 14, 47, 14, 676000, tzinfo=datetime.timezone.utc))
for idx, row in df.iterrows():
langfuse.create_dataset_item(
dataset_name=langfuse_dataset_name,
input={"text": row["question"]},
expected_output={"text": row["expected_answer"]}
)
if idx >= 49: # For this example, we upload only the first 50 items
break
在数据集上运行代理
我们定义一个名为 run_openai_agent() 的辅助函数，该函数：
- 启动 Langfuse 跨度
- 在提示上运行我们的代理
- 在 Langfuse 中记录跟踪 ID
然后，我们循环遍历每个数据集项，运行代理，并将跟踪链接到数据集项。如果需要，我们还可以附加一个快速评估分数。
from agents import Agent, Runner, WebSearchTool
from langfuse import get_client
langfuse = get_client()
dataset_name = "search-dataset_huggingface_openai-agent"
current_run_name = "qna_model_v3_run_05_20" # Identifies this specific evaluation run
agent = Agent(
name="WebSearchAgent",
instructions="You are an agent that can search the web.",
tools=[WebSearchTool(search_context_size= "high")]
)
# 假设 'run_openai_agent' 是您仪器化的应用程序函数
def run_openai_agent(question):
with langfuse.start_as_current_generation(name="qna-llm-call") as generation:
# 模拟 LLM 调用
result = Runner.run_sync(agent, question)
# 使用输入和输出来更新跟踪
generation.update_trace(
input= question,
output=result.final_output,
)
return result.final_output
dataset = langfuse.get_dataset(name=dataset_name) # 获取您预先填充的数据集
for item in dataset.items:
# 使用 item.run() 上下文管理器
with item.run(
run_name=current_run_name,
run_metadata={"model_provider": "OpenAI", "temperature_setting": 0.7},
run_description="Evaluation run for Q&A model v3 on May 20th"
) as root_span: # root_span 是此项目和运行的新跟踪的根跨度。
# 此块中的所有后续 Langfuse 操作都属于此跟踪。
# 调用您的应用程序逻辑
generated_answer = run_openai_agent(question=item.input["text"])
print(item.input)
您可以对不同的项重复此过程：
- 搜索工具（例如，OpenAI 的
WebSearchTool的不同上下文大小） - 模型（gpt-4o-mini、o1 等）
- 工具（搜索 vs. 无搜索）
然后，在 Langfuse 中并排比较它们。在此示例中，我对 50 个数据集问题运行了代理 3 次。对于每次运行，我都使用了 OpenAI 的 WebSearchTool 的上下文大小的不同设置。您可以看到，增加的上下文大小也略微提高了答案的正确性，从 0.89 提高到 0.92。correct_answer 分数是由 LLM 即裁判评估器 创建的，该评估器设置为根据数据集中提供的样本答案来判断问题的正确性。
