OpenAI Agent SDK 技能

使用 OpenAI Agents SDK 构建多 Agent AI 系统。轻量级但强大的框架，支持 Agent 协作、工具调用、交接机制和追踪调试。

何时使用此技能

当用户需要：

• - 构建多 Agent 协作系统
• 实现 Agent 之间的任务交接
• 为 Agent 添加工具（function calling）
• 创建智能路由和编排系统
• 实现人机协作（Human in the Loop）
• 构建语音 Agent（Realtime Agents）
• 需要追踪和调试 Agent 工作流

核心概念

1. Agent（智能体）

配置了指令、工具、防护栏和交接能力的 LLM。

python
from agents import Agent

agent = Agent(
name=Assistant,
instructions=You are a helpful assistant,
tools=[...], # 可选：工具列表
handoffs=[...], # 可选：可交接的其他 agent
model=gpt-4, # 可选：模型选择
)

2. Tools（工具）

让 Agent 执行动作的函数。

python
from agents import function_tool
from typing import Annotated

@function_tool
def get_weather(city: Annotated[str, 城市名称]) -> str:
获取指定城市的天气信息
return f{city}：晴朗，20°C

agent = Agent(tools=[get_weather])

3. Runner（运行器）

执行 Agent 的引擎，支持三种运行模式。

python
from agents import Runner

同步运行

result = Runner.run_sync(agent, Hello)

异步运行（推荐）

result = await Runner.run(agent, Hello)

流式运行

result = Runner.run_streamed(agent, Hello) async for event in result.stream_events(): print(event)

4. Handoffs（交接）

Agent 之间传递任务的机制。

python
french_agent = Agent(name=French, instructions=只说法语)
spanish_agent = Agent(name=Spanish, instructions=只说西班牙语)

triage_agent = Agent(
name=Triage,
instructions=根据语言分配给合适的 agent,
handoffs=[frenchagent, spanishagent],
)

5. Agents as Tools（Agent 作为工具）

将 Agent 封装为工具供其他 Agent 调用。

python
translator = Agent(name=Translator, ...)

orchestrator = Agent(
name=Orchestrator,
tools=[
translator.as_tool(
tool_name=translate,
tool_description=翻译文本
)
],
)

6. Guardrails（防护栏）

输入输出验证和安全检查。

python
from agents import input_guardrail, GuardrailFunctionOutput

@input_guardrail
def check_input(ctx, agent, input):
if 敏感词 in input:
return GuardrailFunctionOutput(
output_info=包含敏感内容,
tripwire_triggered=True
)
return GuardrailFunctionOutput(
output_info=验证通过,
tripwire_triggered=False
)

agent = Agent(inputguardrails=[checkinput])

7. Tracing（追踪）

内置的运行追踪和调试。

python
from agents import trace

with trace(我的工作流):
result = await Runner.run(agent, 任务)

8. Sessions（会话）

自动管理对话历史。

python
from agents import Session

session = Session()
result1 = await Runner.run(agent, 问题1, session=session)
result2 = await Runner.run(agent, 问题2, session=session) # 记住问题1的上下文

快速开始

安装

bash

使用 pip

pip install openai-agents

使用 uv

uv add openai-agents

带语音支持

pip install openai-agents[voice]

带 Redis 会话支持

pip install openai-agents[redis]

设置环境变量

bash
export OPENAIAPIKEY=sk-your-api-key-here

Hello World 示例

python
import asyncio
from agents import Agent, Runner

async def main():
agent = Agent(
name=Assistant,
instructions=You are a helpful assistant,
)

result = await Runner.run(agent, Write a haiku about programming.)
print(result.final_output)

if name == main:
asyncio.run(main())

常见设计模式

模式 1: 路由（Routing）

根据输入类型分配给不同的专业 Agent。

python
sales_agent = Agent(name=Sales, instructions=处理销售相关咨询)
support_agent = Agent(name=Support, instructions=处理技术支持)
technical_agent = Agent(name=Technical, instructions=处理技术问题)

triage_agent = Agent(
name=Triage,
instructions=根据用户请求类型分配给合适的 agent,
handoffs=[salesagent, supportagent, technical_agent],
)

模式 2: 编排者（Orchestrator）

一个主 Agent 协调多个专业 Agent。

python
researcher = Agent(name=Researcher, instructions=搜索和收集信息)
analyst = Agent(name=Analyst, instructions=分析数据)
writer = Agent(name=Writer, instructions=撰写报告)

orchestrator = Agent(
name=Orchestrator,
instructions=协调各个专业 agent 完成复杂任务,
tools=[
researcher.astool(toolname=research, tool_description=搜索信息),
analyst.astool(toolname=analyze, tool_description=分析数据),
writer.astool(toolname=write, tool_description=撰写内容),
],
)

模式 3: 并行执行（Parallelization）

多个 Agent 同时执行独立任务。

python
import asyncio

async def parallel_workflow():
results = await asyncio.gather(
Runner.run(agent1, input),
Runner.run(agent2, input),
Runner.run(agent3, input),
)
return results

模式 4: 人机协作（Human in the Loop）

在关键决策点请求人工确认。

python
from agents import Agent, Runner

agent = Agent(
name=Assistant,
instructions=帮助用户做决策，在重要决策前请求确认,
)

使用流式输出来实现实时交互

result = Runner.run_streamed(agent, 帮我规划旅行) async for event in result.stream_events(): # 检测需要人工确认的事件 if needshumanapproval(event): approval = await gethumaninput() # 继续执行

高级功能

使用其他 LLM

SDK 支持通过 LiteLLM 接入 100+ LLM。

python
from agents import Agent
from openai import AsyncOpenAI

使用 Anthropic Claude

claude_client = AsyncOpenAI( api_key=your-anthropic-key, base_url=https://api.anthropic.com/v1 )

agent = Agent(
name=ClaudeAgent,
model=claude-3-opus-20240229,
modelsettings={client: claudeclient}
)

Realtime Agents（语音 Agent）

python
from agents import Agent, Runner
from agents.voice import VoiceAgent

voice_agent = VoiceAgent(
name=VoiceAssistant,
instructions=You are a helpful voice assistant,
)

实时语音交互

result = await Runner.run(voiceagent, audioinput)

MCP（Model Context Protocol）

python
from agents import Agent
from agents.mcp import MCPServer

mcp_server = MCPServer(path/to/mcp/config)

agent = Agent(
name=MCPAgent,
mcpservers=[mcpserver],
)

最佳实践

1. 异步优先

推荐使用 await Runner.run() 而非 Runner.run_sync()。

2. 类型提示

使用 Pydantic 和 Annotated 提供清晰的类型信息。

python
from pydantic import BaseModel, Field

class WeatherOutput(BaseModel):
city: str = Field(description=城市名称)
temperature: int = Field(description=温度（摄氏度）)
conditions: str = Field(description=天气状况)

@function_tool
def get_weather(city: str) -> WeatherOutput:
return WeatherOutput(city=city, temperature=20, conditions=晴朗)

3. 工具描述

为工具提供清晰的 docstring 和参数描述。

python
@function_tool
def calculate(
expression: Annotated[str, 数学表达式，如 2+2 或 sqrt(16)]
) -> float:

计算数学表达式的结果。
支持基本运算、三角函数、对数等。

return eval(expression)

4. 追踪调试

使用 with trace() 包裹工作流以便调试。

python
from agents import trace

with trace(客户服务流程):
result = await Runner.run(customerserviceagent, user