Qunar技术沙龙

浅谈 OpenAI Agents SDK

文章概览

Image
  • OpenAI  Agents SDK是什么?

  • 核心特性

  • 代码示例

  • 框架对比

  • 总结

  • 参考

保障数据情况
作者介绍:

李杰,2020年2月份加入去哪儿旅行酒店报价中心研发团队,酒店HTC成员,主要负责国内酒店报价实时和离线计算模块的研发和维护。

一、OpenAI  Agents SDK是什么?

    OpenAI Agents SDK是一个轻量级且易于使用的工具包,用于构建基于代理的AI应用程序。 提供了一些基本构建块,包括具备指令和工具的代理(Agents)、用于代理间任务委托的交接(Handoffs)以及用于输入验证的护栏(Guardrails)。

    在官网上OpenAI给出了两个理由使用它,而且也介绍了 Agent Loop、Python-first、Handoffs、Guardrails 、Function tools 和 Tracing 6个特性,对于现阶段的我们对Pyhon-first 和 Function tools 就不做过多的阐述了,但是其他的四个特性还需要进一步详细阐述。

图片1.png

二、核心特性

(一)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# 设置自己的OpenAIKeyexport OPENAI_API_KEY=sk-...

(一)最初天气查询

首先我们先写一个简单的示例,然后再开始一点点的把我们想要了解的特性走一遍。 我们先用python写一个简单的agent,可以查询天气:

图片2.png

from typing import TypedDictfrom agents import Agent, Runner, function_toolimport asyncio
@function_toolasync def fetch_weather(city: str) -> str:    """Fetch the weather for a given location.     Args:        city: The city to fetch the weather for.    """    return "sunny"
# 天气查询Agentweather_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属性说明

我们可以参考源码中的结构,发现有哪些属性我们可以调整或者修改,理论上默认属性值只能够提供简单的使用,如果涉及到自定义的功能那就需要调整更多的参数。

图片3.png

  • instructions:使用instructions属性指定当前agent的系统提示词。可以是字符串,也可以是Agent生成的动态指令,如果是一个函数,将通过上下文和代理实例调用,但是必须返回字符串。

  • name:智能体的名字

  • tools:智能体可以使用工具列表

  • handoff_description:智能体的交接说明,主要用于多个智能体进行交接时,让大模型了解这个智能体能做什么,以及什么时候调用他

  • handoffs:agent可以委派任务的子agent。你可以提供一个 handoffs 的列表,代理在适当的情况下可以选择将任务委派给它们。这样可以实现职责分离和模块化设计。

  • model:可以配置model,默认是gpt4o

  • hooks:接收此代理的各种生命周期事件的回调的类。

  • mcp_servers:agent可以使用的模型上下文协议 (MCP)服务器列表。每次代理运行时,它都会将这些服务器中的工具添加到可用工具列表中。

  • output_guardrails:  生成响应后,对代理的最终输出运行的检查列表。仅当代理生成最终输出时运行。

  • inpur_guardrails:  在生成响应之前,与代理执行并行运行的检查列表。仅当代理是链中的第一个代理时运行。

(二)多个Agent

现在开始构建多个Agent用来观察后面的新特性,例如本段代码示例为一个“天气专家”再加上一个“穿衣专家”两个Agent,回答用户的关于穿衣服的一些回答。 两个agent之间“穿衣专家” 使用 “天气专家”给出的天气信息进行总结天气因素,也就是需要两个专家的“交接”,最终给出对应穿衣风格搭配建议。(后面的tracing模块会对交互细节进行阐述)

图片4.png

from typing import TypedDictfrom agents import Agent, Runner, function_toolimport asyncio
@function_toolasync 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"
# 天气查询Agentweather_agent = Agent(                    name="天气查询专家",                     instructions="你是天气查询专家,用户输入城市名,你返回该城市的天气信息。请用简洁中文回复。",                    tools=[fetch_weather]                    )# 提示穿衣的agentdressing_agent = Agent(    name="穿衣提示专家",    instructions="你是穿衣提示专家,用户输入天气信息,你返回该天气的穿衣建议。请用简洁中文回复。",    handoff_description="当用户询问天气信息时,将任务转交给天气查询专家。我可以根据天气信息,给出穿衣建议。",    handoffs=[weather_agent])
async def main():    result = await Runner.run(dressing_agent, input="我在北京,今天穿什么衣服合适")    print(result.final_output)
if __name__ == "__main__":    asyncio.run(main())
----北京今天晴朗,建议穿轻便的衣物,如T恤和牛仔裤。如果温度较高,可以适当选择短袖。记得带上防晒用具哦!

(三)添加护栏(input_guardrails)

现在已经有了两个“专家”,并且“专家”之间的依赖关系也已经声明了,开始搭建一些边界情况,保障整个项目的“确定性”,就是引入“护栏”(guardrails),为了演示这里使用的是输入“护栏”,保障整个功能只聚焦在穿衣相关的问题上。重点看“guardrail_agent”。


from typing import TypedDictfrom agents import Agent, GuardrailFunctionOutput, InputGuardrail, Runner, function_toolimport 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,)
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,  # 不触发终止    )
@function_toolasync 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,温度20度,湿度50%,风力2级,风向东南"
# 天气查询Agentweather_agent = Agent(    name="天气查询专家",    instructions="你是天气查询专家,用户输入城市名,你返回该城市的天气信息。",    tools=[fetch_weather],    handoff_description="可以解决天气、气候、温度、湿度、风力、风向等天气相关问题。")# 提示穿衣的agentdressing_agent = Agent(    name="穿衣提示专家",    instructions="你是穿衣提示专家需要结合天气信息,给出符合用户需求的穿衣建议。",    input_guardrails=[InputGuardrail(name="聚焦穿衣搭配问题", guardrail_function=dress_guardrail)],    handoffs=[weather_agent])
async def main():    result = await Runner.run(dressing_agent, input="我在北京,需要穿一点异域风格的衣服,如何搭配")    print(result.final_output)
if __name__ == "__main__":    asyncio.run(main())

符合护栏输入的要求:


----
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. **波西米亚风**:   - 上衣:流苏或印花衬衫   - 下装:长裙或阔腿裤   - 配饰:大项链、耳环和宽檐帽
2. **摩洛哥风**:   - 上衣:刺绣或夹花罩衫   - 下装:灯笼裤或围裙裙   - 配饰:皮革腰带和手工编织包
3. **印度风**:   - 上衣:贴身花纹上衣或开襟毛衣   - 下装:绸缎长裙或萨莉   - 配饰:色彩丰富的围巾和镯子
建议选择轻便的材质,如棉或麻,以保持全天的舒适。同时,可以穿一双凉鞋以增加整体风格协调。

在护栏中也可以设置"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": "用户的问题与当前天气情况有关,而不是关于穿衣建议的请求。"    }- 1 new item(s)- 1 raw response(s)- 0 input guardrail result(s)- 0 output guardrail result(s)(See `RunResult` for more details)Traceback (most recent call last):  File "/Users/lijie/Documents/codeSpace/PycharmProjects/openaiSdk/weather.py", line 64, in <module>    asyncio.run(main())  File "/Users/lijie/DevTools/miniconda3/envs/daily_tools/lib/python3.11/asyncio/runners.py", line 190, in run    return runner.run(main)           ^^^^^^^^^^^^^^^^

(四)Tracing

OpenAI的SDK默认开启了Tracing功能,可以支持手动关闭,以满足特定场景下的隐私或资源需求。

from agents import set_tracing_disabledset_tracing_disabled(True)

1、官方展示

访问:https://platform.openai.com/traces

图片5.png

(1)多agent+交接

图片6.png

图片7.png图片8.png

图片9.png

交互可以很好的看清楚对应token消耗,入参和系统提示词以及对应functioncall,两个agent交接展示也是functioncall,时速度很快,但是存在2点展示不太好:

①交接时 上下文展示

②交接时 token有没有消耗

(2)添加护栏对应的Tracing

可以看出使用护栏输入拦截后token可以节省,流程可以提前终止。

图片10.png

图片11.png

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 TypedDictfrom agents import Agent, GuardrailFunctionOutput, InputGuardrail, Runner, function_toolimport asyncio
import mlflowfrom 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,)
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  # 不触发终止    )
@function_toolasync 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,温度20度,湿度50%,风力2级,风向东南"
# 天气查询Agentweather_agent = Agent(    name="天气查询专家",    instructions="你是天气查询专家,用户输入城市名,你返回该城市的天气信息。",    tools=[fetch_weather],    handoff_description="可以解决天气、气候、温度、湿度、风力、风向等天气相关问题。")# 提示穿衣的agentdressing_agent = Agent(    name="穿衣提示专家",    instructions="你是穿衣提示专家需要结合天气信息,给出符合用户需求的穿衣建议。",    input_guardrails=[InputGuardrail(name="聚焦穿衣搭配问题", guardrail_function=dress_guardrail)],    handoffs=[weather_agent])
async def main():    result = await Runner.run(dressing_agent, input="我在北京,需要穿一点异域风格的衣服,如何搭配")    print(result.final_output)
if __name__ == "__main__":    asyncio.run(main())

图片12.png图片13.png

3、对比

Image

(五)MCP使用

介绍完一些基础的用法后,我们看下最近比较火的MCP的接入实现。需要再在项目中插入MCP的定义,我们这里使用高德的MCP进行演示:

图片14.png

from typing import TypedDictfrom agents import Agent, GuardrailFunctionOutput, InputGuardrail, Runner, function_toolimport asynciofrom agents.mcp.server import MCPServerStdio
import mlflowfrom 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,)
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  # 不触发终止    )
@function_toolasync 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,温度20度,湿度50%,风力2级,风向东南"
amap_server = MCPServerStdio(    params={        "command": "npx",        "args": ["-y", "@amap/amap-maps-mcp-server"],        "env": {            "AMAP_MAPS_API_KEY": "需要申请"        }    },    cache_tools_list=True)async def init_servers():    await amap_server.connect()  # 初始化连接
# 地图专家map_agent = Agent(    name="景点推荐专家",    handoff_description="分析酒店地理位置、交通出行便利性、周边游玩资源,给出对应目的地景点信息和景点对应的风格。",    instructions="""景点推荐专家: 结合用户IP对应地址和的目的地给出行方案和目的地的出景点推荐,根据景点给出对应的风格。    1.使用工具获取景点信息。    2.生成对应的景点和景点风格    """,    mcp_servers=[amap_server],)
# 天气查询Agentweather_agent = Agent(    name="天气查询专家",    instructions="你是天气查询专家,用户输入城市名,你返回该城市的天气信息。",    tools=[fetch_weather],    handoff_description="可以解决天气、气候、温度、湿度、风力、风向等天气相关问题。")# 提示穿衣的agentdressing_agent = Agent(    name="穿衣提示专家",    instructions="你是穿衣提示专家需要结合天气信息,给出符合用户需求的穿衣建议。",    input_guardrails=[InputGuardrail(name="聚焦穿衣搭配问题", guardrail_function=dress_guardrail)],    handoffs=[weather_agent, map_agent],)
# 自动生成多个agent交互的流程图from agents.extensions.visualization import draw_graphdraw_graph(dressing_agent, filename="agent_graph")
async def main():    await init_servers()    result = await Runner.run(dressing_agent, input="我要去上海,如何搭配衣服比较适合,景点风格和衣服风格要一致,展示美感")    print(result.final_output)
if __name__ == "__main__":    asyncio.run(main())

OpenAI Agents SD代码示例到这里就告一段落了,大家可以自己尝试使用Deepseek替换一下模型看看效果(ps:deepseek还不支持 openai协议的Json 模版输出,所以需要做些改造)。

四、框架对比

目前市面上对于Agent智能体的搭建有很多方案,我这里只是对OpenAI SDK进行简短的介绍和使用。比如市面上比较火的有 工作流搭建框架 我司的Qmoss平台、开源的Dify、n8n、扣子、飞书 以及可以定制化开发Langchain、langchain4J、 langchain Graph、LlamaIndex等。

Image

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 代理协调。主管控制所有通信流和任务委派,根据当前上下文和任务要求决定调用哪个代理。

图片15.png

另外一个是集群架构(Swarm):各个Agent根据其专长动态地将控制权移交给彼此。系统会记住最后一个处于活动状态的座席,确保在后续交互中,与该座席的对话恢复。

图片16.png

我们现在使用LangGraph的Supervisor架构来实现我们上面的“衣服搭配”demo(ps:专家的名字中文会报错):

图片17.png

# 首先安装依赖pip install -U langgraph "langchain[anthropic]"pip install langgraph-supervisor

使用本地的mlflow记录AI交互信息

import asynciofrom langchain_openai import ChatOpenAIfrom langgraph.prebuilt import create_react_agentfrom langgraph_supervisor import create_supervisorfrom langchain_mcp_adapters.client import MultiServerMCPClientfrom langchain.tools import toolfrom PIL import Imagefrom IPython.display import display
import mlflow
import pretty_util
# Enabling tracing for LangGraph (LangChain)mlflow.langchain.autolog()
# Optional: Set a tracking URI and an experimentmlflow.set_tracking_uri("http://localhost:8080")mlflow.set_experiment("LangGraph")
@tooldef get_weather(city: str) -> str:      """获取指定城市的天气信息。"""    return f"sunny,温度20度,湿度50%,风力2级,风向东南"
async def main():    # model = init_chat_model("gpt-4o",temperature=0,top_p=0.9, max_tokens=1000)    model = ChatOpenAI(model="gpt-4o")    # 创建一个天气查询专家的反应式代理    weather_agent = create_react_agent(        name="weather_agent",        model= model,        tools=[get_weather],        prompt="你是天气查询专家,用户输入城市名,你返回该城市的天气信息。"    )    # 创建一个衣服搭配的专家    dressing_agent = create_react_agent(        name="dressing_agent",        model=model,        prompt="你是一个时尚的衣服搭配专家,针对拍摄场地和场地风格,给出合理的穿衣建议。",        tools=[]    )    map_mcp_client = MultiServerMCPClient(        {            "map": {                "command": "npx",                "args": ["-y", "@amap/amap-maps-mcp-server"],                "env": {                    "AMAP_MAPS_API_KEY": "579ec3c354f6631a9c0ae02bdaa3e45f"                    },                'transport': 'stdio',                }        }    )    map_tools = await map_mcp_client.get_tools()    # 创建一个地图专家    map_agent = create_react_agent(        name="map_agent",        model=model,        prompt="你是地图与出行专家,擅长分析地理位置和景点信息。请关注:\n"        "1. 景点到主要交通枢纽的距离和交通方式\n"        "2. 周边游玩资源分布\n"        "3. 出行路线和时间建议\n"        "4. 使用工具完成用户问题\n"        "请结合用户出行计划,用中文详细解答。",        tools=map_tools    )    # 总管    workflow = create_supervisor(        agents=[weather_agent, dressing_agent,map_agent],        model=model,        prompt="""        你是一个穿衣智能助手,负责协调不同的专家来回答用户的问题。\n        1. 明确有多少个专家参与回答。\n        2. 你需要根据用户的需求,调用不同的专家来获取信息。\n        3. 你需要将各个专家的回答整合成一个完整的答案。\n        """,        tools=[],    )    app = workflow.compile()    st = app.get_graph().draw_mermaid_png()    with open("agent_graph.png", "wb") as f:        f.write(st)
    app.stream({"messages": [{"role": "user", "content": ""}]})    for chunk in app.stream({"messages": [{"role": "user", "content": "我在北京要去上海旅行3天,想去景点,拍点照片。"}]}):        pretty_util.pretty_print_messages(chunk)

if __name__ == "__main__":    asyncio.run(main())
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]        print(f"Update from subgraph {graph_id}:")        print("\n")        is_subgraph = True    for node_name, node_update in update.items():        update_label = f"Update from node {node_name}:"        if is_subgraph:            update_label = "\t" + update_label        print(update_label)        print("\n")        messages = convert_to_messages(node_update["messages"])        if last_message:            messages = messages[-1:]        for m in messages:            pretty_print_message(m, indent=is_subgraph)        print("\n")

返回结果展示:

================================ 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
Successfully transferred to weather_agent

INFO:httpx:HTTP Request: POST https://api.openai.com/v1/chat/completions "HTTP/1.1 200 OK"INFO:httpx:HTTP Request: POST https://api.openai.com/v1/chat/completions "HTTP/1.1 200 OK"Update from node weather_agent:

================================== Ai Message ==================================Name: weather_agent
上海目前的天气情况是晴天,气温在20度,湿度为50%,东南风,风力为2级。这个天气非常适合外出游玩和拍照。您可以考虑穿着轻便舒适的衣物,带上一件薄外套以应对早晚的稍凉天气。
如果您需要关于上海景点的推荐或其他旅行建议,请随时告知!================================== Ai Message ==================================Name: weather_agent
Transferring back to supervisorTool Calls:  transfer_back_to_supervisor (7250caa4-078e-4141-a9b8-4c610bdd440f) Call ID: 7250caa4-078e-4141-a9b8-4c610bdd440f  Args:================================= Tool Message =================================Name: transfer_back_to_supervisor
Successfully transferred back to supervisor

INFO:httpx:HTTP Request: POST https://api.openai.com/v1/chat/completions "HTTP/1.1 200 OK"Update from node supervisor:

================================ 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
Successfully transferred to weather_agent================================== Ai Message ==================================Name: weather_agent
上海目前的天气情况是晴天,气温在20度,湿度为50%,东南风,风力为2级。这个天气非常适合外出游玩和拍照。您可以考虑穿着轻便舒适的衣物,带上一件薄外套以应对早晚的稍凉天气。
如果您需要关于上海景点的推荐或其他旅行建议,请随时告知!================================== Ai Message ==================================Name: weather_agent
Transferring back to supervisorTool Calls:  transfer_back_to_supervisor (7250caa4-078e-4141-a9b8-4c610bdd440f) Call ID: 7250caa4-078e-4141-a9b8-4c610bdd440f  Args:================================= Tool Message =================================Name: transfer_back_to_supervisor
Successfully transferred back to supervisor================================== Ai Message ==================================Name: supervisorTool Calls:  transfer_to_map_agent (call_qm4gzd25ykA0UjrOizx7Z94H) Call ID: call_qm4gzd25ykA0UjrOizx7Z94H  Args:================================= Tool Message =================================Name: transfer_to_map_agent
Successfully transferred to map_agent

INFO:httpx:HTTP Request: POST https://api.openai.com/v1/chat/completions "HTTP/1.1 200 OK"Update from node map_agent:

================================== Ai Message ==================================Name: map_agent
好的,接下来我将为您推荐一些上海的知名景点,每个景点都非常适合拍摄美丽的照片:
1. **外滩** - 上海最具代表性的地标,夜景尤为壮观。2. **东方明珠电视塔** - 登上观光层可以俯瞰整个城市的风光。3. **南京路步行街** - 这里是购物和感受城市活力的好去处。4. **豫园** - 一座经典的江南园林,可以体验到古典的东方建筑美。5. **上海博物馆** - 拥有丰富的中国古代艺术藏品。6. **田子坊** - 富有艺术和创意的小巷,非常适合文艺照片拍摄。7. **上海迪士尼乐园** - 如果时间允许,可以去这里感受乐趣。8. **朱家角古镇** - 感受传统的江南水乡风情。
探索这些地方会让您的旅行更加充实。如果您需要从北京前往上海的具体交通信息和建议,或者关于各个景点的详细咨询,请告知我!================================== Ai Message ==================================Name: map_agent
Transferring back to supervisorTool Calls:  transfer_back_to_supervisor (2cf8761e-c294-4942-ad46-35091a736f1c) Call ID: 2cf8761e-c294-4942-ad46-35091a736f1c  Args:================================= Tool Message =================================Name: transfer_back_to_supervisor
Successfully transferred back to supervisor

INFO:httpx:HTTP Request: POST https://api.openai.com/v1/chat/completions "HTTP/1.1 200 OK"Update from node supervisor:、
================================ 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
Successfully transferred to weather_agent================================== Ai Message ==================================Name: weather_agent
上海目前的天气情况是晴天,气温在20度,湿度为50%,东南风,风力为2级。这个天气非常适合外出游玩和拍照。您可以考虑穿着轻便舒适的衣物,带上一件薄外套以应对早晚的稍凉天气。如果您需要关于上海景点的推荐或其他旅行建议,请随时告知!
================================== Ai Message ==================================Name: weather_agent
Transferring back to supervisorTool Calls:  transfer_back_to_supervisor (7250caa4-078e-4141-a9b8-4c610bdd440f) Call ID: 7250caa4-078e-4141-a9b8-4c610bdd440f  Args:================================= Tool Message =================================Name: transfer_back_to_supervisor
Successfully transferred back to supervisor================================== Ai Message ==================================Name: supervisorTool Calls:  transfer_to_map_agent (call_qm4gzd25ykA0UjrOizx7Z94H) Call ID: call_qm4gzd25ykA0UjrOizx7Z94H  Args:================================= Tool Message =================================Name: transfer_to_map_agent
Successfully transferred to map_agent================================== Ai Message ==================================Name: map_agent
好的,接下来我将为您推荐一些上海的知名景点,每个景点都非常适合拍摄美丽的照片:
1. **外滩** - 上海最具代表性的地标,夜景尤为壮观。2. **东方明珠电视塔** - 登上观光层可以俯瞰整个城市的风光。3. **南京路步行街** - 这里是购物和感受城市活力的好去处。4. **豫园** - 一座经典的江南园林,可以体验到古典的东方建筑美。5. **上海博物馆** - 拥有丰富的中国古代艺术藏品。6. **田子坊** - 富有艺术和创意的小巷,非常适合文艺照片拍摄。7. **上海迪士尼乐园** - 如果时间允许,可以去这里感受乐趣。8. **朱家角古镇** - 感受传统的江南水乡风情。
探索这些地方会让您的旅行更加充实。如果您需要从北京前往上海的具体交通信息和建议,或者关于各个景点的详细咨询,请告知我!================================== Ai Message ==================================Name: map_agent
Transferring back to supervisorTool Calls:  transfer_back_to_supervisor (2cf8761e-c294-4942-ad46-35091a736f1c) Call ID: 2cf8761e-c294-4942-ad46-35091a736f1c  Args:================================= Tool Message =================================Name: transfer_back_to_supervisor
Successfully transferred back to supervisor================================== Ai Message ==================================Name: supervisorTool Calls:  transfer_to_dressing_agent (call_TJ0XVKvFIjEma0DP4muJxoHx) Call ID: call_TJ0XVKvFIjEma0DP4muJxoHx  Args:================================= Tool Message =================================Name: transfer_to_dressing_agent
Successfully transferred to dressing_agent

INFO:httpx:HTTP Request: POST https://api.openai.com/v1/chat/completions "HTTP/1.1 200 OK"Update from node dressing_agent:

================================== Ai Message ==================================Name: dressing_agent
根据上海的天气和您计划访问的景点,我为您提供以下穿搭建议:
1. **外滩和南京路步行街**(城市景观拍照):   - **上装**:选择一件白色衬衫或颜色亮丽的薄毛衣,可以搭配一件浅色风衣。   - **下装**:牛仔裤或修身裤,容易搭配又显气质。   - **鞋子**:舒适的休闲鞋或小白鞋,可以长时间行走。   - **配饰**:墨镜和一顶宽檐帽,特别是拍摄户外照片时可以增添时尚感。
2. **豫园和朱家角古镇**(传统文化与自然景观):   - **上装**:可以选择中式风格的上衣,例如带有刺绣或民族风图案的上衣。   - **下装**:宽松的阔腿裤或者飘逸的长裙,营造古典气息。   - **鞋子**:平底鞋或舒适的凉鞋,便于在古镇小路行走。   - **配饰**:精致的耳环或一些带有东方元素的小饰品。
3. **东方明珠和上海博物馆**(现代城市与文化探索):   - **上装**:选择修身的西装外套或夹克,搭配简单的打底T恤。   - **下装**:商务休闲裤或牛仔裤,都能适应室内室外的环境。   - **鞋子**:乐福鞋或时尚的运动鞋,集时尚与舒适于一体。   - **配饰**:简约设计的手表或手链,增加整体造型的精致感。
总的来说,注重舒适与时尚的平衡,方便您随时随地捕捉到美丽的瞬间!希望这些建议能够帮助到您,如果有其他具体需求,欢迎向我咨询!================================== Ai Message ==================================Name: dressing_agent
Transferring back to supervisorTool Calls:  transfer_back_to_supervisor (9aefd114-5a69-429e-9a44-f5f17f7179f0) Call ID: 9aefd114-5a69-429e-9a44-f5f17f7179f0  Args:================================= Tool Message =================================Name: transfer_back_to_supervisor
Successfully transferred back to supervisor

INFO:httpx:HTTP Request: POST https://api.openai.com/v1/chat/completions "HTTP/1.1 200 OK"Update from node supervisor:

================================ 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
Successfully transferred to weather_agent================================== Ai Message ==================================Name: weather_agent
上海目前的天气情况是晴天,气温在20度,湿度为50%,东南风,风力为2级。这个天气非常适合外出游玩和拍照。您可以考虑穿着轻便舒适的衣物,带上一件薄外套以应对早晚的稍凉天气。
如果您需要关于上海景点的推荐或其他旅行建议,请随时告知!================================== Ai Message ==================================Name: weather_agent
Transferring back to supervisorTool Calls:  transfer_back_to_supervisor (7250caa4-078e-4141-a9b8-4c610bdd440f) Call ID: 7250caa4-078e-4141-a9b8-4c610bdd440f  Args:================================= Tool Message =================================Name: transfer_back_to_supervisor
Successfully transferred back to supervisor================================== Ai Message ==================================Name: supervisorTool Calls:  transfer_to_map_agent (call_qm4gzd25ykA0UjrOizx7Z94H) Call ID: call_qm4gzd25ykA0UjrOizx7Z94H  Args:================================= Tool Message =================================Name: transfer_to_map_agent
Successfully transferred to map_agent================================== Ai Message ==================================Name: map_agent
好的,接下来我将为您推荐一些上海的知名景点,每个景点都非常适合拍摄美丽的照片:
1. **外滩** - 上海最具代表性的地标,夜景尤为壮观。2. **东方明珠电视塔** - 登上观光层可以俯瞰整个城市的风光。3. **南京路步行街** - 这里是购物和感受城市活力的好去处。4. **豫园** - 一座经典的江南园林,可以体验到古典的东方建筑美。5. **上海博物馆** - 拥有丰富的中国古代艺术藏品。6. **田子坊** - 富有艺术和创意的小巷,非常适合文艺照片拍摄。7. **上海迪士尼乐园** - 如果时间允许,可以去这里感受乐趣。8. **朱家角古镇** - 感受传统的江南水乡风情。
探索这些地方会让您的旅行更加充实。如果您需要从北京前往上海的具体交通信息和建议,或者关于各个景点的详细咨询,请告知我!================================== Ai Message ==================================Name: map_agent
Transferring back to supervisorTool Calls:  transfer_back_to_supervisor (2cf8761e-c294-4942-ad46-35091a736f1c) Call ID: 2cf8761e-c294-4942-ad46-35091a736f1c  Args:================================= Tool Message =================================Name: transfer_back_to_supervisor
Successfully transferred back to supervisor================================== Ai Message ==================================Name: supervisorTool Calls:  transfer_to_dressing_agent (call_TJ0XVKvFIjEma0DP4muJxoHx) Call ID: call_TJ0XVKvFIjEma0DP4muJxoHx  Args:================================= Tool Message =================================Name: transfer_to_dressing_agent
Successfully transferred to dressing_agent================================== Ai Message ==================================Name: dressing_agent
根据上海的天气和您计划访问的景点,我为您提供以下穿搭建议:
1. **外滩和南京路步行街**(城市景观拍照):   - **上装**:选择一件白色衬衫或颜色亮丽的薄毛衣,可以搭配一件浅色风衣。   - **下装**:牛仔裤或修身裤,容易搭配又显气质。   - **鞋子**:舒适的休闲鞋或小白鞋,可以长时间行走。   - **配饰**:墨镜和一顶宽檐帽,特别是拍摄户外照片时可以增添时尚感。
2. **豫园和朱家角古镇**(传统文化与自然景观):   - **上装**:可以选择中式风格的上衣,例如带有刺绣或民族风图案的上衣。   - **下装**:宽松的阔腿裤或者飘逸的长裙,营造古典气息。   - **鞋子**:平底鞋或舒适的凉鞋,便于在古镇小路行走。   - **配饰**:精致的耳环或一些带有东方元素的小饰品。
3. **东方明珠和上海博物馆**(现代城市与文化探索):   - **上装**:选择修身的西装外套或夹克,搭配简单的打底T恤。   - **下装**:商务休闲裤或牛仔裤,都能适应室内室外的环境。   - **鞋子**:乐福鞋或时尚的运动鞋,集时尚与舒适于一体。   - **配饰**:简约设计的手表或手链,增加整体造型的精致感。
总的来说,注重舒适与时尚的平衡,方便您随时随地捕捉到美丽的瞬间!希望这些建议能够帮助到您,如果有其他具体需求,欢迎向我咨询!================================== Ai Message ==================================Name: dressing_agent
Transferring back to supervisorTool Calls:  transfer_back_to_supervisor (9aefd114-5a69-429e-9a44-f5f17f7179f0) Call ID: 9aefd114-5a69-429e-9a44-f5f17f7179f0  Args:================================= Tool Message =================================Name: transfer_back_to_supervisor
Successfully transferred back to supervisor================================== Ai Message ==================================Name: supervisor
以下是为您规划的3天上海旅行建议:
1. **天气情况**:   - 上海目前是晴天,气温在20度左右,非常适合旅游和拍照。   - 建议穿着轻便舒适的衣物,并带上一件薄外套以应对早晚的稍凉天气。
2. **推荐景点**:   - 外滩、东方明珠电视塔、南京路步行街、豫园、上海博物馆、田子坊、上海迪士尼乐园、朱家角古镇。   - 这些地方既能感受城市的现代魅力,又能体验传统文化,非常适合拍摄美丽照片。
3. **穿搭建议**:   - 城市景观如外滩:白色衬衫或薄毛衣搭配浅色风衣,牛仔裤或修身裤,舒适的休闲鞋。   - 古典文化如豫园:中式风格的上衣搭配阔腿裤或长裙,平底鞋,带有东方元素的配饰。   - 现代与文化探索如东方明珠:修身西装外套或夹克,搭配商务休闲裤或牛仔裤,乐福鞋。
希望这些建议能为您的上海之旅提供帮助。如果需要更多信息或具体安排,随时告诉我!旅途愉快!

从代码可以看出langgraph的可视化追踪能力还有点薄弱,如果需要达到OpenAI Agent SDK的效果需要可以尝试添加trace、graphstudio或者langsmith进行设计以实现可视化追踪。

五、总结

使用完OpenAI Agent SDK 和市面上的几个成熟框架的对比,个人认为未来AI应用肯定是工作流+多Agent的方向。对于工作流而言,就是让我们快速搭建出来一个MVP版本,感受下AI带来的效果到底如何,是否能达到预期,但是工作流随着业务迭代 如何有效的运维也是需要解决的问题。但是对于工程类项目,则需要定制化的开发,而且现在的AI需要应用到线上,部分业务场景需要保障确定性,这个使用OpenAI Agent SDK、LangGrap等开发框架是一个不错的选择。

六、参考

  1. https://openai.github.io/openai-agents-python/
  2. https://docs.dify.ai/en/introduction
  3. https://docs.llamaindex.ai/en/stable/
  4. https://langchain-ai.github.io/langgraph/concepts/why-langgraph/