核心内容摘要
秘色流光，足尖风华：探寻八重神子踝位的绝美密码
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的执行直到完成任务结束可以调用functioncallMCP等工具来更好的完成任务。
对于许多业务流程涉及一系列需要外部信息或操作的步骤例如查询数据库、调用 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_KEYsk-...一最初天气查询首先我们先写一个简单的示例然后再开始一点点的把我们想要了解的特性走一遍。
我们先用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())
Agent属性说明我们可以参考源码中的结构发现有哪些属性我们可以调整或者修改理论上默认属性值只能够提供简单的使用如果涉及到自定义的功能那就需要调整更多的参数。
instructions使用instructions属性指定当前agent的系统提示词。
可以是字符串也可以是Agent生成的动态指令如果是一个函数将通过上下文和代理实例调用但是必须返回字符串。
name智能体的名字tools智能体可以使用工具列表handoff_description智能体的交接说明主要用于多个智能体进行交接时让大模型了解这个智能体能做什么以及什么时候调用他handoffsagent可以委派任务的子agent。
你可以提供一个 handoffs 的列表代理在适当的情况下可以选择将任务委派给它们。
这样可以实现职责分离和模块化设计。
model可以配置model默认是gpt4ohooks接收此代理的各种生命周期事件的回调的类。
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, wed 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( nameGuardrail check, instructions(请判断用户的问题是否围绕用户穿衣建议相关内容,如果问题属于上述内容请返回 is_dressingTrue并简要说明理由。
), output_typeDressOutput, ) …… 完整代码点击文末下方“阅读原文”查看符合护栏输入的要求---- Guardrail result: RunResult: - Last agent: Agent(nameGuardrail 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度比较舒适可以尝试以下异域风格的搭配
**波西米亚风**: - 上衣流苏或印花衬衫 - 下装长裙或阔腿裤 - 配饰大项链、耳环和宽檐帽 …… 完整代码点击文末下方“阅读原文”查看在护栏中也可以设置tripwire_triggered参数来控制流程要不要终端可以看具体的返回结果不终止任务回答async def main(): result await Runner.run(dressing_agent, input上海天气如何) print(result.final_output) ----- Guardrail result: RunResult: - Last agent: Agent(nameGuardrail 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, contextctx.context) final_output result.final_output_as(DressOutput) print(Guardrail result:, result) # 不再抛异常直接返回判断结果 return GuardrailFunctionOutput( output_infofinal_output, # tripwire_triggeredFalse, # 不触发终止 tripwire_triggeredTrue, # 触发终止 ) ---- Guardrail result: RunResult: - Last agent: Agent(nameGuardrail check, ...) - Final output (DressOutput): { is_dressing: false, reasoning: 用户的问题与当前天气情况有关而不是关于穿衣建议的请求。
} …… 完整代码点击文末下方“阅读原文”查看四TracingOpenAI的SDK默认开启了Tracing功能可以支持手动关闭以满足特定场景下的隐私或资源需求。
from agents import set_tracing_disabledset_tracing_disabled(True)
官方展示访问https://platform.openai.com/traces1多agent交接交互可以很好的看清楚对应token消耗入参和系统提示词以及对应functioncall两个agent交接展示也是functioncall时速度很快但是存在2点展示不太好①交接时 上下文展示②交接时 token有没有消耗2添加护栏对应的Tracing可以看出使用护栏输入拦截后token可以节省流程可以提前终止。
自己本地有一些信息属于敏感信息贸然上传到服务器会对数据安全和流程规范造成挑战。
因此OpenAI的SDK支持多种解决方案①关闭Tracing功能②自己搭建Tracing平台。
这里演示本地运行mlflow用来记录③对敏感数据或者节点关闭tracing。
RunConfig.trace_include_sensitive_data1安装mlflow依赖pip install mlflow # 启动服务 并且在本地使用sqllite存储 mlflow server --host
127.
0.
1 --port 8080 --backend-store-uri sqlite:///mlruns.dbfrom 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:
mlflow.set_experiment(weather) class DressOutput(BaseModel): is_dressing: bool reasoning: str # Guardrail聚焦穿衣建议 guardrail_agent Agent( nameGuardrail check, instructions(请判断用户的问题是否围绕用户穿衣建议相关内容,如果问题属于上述内容请返回 is_dressingTrue并简要说明理由。
), output_typeDressOutput,) …… 完整代码点击文末下方“阅读原文”查看
对比五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:
# mlflow.set_experiment(weather) class DressOutput(BaseModel): is_dressing: bool reasoning: str # Guardrail聚焦穿衣建议 guardrail_agent Agent( nameGuardrail check, instructions(请判断用户的问题是否围绕用户穿衣建议相关内容,如果问题属于上述内容请返回 is_dressingTrue并简要说明理由。
), …… 完整代码点击文末下方“阅读原文”查看OpenAI Agents SD代码示例到这里就告一段落了大家可以自己尝试使用Deepseek替换一下模型看看效果psdeepseek还不支持 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.htmllangchain4j传送门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:
mlflow.set_experiment(LangGraph) …… 完整代码点击文末下方“阅读原文”查看from langchain_core.messages import convert_to_messages def pretty_print_message(message, indentFalse): pretty_message message.pretty_repr(htmlTrue) 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_messageFalse): 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天旅行我将协调以下几个专家的建议
**天气专家weather_agent** - 提供上海未来几天的天气情况以便您选择合适的衣物。
**地图专家map_agent** - 推荐上海的景点以便您可以拍摄美丽的照片。
**穿衣专家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等开发框架是一个不错的选择。