Swarm（实验性、教育用途）

[!重要] Swarm 现已由 OpenAI Agents SDK 取代，它是 Swarm 的生产就绪版本。Agents SDK 具有关键改进，并将由 OpenAI 团队积极维护。

我们建议所有生产环境的使用都迁移到 Agents SDK。

安装

需要 Python 3.10 或更高版本

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

或者

pip install git+https://github.com/openai/swarm.git

使用方法

from swarm import Swarm, Agent

client = Swarm()

def transfer_to_agent_b():
    return agent_b


agent_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(response.messages[-1]["content"])

希望熠熠生辉，
新路优雅交汇，
有何可以帮您？

目录

• 概述
• 示例
• 文档
  • 运行 Swarm
  • 智能体
  • 函数
  • 流式处理
• 评估
• 工具

概述

Swarm 致力于使智能体之间的协调和执行轻量级、高度可控且易于测试。

它通过两种基本抽象实现这一目标：Agent 和交接。一个 Agent 包含 instructions 和 tools，并且可以在任何时候选择将对话交接给另一个 Agent。

这些基本抽象足够强大，能够表达工具与智能体网络之间的丰富动态，使您能够在避免陡峭学习曲线的同时构建可扩展的现实世界解决方案。

[!注意] Swarm 中的智能体与 Assistants API 中的 Assistants 无关。它们之所以名称相似只是为了方便，但实际上完全不相关。Swarm 完全由 Chat Completions API 提供支持，因此在每次调用之间是无状态的。

为什么选择 Swarm

Swarm 探索的是设计上轻量级、可扩展且高度可定制的模式。类似 Swarm 的方法最适合处理大量独立能力以及难以编码到单个提示中的复杂指令的情况。

对于希望使用完全托管的线程以及内置记忆管理和检索功能的开发者来说，Assistants API 是一个很好的选择。然而，Swarm 则是为那些对多智能体编排感兴趣的开发者提供的教育资源。Swarm 几乎完全在客户端运行，而且与 Chat Completions API 类似，在每次调用之间不会存储状态。

示例

请查看 /examples 获取灵感！每个示例的详细信息请参阅其 README 文件。

• basic：基础示例，包括设置、函数调用、交接和上下文变量等基本内容。
• triage_agent：简单的分诊示例，用于将用户引导至合适的智能体。
• weather_agent：简单的函数调用示例。
• airline：一个多智能体系统，用于在航空公司环境中处理不同的客户服务请求。
• support_bot：一个包含用户界面智能体和帮助中心智能体的客服机器人，配备多种工具。
• personal_shopper：一个个人购物助手，可以帮助完成销售并处理订单退款。

文档

运行 Swarm

首先实例化一个 Swarm 客户端（内部只是实例化了一个 OpenAI 客户端）。

from swarm import Swarm

client = Swarm()

client.run()

Swarm 的 run() 函数类似于 Chat Completions API 中的 chat.completions.create() 函数——它接收 messages 参数并返回 messages，且在每次调用之间不会保存任何状态。然而，更重要的是，它还负责处理 Agent 的函数执行、任务交接、上下文变量引用，并且可以在返回给用户之前进行多轮交互。

从本质上讲，Swarm 的 client.run() 实现了以下循环：

1. 从当前 Agent 获取响应
2. 执行工具调用并将结果附加到消息中
3. 如有必要，切换 Agent
4. 如有必要，更新上下文变量
5. 如果没有新的函数调用，则返回结果

参数

参数	类型	描述	默认值
agent	Agent	要调用的（初始）Agent。	(必填)
messages	List	消息对象列表，与 Chat Completions 的 messages 完全相同	(必填)
context_variables	dict	一组额外的上下文变量字典，可供函数和 Agent 指令使用	{}
max_turns	int	允许的最大对话轮次数	float("inf")
model_override	str	可选字符串，用于覆盖 Agent 使用的模型	None
execute_tools	bool	如果为 False，当 Agent 尝试调用函数时，将中断执行并立即返回包含 tool_calls 的消息	True
stream	bool	如果为 True，则启用流式响应	False
debug	bool	如果为 True，则启用调试日志	False

一旦 client.run() 完成（可能经过多次 Agent 和工具的调用），它将返回一个 Response 对象，其中包含所有相关的更新状态。具体来说，包括新的 messages、最后被调用的 Agent 以及最新的 context_variables。您可以将这些值（加上新的用户消息）传递到下一次 client.run() 调用中，以从中断的地方继续交互——这与 chat.completions.create() 的行为非常相似。（run_demo_loop 函数在 /swarm/repl/repl.py 中实现了一个完整的执行循环示例。）

Response 字段

字段	类型	描述
messages	List	在对话过程中生成的消息对象列表。与 Chat Completions 的 messages 非常相似，但增加了一个 sender 字段，用于指示消息来自哪个 Agent。
agent	Agent	最后处理消息的 Agent。
context_variables	dict	与输入变量相同，但包含了任何已发生的变更。

Agents

一个 Agent 简单地封装了一组 instructions 和一组 functions（以及一些额外的设置），并且具备将执行权移交给他者的能力。

尽管我们倾向于将 Agent 比作“做某事的人”，但它也可以用来表示由一组 instructions 和 functions 定义的非常具体的流程或步骤（例如一系列步骤、复杂的检索操作、单一的数据转换步骤等）。这使得 Agent 可以组合成一个由“代理”、“工作流”和“任务”组成的网络，而所有这些都可以用相同的原语来表示。

Agent 字段

字段	类型	描述	默认
name	str	Agent 的名称。	"Agent"
model	str	Agent 使用的模型。	"gpt-4o"
instructions	str 或 func() -> str	Agent 的指令，可以是字符串，也可以是一个返回字符串的可调用函数。	"You are a helpful agent."
functions	List	Agent 可以调用的函数列表。	[]
tool_choice	str	Agent 的工具选择（如果有）。	None

指令

Agent 的 instructions 会直接转换为对话的 system 提示（作为第一条消息）。在任何给定时刻，只有当前活动的 Agent 的 instructions 才会存在（例如，如果发生 Agent 接手，system 提示会发生变化，但聊天历史不会改变）。

agent = Agent(
   instructions="你是一个乐于助人的智能体。"
)

instructions 可以是普通的 str，也可以是一个返回 str 的函数。该函数可以选择性地接收一个 context_variables 参数，这个参数将由传递给 client.run() 的 context_variables 填充。

def instructions(context_variables):
   user_name = context_variables["user_name"]
   return f"帮助用户 {user_name} 完成他们想要做的事情。"

agent = Agent(
   instructions=instructions
)
response = client.run(
   agent=agent,
   messages=[{"role":"user", "content": "你好！"}],
   context_variables={"user_name":"John"}
)
print(response.messages[-1]["content"])

你好 John，今天有什么可以帮您的吗？

函数

• Swarm 的 Agent 可以直接调用 Python 函数。
• 函数通常应返回 str（其他类型的值会被尝试转换为 str）。
• 如果函数返回一个 Agent，执行将转移到该 Agent。
• 如果函数定义了 context_variables 参数，它将由传递给 client.run() 的 context_variables 填充。

def greet(context_variables, language):
   user_name = context_variables["user_name"]
   greeting = "Hola" if language.lower() == "spanish" else "Hello"
   print(f"{greeting}, {user_name}!")
   return "完成"

agent = Agent(
   functions=[greet]
)

client.run(
   agent=agent,
   messages=[{"role": "user", "content": "请使用 greet()。"}],
   context_variables={"user_name": "John"}
)

Hola, John!

• 如果 Agent 调用函数时出现错误（函数不存在、参数错误或运行时错误），错误信息会被添加到聊天记录中，以便 Agent 能够优雅地恢复。
• 如果 Agent 调用了多个函数，这些函数将按照顺序依次执行。

接手与更新上下文变量

Agent 可以通过在函数中返回另一个 Agent 来进行接手。

sales_agent = Agent(name="销售代理")

def transfer_to_sales():
   return sales_agent

agent = Agent(functions=[transfer_to_sales])

response = client.run(agent, [{"role":"user", "content":"把我转接到销售部门。"}])
print(response.agent.name)

销售代理

它也可以通过返回一个更完整的 Result 对象来更新 context_variables。这个对象还可以包含 value 和 agent，这样你就可以让一个函数同时返回值、更新 Agent 并更新上下文变量（或者其中任意一部分）。

sales_agent = Agent(name="销售代理")

def talk_to_sales():
   print("你好，世界！")
   return Result(
       value="完成",
       agent=sales_agent，
       context_variables={"department": "sales"}
   )

agent = Agent(functions=[talk_to_sales])

response = client.run(
   agent=agent,
   messages=[{"role": "user", "content": "把我转接到销售部门"}],
   context_variables={"user_name": "John"}
)
print(response.agent.name)
print(response.context_variables)

销售代理
{'department': 'sales', 'user_name': 'John'}

[!注意] 如果一个 Agent 调用多个函数来进行接手，最终只会采用最后一个接手函数。

函数 Schema

Swarm 会自动将函数转换为 JSON Schema，并将其传递给 Chat Completions 的 tools。

• 文档字符串会变成函数的 description。
• 没有默认值的参数会被设置为 required。
• 类型提示会被映射到参数的 type（默认为 string）。
• 虽然不显式支持每个参数的描述，但如果直接在文档字符串中添加，效果应该类似。（未来可能会增加对文档字符串参数解析的支持）

def greet(name, age: int, location: str = "New York"):
   """向用户问好。调用前请确保获取用户的姓名和年龄。

   Args:
      name: 用户的姓名。
      age: 用户的年龄。
      location: 地球上最好的地方。
   """
   print(f"你好 {name}, 很高兴你在 {age} 岁时身处 {location}！")

{
   "type": "function",
   "function": {
      "name": "greet",
      "description": "向用户问好。调用前请确保获取用户的姓名和年龄。\n\nArgs:\n   name: 用户的姓名。\n   age: 用户的年龄。\n   location: 地球上最好的地方。",
      "parameters": {
         "type": "object",
         "properties": {
            "name": {"type": "string"},
            "age": {"type": "integer"},
            "location": {"type": "string"}
         },
         "required": ["name", "age"]
      }
   }
}

流式处理

stream = client.run(agent, messages, stream=True)
for chunk in stream:
   print(chunk)

使用与 Chat Completions API 流式处理 相同的事件。可参考 /swarm/repl/repl.py 中的 process_and_print_streaming_response 作为示例。

新增了两种事件类型：

• {"delim":"start"} 和 {"delim":"end"}，用于标记每次 Agent 处理单条消息（响应或函数调用）的开始和结束。这有助于识别 Agent 之间的切换。
• {"response": Response} 会在流结束时返回一个 Response 对象，其中包含聚合后的完整响应，以方便使用。

评估

评估对于任何项目都至关重要，我们鼓励开发者使用自己的评估套件来测试其 Swarm 的性能。作为参考，我们在 airline、weather_agent 和 triage_agent 快速入门示例中提供了一些评估方法。更多详情请参阅各项目的 README 文件。

工具

使用 run_demo_loop 来测试你的 Swarm！这将在命令行上运行一个 REPL。支持流式处理。

from swarm.repl import run_demo_loop
...
run_demo_loop(agent, stream=True)

核心贡献者

• Ilan Bigio - ibigio
• James Hills - jhills20
• Shyamal Anadkat - shyamal-anadkat
• Charu Jaiswal - charuj
• Colin Jarvis - colin-openai
• Katia Gil Guzman - katia-openai

Swarm 快速上手指南

重要提示：Swarm 目前处于实验和教育阶段，已被生产级的 OpenAI Agents SDK 取代。建议生产环境迁移至 Agents SDK，本指南仅用于学习多智能体编排原理。

环境准备

• 操作系统：Linux / macOS / Windows
• Python 版本：3.10 或更高版本
• 依赖项：
  • openai Python 库（安装 Swarm 时会自动安装）
  • 有效的 OpenAI API Key（需配置环境变量 OPENAI_API_KEY）

安装步骤

使用 pip 直接从 GitHub 仓库安装：

pip install git+https://github.com/openai/swarm.git

如果网络访问 GitHub 较慢，可尝试使用国内镜像加速（需确保 git 支持 https 克隆）：

pip install git+https://ghp.ci/https://github.com/openai/swarm.git

基本使用

Swarm 的核心概念是 Agent（智能体） 和 Handoffs（移交）。以下是最简单的双智能体交互示例：

1. 创建两个 Agent：Agent A 负责接待，Agent B 专门写俳句。
2. 定义一个函数将对话从 A 移交给 B。
3. 运行客户端并获取结果。

from swarm import Swarm, Agent

# 初始化 Swarm 客户端
client = Swarm()

# 定义移交函数：返回目标 Agent 对象
def transfer_to_agent_b():
    return agent_b

# 定义 Agent A
agent_a = Agent(
    name="Agent A",
    instructions="You are a helpful agent.",
    functions=[transfer_to_agent_b],
)

# 定义 Agent B
agent_b = Agent(
    name="Agent B",
    instructions="Only speak in Haikus.",
)

# 执行对话
response = client.run(
    agent=agent_a,
    messages=[{"role": "user", "content": "I want to talk to agent B."}],
)

# 输出最后一条消息内容
print(response.messages[-1]["content"])

预期输出：

Hope glimmers brightly,
New paths converge gracefully,
What can I assist?

代码解析

• Swarm()：初始化客户端，底层封装了 OpenAI Chat Completions API。
• Agent：包含指令（instructions）和工具函数（functions）。
• client.run()：执行核心循环，自动处理工具调用、智能体移交和上下文更新，直到没有新的函数调用为止。
• 无状态设计：每次调用不保存历史状态，需手动传递 messages 和 context_variables 以延续对话。