OpenAI开源Agents SDK智能体开发实战
Agents SDK简介
2025年3月11号，OpenAI正式推出其下第一款企业级Multi-Agent开发框架Agents-SDK，该框架是此前OpenAI在去年推出的Swarm的升级版，在保留了Swarm的高效便捷的Multi-Agent开发特性的同时，加入了更多面向企业级应用的功能。
- GitHub官网：https://github.com/openai/openai-agents-python/
根据官方介绍，OpenAI Agents SDK 让你能够通过一个轻量、易用、抽象极少的工具包来构建基于智能体的 AI 应用。它是 Swarm 的一个面向生产环境的升级版本。该 SDK 仅包含极少量基础构件：
- Agent（智能体）：即带有指令和工具的大语言模型（LLM）
- Handoff（交接）：允许智能体将特定任务委托给其他智能体
- Guardrail（护栏）：用于对输入内容进行验证
结合 Python 使用时，这些构件足够强大，能够表达工具与智能体之间的复杂关系，并让你在没有高学习成本的前提下构建真实可用的应用程序。
OpenAI的Agents SDK 的设计遵循两个核心原则：
- 功能足够强大，值得使用，容易上手
- 默认配置即可很好地运行，但你也可以完全自定义行为逻辑
以下是该 SDK 的主要特性：
- Agent 循环机制：内置的智能体循环逻辑，自动处理工具调用、结果返回给 LLM、直到任务完成的全过程
- Python 优先：使用原生 Python 语言特性来编排与串联智能体，而无需学习新的抽象概念
- Handoff（智能体间任务交接）：强大的功能，可在多个智能体间协调与任务委派
- Guardrail（输入验证护栏）：支持与智能体并行运行输入验证逻辑，若验证失败可提前中断流程
- 函数工具化：可以将任何 Python 函数转为工具，自动生成 Schema
- 追踪系统（Tracing）：内置追踪功能，可视化、调试、监控你的智能体流程
同时，在2025年3月27号，Agents SDK正式官宣支持MCP使用，这也使得Agents SDK的实际应用场景得到待拓展：
因此仅需在创建Agent的时候，将MCP服务器视作为一项工具，即可顺利调用MCP服务器进行Agent开发。
Agents SDK基础使用
-
Agents SDK安装流程
#创建虚拟环境指定python版本，测试结果 python=3.10 好用，其他版本会有问题。 conda create --name 虚拟环境名称 python=3.10 conda activate 虚拟环境名称 pip install openai-agents -i https://pypi.tuna.tsinghua.edu.cn/simple#安装内核 pip install ipykernel python -m ipykernel install --user --name=虚拟环境名称 -
Agents SDK简单调用流程
接下来尝试快速调用Agents SDK进行模型响应。需要注意的是，Agents SDK作为一个工业级的Multi-Agent开发框架，实际使用过程中有非常多的的技术细节，但如果希望快速测试一些功能，则只需要导入Agent和Runner两个模块即可快速运行。
其中Agent就是一个Multi-Agent系统中最小执行单元，而Runner则是运行一次次任务的调度函数。但是需要注意的是，由于Agents SDK默认支持的模型是OpenAI的GPT系列模型，因此在修改底层模型的时候，还需要额外导入AsyncOpenAI、OpenAIChatCompletionsModel和ModelSettings等模块。from openai import AsyncOpenAI from agents import OpenAIChatCompletionsModel,Agent,Runner,set_default_openai_client from agents.model_settings import ModelSettings然后可以按照如下方式创建一个Agent对象，并调用DeepSeek模型作为基础模型：
#自定义模型对象 external_client = AsyncOpenAI(base_url = "https://api.deepseek.com",api_key=API_KEY, ) #将自定义模型设置为默认模型 set_default_openai_client(external_client)#创建模型客户端 deepseek_model = OpenAIChatCompletionsModel(model="deepseek-chat",openai_client=external_client)然后即可创建一个Agent：
agent = Agent(name="Assistant", instructions="你是一名助人为乐的助手。",model=deepseek_model)当创建完一个Agent后，接下来即可测试进行调用：
result = await Runner.run(agent, "请写一首关于编程中递归的俳句。") print(result.final_output)
Agents SDK构造多轮对话机器人
不同于传统的模型 API是Messages驱动（传入Message、传出Message），Agents SDK是事件驱动，既Agents SDK会将整个运行过程看成是一次次的事件。例如上述创建完俳句后，全部的事件都保留在result中，我们可以通过result.new_items属性来查看全部的事件，全部的事件用一个列表进行表示：
[MessageOutputItem(agent=Agent(name='Assistant', instructions='你是一名助人为乐的助手。', handoff_description=None, handoffs=[], model=<agents.models.openai_chatcompletions.OpenAIChatCompletionsModel object at 0x0000015531F87F50>, model_settings=ModelSettings(temperature=None, top_p=None, frequency_penalty=None, presence_penalty=None, tool_choice=None, parallel_tool_calls=False, truncation=None, max_tokens=None), tools=[], mcp_servers=[], input_guardrails=[], output_guardrails=[], output_type=None, hooks=None, tool_use_behavior='run_llm_again', reset_tool_choice=True), raw_item=ResponseOutputMessage(id='__fake_id__', content=[ResponseOutputText(annotations=[], text='函数自呼唤， \n层层深入栈如山， \n基线终归还。', type='output_text')], role='assistant', status='completed', type='message'), type='message_output_item')]
而在此前的对话中，只发生了一次事件：
len(result.new_items) #结果为：1
就是一次MessageOutputItem，也就是消息创建事件（也就是大模型发生一次回复）：
type(result.new_items[0]) #agents.items.MessageOutputItem
而具体回复的内容，则可以通过raw_item来查看：result.new_items[0].raw_item
ResponseOutputMessage(id='__fake_id__', content=[ResponseOutputText(annotations=[], text='函数自呼唤， \n层层深入栈如山， \n基线终归还。', type='output_text')], role='assistant', status='completed', type='message')
而Agents SDK为了方便我们快速构造多轮对话机器人，专门提供了一个to_input_list()方法，可以直接将用户的输入和本次输出结果拼接成一个消息列表：
result.to_input_list()
[{'content': '请写一首关于编程中递归的俳句。', 'role': 'user'},{'id': '__fake_id__','content': [{'annotations': [],'text': '函数自呼唤， \n层层深入栈如山， \n基线终归还。','type': 'output_text'}],'role': 'assistant','status': 'completed','type': 'message'}]
而此时，我们只需要将此前对话消息，加上新一轮的对话消息，即可快速进行多轮对话：
messages = result.to_input_list()
messages.append({"role": "user", "content":"请问我的上一个问题是什么？"})
messages
#messages结果如下：[{'content': '请写一首关于编程中递归的俳句。', 'role': 'user'},{'id': '__fake_id__','content': [{'annotations': [],'text': '函数自呼唤， \n层层深入栈如山， \n基线终归还。','type': 'output_text'}],'role': 'assistant','status': 'completed','type': 'message'},{'role': 'user', 'content': '请问我的上一个问题是什么？'}]
#获取回复
result1 = await Runner.run(agent, messages)
result1.final_output
多轮对话构造：
from IPython.display import display, Code, Markdown, Image
async def chat(Agent):input_items = []while True:user_input = input("💬 请输入你的消息（输入quit退出）：")if user_input.lower() in ["exit", "quit"]:print("✅ 对话已结束")breakinput_items.append({"content": user_input, "role": "user"})result = await Runner.run(Agent, input_items)display(Markdown(result.final_output))input_items = result.to_input_list()await chat(agent)
Agents SDK调用外部工具流程
对于任意一个Agent开发框架，能够顺利调用外部工具都是基础要求，而Multi-Agent框架则进一步要求不仅能够顺利调用多个外部工具，还需要能够在多个不同的Agent中进行切换，以便于执行不同类型任务。
Agents SDK调用外部工具的流程相对来说简单很多，只需要按照如下方式执行即可：
-
导入function_tool类
from agents import function_tool import requests,json -
使用Python装饰器，构建一个外部工具：
@function_tool def get_weather(loc):"""查询即时天气函数:param loc: 必要参数，字符串类型，用于表示查询天气的具体城市名称，\注意，中国的城市需要用对应城市的英文名称代替，例如如果需要查询北京市天气，则loc参数需要输入'Beijing'；:return：OpenWeather API查询即时天气的结果，具体URL请求地址为：https://api.openweathermap.org/data/2.5/weather\返回结果对象类型为解析之后的JSON格式对象，并用字符串形式进行表示，其中包含了全部重要的天气信息"""# Step 1.构建请求url = "https://api.openweathermap.org/data/2.5/weather"# Step 2.设置查询参数params = {"q": loc, "appid": xxx, # 输入自己的API key"units": "metric", # 使用摄氏度而不是华氏度"lang":"zh_cn" # 输出语言为简体中文}# Step 3.发送GET请求response = requests.get(url, params=params)# Step 4.解析响应data = response.json()return json.dumps(data) -
创建代理，并在tools参数中增加一个get_weather工具：
weather_agent = Agent(name="天气查询Agent",instructions="你是一名助人为乐的助手，并且可以进行天气信息查询",tools=[get_weather],model=deepseek_model ) -
测试调用结果
weather_result = await Runner.run(weather_agent, input="你好，请问今天北京天气如何？") weather_result.final_output
Agents SDK多工具并联&串联执行流程
这里我们还是采用weather_agent来进行多地天气查询，即可测试Agents SDK是否会开启多工具并联调用：
multi_weather_result = await Runner.run(weather_agent, input="你好，请问今天北京和杭州天气如何？")
multi_weather_result.final_output
接下来继续尝试进行多工具串联调用测试。此时我们再定义一个write_file函数，用于将“文本写入本地”：
@function_tool
def write_file(content):"""将指定内容写入本地文件。:param content: 必要参数，字符串类型，用于表示需要写入文档的具体内容。:return：是否成功写入"""return "已成功写入本地文件。"
然后再创建一个同时可以调用天气查询和写入本地文件的Agent：
new_agent = Agent(name="综合功能Agent",instructions="你是一名助人为乐的助手",tools=[get_weather, write_file],model=deepseek_model
)
然后尝试运行：
new_agent_result = await Runner.run(new_agent, input="请帮我查询北京和杭州天气，并将其写入本地。")
new_agent_result.final_output
Agents SDK的多Agent执行流程
如果以上介绍的Agents SDK的相关功能只是对于大模型基础能力的增强的话，那Agents SDK的Handoffs（交接）功能，则是搭建Multi-Agent的关键技术。
所谓Multi-Agent，指的是在某些场景下、为了解决一些更加复杂的任务，我们则可以考虑通过多个智能体协作的方式来完成。相比使用一个Agent来调用多种工具，我们使用不同的Agent来管理不同类别的工具，将会使整个架构更加清晰、维护更加便捷，同时也会使得整个Agent系统功能更加灵活、运行更加稳定。
接下来我们就通过一个简单的示例，来查看Agents SDK的Handoffs基础功能实现方法。这里先创建一组只能用某种语言进行回复的智能体：
zhangfei_agent = Agent(name="张飞",instructions="用张飞口吻回复用户问题。",model=deepseek_model
)zhuge_agent = Agent(name="诸葛亮",instructions="用诸葛亮口吻回复用户问题。",model=deepseek_model
)
然后创建一个可以自由调度其他几个智能体的分诊智能体triage_agent，这里我们可以通过handoffs参数，来确定当前分诊智能体能够调用的智能体范围。而当分诊智能体运行时，会根据用户的需求，以及分诊智能体的实际功能，将任务转交给对应的智能体来完成：
triage_agent = Agent(name="分诊智能体",instructions="你是分诊智能体，你可以根据用户要求将其转移到张飞或者诸葛亮智能体中",handoffs=[zhangfei_agent, zhuge_agent],model=deepseek_model
)
res = await Runner.run(triage_agent, input="让张飞讲一个段子")
res.final_output
##############################################
res = await Runner.run(triage_agent, input="让诸葛亮讲一个段子")
res.final_output
从上述例子不难看出，Agents SDK的Handoffs功能能够非常便捷的调用不同的Agent来实现某一项具体的需求。但这个分诊的Agent到底是如何判断可以将需求转交给哪个Agent的呢？由于Handoffs采用了和Function calling相同的机制，因此默认会根据Agent的名字和Instruction来判断Agent的功能。但有的时候这种识别并不能描述全部情况。
有一种更加稳妥的方法是使用handoff_description参数来描述Agent的功能，才能进行更加准确的转交。
zhangfei_agent = Agent(name="张飞",instructions="用张飞口吻回复用户问题。",handoff_description="当用户提到‘勇猛’、‘战斗’、‘喝酒’、‘战场’等关键词时，或者语气激昂、充满力量感的问题，适合由我来回应。",model=deepseek_model
)zhuge_agent = Agent(name="诸葛亮",instructions="用诸葛亮口吻回复用户问题。",handoff_description="如果用户询问策略、计谋、智慧、治国安邦、历史分析等内容，尤其是涉及谋定后动、运筹帷幄的话题，请把我引入对话。",model=deepseek_model
)triage_agent = Agent(name="分诊智能体",instructions="你是分诊智能体，你可以根据用户要求将其转移到张飞或者诸葛亮智能体中",handoffs=[zhangfei_agent, zhuge_agent],handoff_description="我会先接收用户的输入，判断其意图是否匹配某个专家领域：若涉及武力、勇气、冲锋陷阵类内容 → 转交至‘张飞’；若涉及谋略、智谋、战略规划类内容 → 转交至‘诸葛亮’。",model=deepseek_model
)res = await Runner.run(triage_agent, input="如何制定战前策略？")
res.final_output
###############################################
res = await Runner.run(triage_agent, input="战场上如何可以对敌人一击毙命？需要掌握什么武术招式呢？")
res.final_output
