Lord1Egypt 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。
from agents import Agent
agent = Agent(
name="Assistant",
instructions="You are a helpful assistant",
tools=[...], # 可选：工具列表
handoffs=[...], # 可选：可交接的其他 agent
model="gpt-4", # 可选：模型选择
)
2. Tools（工具）
让 Agent 执行动作的函数。
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 的引擎，支持三种运行模式。
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 之间传递任务的机制。
french_agent = Agent(name="French", instructions="只说法语")
spanish_agent = Agent(name="Spanish", instructions="只说西班牙语")
triage_agent = Agent(
name="Triage",
instructions="根据语言分配给合适的 agent",
handoffs=[french_agent, spanish_agent],
)
5. Agents as Tools（Agent 作为工具）
将 Agent 封装为工具供其他 Agent 调用。
translator = Agent(name="Translator", ...)
orchestrator = Agent(
name="Orchestrator",
tools=[
translator.as_tool(
tool_name="translate",
tool_description="翻译文本"
)
],
)
6. Guardrails（防护栏）
输入输出验证和安全检查。
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(input_guardrails=[check_input])
7. Tracing（追踪）
内置的运行追踪和调试。
from agents import trace
with trace("我的工作流"):
result = await Runner.run(agent, "任务")
8. Sessions（会话）
自动管理对话历史。
from agents import Session
session = Session()
result1 = await Runner.run(agent, "问题1", session=session)
result2 = await Runner.run(agent, "问题2", session=session) # 记住问题1的上下文
快速开始
安装
# 使用 pip
pip install openai-agents
# 使用 uv
uv add openai-agents
# 带语音支持
pip install 'openai-agents[voice]'
# 带 Redis 会话支持
pip install 'openai-agents[redis]'
设置环境变量
export OPENAI_API_KEY="sk-your-api-key-here"
Hello World 示例
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。
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=[sales_agent, support_agent, technical_agent],
)
模式 2: 编排者（Orchestrator）
一个主 Agent 协调多个专业 Agent。
researcher = Agent(name="Researcher", instructions="搜索和收集信息")
analyst = Agent(name="Analyst", instructions="分析数据")
writer = Agent(name="Writer", instructions="撰写报告")
orchestrator = Agent(
name="Orchestrator",
instructions="协调各个专业 agent 完成复杂任务",
tools=[
researcher.as_tool(tool_name="research", tool_description="搜索信息"),
analyst.as_tool(tool_name="analyze", tool_description="分析数据"),
writer.as_tool(tool_name="write", tool_description="撰写内容"),
],
)
模式 3: 并行执行（Parallelization）
多个 Agent 同时执行独立任务。
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）
在关键决策点请求人工确认。
from agents import Agent, Runner
agent = Agent(
name="Assistant",
instructions="帮助用户做决策，在重要决策前请求确认",
)
# 使用流式输出来实现实时交互
result = Runner.run_streamed(agent, "帮我规划旅行")
async for event in result.stream_events():
# 检测需要人工确认的事件
if needs_human_approval(event):
approval = await get_human_input()
# 继续执行
高级功能
使用其他 LLM
SDK 支持通过 LiteLLM 接入 100+ LLM。
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",
model_settings={"client": claude_client}
)
Realtime Agents（语音 Agent）
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(voice_agent, audio_input)
MCP（Model Context Protocol）
from agents import Agent
from agents.mcp import MCPServer
mcp_server = MCPServer("path/to/mcp/config")
agent = Agent(
name="MCPAgent",
mcp_servers=[mcp_server],
)
最佳实践
1. 异步优先
推荐使用 await Runner.run() 而非 Runner.run_sync()。
2. 类型提示
使用 Pydantic 和 Annotated 提供清晰的类型信息。
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 和参数描述。
@function_tool
def calculate(
expression: Annotated[str, "数学表达式，如 '2+2' 或 'sqrt(16)'"]
) -> float:
"""
计算数学表达式的结果。
支持基本运算、三角函数、对数等。
"""
return eval(expression)
4. 追踪调试
使用 with trace() 包裹工作流以便调试。
from agents import trace
with trace("客户服务流程"):
result = await Runner.run(customer_service_agent, user_query)
5. 错误处理
SDK 内置重试机制，可自定义重试策略。
from agents import Agent, Runner
agent = Agent(
name="RobustAgent",
retry_config={
"max_retries": 3,
"retry_delay": 1.0,
}
)
6. 资源清理
使用上下文管理器确保资源正确释放。
async with AgentSession() as session:
result = await Runner.run(agent, input, session=session)
完整示例
示例 1: 多语言客服系统
import asyncio
from agents import Agent, Runner
# 创建专业 Agent
chinese_agent = Agent(
name="中文客服",
instructions="用中文提供客户服务",
handoff_description="中文客服支持"
)
english_agent = Agent(
name="English Support",
instructions="Provide customer service in English",
handoff_description="English support"
)
# 路由 Agent
triage_agent = Agent(
name="Triage",
instructions="根据用户语言分配给合适的客服",
handoffs=[chinese_agent, english_agent],
)
async def main():
result = await Runner.run(triage_agent, "你好，我需要帮助")
print(f"当前 Agent: {result.current_agent.name}")
print(f"回复: {result.final_output}")
if __name__ == "__main__":
asyncio.run(main())
示例 2: 研究助手
import asyncio
from agents import Agent, Runner, function_tool
from typing import Annotated
@function_tool
def search_web(query: Annotated[str, "搜索查询"]) -> str:
"""在网络上搜索信息"""
# 实现搜索逻辑
return f"搜索结果：{query}"
@function_tool
def analyze_data(data: Annotated[str, "要分析的数据"]) -> str:
"""分析数据并返回洞察"""
# 实现分析逻辑
return f"分析结果：{data}"
# 专业 Agent
researcher = Agent(
name="Researcher",
instructions="搜索和收集相关信息",
tools=[search_web],
)
analyst = Agent(
name="Analyst",
instructions="分析数据并提供洞察",
tools=[analyze_data],
)
writer = Agent(
name="Writer",
instructions="基于研究和分析撰写报告",
)
# 编排 Agent
orchestrator = Agent(
name="ResearchOrchestrator",
instructions="协调研究、分析和写作流程",
tools=[
researcher.as_tool(tool_name="research", tool_description="搜索信息"),
analyst.as_tool(tool_name="analyze", tool_description="分析数据"),
writer.as_tool(tool_name="write", tool_description="撰写报告"),
],
)
async def main():
result = await Runner.run(
orchestrator,
"研究人工智能的最新发展趋势并撰写报告"
)
print(result.final_output)
if __name__ == "__main__":
asyncio.run(main())
调试技巧
1. 查看完整对话历史
result = await Runner.run(agent, input)
print(result.to_input_list()) # 完整对话历史
2. 检查当前 Agent
print(result.current_agent) # 当前运行的 agent
print(result.current_agent.name) # Agent 名称
3. 追踪执行过程
from agents import trace
with trace("调试工作流") as t:
result = await Runner.run(agent, input)
# 在 OpenAI 后台查看详细追踪信息
4. 流式调试
result = Runner.run_streamed(agent, input)
async for event in result.stream_events():
print(f"Event: {event.type}")
if hasattr(event, 'data'):
print(f"Data: {event.data}")
性能优化
1. 并行工具调用
# SDK 自动并行调用独立的工具
agent = Agent(
name="ParallelAgent",
tools=[tool1, tool2, tool3],
parallel_tool_calls=True, # 启用并行调用
)
2. 缓存策略
from agents import Agent
agent = Agent(
name="CachedAgent",
cache_settings={
"enable_cache": True,
"cache_ttl": 3600, # 1小时
}
)
3. 批处理
async def batch_process(inputs):
tasks = [Runner.run(agent, input) for input in inputs]
results = await asyncio.gather(*tasks)
return results
常见问题
Q: 如何使用其他 LLM？
A: 通过 LiteLLM 配置其他模型提供商，详见"高级功能"部分。
Q: 如何实现流式输出？
A: 使用 Runner.run_streamed() 并遍历 stream_events()。
Q: 如何保存对话历史？
A: 使用 Session 功能自动管理历史。
Q: 如何调试 Agent 行为？
A: 使用 with trace() 包裹代码，在 OpenAI 后台查看详细执行过程。
Q: 如何限制 Agent 的输出格式？
A: 使用 Pydantic 模型定义输出类型。
from pydantic import BaseModel
class Output(BaseModel):
summary: str
confidence: float
agent = Agent(
name="StructuredAgent",
output_type=Output,
)
学习资源
- 📖 官方文档: https://openai.github.io/openai-agents-python/
- 💻 GitHub: https://github.com/openai/openai-agents-python
- 📦 PyPI: https://pypi.org/project/openai-agents/
- 🎓 示例代码: https://github.com/openai/openai-agents-python/tree/main/examples
版本要求
- Python >= 3.10
- openai-agents >= 0.1.0
相关技能
find-skills- 查找和安装其他技能file-sync- 同步文件到集中存储
注意: 使用此技能需要有效的 OpenAI API Key 或其他支持的 LLM API Key。