开源多智能体开源Swarm

edwin99
edwin99
2024-10-13 00:50
43 阅读
0 评论
文章封面
目录
正在加载目录...
Swarm是一个实验性质的多智能体编排框架，主打特征是工效（ergonomic）与轻量（lightweight）。

项目地址：https://github.com/openai/swarm

我们先来看一个例子。首先安装 Swarm，很简单：
pip install git+ssh://git@github.com/openai/swarm.git
装好这个框架之后，用起来也很方便。以下代码定义了 2 个智能体，而用户的指令是与智能体 B 交谈：
  
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 是一个实验性质的多智能体框架，并不是为生产目的开发的，因此团队表示不会提供任何官方支持。Swarm 关注的重点是让智能体协作和执行变得轻量、高度可控且易于测试。为此，它使用了两种原语抽象：智能体（agent）和交接（handoff）。其中，智能体包含指令和工具，并且在任何时间都可以选择将对话交接给另一个智能体。

该团队表示，这些原语很强大，「足以表达工具和智能体网络之间的丰富动态，让你可以针对真实世界问题构建可扩展的解决方案，同时避免陡峭的学习曲线。」另外，请注意 Swarm 智能体与 Assistants API 中的 Assistants 无关。之所以名字相似，只是为了方便。Swarm 完全由 Chat Completions API 提供支持，因此在调用之间是无状态的。

 

在设计上，Swarm 是轻量级、可扩展且高度可定制的。它最适合处理存在大量独立功能和指令的情况——这些功能和指令很难编码成单个提示词。如果开发者想要寻求完全托管的线程以及内置的内存管理和检索，那么 Assistants API 就已经是很好的选择了。但如果开发者想要完全的透明度，并且能够细粒度地控制上下文、步骤和工具调用，那么 Swarm 才是最佳选择。Swarm （几乎）完全运行在客户端，与 Chat Completions API 非常相似，不会在调用之间存储状态。该团队还展示了一个应用示例，包括天气查询智能体、用于在航空公司环境中处理不同客户服务请求的多智能体设置、客服机器人、可以帮助销售和退款的个人智能体等。具体示例请访问 Swarm 代码库。

 

Swarm核心部件

Swarm 的核心组件包括 client（客户端）、Agent（智能体）、Function（函数）。运行 Swarm 就是从实例化一个 client 开始的（其就是在内部实例化一个 OpenAI 客户端）。​​​​
from swarm import Swarm
client = Swarm()

client.run()
Swarm 的 run() 函数类似于 Chat Completions API 中的 chat.completions.create() 函数——接收消息并返回消息，并且在调用之间不保存任何状态。但重点在于，它还处理 Agent 函数执行、交接、上下文变量引用，并且可以在返回给用户之前进行多轮执行。
究其核心，Swarm 的 client.run() 是实现以下循环：
  1. 先让当前智能体完成一个结果
  2. 执行工具调用并附加结果
  3. 如有必要，切换智能体
  4. 如有必要，更新上下文变量
  5. 如果没有新的函数调用，则返回
 

client().run的参数

client.run() 完成后（可能进行过多次智能体和工具调用），会返回一个响应，其中包含所有相关的已更新状态。具体来说，即包含新消息、最后调用的智能体、最新的上下文变量。你可以将这些值（加上新的用户消息）传递给 client.run() 的下一次执行，以继续上次的交互——就像是 chat.completions.create()

 

响应字段

 

Agent字段

Agent（智能体）就是将一组指令与一组函数封装在一起（再加上一些额外的设置），并且其有能力将执行过程交接给另一个 Agent。

 

Agent指令

Agent instructions 会直接转换成对话的系统提示词（作为第一条消息）。只有当前活动的 Agent 的指令会被使用（当发生智能体交接时，系统提示词会变化，但聊天历史不会）。
agent = Agent(
instructions="You are a helpful agent."
)
instructions 可以是常规字符串，也可以是返回字符串的函数。该函数可以选择性地接收 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?

 

Function

  • Swarm Agent 可以直接调用 Python 函数。
  • 函数通常应返回一个字符串（数值会被转换为字符串）。
  • 如果一个函数返回了一个 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=[print_hello]
)

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

输出消息：

Hola, John!
 
如果某个 Agent 函数调用出错（缺少函数、参数错误等），则会在聊天之中附加一条报错响应，以便 Agent 恢复正常。如果 Ageny 调用多个函数，则按顺序执行它们。
 

 

交接和更新上下文变量

通过在返回的函数中包含一个 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。如果你希望用单个函数返回一个值、更新智能体并更新上下文变量（或三者中的任何组合），它还可以包含一个 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 模式，然后将其传递给聊天补全工具。
  • 文档字符串会转换为函数 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 streaming 一样的事件，但添加了两个事件类型：
  • {"delim":"start"} 和 {"delim":"start"}，用于在 Agent 每次处理单个消息（响应或函数调用）时发出信号。这有助于识别 Agent 之间的切换。
  • 为方便起见，{"response": Response} 将在流的末尾返回带有已聚合的（完整）响应的 Response 对象。
 
LLM 智能体

评论区 (0)

请登录后参与评论

暂无评论，抢沙发吧！