一、OpenAI Agents SDK是什么?
OpenAI Agents SDK是一个轻量级且易于使用的工具包,用于构建基于代理的AI应用程序。 提供了一些基本构建块,包括具备指令和工具的代理(Agents)、用于代理间任务委托的交接(Handoffs)以及用于输入验证的护栏(Guardrails)。
在官网上OpenAI给出了两个理由使用它,而且也介绍了 Agent Loop、Python-first、Handoffs、Guardrails 、Function tools 和 Tracing 6个特性,对于现阶段的我们对Python-first 和 Function tools 就不做过多的阐述了,但是其他的四个特性还需要进一步详细阐述。
二、核心特性
(一)Agent Loop(循环)
支持在没有完成任务时自动的循环agent的执行,直到完成任务结束,可以调用functioncall,MCP等工具来更好的完成任务。对于许多业务流程涉及一系列需要外部信息或操作的步骤(例如,查询数据库、调用 API、根据查询结果生成报告)。Agent Loop 自动化了这个迭代过程,能够自主完成这些复杂、多步任务的智能体成为可能。
(二)Handoffs(交接)
交接的出现允许一个智能体(Agent)将特定任务委托给另一个智能体来完成,是用于在多个智能体之间进行协调和委托。实现了复杂工作流和多智能体协作的关键机制。在大型业务场景中,单一智能体可能难以处理所有类型的任务。通过 Handoffs,可以设计由多个专业智能体组成的系统,每个智能体负责其擅长的领域;将复杂业务流程分解到不同的智能体中,提高了系统的模块化,使得每个智能体更容易开发、测试和维护。
例如在酒店业务中,有些问题需要多个业务方合作才能完成,那么不同业务域可以维护自己的agent,从入口确定是业务范围,涉及到多个智能体交互时,就可以使用Handoff完成交接,更加准确的给出问题的解决思路和方案,甚至是解决问题。
(三)Guardrails(护栏)
这个可以在一定程度上解决不确定性。Guardrails 用于验证智能体输入的能力,它们可以在智能体运行之前并行执行输入验证和检查。如果检查失败,Guardrails 可以让应用快速中止(breaking early)。这有助于确保应用的可靠性,防止因不当输入导致的错误行为。
对于现在大模型的产生,我们都受益于大模型带来的生产力提高,但是还是会出现一些不确定性,与我们的期望有所偏差,随着guardrails的出现,就像是一个家教在辅导学生作业一样,每完成一个任务,老师都会给出一个反馈。
在我们的业务系统中Guardrails可以提升我们应用的可靠性和鲁棒性。对于不合法或恶意输入可能导致错误、安全漏洞或意外行为,Guardrails 通过强制执行输入验证,确保智能体只处理符合预期的输入;还可以节省计算资源和减少成本,如果不符合预期在智能体运行初期就会判别和拒绝;确保了系统的业务逻辑和合规问题。
(四)Tracing(可视化追踪)
在OpenAI的SDK中内置了Tracing能力,可以很直观地解读Agent之间的交互过程,可以看到请求体,响应事件等的,除此之外还支持评估(evaluate)工作流程,甚至可以用来微调(fine-tune)模型以优化应用的性能。
在Agent初期,对于几乎“黑盒”系统来说,我们的掌控感和信任感会特别的弱,而Tracing这个能力就给我们带来一束“曙光”。我们可以在开发阶段,测试阶段甚至是部分线上场景对Agent进行监控和可视化的追踪。可以帮助我们快速定位和解决问题;除此之外OpenAI 的SDK中还支持基于追踪数据,可以进一步微调或蒸馏模型,持续优化智能体在实际业务场景中的效率和效果,例如提高回复的准确性、减少错误操作。
三、代码示例
这些特性激起了我的好奇心,本着对知识的渴望,咱们还是回归到代码实现上!为了快速实现:我们使用OpenAI的GPT-4o模型(默认)。
前期准备:
# 安装依赖 pip install openai-agents # or `uv add openai-agents`, etc # 设置自己的OpenAIKey export OPENAI_API_KEY=sk-...(一)最初天气查询
首先我们先写一个简单的示例,然后再开始一点点的把我们想要了解的特性走一遍。 我们先用python写一个简单的agent,可以查询天气:
from typing import TypedDict from agents import Agent, Runner, function_tool import asyncio @function_tool async def fetch_weather(city: str) -> str: """Fetch the weather for a given location. Args: city: The city to fetch the weather for. """ return "sunny" # 天气查询Agent weather_agent = Agent( name="天气查询专家", instructions="你是天气查询专家,用户输入城市名,你返回该城市的天气信息。请用简洁中文回复。", tools=[fetch_weather] ) async def main(): result = await Runner.run(weather_agent, input="北京") print(result.final_output) if __name__ == "__main__": asyncio.run(main())1、Agent属性说明
我们可以参考源码中的结构,发现有哪些属性我们可以调整或者修改,理论上默认属性值只能够提供简单的使用,如果涉及到自定义的功能那就需要调整更多的参数。
instructions:使用instructions属性指定当前agent的系统提示词。可以是字符串,也可以是Agent生成的动态指令,如果是一个函数,将通过上下文和代理实例调用,但是必须返回字符串。
name:智能体的名字
tools:智能体可以使用工具列表
handoff_description:智能体的交接说明,主要用于多个智能体进行交接时,让大模型了解这个智能体能做什么,以及什么时候调用他
handoffs:agent可以委派任务的子agent。你可以提供一个 handoffs 的列表,代理在适当的情况下可以选择将任务委派给它们。这样可以实现职责分离和模块化设计。
model:可以配置model,默认是gpt4o
hooks:接收此代理的各种生命周期事件的回调的类。
mcp_servers:agent可以使用的模型上下文协议 (MCP)服务器列表。每次代理运行时,它都会将这些服务器中的工具添加到可用工具列表中。
output_guardrails: 生成响应后,对代理的最终输出运行的检查列表。仅当代理生成最终输出时运行。
input_guardrails: 在生成响应之前,与代理执行并行运行的检查列表。仅当代理是链中的第一个代理时运行。
(二)多个Agent
现在开始构建多个Agent用来观察后面的新特性,例如本段代码示例为一个“天气专家”再加上一个“穿衣专家”两个Agent,回答用户的关于穿衣服的一些回答。 两个agent之间“穿衣专家” 使用 “天气专家”给出的天气信息进行总结天气因素,也就是需要两个专家的“交接”,最终给出对应穿衣风格搭配建议。(后面的tracing模块会对交互细节进行阐述)
from typing import TypedDict from agents import Agent, Runner, function_tool import asyncio @function_tool async def fetch_weather(city: str) -> str: """Fetch the weather for a given location. Args: location: The location to fetch the weather for. """ # In real life, we'd fetch the weather from a weather API return "sunny" # 天气查询Agent weather_agent = Agent( name="天气查询专家", instructions="你是天气查询专家,用户输入城市名,你返回该城市的天气信息。请用简洁中文回复。", tools=[fetch_weather] ) …… (完整代码点击文末下方“阅读原文”查看)(三)添加护栏(input_guardrails)
现在已经有了两个“专家”,并且“专家”之间的依赖关系也已经声明了,开始搭建一些边界情况,保障整个项目的“确定性”,就是引入“护栏”(guardrails),为了演示这里使用的是输入“护栏”,保障整个功能只聚焦在穿衣相关的问题上。重点看“guardrail_agent”。
from typing import TypedDict from agents import Agent, GuardrailFunctionOutput, InputGuardrail, Runner, function_tool import asyncio from pydantic import BaseModel class DressOutput(BaseModel): is_dressing: bool reasoning: str # Guardrail聚焦穿衣建议 guardrail_agent = Agent( name="Guardrail check", instructions=("请判断用户的问题是否围绕用户穿衣建议相关内容,如果问题属于上述内容,请返回 is_dressing=True,并简要说明理由。"), output_type=DressOutput, ) …… (完整代码点击文末下方“阅读原文”查看)符合护栏输入的要求:
---- Guardrail result: RunResult: - Last agent: Agent(name="Guardrail check", ...) - Final output (DressOutput): { "is_dressing": true, "reasoning": "用户询问关于在北京穿着异域风格衣服的搭配建议,涉及穿衣搭配问题。" } - 1 new item(s) - 1 raw response(s) - 0 input guardrail result(s) - 0 output guardrail result(s) (See `RunResult` for more details) 在北京今天的天气是晴天,温度约20度,比较舒适,可以尝试以下异域风格的搭配: 1. **波西米亚风**: - 上衣:流苏或印花衬衫 - 下装:长裙或阔腿裤 - 配饰:大项链、耳环和宽檐帽 …… (完整代码点击文末下方“阅读原文”查看)在护栏中也可以设置"tripwire_triggered"参数来控制流程要不要终端,可以看具体的返回结果:
不终止任务,回答:
async def main(): result = await Runner.run(dressing_agent, input="上海天气如何") print(result.final_output) ----- Guardrail result: RunResult: - Last agent: Agent(name="Guardrail check", ...) - Final output (DressOutput): { "is_dressing": false, "reasoning": "用户询问的是上海的天气状况,与穿衣建议无直接关联。" } - 1 new item(s) - 1 raw response(s) - 0 input guardrail result(s) - 0 output guardrail result(s) (See `RunResult` for more details) 上海今天天气晴朗,温度为20度,湿度50%,东南风2级。 [non-fatal] Tracing: request failed: _ssl.c:989: The handshake operation timed o终止任务:
async def dress_guardrail(ctx, agent, input_data): result = await Runner.run(guardrail_agent, input_data, context=ctx.context) final_output = result.final_output_as(DressOutput) print("Guardrail result:", result) # 不再抛异常,直接返回判断结果 return GuardrailFunctionOutput( output_info=final_output, # tripwire_triggered=False, # 不触发终止 tripwire_triggered=True, # 触发终止 ) ---- Guardrail result: RunResult: - Last agent: Agent(name="Guardrail check", ...) - Final output (DressOutput): { "is_dressing": false, "reasoning": "用户的问题与当前天气情况有关,而不是关于穿衣建议的请求。" } …… (完整代码点击文末下方“阅读原文”查看)(四)Tracing
OpenAI的SDK默认开启了Tracing功能,可以支持手动关闭,以满足特定场景下的隐私或资源需求。
from agents import set_tracing_disabledset_tracing_disabled(True)1、官方展示
访问:https://platform.openai.com/traces
(1)多agent+交接
交互可以很好的看清楚对应token消耗,入参和系统提示词以及对应functioncall,两个agent交接展示也是functioncall,时速度很快,但是存在2点展示不太好:
①交接时 上下文展示
②交接时 token有没有消耗
(2)添加护栏对应的Tracing
可以看出使用护栏输入拦截后token可以节省,流程可以提前终止。
2、自己本地
有一些信息属于敏感信息贸然上传到服务器,会对数据安全和流程规范造成挑战。因此OpenAI的SDK支持多种解决方案:
①关闭Tracing功能
②自己搭建Tracing平台。这里演示本地运行mlflow用来记录
③对敏感数据或者节点关闭tracing。 RunConfig.trace_include_sensitive_data
(1)安装mlflow依赖
pip install mlflow # 启动服务 并且在本地使用sqllite存储 mlflow server --host 127.0.0.1 --port 8080 --backend-store-uri sqlite:///mlruns.db
from typing import TypedDict from agents import Agent, GuardrailFunctionOutput, InputGuardrail, Runner, function_tool import asyncio import mlflow from pydantic import BaseModel mlflow.openai.autolog() mlflow.set_tracking_uri("http://localhost:8080") mlflow.set_experiment("weather") class DressOutput(BaseModel): is_dressing: bool reasoning: str # Guardrail聚焦穿衣建议 guardrail_agent = Agent( name="Guardrail check", instructions=("请判断用户的问题是否围绕用户穿衣建议相关内容,如果问题属于上述内容,请返回 is_dressing=True,并简要说明理由。"), output_type=DressOutput,) …… (完整代码点击文末下方“阅读原文”查看)3、对比
(五)MCP使用
介绍完一些基础的用法后,我们看下最近比较火的MCP的接入实现。需要再在项目中插入MCP的定义,我们这里使用高德的MCP进行演示:
from typing import TypedDict from agents import Agent, GuardrailFunctionOutput, InputGuardrail, Runner, function_tool import asyncio from agents.mcp.server import MCPServerStdio import mlflow from pydantic import BaseModel # mlflow.openai.autolog() # mlflow.set_tracking_uri("http://localhost:8080") # mlflow.set_experiment("weather") class DressOutput(BaseModel): is_dressing: bool reasoning: str # Guardrail聚焦穿衣建议 guardrail_agent = Agent( name="Guardrail check", instructions=("请判断用户的问题是否围绕用户穿衣建议相关内容,如果问题属于上述内容,请返回 is_dressing=True,并简要说明理由。"), …… (完整代码点击文末下方“阅读原文”查看)OpenAI Agents SD代码示例到这里就告一段落了,大家可以自己尝试使用Deepseek替换一下模型看看效果(ps:deepseek还不支持 openai协议的Json 模版输出,所以需要做些改造)。
四、框架对比
目前市面上对于Agent智能体的搭建有很多方案,我这里只是对OpenAI SDK进行简短的介绍和使用。比如市面上比较火的有 工作流搭建框架 我司的Qmoss平台、开源的Dify、n8n、扣子、飞书 以及可以定制化开发Langchain、langchain4J、 langchain Graph、LlamaIndex等。
LangGraph 传送门:https://langchain-ai.github.io/langgraph/concepts/why-langgraph/
AutoGen传送门:https://microsoft.github.io/autogen/stable//user-guide/agentchat-user-guide/quickstart.html
langchain4j传送门:https://docs.langchain4j.dev/
对于多个Agent的交互和实现,我们再来看下LangGraph的简单示例,LangGraph 有两种多agnet的模型,多模型之间也是使用“交接”的方式。对于langgraph来说多个agent有两种架构设计:
一是监管架构(Supervisor):各个代理由中央 Supervisor 代理协调。主管控制所有通信流和任务委派,根据当前上下文和任务要求决定调用哪个代理。
另外一个是集群架构(Swarm):各个Agent根据其专长动态地将控制权移交给彼此。系统会记住最后一个处于活动状态的座席,确保在后续交互中,与该座席的对话恢复。
我们现在使用LangGraph的Supervisor架构来实现我们上面的“衣服搭配”demo(ps:专家的名字中文会报错):
# 首先安装依赖 pip install -U langgraph "langchain[anthropic]" pip install langgraph-supervisor使用本地的mlflow记录AI交互信息
import asyncio from langchain_openai import ChatOpenAI from langgraph.prebuilt import create_react_agent from langgraph_supervisor import create_supervisor from langchain_mcp_adapters.client import MultiServerMCPClient from langchain.tools import tool from PIL import Image from IPython.display import display import mlflow import pretty_util # Enabling tracing for LangGraph (LangChain) mlflow.langchain.autolog() # Optional: Set a tracking URI and an experiment mlflow.set_tracking_uri("http://localhost:8080") mlflow.set_experiment("LangGraph") …… (完整代码点击文末下方“阅读原文”查看)from langchain_core.messages import convert_to_messages def pretty_print_message(message, indent=False): pretty_message = message.pretty_repr(html=True) if not indent: print(pretty_message) return indented = "\n".join("\t" + c for c in pretty_message.split("\n")) print(indented) def pretty_print_messages(update, last_message=False): is_subgraph = False if isinstance(update, tuple): ns, update = update # skip parent graph updates in the printouts if len(ns) == 0: return graph_id = ns[-1].split(":")[0] …… (完整代码点击文末下方“阅读原文”查看)返回结果展示:
================================ Human Message ================================= 我在北京要去上海旅行3天,想去景点,拍点照片。 ================================== Ai Message ================================== Name: supervisor 为了帮助您规划这次北京到上海的3天旅行,我将协调以下几个专家的建议: 1. **天气专家(weather_agent)** - 提供上海未来几天的天气情况,以便您选择合适的衣物。 2. **地图专家(map_agent)** - 推荐上海的景点,以便您可以拍摄美丽的照片。 3. **穿衣专家(dressing_agent)** - 根据天气情况为您推荐合适的装扮。 稍等一下,我会先调用天气专家获取上海的天气情况。 Tool Calls: transfer_to_weather_agent (call_07EqbGkxmHZekxs9RCNdn90x) Call ID: call_07EqbGkxmHZekxs9RCNdn90x Args: ================================= Tool Message ================================= Name: transfer_to_weather_agent …… (完整代码点击文末下方“阅读原文”查看)从代码可以看出langgraph的可视化追踪能力还有点薄弱,如果需要达到OpenAI Agent SDK的效果需要可以尝试添加trace、graphstudio或者langsmith进行设计以实现可视化追踪。
五、总结
使用完OpenAI Agent SDK 和市面上的几个成熟框架的对比,个人认为未来AI应用肯定是工作流+多Agent的方向。对于工作流而言,就是让我们快速搭建出来一个MVP版本,感受下AI带来的效果到底如何,是否能达到预期,但是工作流随着业务迭代 如何有效的运维也是需要解决的问题。但是对于工程类项目,则需要定制化的开发,而且现在的AI需要应用到线上,部分业务场景需要保障确定性,这个使用OpenAI Agent SDK、LangGrap等开发框架是一个不错的选择。
六、参考
https://openai.github.io/openai-agents-python/
https://docs.dify.ai/en/introduction
https://docs.llamaindex.ai/en/stable/
https://langchain-ai.github.io/langgraph/concepts/why-langgraph/