OpenAI 开源多智能体框架 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="You are a helpful agent.",
    functions=[transfer_to_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 · Agent · 函数 · 流式传输

Swarm 专注于使 Agent 协调和执行轻量化、高度可控且易于测试。

它通过两个基本抽象实现这一点：Agent 和切换。一个 Agent 包含指令和工具，可以在任何时间点选择将对话切换给另一个 Agent。

这些基本原语强大到足以表达工具和 Agent 网络之间的丰富交互，允许你构建可扩展的现实世界解决方案，同时避免陡峭的学习曲线。

Swarm Agent 与 Assistants API 中的 Assistant 无关。它们名称相似仅是为了方便，但在其他方面完全无关。Swarm 完全由 Chat Completions API 驱动，因此在调用之间是无状态的。

Swarm 探索的模式设计上轻量、可扩展且高度可定制。类似 Swarm 的方法最适合于处理大量独立功能和难以编码到单个 prompt 中的指令的情况。

Assistants API 是为希望获得完全托管的线程和内置内存管理及检索功能的开发者提供的绝佳选择。然而，Swarm 是一个教育资源，供好奇的开发者学习多 Agent 编排。Swarm 几乎完全在客户端运行，并且如同 Chat Completions API，在调用之间不存储状态。

查看 /examples 获取灵感！在其 README 中了解更多关于每个示例的信息。

• basic：设置、函数调用、切换和上下文变量等基础知识的简单示例
• triage_agent：设置基本分流步骤以切换到正确 Agent 的简单示例
• weather_agent：函数调用的简单示例
• airline：处理航空公司背景下不同客户服务请求的多 Agent 设置
• support_bot：客户服务机器人，包括用户界面 Agent 和带有多个工具的帮助中心 Agent
• personal_shopper：个人购物 Agent，可帮助进行销售和订单退款

运行 Swarm

首先通过实例化 Swarm 客户端开始（内部只是实例化 OpenAI 客户端）。

from swarm import Swarm

client = Swarm()

Swarm 的 run() 函数类似于 Chat Completions API 中的 chat.completions.create() 函数——它接受消息并返回消息，在调用之间不保存状态。重要的是，它还处理 Agent 函数执行、切换、上下文变量引用，并可以在返回给用户前执行多个轮次。

在其核心，Swarm 的 client.run() 实现了以下循环：

• 从当前 Agent 获取完成
• 执行工具调用并附加结果
• 必要时切换 Agent
• 如果必要更新上下文变量
• 如果没有新的函数调用，返回

client.run() 完成后（在可能的多个 Agent 和工具调用后），它将返回一个 Response，包含所有相关的更新状态。具体来说，新消息、最后被调用的 Agent，以及最新的 context_variables。你可以将这些值（加上新的用户消息）传递到下一个 client.run() 执行中，以继续交互——就像 chat.completions.create()。(run_demo_loop 函数在 /swarm/repl/repl.py 中实现了一个完整执行循环的示例。)

Agent

一个 Agent 只是将一组指令与一组函数（加上下面的一些额外设置）封装起来，并具有将执行切换给另一个 Agent 的能力。

虽然将 Agent 拟人化为"做 X 的某人"很诱人，但它也可以用来表示由一组指令和函数定义的非常具体的工作流或步骤（例如，一组步骤、复杂检索、单步数据转换等）。这允许 Agent 组成"Agent"、"工作流"和"任务"的网络，全部由同一个基本原语表示。

Agent 指令直接转换为对话的系统 prompt（作为第一条消息）。任何时候只有活跃 Agent 的指令才会出现（例如，如果有 Agent 切换，系统 prompt 会改变，但聊天历史不会。）

agent = Agent(
   instructions="You are a helpful agent."
)

指令可以是常规的 str，也可以是返回 str 的函数。该函数可以选择接收 context_variables 参数，它将由传递给 client.run() 的 context_variables 填充。

def instructions(context_variables):
   user_name = context_variables["user_name"]
   return f"Help the user, {user_name}, do whatever they want."

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

Hi John, how can I assist you today?

函数

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 "Done"

agent = Agent(
   functions=[greet]
)

client.run(
   agent=agent,
   messages=[{"role": "user", "content": "Usa greet() por favor."}],
   context_variables={"user_name": "John"}
)

Hola, John!

如果 Agent 函数调用有错误（缺少函数、错误的参数、错误），错误响应将被附加到聊天中，以便 Agent 可以优雅地恢复。

如果 Agent 调用了多个函数，它们将按该顺序执行。

切换和更新上下文变量

Agent 可以通过在函数中返回另一个 Agent 来切换到它。

sales_agent = Agent(name="Sales Agent")

def transfer_to_sales():
   return sales_agent

agent = Agent(functions=[transfer_to_sales])

response = client.run(agent, [{"role":"user", "content":"Transfer me to sales."}])
print(response.agent.name)

Sales Agent

它也可以通过返回一个更完整的 Result 对象来更新 context_variables。这也可以包含一个值和一个 Agent，以防你想要单个函数返回一个值、更新 Agent 和更新上下文变量（或三者的任何子集）。

sales_agent = Agent(name="Sales Agent")

def talk_to_sales():
   print("Hello, World!")
   return Result(
       value="Done",
       agent=sales_agent,
       context_variables={"department": "sales"}
   )

agent = Agent(functions=[talk_to_sales])

response = client.run(
   agent=agent,
   messages=[{"role": "user", "content": "Transfer me to sales"}],
   context_variables={"user_name": "John"}
)
print(response.agent.name)
print(response.context_variables)

Sales Agent
{'department': 'sales', 'user_name': 'John'}

如果 Agent 调用多个函数来切换到一个 Agent，只有最后的切换函数会被使用。

JSON Schema 转换

Swarm 自动将函数转换为传递给 Chat Completions 工具的 JSON Schema。

• 文档字符串被转换为函数描述。
• 没有默认值的参数被设置为必需。
• 类型提示被映射到参数的类型（默认为字符串）。
• 不明确支持各参数描述，但如果只是在文档字符串中添加应该会类似工作。（将来可能会添加文档字符串参数解析。）

def greet(name, age: int, location: str = "New York"):
   """Greets the user. Make sure to get their name and age before calling.

   Args:
      name: Name of the user.
      age: Age of the user.
      location: Best place on earth.
   """
   print(f"Hello {name}, glad you are {age} in {location}!")

{
   "type": "function",
   "function": {
      "name": "greet",
      "description": "Greets the user. Make sure to get their name and age before calling.\n\nArgs:\n   name: Name of the user.\n   age: Age of the user.\n   location: Best place on earth.",
      "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 快速入门示例中有一些评估 swarm 的示例。有关更多详情，请参见 README。

试用

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

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

贡献者

James Hills - jhills20

Shyamal Anadkat - shyamal-anadkat

Charu Jaiswal - charuj

Colin Jarvis - colin-openai

Katia Gil Guzman - katia-openai