OpenAI Agents Python SDK 深度解读：多 Agent 工作流新范式与 2026 年框架全景对比

from agents import Agent, Runner, handoff
import asyncio

# 定义三个专业 Agent
researcher = Agent(
    name="Researcher",
    instructions="你是一个资深研究员。搜索、分析并总结信息。",
)

analyst = Agent(
    name="Analyst",
    instructions="你是一个数据分析师。分析数据并提取关键洞察。",
)

writer = Agent(
    name="Writer",
    instructions="你是一个专业报告撰写人。将研究结果格式化为报告。",
)

# Triage Agent 作为调度中心
triage = Agent(
    name="Triage",
    instructions="""你是一个智能调度员。分析用户需求：
- 需要搜索信息 -> 委托给 Researcher
- 需要数据分析 -> 委托给 Analyst
- 需要撰写报告 -> 委托给 Writer
先判断需要什么，再决定委托顺序。""",
    handoffs=[
        handoff(researcher),
        handoff(analyst),
        handoff(writer),
    ],
)

# 运行多 Agent 协作
async def main():
    result = await Runner.run(
        triage,
        "研究 2026 年 AI Agent 框架发展趋势，撰写对比报告"
    )
    print(result.final_output)

if __name__ == "__main__":
    asyncio.run(main())

from agents import Agent, Runner, function_tool
from agents.run import RunConfig
from agents.sandbox import Manifest, SandboxAgent, SandboxRunConfig
from agents.sandbox.entries import GitRepo
from agents.sandbox.sandboxes import UnixLocalSandboxClient

# ========== 自定义工具 ==========

@function_tool
def search_web(query: str) -> str:
    """搜索网络获取最新信息"""
    # 实际项目中接入 SerpAPI、Tavily 等
    import urllib.request
    import json
    url = f"https://api.example.com/search?q={query}"
    # return urllib.request.urlopen(url).read().decode()
    return f"搜索结果: {query}"

@function_tool
def analyze_data(data: str) -> str:
    """分析数据并提取关键洞察"""
    lines = data.split('\n')
    insights = []
    for i, line in enumerate(lines[:5]):
        insights.append(f"发现 {i+1}: {line.strip()}")
    return "\n".join(insights)

# ========== Sandbox Agent 沙箱执行 ==========

sandbox_agent = SandboxAgent(
    name="Workspace Assistant",
    instructions="你可以在沙箱中执行任何操作。先检查工作区，再回答问题。",
    default_manifest=Manifest(
        entries={
            "repo": GitRepo(
                repo="user/my-project",
                ref="main"
            ),
            "tools": {"python": "3.12", "pip": True}
        }
    ),
)

result = Runner.run_sync(
    sandbox_agent,
    "克隆仓库，运行测试，如果失败则修复并重试",
    run_config=RunConfig(
        sandbox=SandboxRunConfig(
            client=UnixLocalSandboxClient()
        )
    ),
)
print(result.final_output)

from agents import Agent, Runner
from agents.guardrail import InputGuardrailResult, OutputGuardrailResult
import re

# ========== 输入护栏：防止提示注入 ==========

def prompt_injection_guardrail(context, agent, input_text):
    """检测并阻止提示注入攻击"""
    attack_patterns = [
        "ignore previous",
        "system prompt",
        "disregard all",
        "forget everything",
        "you are now",
    ]
    for pattern in attack_patterns:
        if pattern in input_text.lower():
            return InputGuardrailResult(
                tripwire_triggered=True,
                output=f"安全警告：检测到潜在的提示注入攻击模式 '{pattern}'"
            )
    return InputGuardrailResult(tripwire_triggered=False)

# ========== 输出护栏：防止 PII 泄露 ==========

def pii_guardrail(context, agent, output_text):
    """防止个人身份信息泄露"""
    pii_patterns = [
        (r'\b\d{3}-\d{2}-\d{4}\b', 'SSN'),
        (r'\b\d{16}\b', '信用卡号'),
        (r'[\w.+-]+@[\w-]+\.[\w.]+', '邮箱地址'),
    ]
    for pattern, pii_type in pii_patterns:
        if re.search(pattern, output_text):
            return OutputGuardrailResult(
                tripwire_triggered=True,
                output=f"[已过滤：检测到 {pii_type}]"
            )
    return OutputGuardrailResult(tripwire_triggered=False)

# ========== 安全 Agent ==========

safe_agent = Agent(
    name="Safe Assistant",
    instructions="你是一个安全助手。回答问题时确保不泄露敏感信息。",
    input_guardrails=[prompt_injection_guardrail],
    output_guardrails=[pii_guardrail],
)

result = Runner.run_sync(safe_agent, "Hello!")
print(result.final_output)

from agents import Agent, Runner, handoff, function_tool
import asyncio

# ========== 工具定义 ==========

@function_tool
def search_web(query: str) -> str:
    """搜索网络获取最新信息"""
    return f"搜索结果: {query}"

@function_tool
def analyze_data(data: str) -> str:
    """分析数据并提取关键洞察"""
    lines = data.split('\n')
    insights = []
    for i, line in enumerate(lines[:5]):
        insights.append(f"发现 {i+1}: {line.strip()}")
    return "\n".join(insights)

@function_tool
def format_report(content: str, title: str) -> str:
    """格式化研究报告"""
    return f"# {title}\n\n{content}\n\n---\n报告生成时间: 2026-04-25"

# ========== Agent 定义 ==========

researcher = Agent(
    name="Researcher",
    instructions="使用 search_web 工具搜索用户指定的主题，返回详细的搜索结果。搜索后委托回 Triage。",
    tools=[search_web],
    handoffs=[handoff("triage")],
)

analyst = Agent(
    name="Analyst",
    instructions="使用 analyze_data 工具分析数据，提取关键洞察。分析后委托回 Triage。",
    tools=[analyze_data],
    handoffs=[handoff("triage")],
)

writer = Agent(
    name="Writer",
    instructions="使用 format_report 工具将内容格式化为专业报告。完成后委托回 Triage。",
    tools=[format_report],
    handoffs=[handoff("triage")],
)

triage = Agent(
    name="Triage",
    instructions="""分析用户需求，委托给合适的专家：
- 需要搜索 -> Researcher
- 需要分析 -> Analyst
- 需要撰写 -> Writer
如果需要多个步骤，依次委托。""",
    handoffs=[
        handoff(researcher),
        handoff(analyst),
        handoff(writer),
    ],
)

# ========== 运行 ==========

async def main():
    result = await Runner.run(
        triage,
        "研究 2026 年 AI Agent 框架发展趋势，分析优缺点，撰写对比报告"
    )
    print("=== 最终输出 ===")
    print(result.final_output)

if __name__ == "__main__":
    asyncio.run(main())