swarm Agent框架入门指南：构建与编排多智能体系统的利器 | AI应用开发

Swarm是OpenAI在2024年10月12日宣布开源的一个实验性质的多智能体编排框架。其核心目标是让智能体之间的协调和执行变得更轻量级、更容易控制和测试。Swarm框架的主要特性包括轻量化、易于使用和高度可定制性，非常适合处理大量独立的功能和指令。

Swarm专注于使代理的协调和执行轻量级、高度可控且易于测试。

它通过两个原始抽象：Agents和交接来实现这一点。一个Agent包含instructions和tools，并且在任何时候都可以选择将对话交接给另一个Agent。

这些原始抽象足够强大，可以表达工具和代理网络之间的丰富动态，允许您构建可扩展的、现实世界的解决方案，同时避免陡峭的学习曲线。

一、安装

需要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探索了轻量级、可扩展且高度可定制的设计模式。类似于Swarm的方法最适合处理大量独立能力和难以编码到单个提示中的指令的情况。

Assistants API是寻找完全托管的线程和内置的内存管理和检索的开发者的一个很好的选择。然而，Swarm是一个教育资源，供对多代理编排感兴趣的开发者学习。Swarm（几乎）完全在客户端上运行，并且像Chat Completions API一样，在调用之间不存储状态。

四、运行Swarm

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

from swarm import Swarm

client = Swarm()

client.run()

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

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

1. 从当前代理获取一个完成
2. 执行工具调用并追加结果
3. 如有必要，切换代理
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拟人化为“做X的人”是诱人的，但它也可以用来表示由一组instructions和functions定义的非常具体的工作流程或步骤（例如一组步骤、复杂的检索、单个数据转换步骤等）。这允许Agents被组合成一个由“代理”、“工作流程”和“任务”组成的网络，所有这些都由相同的原始表示。

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 Agents可以直接调用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"):
   ""“向用户打招呼。在调用之前，请确保获取他们的姓名和年龄。

   参数：
      name：用户的姓名。
      age：用户的年龄。
      location：地球上最好的地方。
   ”"
   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流式传输相同的事件。参见process_and_print_streaming_response在/swarm/repl/repl.py中的示例。

增加了两个新的事件类型：

• {"delim":"start"}和{"delim":"end"}，每次Agent处理单个消息（响应或函数调用）时发出信号。这有助于识别Agent之间的切换。
• {"response": Response}将在流的末尾返回一个Response对象，包含聚合的（完整的）响应，以方便使用。

六、总结

Swarm Agent框架的发布标志着多智能体系统开发的一个新阶段。通过优化智能体的协调、执行和测试流程，Swarm使得开发者能够更加高效地管理、互动多个AI智能体。虽然目前该框架仍处于实验阶段，但它为开发者探索多智能体系统提供了新的思路和工具。