Swarm (实验性，教育用途)

[!IMPORTANT]
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"])

希望之光闪耀，
新路径优雅地汇聚，
我能帮您什么？

目录

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

概述

Swarm 专注于使智能体的协调与执行变得轻量级、高度可控且易于测试。

它通过两个基本抽象来实现这一点：Agent（智能体）和交接。一个 Agent 包含 instructions（指令）和 tools（工具），并且可以在任何时候选择将会话交接给另一个 Agent。

这些基本元素足以表达工具与智能体网络之间丰富的动态关系，使您能够构建可扩展的、面向真实世界的解决方案，同时避免陡峭的学习曲线。

[!NOTE]
Swarm 中的 Agent 与 Assistants API 中的 Assistant 无关。它们名称相似是为了方便，但除此之外完全无关。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. 如果没有新的函数调用，则返回

参数

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

Response 字段

智能体

一个 Agent 简单地封装了一组 instructions 和一组 functions（加上下面的一些额外设置），并具有将会话执行交接给另一个 Agent 的能力。

虽然很容易将 Agent 拟人化为“做某事的人”，但它也可以用来表示由一组 instructions 和 functions 定义的非常具体的工作流或步骤（例如，一组步骤、复杂的检索、单步数据转换等）。这使得 Agent 可以组合成一个由“智能体”、“工作流”和“任务”组成的网络，所有这些都由同一个基本元素表示。

Agent 字段

指令

Agent 的 instructions 直接转换为对话的 system 提示（作为第一条消息）。在任何给定时间，只有活动 Agent 的 instructions 会出现（例如，如果有 Agent 交接，system 提示会改变，但聊天历史不会。）

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

instructions 可以是普通的 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 可以通过在 function 中返回另一个 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。这也可以包含一个 value 和一个 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'}

[!NOTE]
如果一个 Agent 调用多个函数来交接给另一个 Agent，只有最后一个交接函数会被使用。

函数模式

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

• 文档字符串被转换为函数的 description。
• 没有默认值的参数被设置为 required。
• 类型提示被映射到参数的 type（默认为 string）。
• 每个参数的描述没有明确支持，但如果只是添加到文档字符串中，应该以类似的方式工作。（未来可能会添加文档字符串参数解析。）

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)

核心贡献者

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