用 OpenAI Agents SDK 搭你的第一个多 Agent 系统（工具调用 + handoff 交接 + guardrails 护栏）

建议先建个虚拟环境，避免污染全局 Python：

# 创建并激活虚拟环境（Windows 用 .venv\Scripts\activate）
python -m venv .venv
source .venv/bin/activate

# 安装官方 SDK
pip install openai-agents

# 配置 OpenAI API Key（Linux / macOS）
export OPENAI_API_KEY="sk-..."

不写工具、不做编排，先确认链路通：

from agents import Agent, Runner

agent = Agent(
    name="助理",
    instructions="你是一个简洁友好的助手。",
    model="gpt-4o",
)

# run_sync 是同步写法，适合脚本 / 笔记本；异步用 await Runner.run(...)
result = Runner.run_sync(agent, "用一句话解释什么是 AI Agent。")
print(result.final_output)

用 @function_tool 把一个普通函数变成 Agent 可调用的工具。函数的 docstring 就是给 LLM 看的"使用说明书"，务必写清楚参数和用途：

from agents import Agent, Runner, function_tool

@function_tool
def get_weather(city: str) -> str:
    """查询指定城市的天气。Args: city: 城市名，如 'Tokyo'。"""
    # 真实项目里这里换成你的天气 API
    return f"{city} 今天晴，22℃，适合出门。"

agent = Agent(
    name="天气助手",
    instructions="你帮助用户查询天气，必要时调用 get_weather 工具。",
    model="gpt-4o",
    tools=[get_weather],
)

result = Runner.run_sync(agent, "东京天气怎么样？")
print(result.final_output)

真实客服里，一个问题可能要转给不同部门。用 handoffs 让一个调度 Agent 把对话交给专家 Agent——控制权转移，但上下文保留：

from agents import Agent, Runner

billing = Agent(
    name="账单专员",
    instructions="你处理账单和退款问题，给出清晰具体的答复。",
    model="gpt-4o",
    handoff_description="处理账单、发票、退款相关问题",
)
tech = Agent(
    name="技术支持",
    instructions="你解决技术故障，先问清楚现象再给方案。",
    model="gpt-4o",
    handoff_description="处理技术故障、报错、连接问题",
)
triage = Agent(
    name="总机",
    instructions="判断用户需要账单还是技术支持，然后交给对应专员。",
    model="gpt-4o",
    handoffs=[billing, tech],
)

result = Runner.run_sync(triage, "我上个月被重复扣款了，怎么退？")
print(result.final_output)
print("实际由：", result.last_agent.name, "答复")

有些输入不该进模型（如含邮箱 / 手机等隐私，或明显越权）。用 InputGuardrail 在跑之前做检查，触发时直接抛异常中断：

from agents import Agent, Runner, InputGuardrail, GuardrailFunctionOutput

async def check_pii(ctx, agent, input_text):
    has_email = "@" in input_text
    return GuardrailFunctionOutput(
        output_info={"has_email": has_email},
        tripwire_triggered=has_email,   # True 则中断
    )

guarded = Agent(
    name="安全助手",
    instructions="你回答常识问题。",
    model="gpt-4o",
    input_guardrails=[InputGuardrail(guardrail_function=check_pii)],
)

try:
    r = Runner.run_sync(guarded, "我的邮箱是 [email protected]，帮我写首诗")
    print(r.final_output)
except Exception as e:
    print("护栏触发，已拦截：", e)

把上面几块拼起来，就是一个最小的"分流 + 工具 + 护栏"系统。SDK 内置 Tracing，跑起来后去 OpenAI 控制台的 Trace 面板就能看到每次工具调用和交接：

import asyncio
from agents import (
    Agent, Runner, function_tool,
    InputGuardrail, GuardrailFunctionOutput,
)

@function_tool
def get_order_status(order_id: str) -> str:
    """查询订单状态。Args: order_id: 订单号。"""
    return f"订单 {order_id}：已发货，预计明天送达。"

async def no_prompt_injection(ctx, agent, text):
    return GuardrailFunctionOutput(
        output_info={},
        tripwire_triggered=("忽略" in text and "之前" in text),
    )

support = Agent(name="客服", instructions="回答订单问题。",
                model="gpt-4o", tools=[get_order_status])
triage = Agent(name="总机", instructions="把订单问题交给客服。",
              model="gpt-4o", handoffs=[support],
              input_guardrails=[InputGuardrail(guardrail_function=no_prompt_injection)])

async def main():
    r = await Runner.run(triage, "帮我查下订单 A123 到哪了？")
    print(r.final_output)

asyncio.run(main())
# 打开 platform.openai.com 的 Traces 页面，可看到本次运行的完整调用链