第一个 Agent 程序

2026-04-16 AI Agent 编程 4.1k 字 15 分钟阅读
📑 目录
  1. 最小示例
  2. Runner 事件循环的底层机制
  3. 三个核心对象
    1. 1. Agent
    2. 2. Runner
    3. 3. RunResult
  4. 带工具的 Agent
  5. 多轮对话
  6. Agent 执行流程时序图
  7. 从同步到异步：执行模型的演进
  8. 下一步
  9. 完整实战示例：生产级 Agent 初始化模板
  10. 常见问题与调试
    1. 异步代码的测试陷阱
  11. 生产环境部署与性能优化
    1. 生产环境初始化的实践要点
    2. 健康检查端点的关键指标
    3. 并发请求处理的架构考量
    4. 运维团队的协作建议

最小示例

Runner 事件循环的底层机制

Runner.run() 表面上只是一个简单的异步调用，但内部实现了一个完整的状态机循环。当调用发生时，Runner 会执行以下步骤：

  1. 初始化运行上下文：创建 RunContextWrapper，注入用户提供的上下文对象和环境配置。
  2. 加载会话历史：如果提供了 session，从持久化存储中加载历史消息并构建输入列表。
  3. LLM 调用：将 system prompt、历史消息和当前输入打包发送给模型。
  4. 响应解析：解析模型的响应，判断是文本输出、工具调用请求还是 Handoff 请求。
  5. 工具执行：如果是工具调用，并发执行所有工具（Agents SDK 默认会并行执行同一轮中的所有工具调用），收集结果。
  6. 循环判断：将工具结果回传给模型，再次调用 LLM，重复 3-5 步，直到模型产生最终输出或达到 max_turns 限制。
  7. 输出封装：将最终结果包装为 RunResult，包含 final_output、完整的消息历史、追踪数据等。

这个循环的设计借鉴了 React 模式和 ReAct 论文中的"思考-行动-观察"范式。与 LangChain 的 AgentExecutor 相比，Agents SDK 的 Runner 更轻量，没有复杂的中间件链，但也因此缺乏一些开箱即用的重试和回退机制。

在性能层面，每次循环都会产生一次 LLM API 调用，这意味着延迟和 token 消耗都会随着对话轮数线性增长。对于复杂任务，max_turns 的设置需要在"任务完成度"和"成本可控性"之间做出权衡。

一个 Agents SDK 程序只需要三行核心代码：

from agents import Agent, Runner

agent = Agent(name="Assistant", instructions="Be helpful.")
result = Runner.run_sync(agent, "Hello!")
print(result.final_output)

三个核心对象

1. Agent

Agent 是配置了指令和能力的 LLM：

from agents import Agent

agent = Agent(
    name="Math Tutor",
    instructions="You help students with math problems. Explain step by step.",
    model="gpt-5-nano",
)

2. Runner

Runner 负责执行 Agent 的对话循环：

from agents import Runner

# 同步执行
result = Runner.run_sync(agent, "What is 2+2?")

# 异步执行
result = await Runner.run(agent, "What is 2+2?")

# 流式执行
result = await Runner.run_streamed(agent, "What is 2+2?")

3. RunResult

RunResult 包含执行结果和完整上下文：

print(result.final_output)      # 最终输出文本
print(result.raw_responses)     # 原始 LLM 响应
print(result.to_input_list())   # 转为下一轮输入

带工具的 Agent

from agents import Agent, Runner, function_tool

@function_tool
def get_weather(city: str) -> str:
    """Get weather for a city."""
    return f"The weather in {city} is sunny."

agent = Agent(
    name="Weather Bot",
    instructions="Help users with weather queries.",
    tools=[get_weather],
)

result = Runner.run_sync(agent, "What's the weather in Tokyo?")
print(result.final_output)

多轮对话

# 第一轮
result = Runner.run_sync(agent, "Hello, my name is Alice.")

# 第二轮：传入历史上下文
result2 = Runner.run_sync(agent, result.to_input_list() + [{"role": "user", "content": "What's my name?"}])
print(result2.final_output)  # "Your name is Alice."

Agent 执行流程时序图

下图展示了从用户输入到获得最终输出的完整执行时序：

sequenceDiagram
    participant U as 用户
    participant R as Runner
    participant A as Agent
    participant L as LLM API
    participant T as 工具函数

    U->>R: 输入: 北京今天天气如何?
    R->>A: 加载 Agent 配置
    A->>L: 发送 prompt
    L-->>A: 响应: 需要调用工具
    A->>T: 调用 weather_tool
    T-->>A: 返回: 25°C, 晴
    A->>L: 发送工具结果
    L-->>A: 最终回复
    A-->>R: 输出结果对象
    R-->>U: 返回最终输出

时序图中的关键洞察在于 Runner 的编排者角色：它负责管理 Agent 与外部系统之间的所有交互，开发者无需手动处理 HTTP 请求、重试逻辑或对话状态维护。

从同步到异步：执行模型的演进

初学者通常从 Runner.run_sync() 开始，这在原型验证阶段完全够用。但当业务需要处理并发请求时，同步模型的缺陷会迅速暴露：每个请求都会阻塞一个线程，Python 的 GIL 使得多线程无法真正并行执行 CPU 密集型任务。

切换到异步模型后，吞吐量可以提升一个数量级。Runner.run() 返回一个协程，可以被事件循环调度。在 FastAPI 等异步 Web 框架中，这意味着单个 worker 进程可以同时处理数百个并发的 Agent 请求，而不会耗尽线程池。

from fastapi import FastAPI
from agents import Agent, Runner

app = FastAPI()
agent = Agent(name="APIBot", instructions="简洁回答")

@app.post("/chat")
async def chat(message: str):
    result = await Runner.run(agent, message)
    return {"reply": result.final_output}

异步执行的另一个优势是资源复用。HTTP 连接池在异步模式下可以被多个请求共享，减少了 TCP 握手开销。对于需要频繁调用 OpenAI API 的应用，建议使用 aiohttp 的持久连接功能，将连接复用率提升到 90% 以上。

调试异步 Agent 时，常见的陷阱是忘记 await 协程。Python 3.11+ 的 asyncio 会在调试模式下发出警告，建议在开发环境设置 PYTHONASYNCIODEBUG=1 环境变量，及时发现未等待的协程和事件循环阻塞问题。

下一步

理解了基本用法后，让我们系统梳理 SDK 核心概念一览。

完整实战示例：生产级 Agent 初始化模板

以下是一个包含完整错误处理、日志记录和优雅关闭的生产级 Agent 模板：

import asyncio
import logging
import sys
from typing import Any

from agents import Agent, Runner, RunContextWrapper, function_tool


# 配置结构化日志
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    handlers=[logging.StreamHandler(sys.stdout)])

常见问题与调试

问题一：RuntimeError: Event loop is closed

在 Jupyter Notebook 或某些 Web 框架中，事件循环可能已被其他库管理。解决方案是使用 nest_asyncio：
python
import nest_asyncio
nest_asyncio.apply()


或者在 FastAPI 中直接使用异步路由，让框架管理事件循环。

**问题二：Agent 陷入无限循环**

当工具返回的结果不能让模型满意时，模型可能反复调用工具。调试方法：

1. 设置 `max_turns=5` 限制循环次数。
2. 在工具返回中添加明确的终止信号，如 `"NO_MORE_ACTION_NEEDED"`。
3. 查看 trace 中的完整消息历史，定位模型反复调用的原因。

**问题三：`RunResult` 为空或 None**

通常是因为模型输出了空字符串，或者工具异常导致流程中断。检查 `result.raw_responses` 和 `result.to_input_list()` 可以获取完整的调试信息。

**调试技巧**

Agents SDK 的 trace 系统是最强大的调试工具。每次运行都会产生一个 trace ID，你可以在 OpenAI 平台的 Trace Viewer 中查看完整的调用链、每个工具的执行耗时、以及每次 LLM 调用的输入输出。对于本地调试，可以临时设置环境变量 `OPENAI_AGENTS_DEBUG=1` 开启详细日志。

## 与其他方案对比

| 维度 | Agents SDK Runner | LangChain AgentExecutor | AutoGen ConversableAgent |
|------|-------------------|------------------------|--------------------------|
| 启动成本 | 低（单函数调用） | 中（需组装 chain） | 高（需注册多个 agent） |
| 循环控制 | 内置，可配 max_turns | 可自定义中间件 | 需自行实现终止条件 |
| 上下文注入 | 原生泛型支持 | 依赖 RunnableConfig | 通过 message 传递 |
| 调试体验 | Trace Viewer 可视化 | LangSmith 追踪 | 日志输出 |

对于从 0 开始构建 Agent 应用的开发者，Agents SDK 的 Runner 提供了最简洁的入口。LangChain 的灵活性更高，但需要理解 LCEL（LangChain Expression Language）的概念。AutoGen 的多 Agent 对话模式更强大，但单 Agent 运行的复杂度也更高。建议新手从 Agents SDK 开始，遇到复杂多 Agent 场景时再评估 AutoGen。

## Runner 执行生命周期与测试策略

<pre class="mermaid">sequenceDiagram
    participant U as 用户代码
    participant R as Runner
    participant LLM as OpenAI API
    participant T as 工具函数
    U->>R: run_sync(agent, input)
    R->>R: 创建 RunContextWrapper
    R->>LLM: 发送 system + user 消息
    LLM-->>R: 返回 assistant 消息
    alt 需要工具调用
        R->>T: 并发执行工具
        T-->>R: 返回结果
        R->>LLM: 发送 tool_result
        LLM-->>R: 返回最终回复
    end
    R-->>U: 返回 RunResult</pre>

该时序图清晰展现了 Runner 从接收输入到返回结果的全过程。理解这一生命周期对于编写健壮的测试用例至关重要，特别是工具调用和重试逻辑的测试。每次 LLM 往返都是一个潜在的失败点，测试覆盖必须穿透到这些边界。忽视了循环内部的中间状态，往往会导致线上问题难以复现，调试时陷入"黑盒猜测"的困境。

### Agent 程序的单元测试模式

直接调用 `Runner.run_sync` 会触发真实的 LLM API，这在单元测试中既不经济也不稳定。推荐采用**分层测试策略**：对工具函数进行纯单元测试，对 Agent 编排逻辑进行集成测试。以下是一个使用 `unittest.mock` 模拟 Runner 的示例：

```python
import unittest
from unittest.mock import patch, MagicMock
from agents import Agent, Runner


class TestAgentOrchestration(unittest.TestCase):
    def test_weather_agent_invokes_tool(self):
        """验证 Weather Agent 在收到天气查询时会调用 get_weather 工具。"""
        agent = Agent(
            name="Weather Bot",
            instructions="Help with weather.",
            tools=[self.mock_get_weather],
        )

        # Mock Runner.run_sync 的返回结果
        mock_result = MagicMock()
        mock_result.final_output = "The weather in Tokyo is sunny."
        mock_result.raw_responses = []

        with patch.object(Runner, "run_sync", return_value=mock_result):
            result = Runner.run_sync(agent, "Tokyo weather")

        self.assertIn("sunny", result.final_output)

    def mock_get_weather(self, city: str) -> str:
        return f"Mock weather for {city}"


class TestToolFunctions(unittest.TestCase):
    def test_get_weather_normalizes_city_name(self):
        """工具函数的纯单元测试：验证输入清洗逻辑。"""
        # 这里测试工具内部的业务逻辑，不涉及 LLM
        raw = "  tokyo  "
        normalized = raw.strip().title()
        self.assertEqual(normalized, "Tokyo")

分层测试的核心思想是将不可控的外部依赖（LLM API）与可控的业务逻辑（工具函数、Agent 配置）解耦。对于必须测试真实 LLM 交互的场景，建议单独维护一个标记为 slow 的 pytest 测试集，仅在 nightly build 中运行，避免拖慢日常开发的反馈循环。同时，应在 CI 中为 slow 测试配置独立的并发 runner，防止阻塞常规构建流水线。

异步代码的测试陷阱

Runner.run 返回协程对象，测试异步代码需要额外注意事件循环的管理。pytest 的 pytest-asyncio 插件是标准方案，但默认模式下每个测试用例会创建独立的事件循环，这与 Agents SDK 内部可能复用全局循环的行为存在差异。更稳健的做法是使用 asyncio.run() 在测试函数内部显式管理循环生命周期：

import asyncio
import pytest
from agents import Agent, Runner


@pytest.fixture
def echo_agent():
    return Agent(name="Echo", instructions="Repeat the user input.")


def test_async_runner_with_explicit_loop(echo_agent):
    """使用显式事件循环测试异步 Runner，避免与 pytest-asyncio 的循环策略冲突。"""
    async def _run():
        result = await Runner.run(echo_agent, "hello")
        return result.final_output

    output = asyncio.run(_run())
    assert "hello" in output.lower()

显式事件循环的好处在于测试行为与生产环境完全一致。当 Agent 内部注册了需要在循环关闭时清理的资源（如 HTTP 连接池），这种一致性尤为重要。建议在团队测试规范中统一约定异步测试的写法，避免因事件循环策略不一致导致的 flaky test。同时，对 RunResult 的断言不应只检查 final_output，还应验证 raw_responses 的轮数是否符合预期，以捕获潜在的无限循环风险。对于多轮对话场景，可以通过构造伪造的 raw_responses 来模拟工具调用循环，测试 Agent 在极端轮数下的行为稳定性。测试覆盖率的目标不应止步于行覆盖，而应追求分支覆盖和状态覆盖，确保每个可能的循环出口都被验证。在实际项目中，建议将测试金字塔的比例控制在单元测试七成、集成测试两成、端到端测试一成，以获得最佳的投入产出比。

生产环境部署与性能优化

生产环境初始化的实践要点

将本章节的技术应用到生产环境时，首要考虑的是稳定性与可观测性。建议采用渐进式 rollout 策略：先在开发环境验证核心逻辑，再迁移到预发布环境进行压力测试，最后才全量上线。部署过程中应配置完善的日志收集和指标监控，确保任何问题都能被快速发现和定位。

具体来说，需要在基础设施层面做好以下准备：容器资源限制（CPU/内存）、网络策略配置（防火墙规则、服务网格）、持久化存储选型（SSD vs 标准盘）以及备份恢复方案。对于高可用要求严格的场景，建议部署多实例并配置负载均衡，避免单点故障导致服务中断。

健康检查端点的关键指标

监控是生产系统的生命线。针对本章节涉及的功能，建议重点跟踪以下指标：请求延迟（P50/P95/P99）、错误率（4xx/5xx/超时）、吞吐量（QPS/TPS）以及资源利用率（CPU/内存/磁盘/网络）。这些指标应接入统一的监控大盘，并设置合理的告警阈值。

除了基础指标，还应关注业务层面的指标。例如功能成功率、用户满意度、成本消耗趋势等。通过将技术指标与业务指标关联分析，可以更准确地评估系统改进的实际价值，避免陷入"为了优化而优化"的陷阱。

并发请求处理的架构考量

随着业务规模增长，单实例部署很快会成为瓶颈。扩展性设计应在项目初期就纳入考量，而非事后补救。水平扩展通常比垂直扩展更具成本效益，但也引入了分布式系统的复杂性（数据一致性、服务发现、负载均衡等）。

在扩展过程中，建议遵循"无状态优先"原则：将状态外置到独立的存储层（如 Redis、PostgreSQL），使计算层可以随时水平扩容。对于无法避免的状态（如会话、缓存），采用分布式一致性协议或最终一致性模型来管理。定期进行容量规划和压力测试，确保系统在流量峰值时仍能稳定运行。

运维团队的协作建议

技术方案的落地离不开高效的团队协作。建议建立清晰的运维手册（Runbook），涵盖常见故障的诊断步骤、应急处理流程和升级路径。同时，通过定期的复盘会议，将线上事故转化为团队的学习素材，持续完善系统的健壮性。

在工具链方面，推荐将本章节的配置和脚本纳入版本控制（Git），并使用 Infrastructure as Code（IaC）工具（如 Terraform、Ansible）管理基础设施变更。这不仅能提高部署效率，还能确保环境一致性，减少"在我机器上能跑"的问题。

初学者常犯的错误是忽略异常处理。在生产环境中，API 密钥失效、网络超时、模型服务不可用都是常态。务必在代码外层包裹健壮的 try-catch 逻辑，并实现指数退避重试。

日志记录是排查线上问题的重要手段。建议在 Agent 运行链路的关键节点（输入接收、工具调用、输出生成）都打印结构化日志，方便后续通过日志平台快速定位异常。

在将 Agent 接入 Web 框架时，建议将 Agent 的运行逻辑封装为独立的服务层，而非直接嵌入路由处理函数中。这样可以使业务逻辑与 HTTP 协议解耦，便于后续单元测试和逻辑复用。

对于初学者来说，理解异步编程模型是掌握 Agents SDK 的关键。asyncio 的事件循环机制使得单线程也能处理高并发 I/O，这与传统的多线程模型有本质区别。