项目概述

OpenAI Agents SDK 是一个轻量级但功能强大的框架，用于构建多智能体工作流。它主要利用大语言模型（LLM），通过配置智能体、交接、护栏和跟踪等功能，实现复杂的工作流管理。以下是对其各个部分运行过程和代码流程的详细介绍。

项目链接：OpenAI Agent SDK项目地址

核心架构

项目整体架构图

根据提供的代码库信息，OpenAI Agents SDK 项目的整体架构可以分为以下几个主要模块：

架构图说明：

1. 应用层（Application）：用户通过输入触发 Runner 开始执行代理流程。
2. 代理层（Agents）：
   • Agent 是核心组件，包含指令、工具、交接和护栏等配置。
   • Tool 是代理可以调用的工具，用于执行特定任务。
   • Handoff 允许代理将控制权转移给其他代理。
   • Guardrail 用于输入和输出的验证，确保代理的安全性。
3. 模型层（Models）：Agent 调用 Model 来生成响应。
4. 跟踪层（Tracing）：Runner 在执行过程中会生成跟踪信息，方便调试和优化。
5. 输出层（Output）：最终生成 RunResult 作为执行结果。

项目整体流程图

流程图说明：

1. 用户输入触发 Runner 初始化，并选择起始 Agent。
2. 对输入进行护栏检查，如果未通过则抛出异常。
3. 调用 Model 生成响应。
4. 判断是否有最终输出，如果有则进行输出护栏检查。
5. 如果没有最终输出，判断是否有交接或工具调用，根据情况进行相应处理。
6. 输出护栏检查通过后，返回 RunResult；否则抛出异常。

核心概念

1. 智能体（Agents）

智能体是配置了指令、工具、护栏和交接功能的大语言模型。在代码中，Agent 类是核心，它包含了智能体的基本信息。

from agents import Agent

agent = Agent(name="Assistant", instructions="You are a helpful assistant")

• 运行过程：当调用 Runner.run() 或 Runner.run_sync() 时，会启动一个循环，不断与 LLM 交互，直到得到最终输出。
• 代码流程：
  • Runner.run() 或 Runner.run_sync() 调用 RunImpl.execute_tools_and_side_effects() 方法。
  • 在 execute_tools_and_side_effects() 中，首先处理模型响应，包括解析工具调用、交接等信息。
  • 然后根据响应结果，决定是继续运行、进行交接还是返回最终输出。

2. 交接（Handoffs）

交接允许智能体将特定任务的控制权转移给其他智能体。

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]
)

• 运行过程：当 triage_agent 收到请求时，根据请求的语言将任务交接给 spanish_agent 或 english_agent。
• 代码流程：
  • 在 RunImpl.execute_tools_and_side_effects() 中，处理模型响应时，如果发现有交接信息，调用 execute_handoffs() 方法。
  • execute_handoffs() 方法会将控制权转移到新的智能体，并重新开始循环。

3. 护栏（Guardrails）

护栏是可配置的安全检查，用于输入和输出验证。

from agents import InputGuardrail, GuardrailFunctionOutput
from pydantic import BaseModel

class HomeworkOutput(BaseModel):
    is_homework: bool
    reasoning: str

guardrail_agent = Agent(
    name="Guardrail check",
    instructions="Check if the user is asking about homework.",
    output_type=HomeworkOutput
)

• 运行过程：在智能体运行过程中，护栏会对输入和输出进行验证，如果不符合条件，可能会中断流程。
• 代码流程：
  • 在 RunImpl.execute_tools_and_side_effects() 中，处理最终输出时，会根据输出类型进行验证。
  • 如果输出类型是结构化的，会调用 output_schema.validate_json() 方法进行验证。

4. 跟踪（Tracing）

跟踪是内置的智能体运行跟踪功能，可用于查看、调试和优化工作流。

from agents.tracing.setup import TraceProvider

trace_provider = TraceProvider()

• 运行过程：在智能体运行过程中，会创建跟踪信息并发送到指定的后端。
• 代码流程：
  • TraceProvider 类负责管理跟踪处理器。
  • 在创建跟踪或跨度时，会调用相应的处理器进行处理。
  • 处理器会在跟踪开始、结束，跨度开始、结束等事件发生时进行相应的操作。

项目使用步骤

1. 环境搭建

python -m venv env
source env/bin/activate

• 运行过程：创建一个虚拟环境并激活它，确保项目的依赖与系统环境隔离。
• 代码流程：这是 Python 标准的虚拟环境创建和激活步骤，由 Python 解释器和操作系统完成。

2. 安装 SDK

pip install openai-agents

• 运行过程：使用 pip 工具从 Python Package Index（PyPI）下载并安装 openai-agents 包。
• 代码流程：pip 会解析依赖关系，下载所需的包，并将其安装到虚拟环境中。

3. 示例代码

Hello world 示例

from agents import Agent, Runner

agent = Agent(name="Assistant", instructions="You are a helpful assistant")

result = Runner.run_sync(agent, "Write a haiku about recursion in programming.")
print(result.final_output)

• 运行过程：创建一个简单的智能体，调用 Runner.run_sync() 方法与 LLM 交互，获取最终输出并打印。
• 代码流程：
  • 创建 Agent 对象，设置名称和指令。
  • 调用 Runner.run_sync() 方法，该方法会调用 RunImpl.execute_tools_and_side_effects() 方法进行处理。
  • 在 execute_tools_and_side_effects() 中，与 LLM 交互，处理响应，直到得到最终输出。
  • 返回最终输出并打印。

交接示例

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],
)


async def main():
    result = await Runner.run(triage_agent, input="Hola, ¿cómo estás?")
    print(result.final_output)


if __name__ == "__main__":
    asyncio.run(main())

• 运行过程：创建三个智能体，其中 triage_agent 负责根据请求语言进行交接，调用 Runner.run() 方法处理请求，最终输出结果。
• 代码流程：
  • 创建三个 Agent 对象，设置名称、指令和交接信息。
  • 调用 Runner.run() 方法，该方法会调用 RunImpl.execute_tools_and_side_effects() 方法进行处理。
  • 在 execute_tools_and_side_effects() 中，处理模型响应时发现交接信息，调用 execute_handoffs() 方法将控制权转移到 spanish_agent。
  • spanish_agent 处理请求并返回最终输出。
  • 返回最终输出并打印。

函数示例

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 Tokyo?")
    print(result.final_output)


if __name__ == "__main__":
    asyncio.run(main())

• 运行过程：创建一个带有工具的智能体，调用 Runner.run() 方法处理请求，智能体调用工具获取天气信息并返回最终输出。
• 代码流程：
  • 使用 @function_tool 装饰器将 get_weather 函数转换为工具。
  • 创建 Agent 对象，设置名称、指令和工具。
  • 调用 Runner.run() 方法，该方法会调用 RunImpl.execute_tools_and_side_effects() 方法进行处理。
  • 在 execute_tools_and_side_effects() 中，处理模型响应时发现工具调用，调用 execute_function_tool_calls() 方法执行工具。
  • 工具执行后返回结果，继续与 LLM 交互，直到得到最终输出。
  • 返回最终输出并打印。

智能体循环

当调用 Runner.run() 时，会运行一个循环，直到得到最终输出。

# 伪代码表示循环过程
while True:
    # 1. 调用 LLM
    response = call_llm(agent, message_history)
    # 2. 处理响应
    processed_response = process_response(response)
    # 3. 检查是否有最终输出
    if processed_response.has_final_output():
        return processed_response.final_output
    # 4. 检查是否有交接
    if processed_response.has_handoff():
        agent = processed_response.handoff_agent
        continue
    # 5. 处理工具调用
    process_tool_calls(processed_response.tool_calls)
    # 6. 更新消息历史
    update_message_history(processed_response)

• 运行过程：
  1. 使用智能体的模型和设置以及消息历史调用 LLM。
  2. 处理 LLM 返回的响应，包括解析工具调用、交接等信息。
  3. 如果响应有最终输出，返回并结束循环。
  4. 如果响应有交接，切换到新的智能体并回到步骤 1。
  5. 处理工具调用（如果有）并追加工具响应消息，然后回到步骤 1。
• 代码流程：
  • 在 Runner.run() 方法中，调用 RunImpl.execute_tools_and_side_effects() 方法，该方法内部实现了循环逻辑。
  • 在 execute_tools_and_side_effects() 中，使用 ModelResponse 对象表示 LLM 的响应，使用 ProcessedResponse 对象表示处理后的响应。
  • 根据 ProcessedResponse 对象的属性，判断是否有最终输出、交接或工具调用，并进行相应的处理。

最终输出

最终输出是智能体在循环中产生的最后结果。

# 判断最终输出的伪代码
if agent.output_type:
    # 有输出类型，检查是否为结构化输出
    if response.has_structured_output(agent.output_type):
        final_output = response.structured_output
else:
    # 无输出类型，检查是否无工具调用或交接
    if not response.has_tool_calls() and not response.has_handoff():
        final_output = response.message

• 运行过程：
  1. 如果智能体设置了 output_type，最终输出是当 LLM 返回符合该类型的结构化输出时。
  2. 如果没有 output_type（即纯文本响应），则第一个没有任何工具调用或交接的 LLM 响应被视为最终输出。
• 代码流程：
  • 在 RunImpl.execute_tools_and_side_effects() 中，处理最终输出时，根据 output_schema 判断是否有输出类型。
  • 如果有输出类型，调用 output_schema.validate_json() 方法验证是否为结构化输出。
  • 如果没有输出类型，检查是否有工具调用或交接，如果没有则将消息作为最终输出。

开发流程

1. 安装 uv

uv --version

• 运行过程：检查 uv 工具是否安装。
• 代码流程：由操作系统执行命令，检查 uv 工具的版本信息。

2. 安装依赖

make sync

• 运行过程：使用 make 工具执行 sync 目标，安装项目的依赖。
• 代码流程：make 工具会读取 Makefile 文件，执行相应的命令，安装依赖。

3. 代码检查和测试

make tests  # run tests
make mypy   # run typechecker
make lint   # run linter

• 运行过程：
  • make tests：运行项目的测试代码，确保功能的正确性。
  • make mypy：运行类型检查器，检查代码中的类型错误。
  • make lint：运行代码检查工具，检查代码的风格和规范。
• 代码流程：
  • make tests 会执行 pytest 命令，运行 tests 目录下的测试代码。
  • make mypy 会执行 mypy 命令，对项目代码进行类型检查。
  • make lint 会执行 ruff 命令，对项目代码进行风格和规范检查。

总结

OpenAI Agents SDK 通过智能体、交接、护栏和跟踪等机制，构建了一个灵活、可扩展的多智能体工作流框架。其代码流程清晰，通过 Runner 和 RunImpl 类实现了智能体的运行和管理，通过 TraceProvider 类实现了跟踪功能，通过 Agent 类实现了智能体的配置和管理。开发者可以根据自己的需求，配置不同的智能体和工具，构建复杂的工作流。