点击上方蓝字加入我们


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输出。
护栏的使用分为几个步骤:
-
创建一个护栏Agent,这个Agent接收一个输入并作出判断,输出判断结果。
这里我们增加一个内容审核的Agent:
...
class SensitiveCheckOutput(BaseModel):
is_sensitive: bool
reasoning: str
input_guardrail_agent = Agent(
name="内容审核",
instructions="""检查用户输入是否包含政治话题。
如果内容包含以下内容,返回true:
- 有争议的政治讨论
- 敏感的地缘政治问题
- 极端政治观点
请为决定提供简短的理由。""",
model=model,
output_type=SensitiveCheckOutput,
)
-
创建护栏函数,这个函数接收与“被保护”的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,
)
-
给“被保护”的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应用开发的全新世界。
更多细节,点击链接了解
