Swarm - Agent 编排工具

---

一、关于 Swarm（实验性、教育性）

• github : https://github.com/openai/swarm

---

Swarm 是探索人体工程学、轻量级 多代理编排的 教育框架。(An educational framework exploring ergonomic, lightweight multi-agent orchestration.)

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

它通过两个基本抽象来实现这一点：Agent和切换。Agent包含instructions和tools，并且可以 随时选择将对话 交给另一个Agent。

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

注：群代理与助理应用编程接口中的助理无关。为了方便起见，它们的命名类似，但在其他方面完全不相关。群完全由聊天完成应用编程接口提供支持，因此在调用之间是无状态的。

---

警告：

Swarm目前是一个实验性的示例框架，旨在探索多智能体系统的人体工程学界面。它不打算在生产中使用，因此没有官方支持。（这也意味着我们不会审查PR或问题！）

Swarm的主要目标是展示编排代理：切换和例程食谱中探索的切换和例程模式。它不意味着是一个独立的库，主要用于教育目的。

---

为什么选择蜂群

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

对于寻找完全托管线程和内置内存管理和检索的开发人员来说，助理应用编程接口是一个很好的选择。然而，对于想了解多代理编排的开发人员来说，Swarm是一个教育资源。Swarm（几乎）完全在客户端上运行，并且很像聊天完成应用编程接口，不存储调用之间的状态。

---

文档

---

二、安装使用

安装

需要Python 3.10+

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

---

或者

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

---

基本用法

如果你想使用 Qwen 等 API 替代默认的 OpenAI，可参考文章：https://ezcode.blog.csdn.net/article/details/143300605

from swarm import Swarm, Agentclient = Swarm()def transfer_to_agent_b():return agent_bagent_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?

---

其它示例

看看/examples的灵感！在自述文件中了解更多关于每一个的信息。

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

---

三、Running Swarm

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

from swarm import Swarmclient = Swarm()

---

client.run()

Swarm的run()函数类似于Chat Completions API中的chat.completions.create()函数 —— 它接受messages并返回messages，并且在调用之间不保存任何状态。

然而，重要的是，它还处理代理 函数的执行、移交、上下文变量引用，并且可以在返回给用户之前 进行多次轮换。

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

1. 从当前代理处获得完成
2. 执行工具调用并附加结果
3. 必要时切换代理
4. 如有必要，更新上下文变量
5. 如果没有新的函数调用，返回

---

Arguments

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

---

一旦client.run() 完成（在可能多次调用代理和工具之后），它将返回一个Response，其中包含所有相关的更新状态。

具体来说，新messages、最后一个要调用的Agent和最新的context_variables。

您可以将这些值（加上新用户消息）传递到client.run()的下一次执行中，以继续它中断的交互——很像chat.completions.create()。

（run_demo_loop函数在/swarm/repl/repl.py.）中实现完整执行循环的示例

---

Response字段

字段	类型	描述
messages	List	对话期间生成的消息对象列表。非常类似于聊天完成messages，但有一个sender字段，指示消息来自哪个Agent。
agent	Agent	处理消息的最后一个代理。
context_variables	dict	与输入变量相同，加上任何更改。

---

四、Agent

一个Agent只是用一组functions封装了一组instructions（加上下面的一些附加设置），并且能够将执行交给另一个Agent。

虽然很容易将Agent拟人化为“做X的人”，但它也可以用来表示由一组instructions和functions定义的非常具体的工作流或步骤（例如一组步骤、复杂的检索、单步数据转换等）。

这允许Agent被组成一个由“代理”、“工作流”和“任务”组成的网络，所有这些都由同一个原语表示。

---

Fields Agent

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

---

指令

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?

---

函数

• SwarmAgent可以直接调用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中返回另一个function来传递。

sales_agent = Agent(name="Sales Agent")def transfer_to_sales():return sales_agentagent = Agent(functions=[transfer_to_sales])response = client.run(agent, [{"role":"user", "content":"Transfer me to sales."}])
print(response.agent.name)

---

Sales Agent

---

它还可以通过返回更完整的context_variables对象来更新Result。这也可以包含一个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'}

---

注：如果Agent调用多个函数来移交给Agent，则仅使用最后一个移交函数。

---

函数模式

Swarm自动将函数转换为JSON模式，并将其传递给聊天完成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)

---

使用与聊天完成API流相同的事件。请参见/swarm/repl/repl.py中的process_and_print_streaming_response作为示例。

添加了两种新的事件类型：

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

---

评估

评估对任何项目都是至关重要的，我们鼓励开发人员带上他们自己的eval套件来测试他们的集群的性能。作为参考，我们有一些关于如何在airline、weather_agent和triage_agent快速入门示例中使用eval swarm的示例。有关更多详细信息，请参阅READMEs。

---

工具

使用该run_demo_loop测试您的集群！这将在您的命令行上运行REPL。支持流式传输。

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

---

2024-10-28(一)