OpenAI Agents SDK完全教程

前言

2025年3月，OpenAI正式发布了Agents SDK（前身为Swarm实验项目），这是一个开源的轻量级多Agent编排框架。与LangChain、CrewAI等通用框架不同，OpenAI Agents SDK专注于为开发者提供最简洁、最Pythonic的方式来构建多Agent协作系统。本教程将从零开始，带你深入理解并掌握这一强大工具。

一、OpenAI Agents SDK概述

1.1 为什么需要Agents SDK？

在AI Agent开发领域，开发者面临的核心挑战包括：

• 多Agent协作：如何让多个Agent高效地分工合作
• 工具调用：如何安全地让Agent使用外部工具
• 安全控制：如何确保Agent行为在可控范围内
• 上下文管理：如何在多轮对话中维护状态

OpenAI Agents SDK正是为解决这些问题而设计的。它提供了四大核心原语：

1.2 与其他框架的对比

# 对比概览
comparison = {
    "OpenAI Agents SDK": {
        "定位": "轻量级多Agent编排",
        "特点": "原生OpenAI集成、简洁API",
        "适合": "已使用OpenAI API的团队"
    },
    "LangChain": {
        "定位": "通用LLM应用框架",
        "特点": "生态丰富、支持多模型",
        "适合": "需要灵活切换模型的项目"
    },
    "CrewAI": {
        "定位": "角色扮演式多Agent",
        "特点": "角色定义直观",
        "适合": "模拟团队协作场景"
    }
}

二、环境安装与配置

2.1 安装

# 推荐使用Python 3.10+
pip install openai-agents

# 或者从源码安装
git clone https://github.com/openai/openai-agents-python.git
cd openai-agents-python
pip install -e .

2.2 配置API密钥

# 方式一：环境变量（推荐）
export OPENAI_API_KEY="sk-your-api-key-here"

# 方式二：.env文件
echo 'OPENAI_API_KEY=sk-your-api-key-here' > .env

# 方式三：代码中直接设置（不推荐用于生产环境）
import os
os.environ["OPENAI_API_KEY"] = "sk-your-api-key-here"

2.3 验证安装

from agents import Agent, Runner

# 创建最简单的Agent
agent = Agent(
    name="测试助手",
    instructions="你是一个友好的AI助手。"
)

# 同步运行
result = Runner.run_sync(agent, "你好，请介绍一下你自己。")
print(result.final_output)

三、Agent基础架构详解

3.1 Agent核心参数

from agents import Agent, ModelSettings

agent = Agent(
    name="代码审查员",                    # Agent名称
    instructions="你是一位资深代码审查员...",  # 系统提示词
    model="gpt-4o",                       # 使用的模型
    model_settings=ModelSettings(          # 模型参数
        temperature=0.3,
        max_tokens=4096,
        top_p=0.95
    ),
    tools=[],                             # 工具列表
    handoffs=[],                          # 可切换的Agent列表
    input_guardrails=[],                  # 输入护栏
    output_guardrails=[],                 # 输出护栏
    output_type=None                      # 结构化输出类型
)

3.2 动态指令

指令可以是字符串，也可以是函数，实现动态系统提示：

def dynamic_instructions(context, agent):
    """根据上下文动态生成指令"""
    user_name = context.context.get("user_name", "用户")
    user_role = context.context.get("user_role", "普通用户")
    
    base_instruction = f"你是{user_name}的专属AI助手。"
    
    if user_role == "admin":
        base_instruction += "用户是管理员，你可以提供系统级操作建议。"
    else:
        base_instruction += "用户是普通用户，请提供基础帮助。"
    
    return base_instruction

agent = Agent(
    name="动态助手",
    instructions=dynamic_instructions,
)

3.3 结构化输出

from pydantic import BaseModel, Field
from typing import List

class CodeReviewResult(BaseModel):
    """代码审查结果"""
    score: int = Field(description="代码质量评分(1-10)", ge=1, le=10)
    issues: List[str] = Field(description="发现的问题列表")
    suggestions: List[str] = Field(description="改进建议")
    summary: str = Field(description="总体评价")

reviewer = Agent(
    name="代码审查员",
    instructions="审查用户提交的代码，给出详细评价。",
    output_type=CodeReviewResult,  # 指定结构化输出
)

# 运行
result = Runner.run_sync(reviewer, "请审查这段代码：\nprint('hello')")
review = result.final_output  # 自动解析为CodeReviewResult对象
print(f"评分: {review.score}/10")
print(f"问题: {review.issues}")

四、工具函数定义与调用

4.1 基本工具定义

使用@function_tool装饰器将普通Python函数转化为Agent可用的工具：

from agents import Agent, Runner, function_tool

@function_tool
def get_weather(city: str, unit: str = "celsius") -> str:
    """获取指定城市的天气信息。
    
    Args:
        city: 城市名称，如"北京"、"上海"
        unit: 温度单位，celsius或fahrenheit
    """
    # 模拟天气API调用
    weather_data = {
        "北京": {"temp": 22, "condition": "晴"},
        "上海": {"temp": 25, "condition": "多云"},
        "深圳": {"temp": 28, "condition": "阵雨"},
    }
    
    if city not in weather_data:
        return f"抱歉，暂不支持{city}的天气查询。"
    
    data = weather_data[city]
    temp = data["temp"]
    if unit == "fahrenheit":
        temp = temp * 9/5 + 32
    
    return f"{city}当前天气：{data['condition']}，温度{temp}{'°F' if unit == 'fahrenheit' else '°C'}"

@function_tool
def search_database(query: str, limit: int = 5) -> str:
    """搜索知识库。
    
    Args:
        query: 搜索关键词
        limit: 返回结果数量上限
    """
    # 模拟数据库搜索
    return f"找到{limit}条与'{query}'相关的结果。"

# 创建带工具的Agent
weather_agent = Agent(
    name="天气助手",
    instructions="你是一个天气查询助手，使用工具获取天气信息。",
    tools=[get_weather, search_database],
)

result = Runner.run_sync(weather_agent, "北京今天天气怎么样？")
print(result.final_output)

4.2 工具的高级用法

import httpx
from agents import function_tool

@function_tool
async def fetch_webpage(url: str) -> str:
    """抓取网页内容。
    
    Args:
        url: 要抓取的网页URL
    """
    async with httpx.AsyncClient() as client:
        response = await client.get(url, timeout=10)
        return response.text[:5000]  # 限制返回长度

@function_tool
def calculate(expression: str) -> str:
    """安全地计算数学表达式。
    
    Args:
        expression: 数学表达式，如 "2 + 3 * 4"
    """
    import ast
    import operator
    
    ops = {
        ast.Add: operator.add,
        ast.Sub: operator.sub,
        ast.Mult: operator.mul,
        ast.Div: operator.truediv,
    }
    
    try:
        tree = ast.parse(expression, mode='eval')
        # 安全地评估表达式
        result = _safe_eval(tree.body, ops)
        return str(result)
    except Exception as e:
        return f"计算错误: {e}"

五、Handoffs多Agent切换

Handoffs是Agents SDK最强大的特性之一，允许Agent在运行时将任务转交给其他Agent。

5.1 基本Handoff配置

from agents import Agent, Runner

# 定义专业Agent
chinese_agent = Agent(
    name="中文客服",
    instructions="你是中文客服专员，用中文回答用户问题。当用户需要技术支持时，转交给技术支持Agent。"
)

tech_agent = Agent(
    name="技术支持",
    instructions="你是技术支持工程师，帮助用户解决技术问题。"
)

sales_agent = Agent(
    name="销售顾问",
    instructions="你是销售顾问，帮助用户了解产品信息和价格。"
)

# 配置Handoff
chinese_agent.handoffs = [
    tech_agent,    # 简单handoff
    sales_agent,   # 简单handoff
]

# 运行
result = Runner.run_sync(chinese_agent, "我的产品出现了技术故障，怎么办？")
print(result.final_output)
# Agent会自动判断应该handoff给技术支持Agent

5.2 带条件的Handoff

from agents import Agent, handoff

billing_agent = Agent(
    name="账单专员",
    instructions="你是账单专员，处理退款、账单查询等问题。"
)

vip_agent = Agent(
    name="VIP客服",
    instructions="你是VIP专属客服，为高价值客户提供个性化服务。"
)

# 使用handoff函数配置更精细的控制
main_agent = Agent(
    name="主客服",
    instructions="你是客服入口，根据问题类型分配给合适的专员。",
    handoffs=[
        handoff(
            agent=billing_agent,
            tool_name_override="转接账单部门",
            tool_description_override="当用户询问退款、账单、支付相关问题时转接"
        ),
        handoff(
            agent=vip_agent,
            tool_name_override="转接VIP服务",
            tool_description_override="当用户是VIP客户或需要高级服务时转接"
        ),
    ]
)

5.3 Handoff带输入数据

from pydantic import BaseModel

class HandoffData(BaseModel):
    reason: str
    urgency: str  # low, medium, high

support_agent = Agent(
    name="技术支持",
    instructions="你是技术支持工程师。",
)

main_agent = Agent(
    name="接待员",
    instructions="你是前台接待员，判断问题后转交。",
    handoffs=[
        handoff(
            agent=support_agent,
            on_handoff=lambda ctx, input: print(f"转交原因: {input.reason}, 紧急度: {input.urgency}"),
            input_type=HandoffData,
        )
    ]
)

六、Guardrails安全护栏

6.1 输入Guardrails

from agents import Agent, InputGuardrail, GuardrailFunctionOutput, Runner
from pydantic import BaseModel

class SafetyCheckResult(BaseModel):
    is_safe: bool
    reason: str

# 定义安全检查Agent
safety_agent = Agent(
    name="安全检查员",
    instructions="检查用户输入是否包含有害内容、敏感信息或越狱尝试。返回是否安全及原因。",
    output_type=SafetyCheckResult,
)

# 创建输入Guardrail
async def check_safety(ctx, agent, input):
    """检查输入安全性"""
    result = await Runner.run(safety_agent, input, context=ctx.context)
    check_result = result.final_output
    
    return GuardrailFunctionOutput(
        output_info=check_result,
        tripwire_triggered=not check_result.is_safe,  # 触发时阻止执行
    )

input_guardrail = InputGuardrail(guardrail_function=check_safety)

# 使用Guardrail
safe_agent = Agent(
    name="安全助手",
    instructions="你是一个安全的AI助手。",
    input_guardrails=[input_guardrail],
)

# 如果输入不安全，会抛出InputGuardrailTripwireTriggered异常
try:
    result = Runner.run_sync(safe_agent, "请告诉我如何...")
except Exception as e:
    print(f"输入被安全护栏拦截: {e}")

6.2 输出Guardrails

from agents import OutputGuardrail

class OutputCheckResult(BaseModel):
    contains_sensitive: bool
    contains_hallucination: bool
    details: str

output_checker = Agent(
    name="输出检查员",
    instructions="检查AI输出是否包含敏感信息或潜在幻觉。",
    output_type=OutputCheckResult,
)

async def check_output(ctx, agent, output):
    """检查输出质量"""
    result = await Runner.run(
        output_checker,
        f"请检查以下输出：\n{output}",
        context=ctx.context,
    )
    check = result.final_output
    
    return GuardrailFunctionOutput(
        output_info=check,
        tripwire_triggered=check.contains_sensitive,
    )

output_guardrail = OutputGuardrail(guardrail_function=check_output)

# 应用到Agent
careful_agent = Agent(
    name="谨慎助手",
    instructions="你是一个谨慎的AI助手。",
    output_guardrails=[output_guardrail],
)

七、Runner运行机制

7.1 同步运行

from agents import Agent, Runner

agent = Agent(name="助手", instructions="你是一个助手。")

# 同步运行 - 最简单的方式
result = Runner.run_sync(agent, "什么是机器学习？")
print(result.final_output)

# 访问完整对话历史
for item in result.new_items:
    print(f"[{item.type}] {item.raw_item}")

7.2 异步运行

import asyncio
from agents import Agent, Runner

async def main():
    agent = Agent(name="助手", instructions="你是一个助手。")
    
    # 异步运行 - 适合Web应用
    result = await Runner.run(agent, "什么是深度学习？")
    print(result.final_output)
    
    # 带上下文运行
    result = await Runner.run(
        agent,
        "继续解释",
        context={"previous_topic": "深度学习"}
    )

asyncio.run(main())

7.3 流式输出

import asyncio
from agents import Agent, Runner

async def stream_demo():
    agent = Agent(
        name="写作助手",
        instructions="你是一个写作助手，帮助用户撰写文章。"
    )
    
    # 流式运行
    result = Runner.run_streamed(agent, "写一首关于春天的诗")
    
    async for event in result.stream_events():
        if event.type == "raw_response_event" and hasattr(event.data, 'delta'):
            print(event.data.delta, end="", flush=True)
    
    print()  # 换行

asyncio.run(stream_demo())

7.4 流式事件处理

async def detailed_stream():
    agent = Agent(name="助手", instructions="你是一个助手。")
    result = Runner.run_streamed(agent, "解释量子计算")
    
    async for event in result.stream_events():
        # 处理不同类型的事件
        if event.type == "raw_response_event":
            # LLM原始输出
            pass
        elif event.type == "run_item_stream_event":
            # 运行项事件（工具调用、handoff等）
            if event.item.type == "tool_call_item":
                print(f"\n[调用工具] {event.item.raw_item.name}")
            elif event.item.type == "handoff_call_item":
                print(f"\n[切换Agent] -> {event.item.target_agent.name}")
        elif event.type == "agent_updated_stream_event":
            print(f"\n[Agent更新] 当前Agent: {event.new_agent.name}")

八、上下文管理与会话记忆

8.1 Context对象

from agents import Agent, Runner, RunContextWrapper
from dataclasses import dataclass

@dataclass
class UserContext:
    user_id: str
    user_name: str
    session_history: list
    preferences: dict

# 在Agent指令中使用上下文
def get_instructions(ctx: RunContextWrapper[UserContext], agent: Agent) -> str:
    user = ctx.context
    return f"""你是{user.user_name}的AI助手。
用户偏好：{user.preferences}
请根据用户偏好调整你的回答风格。"""

agent = Agent(
    name="个性化助手",
    instructions=get_instructions,
)

# 运行时传入上下文
context = UserContext(
    user_id="user_123",
    user_name="张三",
    session_history=[],
    preferences={"language": "中文", "style": "简洁"}
)

result = Runner.run_sync(agent, "推荐一本书", context=context)
print(result.final_output)

8.2 实现会话记忆

from agents import Agent, Runner, ItemHelpers
from typing import List, Dict

class ConversationMemory:
    """会话记忆管理器"""
    
    def __init__(self, max_history: int = 20):
        self.history: List[Dict] = []
        self.max_history = max_history
    
    def add_message(self, role: str, content: str):
        self.history.append({"role": role, "content": content})
        # 保持历史在限制范围内
        if len(self.history) > self.max_history:
            self.history = self.history[-self.max_history:]
    
    def get_messages(self) -> List[Dict]:
        return self.history.copy()

class ChatSession:
    """聊天会话"""
    
    def __init__(self, agent: Agent):
        self.agent = agent
        self.memory = ConversationMemory()
        self.conversation_id = None
    
    async def chat(self, user_message: str) -> str:
        # 添加用户消息到记忆
        self.memory.add_message("user", user_message)
        
        # 构建输入（包含历史）
        input_messages = self.memory.get_messages()
        
        # 运行Agent
        result = await Runner.run(
            self.agent,
            input_messages,
        )
        
        # 保存Agent回复到记忆
        assistant_reply = result.final_output
        self.memory.add_message("assistant", assistant_reply)
        
        # 保存conversation_id用于恢复
        self.conversation_id = result.conversation_id
        
        return assistant_reply

# 使用示例
async def chat_demo():
    agent = Agent(
        name="记忆助手",
        instructions="你是一个有记忆力的AI助手，能记住之前的对话内容。"
    )
    
    session = ChatSession(agent)
    
    # 多轮对话
    print(await session.chat("我叫小明，我喜欢编程"))
    print(await session.chat("我之前说了什么？"))
    print(await session.chat("你还记得我的名字吗？"))

九、实战案例一：智能客服系统

9.1 系统设计

"""
智能客服系统 - 多Agent协作
包含：前台接待、技术支持、账单服务、升级处理
"""
from agents import Agent, Runner, function_tool, handoff, InputGuardrail, GuardrailFunctionOutput
from pydantic import BaseModel
from typing import Optional
import asyncio

# ============ 数据模型 ============

class CustomerInfo(BaseModel):
    customer_id: str
    name: str
    vip_level: int = 0
    issue_history: list = []

class TicketInfo(BaseModel):
    ticket_id: str
    category: str  # tech, billing, general
    priority: str  # low, medium, high
    description: str

# ============ 工具函数 ============

@function_tool
def lookup_customer(customer_id: str) -> str:
    """查询客户信息。
    
    Args:
        customer_id: 客户ID
    """
    # 模拟数据库查询
    customers = {
        "C001": {"name": "张三", "vip": 3, "history": ["退款申请", "技术咨询"]},
        "C002": {"name": "李四", "vip": 1, "history": ["产品咨询"]},
    }
    if customer_id in customers:
        info = customers[customer_id]
        return f"客户: {info['name']}, VIP等级: {info['vip']}, 历史: {info['history']}"
    return "未找到该客户信息"

@function_tool
def create_ticket(customer_id: str, category: str, description: str) -> str:
    """创建工单。
    
    Args:
        customer_id: 客户ID
        category: 工单分类 (tech/billing/general)
        description: 问题描述
    """
    import random
    ticket_id = f"TK-{random.randint(10000, 99999)}"
    return f"工单已创建: {ticket_id}, 分类: {category}"

@function_tool
def check_order_status(order_id: str) -> str:
    """查询订单状态。
    
    Args:
        order_id: 订单ID
    """
    statuses = {
        "ORD001": "已发货，预计明天到达",
        "ORD002": "处理中，预计3天内发货",
    }
    return statuses.get(order_id, "未找到该订单")

# ============ 安全护栏 ============

profanity_agent = Agent(
    name="内容审核",
    instructions="检查输入是否包含辱骂、威胁或不当内容。返回is_safe和reason。",
    output_type=BaseModel.from_fields(
        "SafetyResult",
        is_safe=(bool, ...),
        reason=(str, ...)
    ),
)

async def customer_input_guardrail(ctx, agent, input):
    result = await Runner.run(profanity_agent, input)
    return GuardrailFunctionOutput(
        output_info=result.final_output,
        tripwire_triggered=not result.final_output.is_safe,
    )

# ============ 专业Agent ============

tech_support = Agent(
    name="技术支持",
    instructions="""你是技术支持工程师。
- 诊断技术问题，提供解决方案
- 如果问题复杂，创建技术工单
- 保持专业但友好的语气""",
    tools=[create_ticket],
)

billing_support = Agent(
    name="账单服务",
    instructions="""你是账单服务专员。
- 处理退款、账单查询、支付问题
- 查询订单状态
- 如需退款，创建账单工单""",
    tools=[check_order_status, create_ticket],
)

escalation_agent = Agent(
    name="升级处理",
    instructions="""你是客服主管。
- 处理需要升级的复杂问题
- 处理VIP客户的特殊需求
- 处理投诉和不满""",
    tools=[create_ticket],
)

# ============ 主客服Agent ============

customer_service = Agent(
    name="智能客服",
    instructions="""你是智能客服系统的前台接待员。
你的职责：
1. 首先问候客户，了解问题
2. 查询客户信息（使用lookup_customer工具）
3. 根据问题类型转接给合适的专员：
   - 技术问题 -> 技术支持
   - 账单/订单问题 -> 账单服务
   - 复杂问题/投诉 -> 升级处理
4. VIP3+客户优先转接升级处理

始终保持友好、耐心的态度。""",
    tools=[lookup_customer],
    handoffs=[tech_support, billing_support, escalation_agent],
    input_guardrails=[InputGuardrail(guardrail_function=customer_input_guardrail)],
)

# ============ 运行客服系统 ============

async def customer_service_demo():
    """演示客服系统"""
    print("=" * 50)
    print("智能客服系统已启动")
    print("=" * 50)
    
    # 模拟对话
    conversations = [
        "你好，我的订单ORD001还没收到，能帮我查一下吗？",
        "我的产品出现了蓝屏问题，怎么解决？",
        "我要投诉！你们的服务太差了！",
    ]
    
    for msg in conversations:
        print(f"\n客户: {msg}")
        result = await Runner.run(customer_service, msg)
        print(f"客服: {result.final_output}")
        print("-" * 30)

# asyncio.run(customer_service_demo())

十、实战案例二：研究助手系统

10.1 完整研究助手

"""
研究助手系统 - 帮助用户进行深度研究
包含：研究规划、信息收集、分析总结
"""
import asyncio
from agents import Agent, Runner, function_tool
from pydantic import BaseModel
from typing import List

# ============ 工具 ============

@function_tool
def search_papers(query: str, max_results: int = 5) -> str:
    """搜索学术论文。
    
    Args:
        query: 搜索关键词
        max_results: 最大返回数量
    """
    # 模拟论文搜索
    papers = [
        f"论文{i+1}: 《{query}的最新研究进展》 - 发表于2024年"
        for i in range(max_results)
    ]
    return "\n".join(papers)

@function_tool
def search_web(query: str) -> str:
    """搜索互联网信息。
    
    Args:
        query: 搜索关键词
    """
    return f"搜索'{query}'的结果：找到相关信息10条。"

@function_tool
def save_research_notes(topic: str, notes: str) -> str:
    """保存研究笔记。
    
    Args:
        topic: 研究主题
        notes: 笔记内容
    """
    # 实际应用中会保存到数据库
    return f"笔记已保存 - 主题: {topic}"

class ResearchReport(BaseModel):
    """研究报告结构"""
    title: str
    summary: str
    key_findings: List[str]
    sources: List[str]
    recommendations: List[str]

# ============ 研究Agent ============

research_planner = Agent(
    name="研究规划师",
    instructions="""你是研究规划师，负责：
1. 理解用户的研究需求
2. 制定研究计划和步骤
3. 确定需要搜索的关键词
4. 输出结构化的研究计划""",
    output_type=BaseModel.from_fields(
        "ResearchPlan",
        topic=(str, ...),
        steps=(List[str], ...),
        keywords=(List[str], ...),
    ),
)

information_collector = Agent(
    name="信息收集员",
    instructions="""你是信息收集专员，负责：
1. 使用搜索工具收集相关信息
2. 整理和筛选有价值的信息
3. 保存重要发现到研究笔记""",
    tools=[search_papers, search_web, save_research_notes],
)

research_analyst = Agent(
    name="研究分析师",
    instructions="""你是研究分析师，负责：
1. 分析收集到的信息
2. 提炼关键发现
3. 撰写研究报告""",
    output_type=ResearchReport,
)

# ============ 主研究Agent ============

research_assistant = Agent(
    name="研究助手",
    instructions="""你是智能研究助手，协调研究流程：
1. 首先理解用户的研究需求
2. 制定研究计划
3. 收集相关信息
4. 分析并输出研究报告

你可以将子任务分配给专业Agent。""",
    handoffs=[research_planner, information_collector, research_analyst],
)

async def research_demo():
    """研究助手演示"""
    print("研究助手启动...\n")
    
    result = await Runner.run(
        research_assistant,
        "帮我研究一下2024年大语言模型的最新发展趋势"
    )
    
    print("研究报告：")
    print(result.final_output)

# asyncio.run(research_demo())

十一、性能优化与成本控制

11.1 Token使用监控

from agents import Agent, Runner, Usage

async def monitor_usage():
    agent = Agent(name="助手", instructions="你是一个助手。")
    
    result = await Runner.run(agent, "简要介绍人工智能。")
    
    # 查看token使用情况
    usage = result.context_wrapper.usage
    print(f"输入Token: {usage.input_tokens}")
    print(f"输出Token: {usage.output_tokens}")
    print(f"总计Token: {usage.total_tokens}")
    
    # 计算成本 (GPT-4o价格)
    input_cost = usage.input_tokens * 0.0025 / 1000
    output_cost = usage.output_tokens * 0.01 / 1000
    total_cost = input_cost + output_cost
    print(f"预估成本: ${total_cost:.4f}")

11.2 降低成本的策略

# 策略1: 使用小模型处理简单任务
simple_agent = Agent(
    name="简单助手",
    instructions="你是一个助手。",
    model="gpt-4o-mini",  # 使用更便宜的模型
)

# 策略2: 限制输出长度
agent = Agent(
    name="简洁助手",
    instructions="用不超过100字回答问题。",  # 通过指令控制
)

# 策略3: 缓存机制
from functools import lru_cache

@lru_cache(maxsize=100)
def cached_query(query: str) -> str:
    """缓存常见查询结果"""
    # 实际实现中会调用Agent
    pass

# 策略4: 批量处理
async def batch_process(queries: list):
    """批量处理多个查询"""
    agent = Agent(name="助手", instructions="你是一个助手。")
    tasks = [Runner.run(agent, q) for q in queries]
    results = await asyncio.gather(*tasks)
    return [r.final_output for r in results]

11.3 生产环境配置

from agents import Agent, ModelSettings

# 生产环境推荐配置
production_agent = Agent(
    name="生产助手",
    instructions="...",
    model="gpt-4o",
    model_settings=ModelSettings(
        temperature=0.1,        # 低温度，输出更稳定
        max_tokens=2048,        # 限制输出长度
        top_p=0.9,
    ),
    # 设置最大轮次防止无限循环
    # 通过Runner的max_turns参数控制
)

# 使用时限制轮次
async def safe_run(agent, message, max_turns=10):
    """安全运行，防止无限循环"""
    result = await Runner.run(
        agent,
        message,
        max_turns=max_turns,
    )
    return result.final_output

十二、与LangChain/CrewAI对比

12.1 代码风格对比

# ===== OpenAI Agents SDK =====
from agents import Agent, Runner

agent = Agent(name="助手", instructions="...", tools=[my_tool])
result = Runner.run_sync(agent, "你好")
print(result.final_output)

# ===== LangChain =====
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")
agent = create_openai_tools_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools)
result = executor.invoke({"input": "你好"})
print(result["output"])

# ===== CrewAI =====
from crewai import Agent, Task, Crew

agent = Agent(role="助手", goal="帮助用户", backstory="...", tools=[my_tool])
task = Task(description="你好", agent=agent)
crew = Crew(agents=[agent], tasks=[task])
result = crew.kickoff()

12.2 选型建议

十三、企业级部署方案

13.1 FastAPI集成

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from agents import Agent, Runner
import asyncio

app = FastAPI()

# 全局Agent配置
agent = Agent(
    name="API助手",
    instructions="你是一个API服务的AI助手。"
)

class ChatRequest(BaseModel):
    message: str
    conversation_id: str = None

class ChatResponse(BaseModel):
    reply: str
    conversation_id: str

@app.post("/chat", response_model=ChatResponse)
async def chat_endpoint(request: ChatRequest):
    try:
        result = await Runner.run(
            agent,
            request.message,
            max_turns=5,
        )
        return ChatResponse(
            reply=result.final_output,
            conversation_id=result.conversation_id or "",
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

# 健康检查
@app.get("/health")
async def health():
    return {"status": "ok"}

13.2 日志与监控

import logging
from agents import Agent, Runner

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("agents")

async def logged_run(agent, message):
    """带日志的Agent运行"""
    logger.info(f"Agent '{agent.name}' 收到消息: {message[:100]}...")
    
    try:
        result = await Runner.run(agent, message)
        logger.info(f"Agent '{agent.name}' 响应完成, tokens: {result.context_wrapper.usage}")
        return result.final_output
    except Exception as e:
        logger.error(f"Agent '{agent.name}' 运行失败: {e}")
        raise

十四、总结

OpenAI Agents SDK为开发者提供了一个简洁、高效的多Agent开发框架。通过本教程，你已经掌握了：

1. Agent基础架构：创建和配置Agent，定义指令和工具
2. Handoffs机制：实现多Agent之间的任务切换
3. Guardrails安全：构建输入输出的安全防护
4. 工具系统：定义和管理Agent可调用的工具函数
5. Runner运行：同步、异步和流式三种运行方式
6. 上下文管理：维护会话状态和用户上下文
7. 实战案例：客服系统和研究助手的完整实现
8. 生产部署：企业级应用的部署和优化方案

推荐学习路径

入门 → Agent基础 → 工具定义 → Handoffs → Guardrails → 流式输出 → 企业部署

官方资源

• GitHub仓库：openai/openai-agents-python
• 官方文档：OpenAI Agents SDK Documentation
• 示例代码：examples/目录

本教程持续更新中，如有问题欢迎反馈。