OpenAI Agents SDK 中文文档 中文教程 （3）

英文文档原文详见 OpenAI Agents SDK

/

本文是OpenAI-agents-sdk-python使用翻译软件翻译后的中文文档/教程。分多个帖子发布，帖子的目录如下：

(1) OpenAI 代理 SDK， 介绍及快速入门

(2)OpenAI agents sdk, agents，运行agents，结果，流，工具，交接

(3) OpenAi agents sdk, 跟踪，上下文管理，护栏

跟踪

代理 SDK 包括内置跟踪功能，可收集代理运行期间事件的全面记录：LLM 生成、工具调用、切换、防护机制，甚至发生的自定义事件。使用 Traces 控制面板，您可以在开发和生产期间调试、可视化和监控您的工作流程。

注意

默认情况下，跟踪处于启用状态。有两种方法可以禁用跟踪：

1. 您可以通过设置 env varOPENAI_AGENTS_DISABLE_TRACING=1
2. 您可以通过将 agents.run.RunConfig.tracing_disabled 设置为True

跟踪和跨度

• 跟踪表示 “workflow” 的单个端到端作。它们由 Span 组成。跟踪具有以下属性：
  • workflow_name：这是逻辑工作流或应用程序。例如，“代码生成”或“客户服务”。
  • trace_id：跟踪的唯一 ID。如果您未传递 1 个，则自动生成。必须具有格式 .trace_<32_alphanumeric>
  • group_id：可选的组 ID，用于链接来自同一对话的多个跟踪。例如，您可以使用聊天会话 ID。
  • disabled：如果为 True，则不会记录跟踪。
  • metadata：跟踪的可选元数据。
• Span 表示具有开始时间和结束时间的作。Span 具有：
  • started_at和时间戳。ended_at
  • trace_id来表示它们所属的跟踪
  • parent_id，它指向此 Span 的父 Span（如果有）
  • span_data，这是有关 Span 的信息。例如，包含有关 Agent 的信息，包含有关 LLM 生成的信息，等等。AgentSpanDataGenerationSpanData

默认跟踪

默认情况下，SDK 会跟踪以下内容：

• 整个包裹在一个 .Runner.{run, run_sync, run_streamed}()trace()
• 每次代理运行时，都会将其包装在agent_span()
• LLM 代被包装在generation_span()
• 函数工具调用都包装在function_span()
• 护栏被包裹在guardrail_span()
• Handoff 被包裹在handoff_span()

默认情况下，跟踪名为 “Agent trace”。如果使用 ，则可以设置此名称，也可以使用 RunConfig 配置名称和其他属性。trace

此外，您还可以设置自定义跟踪处理器以将跟踪推送到其他目标（作为替代目标或辅助目标）。

更高级别的跟踪

有时，您可能希望多个调用成为单个跟踪的一部分。您可以通过将整个代码包装在 .run()trace()

from agents import Agent, Runner, trace

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}")

创建跟踪

您可以使用 trace（） 函数创建跟踪。需要启动和完成跟踪。您有两种选择：

1. 建议：将跟踪用作上下文管理器，即 .这将在正确的时间自动开始和结束跟踪。with trace(...) as my_trace
2. 您还可以手动调用 trace.start（） 和 trace.finish（）。

当前跟踪是通过 Python contextvar 跟踪的。这意味着它会自动使用并发。如果您手动启动/结束跟踪，则需要传递 和 to / 来更新当前跟踪。mark_as_currentreset_currentstart()finish()

创建 Span

您可以使用各种 *_span（） 方法创建一个 span。通常，您不需要手动创建 span。custom_span（） 函数可用于跟踪自定义范围信息。

Span 自动成为当前跟踪的一部分，并嵌套在最近的当前 Span 下，该 Span 通过 Python contextvar 进行跟踪。

敏感数据

某些 span 会跟踪潜在的敏感数据。例如，存储 LLM 生成的 inputs/outputs，并存储函数调用的 inputs/outputs。这些可能包含敏感数据，因此您可以通过 RunConfig.trace_include_sensitive_data 禁用捕获该数据。generation_span()function_span()

自定义跟踪处理器

跟踪的高级体系结构是：

• 在初始化时，我们创建一个全局 TraceProvider，它负责创建跟踪。
• 我们使用 BatchTraceProcessor 进行配置，该处理器将跟踪/跨度批量发送到 BackendSpanExporter，后者将跨度和跟踪批量导出到 OpenAI 后端。TraceProvider

要自定义此默认设置，将跟踪发送到备用或其他后端或修改导出器行为，您有两个选项：

1. add_trace_processor（） 允许您添加一个额外的跟踪处理器，该处理器将在跟踪和 span 准备就绪时接收它们。除了将跟踪发送到 OpenAI 的后端之外，您还可以进行自己的处理。
2. set_trace_processors（） 允许您将默认处理器替换为自己的跟踪处理器。这意味着跟踪不会发送到 OpenAI 后端，除非您包含这样做的 a。TracingProcessor

外部跟踪处理器包括：

• 智囊团
• Pydantic Logfire
• 代理运营
• 关键词 AI

上下文管理

Context 是一个重载的术语。您可能关心的上下文主要有两类：

1. 代码本地可用的上下文：这是工具函数运行时、生命周期钩子等回调期间可能需要的数据和依赖项。on_handoff
2. LLM 可用的上下文：这是 LLM 在生成响应时看到的数据。

本地上下文

这是通过 RunContextWrapper 类和其中的 context 属性表示的。其工作方式是：

1. 您可以创建所需的任何 Python 对象。一种常见的模式是使用 dataclass 或 Pydantic 对象。
2. 您将该对象传递给各种 run 方法（例如 .Runner.run(..., **context=whatever**))
3. 您的所有工具调用、生命周期钩子等都将传递一个包装对象，其中 表示您的上下文对象类型，您可以通过 .RunContextWrapper[T]Twrapper.context

需要注意的最重要的一点是：给定代理运行的每个代理、工具功能、生命周期等都必须使用相同类型的上下文。

您可以将上下文用于以下作：

• 运行的上下文数据（例如，用户名/uid 或有关用户的其他信息等）
• 依赖项（例如 Logger 对象、数据获取器等）
• 帮助程序函数

注意

上下文对象不会发送到 LLM。它纯粹是一个本地对象，你可以从中读取、写入和调用方法。

import asyncio
from dataclasses import dataclass

from agents import Agent, RunContextWrapper, Runner, function_tool

@dataclass
class UserInfo:  

name: str uid: int @function_tool async def fetch_user_age(wrapper: RunContextWrapper[UserInfo]) -> str: return f"User {wrapper.context.name} is 47 years old" async def main(): user_info = UserInfo(name="John", uid=123) agent = Agent[UserInfo]( name="Assistant", tools=[fetch_user_age], ) result = await Runner.run( starting_agent=agent, input="What is the age of the user?", context=user_info, ) print(result.final_output) # The user John is 47 years old. if __name__ == "__main__": asyncio.run(main())

代理/LLM 上下文

调用 LLM 时，它唯一可以看到的数据来自对话历史记录。这意味着，如果要向 LLM 提供一些新数据，则必须以使其在该历史记录中可用的方式进行。有几种方法可以做到这一点：

1. 您可以将其添加到代理 。这也称为 “系统提示符” 或 “开发人员消息”。系统提示符可以是静态字符串，也可以是接收上下文并输出字符串的动态函数。对于始终有用的信息（例如，用户名或当前日期），这是一种常见的策略。instructions
2. 在调用函数时将其添加到 中。这与策略类似，但允许您拥有位于指挥链中较低位置的消息。inputRunner.runinstructions
3. 通过函数工具公开它。这对于按需上下文非常有用 - LLM 决定何时需要一些数据，并可以调用该工具来获取该数据。
4. 使用检索或 Web 搜索。这些是能够从文件或数据库（检索）或 Web （Web 搜索）获取相关数据的特殊工具。这对于将响应“接地”在相关上下文数据中非常有用。

护栏

防护机制与您的代理并行运行，使您能够对用户输入进行检查和验证。例如，假设您有一个代理，它使用非常智能（因此速度慢/成本高）的模型来帮助处理客户请求。您不希望恶意用户要求模型帮助他们完成数学作业。因此，您可以使用快速/便宜的模型运行护栏。如果护栏检测到恶意使用，它可能会立即引发错误，从而阻止昂贵的模型运行并节省您的时间/金钱。

护栏有两种：

1. 输入护栏在初始用户输入上运行
2. 输出护栏在最终代理输出上运行

输入护栏

输入护栏分 3 个步骤运行：

1. 首先，护栏接收传递给代理的相同输入。
2. 接下来，护栏函数运行以生成 GuardrailFunctionOutput，然后将其包装在 InputGuardrailResult 中
3. 最后，我们检查 .tripwire_triggered 是否为 true。如果为 true，则会引发 InputGuardrailTripwireTriggered 异常，以便您可以适当地响应用户或处理异常。

注意

输入护栏旨在根据用户输入运行，因此仅当代理是第一个代理时，代理的护栏才会运行。您可能想知道，为什么代理上的属性而不是传递给 ？这是因为护栏往往与实际的 Agent 相关 - 您将为不同的 Agent 运行不同的护栏，因此将代码放在一起有助于提高可读性。guardrailsRunner.run

输出护栏

输出护栏分 3 个步骤运行：

1. 首先，护栏接收传递给代理的相同输入。
2. 接下来，护栏函数运行以生成 GuardrailFunctionOutput，然后将其包装在 OutputGuardrailResult 中
3. 最后，我们检查 .tripwire_triggered 是否为 true。如果为 true，则会引发 OutputGuardrailTripwireTriggered 异常，以便您可以适当地响应用户或处理异常。

注意

输出护栏旨在针对最终代理输入运行，因此仅当代理是最后一个代理时，代理的护栏才会运行。与输入护栏类似，我们这样做是因为护栏往往与实际的 Agent 相关 - 您将为不同的 Agent 运行不同的护栏，因此将代码放在一起有助于提高可读性。

绊线

如果输入或输出未通过护栏，护栏可以通过绊线发出信号。一旦我们看到触发了绊线的护栏，我们就会立即引发异常并停止 Agent 执行。{Input,Output}GuardrailTripwireTriggered

实施护栏

您需要提供一个接收输入并返回 GuardrailFunctionOutput 的函数。在此示例中，我们将通过在后台运行 Agent 来实现此目的。

from pydantic import BaseModel
from agents import (
    Agent,
    GuardrailFunctionOutput,
    InputGuardrailTripwireTriggered,
    RunContextWrapper,
    Runner,
    TResponseInputItem,
    input_guardrail,
)

class MathHomeworkOutput(BaseModel):
    is_math_homework: bool
    reasoning: str

guardrail_agent = Agent( 

name="Guardrail check", instructions="Check if the user is asking you to do their math homework.", output_type=MathHomeworkOutput, ) @input_guardrail async def math_guardrail( ctx: RunContextWrapper[None], agent: Agent, input: str | list[TResponseInputItem] ) -> GuardrailFunctionOutput: result = await Runner.run(guardrail_agent, input, context=ctx.context) return GuardrailFunctionOutput( output_info=result.final_output, tripwire_triggered=result.final_output.is_math_homework, ) agent = Agent( name="Customer support agent", instructions="You are a customer support agent. You help customers with their questions.", input_guardrails=[math_guardrail], ) async def main(): # This should trip the guardrail try: await Runner.run(agent, "Hello, can you help me solve for x: 2x + 3 = 11?") print("Guardrail didn't trip - this is unexpected") except InputGuardrailTripwireTriggered: print("Math homework guardrail tripped")

输出护栏类似。

from pydantic import BaseModel
from agents import (
    Agent,
    GuardrailFunctionOutput,
    OutputGuardrailTripwireTriggered,
    RunContextWrapper,
    Runner,
    output_guardrail,
)
class MessageOutput(BaseModel): 

response: str class MathOutput(BaseModel): is_math: bool reasoning: str guardrail_agent = Agent( name="Guardrail check", instructions="Check if the output includes any math.", output_type=MathOutput, ) @output_guardrail async def math_guardrail( ctx: RunContextWrapper, agent: Agent, output: MessageOutput ) -> GuardrailFunctionOutput: result = await Runner.run(guardrail_agent, output.response, context=ctx.context) return GuardrailFunctionOutput( output_info=result.final_output, tripwire_triggered=result.final_output.is_math, ) agent = Agent( name="Customer support agent", instructions="You are a customer support agent. You help customers with their questions.", output_guardrails=[math_guardrail], output_type=MessageOutput, ) async def main(): # This should trip the guardrail try: await Runner.run(agent, "Hello, can you help me solve for x: 2x + 3 = 11?") print("Guardrail didn't trip - this is unexpected") except OutputGuardrailTripwireTriggered: print("Math output guardrail tripped")