OpenAI开源多Agent框架Swarm：熬夜实测整理出几大亮点

就在昨天，OpenAI开源了一个多智能体编排框架Swarm项目，上线几小时便获得2.8k Star。我马上就拉下来跑了一下，看看有没有别人说的那么厉害。跑完发现，虽然没有太多新东西，但还是有挺多亮点的。

我们先来简单了解一下 Swarm。

Swarm 是一个专注于使代理协调和执行轻量级、高度可控且易于测试的多智能体编排框架。

由于OpenAI官方也说到这是一个实验性质的，而非用于生产的项目，所以Swarm是一个轻量级多代理编排框架，主打特征是工效（ergonomic）与轻量（lightweight）。

Swarm 目前是一个实验性的示例框架，旨在探索多智能体系统的人体工程学接口。它不适用于生产环境，因此没有官方支持。（这也意味着我们不会审查拉取请求或问题！）

Swarm 的主要目标是展示在”编排智能体：交接与例程”指南中探讨的交接和例程模式。它并不是一个独立的库，主要用于教育目的。

——来自OpenAI的官方声明

跑完之后，我觉得相比目前市面上LangGraph，CrewAI等多智能体框架上手较为简单，与调用OpenAI官方的Chat Completions API非常相似。当前的 Swarm 版本更多像是一套从零到一搭建基础Agent框架的教学示例，没有特别多新意，当然它也有许多亮点。

亮点1：Swarm 的完全透明性

首先是 Swarm 的完全透明性，它可以对上下文、步骤和工具调用进行完全透明的打印，非常适合在Agent执行过程中想知道其中每一步发生了什么的开发人员。

亮点2：Swarm 的handoffs 机制

其次是它的handoffs 机制（对话交接）：允许让对话在Agent之间转移执行，在任何时间都可以选择将对话交接给另一个智能体。也就是说我和AgentA说：“我想和Agent B对话”，接下来的对话Agent A 便会把对话交给AgentB进行。

为了更直观的感受这个对话交接机制，接下来我将从零到一部署Swarm框架并进行简单的示例演示。并通过解析源代码详细的了解这背后的运行机制。

项目地址：https://github.com/openai/swarm/

首先根据官网指示，安装Swarm的环境依赖包：

pip install git+ssh://git@github.com/openai/swarm.git

下载完后我们就可以运行以下这段代码：

from swarm import Swarm, Agentclient = Swarm()def transfer_to_agent_b(): return agent_bagent_a = Agent( name=”Agent A”, instructions=”你是一个有用的助手”, functions=[transfer_to_agent_b],)agent_b = Agent( name=”Agent B”, instructions=”仅仅使用繁体字说话”,)response = client.run( agent=agent_a, messages=[{“role”: “user”, “content”: “我想和Agent B说话”}],)print(“User和AgentA说: 我想和Agent B说话。”)print(“Agent：”,response.messages[-1][“content”])

可以从以上代码中看到 client.run 执行的智能体是agent_a，编写一个返回 agent_b 的函数作为Agent_a 的函数调用。其中为了更好的区分开一会是 agent_a回复还是 agent_b 回复，我们让 agent_a 用简体中文，让 agent_b 用繁体字进行输出。点击运行执行这段代码后便输出以下内容：

可以看到结果，也就是AgentA 执行了我们的诉求并将对话交给了AgentB进行，一会我们将详细讨论这部分背后的实现原理。

从以上两个亮点都可以看出 Swarm 框架提供了一种灵活的方式来协调多个 AI 代理，允许进行复杂的交互和工作流程，同时保持实现的轻量级和可控。

接下来我们来了解一下Swarm的handoffs机制。

Swarm 解读——handoffs 机制

首先我为大家整理了一个表格，我们先根据以下的表格简单了解一下Swarm框架的关键组件、用法以及总体的执行流程。

📦概述：

轻量级多代理编排框架

专注于代理协调和执行

使用代理和切换作为关键原语

由 Chat Completions API 提供支持，通话之间无状态

🛠️关键组件：

Swarm 客户端 – 运行代理的主界面

Agent – 封装指令和函数

函数 – 代理可以调用的 Python 函数

交接 – 允许在代理之间转移执行

🔧用法：

通过 pip 从 GitHub repo 安装

创建 Swarm 客户端

定义代理的指令和功能

使用初始代理和消息运行客户端

处理函数调用、切换、上下文变量

🔁执行流程：

从当前代理获取完成信息

执行工具调用

如果发生交接，请切换代理

更新上下文变量

如果没有新的函数调用则返回

💻特点：

流媒体支持

调试日志

上下文变量

从 Python 函数生成函数模式

灵活的代理指令（字符串或函数）

问题：我明明是和agentA发消息的，为什么是agentB回我，因为给agentA加了一个函数transfer_to_agent_b，这一切是怎么发生的？

Swarm的核心逻辑代码部分在core.py文件，基于Swarm类的核心逻辑：

1、初始化：程序首先创建了Swarm客户端，以及两个Agent（A和B）。Agent A被赋予了一个特殊的函数 transfer_to_agent_b 。

client = Swarm() #创建了Swarm客户端

def transfer_to_agent_b(): return agent_b agent_a = Agent( name=”Agent A”, instructions=”你是一个有用的助手”, functions=[transfer_to_agent_b], #Agent A被赋予了一个特殊的函数transfer_to_agent_b。)

agent_b = Agent( name=”Agent B”, instructions=”仅仅使用繁体字说话”,)

2、开始对话：client.run() 方法被调用，初始Agent是Agent A，用户消息是”我想和Agent B说话”。

response = client.run( agent=agent_a, messages=[{“role”: “user”, “content”: “我想和Agent B说话”}],)

3、Agent A的响应：Agent A收到这个消息后，很可能理解了用户想与Agent B交谈的请求。进入run函数。

源代码 core.py/run 函数：def run( self, agent: Agent, messages: List, context_variables: dict = {}, model_override: str = None, stream: bool = False, debug: bool = False, max_turns: int = float(“inf”), execute_tools: bool = True,) -> Response: if stream: return self.run_and_stream( agent=agent, messages=messages, context_variables=context_variables, model_override=model_override, debug=debug, max_turns=max_turns, execute_tools=execute_tools, ) active_agent = agent context_variables = copy.deepcopy(context_variables) history = copy.deepcopy(messages) init_len = len(messages)

while len(history) – init_len < max_turns and active_agent: # get completion with current history, agent completion = self.get_chat_completion( agent=active_agent, history=history, context_variables=context_variables, model_override=model_override, stream=stream, debug=debug, ) message = completion.choices[0].message debug_print(debug, “Received completion:”, message) message.sender = active_agent.name history.append( json.loads(message.model_dump_json()) ) # to avoid OpenAI types (?)

if not message.tool_calls or not execute_tools: debug_print(debug, “Ending turn.”) break

# handle function calls, updating context_variables, and switching agents partial_response = self.handle_tool_calls( message.tool_calls, active_agent.functions, context_variables, debug ) history.extend(partial_response.messages) context_variables.update(partial_response.context_variables) if partial_response.agent: active_agent = partial_response.agent

return Response( messages=history[init_len:], agent=active_agent, context_variables=context_variables, )

以上第28、51、56、57行便是Agent 进行对话交接机制的核心部分，其中 active_agent 便是主导对话的Agent变量。

a.首先进入循环，从这行开始：

while len(history) – init_len < max_turns and active_agent:

这个循环会持续执行，直到达到最大回合数（max_turns）或者 active_agent 变为 None。

b.循环的主要步骤：

获取聊天完成（completion）

将新消息添加到历史记录

检查是否有工具调用（tool calls）

如果有工具调用，处理它们并可能切换 agent

更新上下文变量

c.循环的结束条件：

达到最大回合数

active_agent 变为 None

没有工具调用或 execute_tools 为 False

d.循环结束后：

返回一个 Response对象，包含了整个对话的历史、最后活跃的 agent 和上下文变量。

在我们前面的代码示例中，这个循环执行了两次：

– 第一次使用 AgentA，触发了 transfer_to_agent_b 函数，返回agent_b。handle_tool_calls 方法处理函数调用。当 transfer_to_agent_b 函数被调用时，它返回agent_b。这导致：

if partial_response.agent: active_agent = partial_response.agent

这行代码将活跃的Agent从A切换到B。

– 第二次使用 AgentB，active_agent 变为 None，退出循环，生成了最终的回复。

为什么选择 Swarm？

Swarm 是轻量级的、可扩展的，并且通过设计高度可定制。它最适合于处理大量难以编码为单个提示符的独立功能和指令的情况。

对于寻求完全托管线程和内置内存管理和检索的开发人员来说，Assistants API 是一个不错的选择。但是，Swarm 最适合希望对上下文、步骤和工具调用进行完全透明和精细控制的开发人员。Swarm （几乎）完全在客户端上运行，并且与 Chat Completions API 非常相似，它不会在调用之间存储状态。

—— Swarm 的官方文档

THE END

AI资讯