理解这10个核心概念,你就学完了OpenAI最新的开源Agents SDK

点击上方蓝字加入我们

OpenA推出了全新的开源Agents SDK,这是一个独立的Agent开发工具,也是之前实验性的Swam的升级版,Agents SDK保留了其简洁易用的设计特点,同时在诸多方面进行了增强,使其具备了开发生产级Agent的能力。而且,它不依赖于OpenAI模型。

本文为大家详解Agents SDK中的十个关键概念或特性,并用一个案例贯穿始终 ,助你一文掌握这个强大的新工具。
本文代码地址:https://github.com/pingcy/agent_openai_sdk】

01



Models(大模型)

让我们首先从配置模型开始。

作为OpenAI的产品,其默认模型为自家模型。不过Agents SDK目前支持兼容OpenAI API协议的第三方模型。模型配置方法如下:

你需要配置一个client与model(第三方暂时只支持ChatCompletions类型的model),参考如下配置:

...
openai_client =
AsyncOpenAI(
    api_key = API_KEY,
    base_url = BASE_URL,
)

model =
OpenAIChatCompletionsModel(
    model='你的模型名字',
    openai_client=openai_client
)

其中API_KEY与BASE_URL设定为模型供应商提供的Key与端点。这是最灵活的一种方式,可以随意使用多个模型。如果你是经过统一的API网关使用模型,也可以设定全局的client与model类型,后续使用时只需要指定模型名称即可。如下:

set_default_openai_client(client=client, use_for_tracing=False)
set_default_openai_api("chat_completions")

需要注意的是,由于Agents SDK默认会把跟踪信息发送OpenAI,因此如果使用第三方API_KEY,会导致异常。所以需要暂时关闭全局跟踪:

set_tracing_disabled(disabled=True)

02


Agents(智能体)

Agent是Agents SDK中最核心的抽象。通过对它进行配置,你就可以快速创建一个属于自己的Agent。最基本的配置为:name(名称),instructions(系统提示),model(大模型):

main_agent = Agent(
        name="MainAssistant",
        instructions=(
            "通过回答问题来协助用户。" 
        ),
        model=model
    )

没错,这就是一个基本的Agent了,尽管它还只能简单的依赖LLM来回答问题,但后面我们会不断完善它。

此外,Agents SDK提供了一个创建Agent的快捷方法:直接克隆一个Agent,同时更改必要的属性:

#复制一个Agent,同时更改name和instructions,其他不变
clone_agent = main_agent.
clone (
    name="MainAssistantClone",
    instructions="通过回答问题来协助用户。但你只会说英语。",)

更多的模型参数将在后面的概念中进行丰富。

03



Runner(运行)

有了Agent,那么如何让它运行起来?

与其他框架不一样的是,你不能直接使用agent.run()或agent.chat()来调用Agent,你需要使用Runner类。它提供了多种方法来启动与管理Agent的运行,包括普通运行与流式运行。

【普通运行】

包含同步(run_sync)与异步(run)方式,同步是异步的包装,并无本质区别:

result = Runner.run_sync(
    main_agent,
    "你好!"
)
print(f'AI: {result.final_output}')

【流式运行】

一般框架都借助于事件流机制来实现Streaming输出,由于Agent系统运行中会产生很多跟踪事件,所以你需要对事件类型判断,将每次响应结果的增量部分实时的输出,产生流式效果。方式如下:

# 运行智能体并获取结果
result = Runner.run_streamed(
    main_agent,
    '你好' 
)

async for event in result.stream_events():
  if event.type == "raw_response_event" and isinstance(event.data, ResponseTextDeltaEvent):
    print(event.data.delta, end="", flush=True)

有必要的话你也可以通过事件判断,把运行的中间过程也用流式输出到前端(如Tool调用),这有助于向前端及时反馈运行进度,优化体验。

现在可以看到Agent的工作效果:

04



Tools(工具)

Tools(工具)使用是Agent的必备技能之一,在Agents SDK中有三种类型的工具:

  • 托管工具。OpenAI的云端工具,包括WebSearch(搜索),FileSearch(OpenAI的云端向量搜索)、ComputerTool(计算机任务),只能支持OpenAI模型。


  • 函数工具。这是基本上所有Agent框架都会支持的类型,即遵循一定规范的自定义函数,可用来实现自定义逻辑。


  • Agent工具。把一个Agent再包装成一个工具,这也为构建一个多Agent系统提供了基础。

【函数工具】

我们借助函数工具给Agent增加搜索能力:

@function_tool
def search_web(query_str:str):
    """
    使用Tavily进行网络搜索并返回结果
    Args:
        query_str: 搜索关键词
    Returns:
        result_str: 搜索的相关结果
    """

...
main_agent = Agent(
          ...
        tools = [search_web]
)

创建一个search_web函数,然后用@function_tool装饰成Agent使用的Tool,最后配置到Agent即可。

现在,Agent已经会自己做网络搜索了:

【Agent工具】

我们创建一个问题细化的Agent,并让它成为主Agent的一个Tool:

...    refine_agent = Agent(
      name="RefineAssistant",
      instructions="请根据输入的问题,生成3个细化的相关问题,不需要解释,只需要列出问题",
      model=model
    )

    main_agent = Agent[UserInfo](
        tools = [refine_agent.as_tool(tool_name="refine_question",tool_description="负责细化问题,以获得更多细节"),
                 search_web]    )

测试这个工具的使用:

05



Context(上下文)

Context是Agent系统运行过程中的全局状态对象,可在运行开始时注入,并在运行过程中传递到每一个关联的Agent与Tool等(类似Langgraph的State与LlamaIndex Workflow中的Context)。

  • Context可以放入任意内容,常见的是dataclass或pydantic对象

  • Context在Run方法时注入初始的Context

  • 后续你可以在Tool调用等过程中访问这个Context

  • Context会通过RunContextWrapper包装后传递


在我们的Demo中,如果希望在Agent运行过程中有一个用户信息的上下文(比如在一个多用户Agent系统中),你应该这么做:

#定义上下文类型
@dataclass
class UserInfo:
    UserId: str
    UserName: str

#在创建agent时指定上下文类型
main_agent = Agent[UserInfo](...)

# 搜索
@function_tool
def search_web(wrapper: RunContextWrapper[UserInfo],query_str:str):
...
    print(f'Start web search for 【{wrapper.context.UserName}】 with {query_str}...')
...

#运行时传入这个上下文实例
result = Runner.run_sync(
  main_agent, user_input,
  context=UserInfo('ID001','张三')
)

在Tool中对传入的Context信息做打印来验证:

在实际应用中,你可以在Agent运行中根据需要灵活应用Context,比如根据用户ID获得个性化信息等。

06



StructuredOutput(结构化输出)

很多时候我们都需要Agent做结构化输出,特别是在企业场景中,结构化输出对于后续的处理非常重要,无论是调用API还是输入给其他Agent处理。OpenAI Agents SDK可以方便的做结构化输出:

class Answer(BaseModel):
    '''
    用于定义回答的数据结构
    answer_chinese: 中文回答
    answer_english: 英文回答
    source: 回答参考知识来源
    '''

    answer_chinese: str
    answer_english: str
    source: str
...
main_agent = Agent[UserInfo](
  ...
        output_type=Answer
)

现在输出不再是一段文本,而是一个Answer对象:

不过很遗憾的是,目前结构化输出仅限于天然支持Structured Output的大模型。否则可能会出现如下错误:

如果你必须要使用其他模型,变通方法是通过Prompt让LLM做JSON格式的字符串输出后再自行转换,但有一定的失败概率。

07



Handoffs(转交)

转交是一种多Agent协作的范式:一个Agent可以将用户的请求委托给其他的Agent。

继续对我们的Agent改造,要求它把“不擅长”的数学问题转交给另外的Agent:

...
# 创建数学专家智能体实例
math_agent = Agent(
name="MathAssistant",
instructions="你是一个数学助手,专门解答数学相关的问题",
model=model,
tools=[calculator],
)

...
main_agent = Agent[UserInfo](
...
        instructions=(
            "通过中英文双语回答来协助用户。" 
            "如果询问数学问题,请交给MathAssistant。" 
        ),
        handoffs=[math_agent],
...
    )

这里创建了一个新的math_agent,使用工具calculator专门处理数学问题。现在运行Agent,并输入一个数学问题:

这里打印了Last Agent(最后处理的Agent,访问result.last_agent可得)的名字,可以看到它是MathAssistant,说明控制权已经发生了转移。

转交的基本过程总结如下:

  •  用户调用start_agent
  •  start_agent根据指令评估用户输入任务
  •  如果输入任务与某个可转交的Agent匹配,则转给对应的Agent
  •  控制权交给对应的Agent,如有必要,它也可以再次转交


转交模式的好处是:
  • 有利于模块化分工,多个Agent各司其职,互相协作

  • 有利于维护调试,单个Agent独立测试,最后组装

  • 有利于未来扩展,随时扩展新的Agent,并将部分任务转移


显然,转交可以用来实现多Agent的路由模式。

注意区分转交与工具Agent使用:

工具使用的控制权始终在主Agent;而转交模式的控制权会发生转移。比如你可以让两个翻译不同语言的智能体作为Tool使用,以实现一次翻译两种语言的功能;但是在转交模式下,你一旦转交出去,就会交给目标Agent来输出。

08



Guardrails(护栏)

Guardrails(护栏)是OpenAI Agents SDK非常有特点的设计。护栏是Agent运行的检查程序,通过快速的验证用户输入输出来保护您的Agent系统,以避免不必要的风险,有助于尽早检测恶意或不当请求,从而节省资源。护栏分为两种:

  • 输入护栏:验证初始用户输入。

  • 输出护栏:验证最终Agent输出。


护栏的使用分为几个步骤:

  1. 创建一个护栏Agent,这个Agent接收一个输入并作出判断,输出判断结果。

    这里我们增加一个内容审核的Agent:


...
class SensitiveCheckOutput(BaseModel):
    is_sensitive: bool
    reasoning: str

input_guardrail_agent = Agent(
    name="内容审核",
    instructions="""检查用户输入是否包含政治话题。
    如果内容包含以下内容,返回true:
    - 有争议的政治讨论
    - 敏感的地缘政治问题
    - 极端政治观点
    请为决定提供简短的理由。"""
,
    model=model,
    output_type=SensitiveCheckOutput,
)
  1. 创建护栏函数,这个函数接收与“被保护”的Agent相同的输入与上下文,并调用护栏Agent进行检查。其返回结果必须是GuardrailFunctionOutput类型,其中用tripwire_triggered属性来指示该请求是否要被拦截:


...
@input_guardrail
async def input_guardrail(
    ctx: RunContextWrapper[None], agent: Agent, input: str
)
 -> GuardrailFunctionOutput:

    result = await Runner.run(input_guardrail_agent, input, context=ctx.context)

    return GuardrailFunctionOutput(
        output_info=result.final_output,
        tripwire_triggered=result.final_output.is_sensitive,
    )
  1. 给“被保护”的Agent增加这个护栏。剩下的则交给框架:


...
main_agent = Agent[UserInfo](
    ...
    input_guardrails=[input_guardrail],
)

现在,如果你的话题过于“敏感”,将会被拦截:

你可以使用同样的方法对输出做检查,以识别与拦截AI可能的危险输出。需要注意的是输入护栏只会在Agent是流程的第一个Agent时被调用,而输出护栏只会在Agent是最后一个Agent时被调用。

09



Tracing(跟踪)

Agent系统的运行与调试是常见的一大麻烦。特别是基于高度抽象的框架开发的系统,由于隐藏了很多内部细节,导致可观测性较差,一般需要借助于专门的工程化平台如LangSmith,LangFuse等。OpenAI Agents SDK内置了较完善的Tracing机制,实现对运行过程中事件的跟踪,包括LLM调用、Tools调用、Agents转交等。

【Tracing开关】

可以使用如下方式全局关闭Tracing:

set_tracing_disabled(disabled=True)

你也可以在某次运行时关闭Trace:

result = Runner.run_sync(
                  main_agent,
                  user_input,
                  run_config=RunConfig(tracing_disabled=True)
)

【使用第三方跟踪器】

除非你使用OpenAI的API_Key,否则我们需要借助第三方跟踪器。这需要自定义实现一个TracingProcessor,然后调用add_trace_processor或者set_default_processor来增加或者替换默认的跟踪处理器。不过OpenAI已经支持了开箱即用的部分第三方工程平台,可以把Tracing信息发送到这些平台。

这里使用Pydantic的Logfire来跟踪我们的Agent(需要在https://logfire.pydantic.dev/申请注册),方法如下:

import logfire
logfire.configure(console=False)
logfire.instrument_openai_agents()

...
            #关闭默认的跟踪器
            set_trace_processors([])
            
            #自定义跟踪的workflow名称
            with trace(workflow_name="Test Workflow"):
                ....

现在可以在后台看到完整的Workflow过程及其每一步的调用细节:

【两个概念】

为了更好的观测与理解Tracing信息,你需要了解两个概念:

  • Traces:代表一次完整的端到端Workflow运行流程跟踪。

  • Spans:一次Trace中某个操作的跟踪,比如某次Tool调用。


所以,如果你编排了一个顺序执行的流程(涉及多个Agent的调用),可以用一个Trace来完整的跟踪其运行:

...
with trace(workflow_name="Test Workflow"):
   result1 = Runner.run_sync(
      agent1,
      user_input
   )
   #result1的输出作为agent2的输入
   result2 = Runner.run_sync(
      agent2,
      result1.final_output  )

这里两个Agent运行的跟踪信息都将被组织到"Test Workflow"中,你可以在跟踪平台证实这一点。

10



Orchestrating(编排)

无论是单智能体内部流程,还是多智能体的协作,都涉及到编排问题。编排代表应用中工作步骤与流程的规划与执行。目前Agent的编排有两种形式:

  • 借助LLM自身的规划能力:在一些观点里也被认为是真正的Agent,即由AI自主完成任务步骤的规划与执行,但目前LLM的能力还无法满足所有任务的要求。


  • 借助编码的工作流编排:这种编排方式牺牲了一定的通用型与灵活性,但增加了可靠性与可预测性,这在企业应用领域特别重要。很多开发框架提供了这样的能力,比如LangGraph,Llamaindex Workflows等。


不过,实际应用中往往是两种方式的结合。比如在一个程序编排的Workflow的局部环节,使用具有简单规划能力的Agent来完成某个子任务,是完全可能的。

从上面的Demo过程可以看到OpenAI Agents SDK目前与流程编排相关的支持:

  • 借助于工具使用,自主采取行动,获取数据,直到完成任务

  • 借助Handoffs可以把任务转交给其他Agent,实现多Agent的工作协作

  • 可以把Agent转化为Tool,实现多Agent间类似指挥者-执行者的工作模式

  • 借助Guardrails,也实现了一种Agent之间的协作


OpenAI没有提供类似LangGraph的程序编排框架,而是选择把这部分交给开发者。一些常见的多Agent工作流模式实现如下:

  • 顺序流程:一个Agent的输出作为另一个Agent的输入,串接起来即可。OpenAI提供了to_input_list函数来帮助串接历史消息:

...
#一个Agent进行生成,一个Agent进行评价
result = Runner.run_sync(
    main_agent, user_input,
    context=UserInfo('ID001','张三')
)
result = Runner.run_sync(
    rate_agent,
    result.to_input_list()+[{"role":"user","content":"请给出你对上述回答的评价"}])
  • 并行模式:如果任务有多个相互独立的步骤,或者需要在多个回复之间挑选,可以使用asyncio.gather来搜集多个并行任务的结果。

  • 路由模式:这可以借助Handoffs来实现,一个路由Agent根据规则把任务转交到其他Agent,这就是一种路由。

  • Supervisor模式:借助Agent的as_tool功能,把多个工作Agent转化为Tool给一个管理Agent协调使用即可。

  • 反思模式:一个Agent生成回复,然后使用第二个Agent提供评价与反馈。通过循环迭代,直到达到设置的条件。



结束语



整体上Open Agents SDK给我们的印象是:

  • 适合Agent特别是多Agent系统的快速构建

  • 简洁易用,适合快速上手,学习门槛低

  • 专注在最小集核心功能,其他的交给开发者

  • 更完善的生产级功能,如Tracing、Guardrails


同时,我们也期待未来的一些改进有:

  • 第三方模型更好的支持,特别是结构化输出

  • 更方便的TracingProcessor定制

  • 长期记忆能力

  • 与其他开发框架更简洁的集成

  • 更强大的流程编排能力


以上就是对OpenAI最新的Agents SDK核心能力的全解析,希望能够帮到大家,喜欢就点个赞吧!


THE END


福利时间

为了帮助LLM开发人员更系统性与更深入的学习RAG应用,特别是企业级的RAG应用场景下,当前主流的优化方法与技术实现,我们编写了《基于大模型的RAG应用开发与优化 — 构建企业级LLM应用》这本长达500页的开发与优化指南,与大家一起来深入到LLM应用开发的全新世界。

更多细节,点击链接了解


交流请识别以下名片

请使用浏览器的分享功能分享到微信等