一、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 就不做过多的阐述了,但是其他的四个特性还需要进一步详细阐述。

图片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
# 设置自己的OpenAIKey
export OPENAI_API_KEY=sk-...

(一)最初天气查询

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

图片2.png

    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属性说明

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

    图片3.png

    • 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模块会对交互细节进行阐述)

    图片4.png

      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

          图片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 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,)
          
          
          
          ……
          (完整代码点击文末下方“阅读原文”查看)

          图片12.png

          图片13.png

          3、对比

          图片

          (五)MCP使用

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

          图片14.png

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

            图片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 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等开发框架是一个不错的选择。

            六、参考

            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/

            Logo

            小龙虾开发者社区是 CSDN 旗下专注 OpenClaw 生态的官方阵地,聚焦技能开发、插件实践与部署教程,为开发者提供可直接落地的方案、工具与交流平台,助力高效构建与落地 AI 应用

            更多推荐