OpenAI Agent SDK

Build multi-agent AI systems with OpenAI Agents SDK. Create, orchestrate, and manage AI agents with tools, handoffs, guardrails, and tracing. Supports 100+ LLMs via LiteLLM.

sendwealth@sendwealth

Install

openclaw skills install @sendwealth/openai-agent-sdk

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=[french_agent, spanish_agent],
)

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(input_guardrails=[check_input])

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 OPENAI_API_KEY="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=[sales_agent, support_agent, 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.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 同时执行独立任务。

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 needs_human_approval(event):
        approval = await get_human_input()
        # 继续执行

高级功能

使用其他 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",
    model_settings={"client": claude_client}
)

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(voice_agent, audio_input)

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",
    mcp_servers=[mcp_server],
)

最佳实践

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(customer_service_agent, user_query)

5. 错误处理

SDK 内置重试机制，可自定义重试策略。

python

from agents import Agent, Runner

agent = Agent(
    name="RobustAgent",
    retry_config={
        "max_retries": 3,
        "retry_delay": 1.0,
    }
)

6. 资源清理

使用上下文管理器确保资源正确释放。

python

async with AgentSession() as session:
    result = await Runner.run(agent, input, session=session)

完整示例

示例 1: 多语言客服系统

python

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: 研究助手

python

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. 查看完整对话历史

python

result = await Runner.run(agent, input)
print(result.to_input_list())  # 完整对话历史

2. 检查当前 Agent

python

print(result.current_agent)  # 当前运行的 agent
print(result.current_agent.name)  # Agent 名称

3. 追踪执行过程

python

from agents import trace

with trace("调试工作流") as t:
    result = await Runner.run(agent, input)
    # 在 OpenAI 后台查看详细追踪信息

4. 流式调试

python

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. 并行工具调用

python

# SDK 自动并行调用独立的工具
agent = Agent(
    name="ParallelAgent",
    tools=[tool1, tool2, tool3],
    parallel_tool_calls=True,  # 启用并行调用
)

2. 缓存策略

python

from agents import Agent

agent = Agent(
    name="CachedAgent",
    cache_settings={
        "enable_cache": True,
        "cache_ttl": 3600,  # 1小时
    }
)

3. 批处理

python

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 模型定义输出类型。

python

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

OpenAI Agent SDK

Install

OpenAI Agent SDK 技能

何时使用此技能

核心概念

1. Agent（智能体）

2. Tools（工具）

3. Runner（运行器）

4. Handoffs（交接）

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

6. Guardrails（防护栏）

7. Tracing（追踪）

8. Sessions（会话）

快速开始

安装

设置环境变量

Hello World 示例

常见设计模式

模式 1: 路由（Routing）

模式 2: 编排者（Orchestrator）

模式 3: 并行执行（Parallelization）

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

高级功能

使用其他 LLM

Realtime Agents（语音 Agent）

MCP（Model Context Protocol）

最佳实践

1. 异步优先

2. 类型提示

3. 工具描述

4. 追踪调试

5. 错误处理

6. 资源清理

完整示例

示例 1: 多语言客服系统

示例 2: 研究助手

调试技巧

1. 查看完整对话历史

2. 检查当前 Agent

3. 追踪执行过程

4. 流式调试

性能优化

1. 并行工具调用

2. 缓存策略

3. 批处理

常见问题

Q: 如何使用其他 LLM？

Q: 如何实现流式输出？

Q: 如何保存对话历史？

Q: 如何调试 Agent 行为？

Q: 如何限制 Agent 的输出格式？

学习资源

版本要求

相关技能

Related skills