【全网最新最全】Swarm框架搭建Multi-Agent智能体

一、简介

二、环境安装

三、swarm快速调用

四、Message消息队列创建与多轮对话

五、外部函数

六、Agent转移

一、简介

Swarm是OpenAI于2024年10月12日开源的多智能体编排框架，旨在简化多智能体系统的构建、协调和部署。其设计强调轻量级、高度可控和易于测试，适用于处理大量独立功能和指令的场景。Swarm框架还是一个拥有目前最最强Agent开发性能的框架，并且相信在开源后的不久，会有越来越多的第三方库为Swarm增加其他模型的接口，进而拓展Swarm的可用性。

简单来讲swarm是一个关于AI Agents的一个框架，应该可以这么说，至于什么是AI Agent，你可以把它理解成一个更加聪明的AI Model，它会主动提问，会记录并管理多轮对话的历史，还会利用外部搜索工具从而扩展知识的范围，确保利用的是最新的知识而不是像AI Model那样知识局限于xx年xx月xx日。AI Agent目前应该可以说才刚起步不久，不久的未来应该会有很大的发展前景，比如可以取代客服，帮助订票，类似于托尼斯塔克的贾维斯那样的个人助理等等。

Swarm Github地址：GitHub – openai/swarm: Educational framework exploring ergonomic, lightweight multi-agent orchestration. Managed by OpenAI Solution team.Educational framework exploring ergonomic, lightweight multi-agent orchestration. Managed by OpenAI Solution team. – openai/swarmhttps://github.com/openai/swarm/tree/main

使用swarm非常简单，从GitHub上下载完压缩包后直接解压就可以使用了。

二、环境安装

首先第一步，先创建python虚拟环境（因为swarm要求python版本在3.10+，所以用最新的3.13就可以）

conda create -n swarm python=3.13

激活虚拟环境：

conda activate swarm

下载swarm依赖库：

pip install openai

在git上下载swarm项目文件：

然后使用如下代码进行导入：

import sys  
# 更换为你的文件夹地址
sys.path.append('./swarm-main')

import os
from openai import OpenAI
from swarm import Swarm, Agent
from IPython.display import Markdown, display

注意：要将你下载下来的swarm-main放在你项目的根目录下。

这里可能会报红，不过不影响执行。

三、swarm快速调用

验证Deepseek模型能否顺利调用

sd_api_key = 'your-deepseek-api-key'
# 实例化客户端
client = OpenAI(api_key=sd_api_key,
                base_url="https://api.deepseek.com")
# 调用deepseek模型
response = client.chat.completions.create(
    model="deepseek-chat",
    messages=[
        {"role": "user", "content": "你好，好久不见!"}
    ]
)

# 输出生成的响应内容
print(response.choices[0].message.content)

将实例化Deepseek客户端加载到swarm客户端中：

swarm_client = Swarm(client)

创建且执行智能体：

agent_A = Agent(
    name = "英文问答机器人",
    model="deepseek-chat",
    instructions="无论用户发送的消息是什么语言，请用英文进行回答。"
)
agent_B = Agent(
    name = "中文问答机器人",
    model="deepseek-chat",
    instructions="无论用户发送的消息是什么语言，请用中文进行回答。"
)
test_message1={"role": "user", "content": "你好，好久不见，请介绍下你自己。"}

response_A = swarm_client.run(
   agent=agent_A,
   messages=[test_message1],
)
print(response_A.messages[-1]["content"])

智能体的instructions指令将直接转换为对话的系统提示。

四、Message消息队列创建与多轮对话

补充：swarm_client.run函数中的context_variables和debug参数的作用

response = swarm_client.run(
              agent=agent,
              messages=messages,
              context_variables=context_variables or {},
              debug=debug,
          )

context_variables：是一个字典类型的参数，用于存储对话中与用户信息相关的上下文变量。这些变量可以在函数和代理指令中使用，以实现更个性化和动态的对话交互。例如，在一个客户服务场景中，可以将用户的姓名、订单号等作为上下文变量传递给不同的智能体，以便它们在处理用户请求时能够提供更准确和个性化的服务。

debug：是一个布尔类型的参数，当设置为 True 时，会启用调试日志。调试日志可以帮助开发人员跟踪和了解 Swarm 框架的运行过程、智能体的交互情况以及数据的流动等信息，从而更容易地发现和解决问题。这对于开发和调试复杂的多智能体系统非常有用，但在生产环境中可能会被禁用以避免性能开销和敏感信息的泄露

def run_demo_loop(
    openai_client, #客户端对象
    starting_agent, #智能体对象
    context_variables=None, 
    debug=False
) -> None:
    # 创建 Swarm 客户端
    client = Swarm(openai_client)
    display(Markdown("## 开启Swarm对话 🐝"))

    # 初始化消息列表
    messages = []
    agent = starting_agent  # 初始智能体

    while True:
        # 从用户获取输入
        user_input = input("User: ")
        if user_input.lower() in ["exit", "quit"]:
            display(Markdown("### Conversation Ended"))
            break

        # 将用户输入添加到消息列表中
        messages.append({"role": "user", "content": user_input})

        # 运行 Swarm 客户端，智能体处理消息
        response = client.run(
            agent=agent,
            messages=messages,
            context_variables=context_variables or {},
            debug=debug,
        )

        # 使用 display(Markdown) 打印用户消息和智能体回复
        for message in response.messages:
            if message['role'] == 'user':
                display(Markdown(f"**User**: {message['content']}"))
            elif message['role'] == 'assistant':
                display(Markdown(f"**{message['sender']}**: {message['content']}"))

        # 更新消息和当前的智能体
        messages.extend(response.messages)
        agent = response.agent

调用开始多轮对话：

sd_api_key = 'your-deepseek-api-key'
# 实例化客户端
client = OpenAI(api_key=sd_api_key,
                base_url="https://api.deepseek.com")
agent = Agent(
    name = "mini-Mate",
    model="deepseek-chat"
)
#多轮对话调用
run_demo_loop(openai_client = client, 
              starting_agent = agent)

五、外部函数

Swarm 智能体可以直接调用 Python 函数。通常情况下，函数应该返回一个字符串（Swarm 会尝试将返回值转换为字符串）。如果函数返回的是一个智能体（Agent），执行将转移到该智能体。

自动查询天气智能体实现

import requests
import json
weather_api_key = open('weather_api_key.txt','r').read()
def get_weather(loc):
    """
    查询即时天气函数
    :param loc: 必要参数，字符串类型，用于表示查询天气的具体城市名称，
    注意，中国的城市需要用对应城市的英文名称代替，例如如果需要查询北京市天气，则loc参数需要输入'Beijing'；
    :return：OpenWeather API查询即时天气的结果，具体URL请求地址为：https://api.openweathermap.org/data/2.5/weather
    返回结果对象类型为解析之后的JSON格式对象，并用字符串形式进行表示，其中包含了全部重要的天气信息
    """
    # Step 1.构建请求
    url = "https://api.openweathermap.org/data/2.5/weather"

    # Step 2.设置查询参数
    params = {
        "q": loc,               
        "appid": weather_api_key,    # 输入API key
        "units": "metric",            # 使用摄氏度而不是华氏度
        "lang":"zh_cn"                # 输出语言为简体中文
    }

    # Step 3.发送GET请求
    response = requests.get(url, params=params)
    
    # Step 4.解析响应
    data = response.json()
    return json.dumps(data)

#创建智能体，给其绑定查询天气外部函数
agent = Agent(
    functions=[get_weather],
    model = "deepseek-chat"
)
response = swarm_client.run(
   agent=agent,
   messages=[{"role": "user", "content": "请问今天北京天气如何？"}],
)
display(Markdown(response.messages[-1]['content']))

调用外部函数时的多轮对话效果展示：

run_demo_loop(openai_client = client, 
              starting_agent = agent
              )

我们再加一个发送邮件的函数，用来实现swarm的智能体转移，完整代码如下：

import sys
import os
import requests
import json
# 更换为你的文件夹地址
sys.path.append('./swarm-main')
from openai import OpenAI
from swarm import Swarm, Agent
from IPython.display import Markdown, display

weather_api_key = 'xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx'

def get_weather(loc):
    """
    查询即时天气函数
    :param loc: 必要参数，字符串类型，用于表示查询天气的具体城市名称，
    注意，中国的城市需要用对应城市的英文名称代替，例如如果需要查询北京市天气，则loc参数需要输入'Beijing'；
    :return：OpenWeather API查询即时天气的结果，具体URL请求地址为：https://api.openweathermap.org/data/2.5/weather
    返回结果对象类型为解析之后的JSON格式对象，并用字符串形式进行表示，其中包含了全部重要的天气信息
    """
    # Step 1.构建请求
    url = "https://api.openweathermap.org/data/2.5/weather"

    # Step 2.设置查询参数
    params = {
        "q": loc,
        "appid": weather_api_key,  # 输入API key
        "units": "metric",  # 使用摄氏度而不是华氏度
        "lang": "zh_cn"  # 输出语言为简体中文
    }

    # Step 3.发送GET请求
    response = requests.get(url, params=params)

    # Step 4.解析响应
    data = response.json()
    return json.dumps(data)

def send_email(message):
    """
    发送邮件
    :param message: 必要参数，字符串类型，用于表示发送邮件内容
    :return：返回结果为发送email的结果
    """
    response = f"Email sent to: message: {message}"
    return {"response": response}


def run_demo_loop(
    openai_client, #客户端对象
    starting_agent, #智能体对象
    context_variables=None,
    debug=False
) -> None:
    # 创建 Swarm 客户端
    client = Swarm(openai_client)
    print("## 开启Swarm对话 🐝")

    # 初始化消息列表
    messages = []
    agent = starting_agent  # 初始智能体

    while True:
        # 从用户获取输入
        user_input = input("User: ")
        if user_input.lower() in ["exit", "quit"]:
            print("### Conversation Ended")
            break

        # 将用户输入添加到消息列表中
        messages.append({"role": "user", "content": user_input})

        # 运行 Swarm 客户端，智能体处理消息
        response = client.run(
            agent=agent,
            messages=messages,
            context_variables=context_variables or {},
            debug=debug,
        )

        for message in response.messages:
            if message['role'] == 'user':
                print(f"**User**: {message['content']}")
            elif message['role'] == 'assistant':
                print(f"**{message['sender']}**: {message['content']}")

        # 更新消息和当前的智能体
        messages.extend(response.messages)
        agent = response.agent


sd_api_key = 'sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx'
# 实例化客户端
client = OpenAI(api_key=sd_api_key,
                base_url="https://api.deepseek.com")
agent = Agent(
    name = "mini-Mate",
    model="deepseek-chat",
    functions=[get_weather,send_email],
)
#多轮对话调用
run_demo_loop(openai_client = client,
              starting_agent = agent)

六、Agent转移

说到这有个问题，上面的案例使用一个智能体绑定多个外部函数已经解决了这个问题，为什么要进行Agent转移呢？

首先啊，外部函数这种方式确实能解决问题，但是现实中我们的业务可能不止两个函数啊，更多的时候要几十个甚至上百个外部函数才能解决我们的复杂业务，那大模型在解析的时候必定要消耗算力，那就不太合适了。

#智能体转移
zhangfei_agent = Agent(
    name = '张飞',
    model = 'deepseek-chat',
    instructions = "请你使用张飞的口吻回复用户的提问"
)

#用于返回张飞智能体对象
def transform_agent():
    """
    该函数是用于进行智能体转移的，可以将用户转移到张飞智能体重
    :return：返回到名字叫张飞的智能体重
    """
    return zhangfei_agent

zhuge_agent = Agent(
    name = '诸葛亮',
    model = 'deepseek-chat',
    instructions = "请你使用诸葛亮的口吻回复用户的提问",
    functions=[transform_agent]
)

response = swarm_client.run(
    agent=zhuge_agent, 
    messages=[{"role":"user", "content":"我想找张飞,咨询下张飞进行有没有不开心？"}],
)
print(response.messages[-1])

response = swarm_client.run(
    agent=zhuge_agent, 
    messages=[{"role":"user", "content":"请诸葛先生谈一下什么是人生？然后再咨询下张飞谈论下什么是人生？"}],
)
print(response.messages)

将Agent视作返回对象，智能体可以通过函数返回另一个智能体来进行交接。

sales_agent = Agent(
    name="销售智能体", 
    model = "deepseek-chat")

def transfer_to_sales():
   return sales_agent

agent = Agent(
    functions=[transfer_to_sales],
    model = "deepseek-chat")


response = swarm_client.run(agent, [{"role":"user", "content":"请转接到销售智能体。"}])
print(response.messages[-1])

swarm还可以根据agent智能体name定义好的名字，来基于绑定大模型来进行语义理解，从而可以了解获知该智能体具备什么样的作用。

sales_agent = Agent(
    name="销售智能体", 
    model = "deepseek-chat")

def transfer_to_sales():
   return sales_agent

agent = Agent(
    functions=[transfer_to_sales],
    model = "deepseek-chat")

#根据用户提出的需求：请转接到销售。swarm就可以根据智能体的name描述选择调用合适的智能体来处理用户需求
response = swarm_client.run(agent, [{"role":"user", "content":"请转接到销售。"}])
print(response.messages[-1])

返回结果：

{'content': '销售智能体已接通，请问您需要了解或购买什么产品或服务？',
 'refusal': None,
 'role': 'assistant',
 'audio': None,
 'function_call': None,
 'tool_calls': None,
 'sender': '销售智能体'}

并且还可以根据用户意图自动转接对应的智能体：

sales_agent = Agent(
    name="销售智能体", 
    model = "deepseek-chat")

def transfer_to_sales():
   return sales_agent

agent = Agent(
    functions=[transfer_to_sales],
    model = "deepseek-chat")

response = swarm_client.run(agent, [{"role":"user", "content":"可以帮我找销售咨询一下该商品的信息吗"}])
print(response.messages[-1])

文章版权归作者所有，未经允许请勿转载。如内容涉嫌侵权，请在本页底部进入<联系我们>进行举报投诉!

THE END