OpenAI Agents SDK 上手

OpenAI Agents SDK 是 OpenAI 于 2025 年开源的轻量级多智能体框架,是实验项目 Swarm 的正式继任,理念延续(Agent 之间靠 handoff 交接),但工程化完备得多:自带工具调用、护栏、会话管理、内置追踪(tracing)。适合把前几章手写的"循环控制 + 工具注册 + 会话"直接换成官方维护的骨架,专注写业务逻辑。

安装与配置

需要 Python 3.10+,安装后设置 OpenAI API Key(PowerShell 用 $env:):

pip install openai-agents          # 校对版本 0.2x,迭代很快,建议装最新
$env:OPENAI_API_KEY = "sk-..."     # Windows PowerShell

核心概念速览

  • Agent:一个 LLM + instructions + 可选的 tools / handoffs / guardrails / 输出格式;
  • Runner:负责执行 Agent(含工具循环与交接),await Runner.run() 异步、Runner.run_sync() 同步;
  • Handoffs(交接):把任务交给更专业的另一个 Agent,交接双方共享会话;
  • Guardrails(护栏):对用户输入 / Agent 输出做检查,命中即抛异常终止;
  • Sessions(会话):把多轮历史自动存取,省去手动拼接消息。

最小示例:跑通一个 Agent

Agent 就是"指令 + 名字",用 Runner 执行后从 final_output 取最终文本:

import asyncio
from agents import Agent, Runner

agent = Agent(
    name="历史老师",
    instructions="用简明中文回答历史问题。",
)

async def main():
    result = await Runner.run(agent, "罗马帝国是哪年灭亡的?")
    print(result.final_output)      # 输出:公元 476 年,西罗马帝国灭亡(内容依模型而定)

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

Runner 会自己完成"模型返回工具调用 → 执行工具 → 再次调用模型"的循环,直到模型给出最终文本。

多工具 Agent 最小示例

function_tool 装饰普通函数即可注册工具,函数 docstring 与类型注解会自动生成工具 Schema(沿用第 05 章"描述要清晰"的结论,docstring 要写清用途):

import asyncio
from agents import Agent, Runner, function_tool

@function_tool
def get_weather(city: str) -> str:
    """查询指定城市当前天气,city 为城市名。"""
    return f"{city}:晴,26℃,微风"   # 示例返回值,可换成真实天气 API

@function_tool
def get_stock(symbol: str) -> str:
    """查询股票最新价格,symbol 为股票代码如 600519。"""
    return f"{symbol} 收盘价 180.5 元"  # 示例返回值

agent = Agent(
    name="助手",
    instructions="你是一个生活助手,需要查天气/股票时使用工具回答。",
    tools=[get_weather, get_stock],
)

async def main():
    result = await Runner.run(agent, "北京今天适合出门吗?顺便看看 600519 多少钱。")
    print(result.final_output)      # 输出:北京晴 26℃ 适合出门;600519 收盘 180.5 元(内容依模型而定)

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

两个 Agent 交接(handoff)

给"客服/总控"Agent 挂上专业 Agent,它会按问题把会话整体交接出去;handoff_description 帮总控判断何时交接:

import asyncio
from agents import Agent, Runner

refund_agent = Agent(
    name="退款专员",
    handoff_description="处理退款、投诉等售后问题",
    instructions="你负责退款流程,语气耐心,先核对订单号。",
)
price_agent = Agent(
    name="价格专员",
    handoff_description="回答价格、优惠、满减问题",
    instructions="你负责介绍价格与优惠活动。",
)
triage = Agent(
    name="总台",
    instructions="先判断用户诉求,再交接给合适专员。",
    handoffs=[refund_agent, price_agent],
)

async def main():
    result = await Runner.run(triage, "我上个月买的耳机坏了,想退款。")
    print(result.final_output)            # 输出:退款专员给出的处理答复(内容依模型而定)
    print(result.last_agent.name)         # 输出:退款专员

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

Guardrails:给 Agent 加输入护栏

护栏适合拦截"不该进模型"的输入(省 token、防滥用)。把护栏函数挂到 Agent 的 input_guardrails,函数返回带 tripwire_triggered 的结果,命中后 Runner 抛 InputGuardrailTripwireTriggered

from agents import Agent, GuardrailFunctionOutput, Runner, input_guardrail

@input_guardrail
async def no_math_guardrail(ctx, agent, input):
    """示意骨架:正式实现时在此调一个小模型或规则判断 input 是否命中禁区。"""
    blocked = "帮我做作业" in input            # 简单关键词规则,仅示意
    return GuardrailFunctionOutput(output_info={"blocked": blocked},
                                   tripwire_triggered=blocked)

agent = Agent(
    name="客服",
    instructions="你是客服助手。",
    input_guardrails=[no_math_guardrail],     # 命中护栏会抛异常而非继续跑
)

输入/输出护栏的完整签名与异常类型随版本演进明显,动手前以官方 docs/guardrails 页为准。

Sessions:多轮会话自动管理

Runner.run(..., session=session) 传入会话对象后,SDK 自动读取历史并在每轮结束写入新内容,无需手动拼 to_input_list()

import asyncio
from agents import Agent, Runner, SQLiteSession

agent = Agent(name="问答助手", instructions="回答尽量简短。")
session = SQLiteSession("conversation_001", db_path="sessions.db")  # 落盘到 sqlite 文件

async def main():
    r1 = await Runner.run(agent, "金门大桥在哪个城市?", session=session)
    print(r1.final_output)                    # 输出:旧金山
    r2 = await Runner.run(agent, "它属于哪个州?", session=session)   # 记得上文
    print(r2.final_output)                    # 输出:加利福尼亚州

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

非 OpenAI 模型与本教程核对说明

  • SDK 默认走 OpenAI 的 Responses/Chat Completions API。换用 deepseek 等"OpenAI 兼容"模型,需按官方 docs/models 页用 Chat Completions 兼容封装:OpenAIChatCompletionsModel(model="deepseek-chat", openai_client=AsyncOpenAI(base_url="https://api.deepseek.com", api_key=...)),再把它作为 Agent(model=...);是否支持工具调用取决于该兼容端点,具体以官方文档为准(各版本导入路径有变,勿照抄旧博客)。
  • 本节核对于官方仓库 20252026 快速迭代期(openai-agents 0.2x,docs/quickstart、handoffs、sessions、guardrails 页面),function_toolRunner.run_sync、session 参数等在 0.1x0.2x 均可用;SDK 仍在高频发版,接口若有出入一律以 openai.github.io/openai-agents-python 为准。 小结:安装 openai-agents 后,用 Agent + Runner 即可获得工具循环;多智能体靠 handoffs 交接、输入输出靠 guardrails 把关、多轮记忆靠 session;非 OpenAI 模型走 Chat Completions 兼容封装,改动前先查官方文档版本。
笔记加载中…