1. 前言

不知不觉中 Multi-Agent 实践系列已经更新到第7期，在之前的文章中已经介绍了众多 AgentScope 支持的有趣的应用。但过去的文章中所介绍的案例都是运行在单台机器上以单进程形式运行，受限于 Python 的全局解释器锁（参见结尾wiki链接），实际只能有效利用一个 CPU 的计算资源，并且无法支持多个用户从自己的电脑上接入同一个 Multi-Agent 应用进行交互。

为了提高运行效率并支持多用户接入同一个应用中，AgentScope 提供了分布式模式，支持将同一个应用中的多个 Agent 运行在不同的进程甚至不同的机器上。同时为了让更多完全没有分布式程序编写经验的小白也能轻松上手，AgentScope 提供了简单的转化函数，能够将已经写好的单机单进程应用扩展到多进程、分布式的场景中。

欢迎关注AgentScope，在github上为我们star 🌟

准备工作

参照 GitHub 中的说明安装 AgentScope 的分布式版本

git clone https://github.com/modelscope/agentscope
cd agentscope
# On windows
pip install -e .[distribute]
# On mac or linux
pip install -e .\[distribute\]

准备模型配置文件（与之前的文章相同，这里以 Qwen 模型为例，其他的模型配置方法请参考我们的教程

// model_configs.json
[
  {
    "model_type": "dashscope_chat",
    "config_name": "tongyi_qwen_config",
    "model_name": "qwen-max",
    "api_key": "************"  // 在这里填写你DashScope中的API-KEY
  }
]

接下来就可以开始编写自己的多进程分布式 Multi-Agent 应用了，首先让我们从最简单的样例开始，将一个已经编写好的单进程 Multi-Agent 应用转化为多进程版本。

2. 转化现有的 Multi-Agent 应用

AgentScope 提供了 to_dist函数将已经编写好的 Agent 转化为对应的分布式版本。这里我们以第一期实践课程中介绍过的 conversation.py 为例，该应用中有两个 Agent，分别为 UserAgent 和 DialogAgent，如果想要将 DialogAgent 转移到一个独立的进程去运行，只需要添加 dialog_agent = dialog_agent.to_dist()。之后 DialogAgent 就会以一个单独的进程形式运行，而不会跟 UserAgent 争夺计算资源了，修改后的部分代码如下。

dialog_agent = DialogAgent(
        name="Assistant",
        sys_prompt="You're a helpful assistant.",
        model_config_name="tongyi_qwen_config",
    )
    # Add to_dist()
    dialog_agent = dialog_agent.to_dist()
    user_agent = UserAgent()

是不是觉得很简单？更好的消息是 AgentScope 中实现的所有 Agent 均支持上述 to_dist()方法。

上面使用的这种不填写任何参数的 to_dist()方法会自动创建一个子进程，并将被调用的 Agent 对象转移到该子进程中运行。当然上面提到的这个对话的应用由于涉及到发言顺序并不能体现出运行效率的提升，接下来让我们以大模型辅助搜索为例看看多进程所能获得的加速效果。

3. Multi-Agent 并行加速样例：大模型辅助搜索

该应用将用户的问题转化为关键词调用搜索引擎，然后爬取返回的一系列网页并从中寻找答案，共涉及到三类 Agent，分别是用户UserAgent，负责搜索的SearcherAgent 以及负责爬取网页的 AnswererAgent。搜索引擎返回的网页链接众多，为了提高性能需要多个 AnswererAgent 共同爬取，但如果还是使用传统的单进程模式将完全无法获得加速效果，为此必须引入 AgentScope 的分布式多进程模式。

38cffc92-17c5-4d1e-9fc2-b404cb1bd06b[1].png

第一步：准备工作

为了能够使用搜索功能，需要提供 Bing 或 Google 的相关参数。如果使用 Bing，需要提供 API key，而如果使用 Google，需要提供 API key 和 CSE id。

第二步：运行案例

下面的代码就是该应用的主流程，运行该应用需要提供几个参数：

--num-workers应用中所运行的 AnswererAgent 数量
--search-engine使用的搜索引擎类型，当前支持 bing 或 google
--api-keyBing 或 Google 所需的 API key
--cse-idGoogle 所需的 CSE id（如使用 Bing 可忽略该项）
--use-dist开启多进程模式

# -*- coding: utf-8 -*-
"""An example use multiple agents to search the Internet for answers"""
import time
import argparse
from loguru import logger
from searcher_agent import SearcherAgent
from answerer_agent import AnswererAgent
import agentscope
from agentscope.agents.user_agent import UserAgent
from agentscope.message import Msg
def parse_args() -> argparse.Namespace:
    # 处理上文提到的输入参数...
    # 在此省略一些代码，完整版代码见GitHub仓库
if __name__ == "__main__":
    args = parse_args()
    agentscope.init(
        model_configs="configs/model_configs.json",
    )
    WORKER_NUM = 3
    searcher = SearcherAgent(
        name="Searcher",
        model_config_name="my_model",
        result_num=args.num_workers,
        search_engine_type=args.search_engine,
        api_key=args.api_key,
        cse_id=args.cse_id,
    )
    answerers = []
    for i in range(args.num_workers):
        answerer = AnswererAgent(
            name=f"Answerer-{i}",
            model_config_name="my_model",
        )
        if args.use_dist:
            answerer = answerer.to_dist(lazy_launch=False)
        answerers.append(answerer)
    user_agent = UserAgent()
    msg = user_agent()
    while not msg.content == "exit":
        start_time = time.time()
        msg = searcher(msg)
        results = []
        for page, worker in zip(msg.content, answerers):
            results.append(worker(Msg(**page)))
        for result in results:
            logger.chat(result)
        end_time = time.time()
        logger.chat(
            Msg(
                name="system",
                role="system",
                content=f"Finish in [{end_time - start_time}]s",
            ),
        )
        msg = user_agent()

如果仔细阅读上述代码可以发现，从单进程到多进程模式的转化只需要调用answerer.to_dist(...)即可，这里的 lazy_launch参数是为了让对应 Agent 立即转移到子进程中，而不是等到被调用时才转移，从而避免影响运行过程中的效率。

如果想以多进程模式运行该应用，使用 Bing 搜索并启动 10 个爬取网页的Agent，只需要使用如下指令：

python main.py --num-workers 10 --search-engine bing --api-key xxxxx --use-dist

应用启动后只需要在命令行的 User input 中输入你想要问的问题即可，该应用会自动调用搜索引擎并借助大模型从搜索引擎返回的各网页中寻找问题的答案，例如下面就是一个多进程模式下的运行案例：

User input: Sora 是什么时候发布的
...
Answerer-2: Sora是在2月16日凌晨由美国人工智能公司OpenAI发布的。
Answerer-2: https://www.stcn.com/article/detail/1121797.html
Answerer-1: OpenAI Sora是在2024年2月15日正式发布的。
Answerer-1: https://shengpu.cc/sora-release-timeline/
...
system: Completed in [13.2] s

可以看到运行耗时为 13.2s，而下面的案例是以单进程模式运行的，其耗时长达 51.3 s

User input: Sora 是什么时候发布的
...
Answerer-1: OpenAI Sora是在2024年2月15日正式发布的。
Answerer-1: https://shengpu.cc/sora-release-timeline/
Answerer-2: Sora是在2月16日凌晨由美国人工智能公司OpenAI发布的。
Answerer-2: https://www.stcn.com/article/detail/1121797.html
...
system: Completed in [51.3] s

AgentScope 的多进程模式相比单进程版本获得了 4 倍的加速效果。

以上的案例以及之前的文章中提到的应用都只允许一个用户参与，接下来我们将以辩论赛场景为例，展示 AgentScope 的多用户协同能力。

4. Multi-Agent 分布式&多用户协同：辩论赛

在该应用中包含三个参与方，分别为正方、反方以及裁判，其中正反双方就一个辩题展开辩论，而裁判将根据辩论的内容决定获胜方。辩论共分为三轮，每轮按照正方、反方、裁判的顺序依次发言，三轮结束后裁判将宣布最终的胜利者。该应用允许用户作为正方或是反方参与到辩论中，甚至正反双方都可以是实际用户。

第一步：准备工作

为了获得最好的多用户体验，推荐准备两台不同的电脑，确保这两台电脑可以通过 IP 互相连通并安装 AgentScope 的分布式版本。如果确实不具备该条件，也可以在一台电脑上打开两个命令行界面来模拟两台电脑的场景。

第二步：运行程序

由于该应用中不同 Agent 可能运行在不同的机器上，因此启动方式与上面提到的两个案例有所差异，需要在两台机器（或是两个命令行界面）单独启动正方和反方的 Agent 服务器。和前一个并行的例子相比，这里我们需要一个Agent 服务器启动函数（片段如下）。完整的代码可以从 GitHub 仓库的 examples/distributed_debate目录下获取。

def setup_server(parsed_args: argparse.Namespace) -> None:
    """Setup rpc server for participant agent"""
    # ...
    # 在此省略一些代码，完整版代码见GitHub仓库
    server_launcher = RpcAgentServerLauncher(
        agent_class=agent_class,
        agent_kwargs=config,
        host=host,
        port=port,
    )
    server_launcher.launch(in_subprocess=False)
    server_launcher.wait_until_terminate()

代码中真正用于运行 Agent 服务器的代码只有以上片段。即使用 Agent 的 Class 、Agent 初始化参数以及 host 和 ip初始化一个 RpcAgentServerLauncher，之后调用 launch即可。这里为了让 Agent 正确获取到来自命令行的输入，将 in_subprocess设置为了 False。

下面将举例说明如何通过调用上述脚本运行 Agent 服务器。

例如，想要自己作为正方加入辩论，可以用如下命令启动正方的 Agent 服务器，注意如果正反双方位于不同的机器，需要将如下命令中--pro-host的值，也就是localhost换为当前机器的IP，并确保其他机器可通过该 IP 访问到此机器：

python distributed_debate.py --role pro --pro-host localhost --pro-port 12011 --is-human

如果想用让另一个用户来充当反方，则可以使用如下命令启动反方的 Agent 服务器，这里同样要注意正确填写--con-host的值，如果想让大模型来充当参与方，则去掉 --is-human即可。

python distributed_debate.py --role con --con-host localhost --con-port 12012 --is-human

上面启动的两个 Agent 服务器都会长期运行，如果想要关闭，可以使用 Ctrl + C，或是直接关闭对应的命令行窗口，但是请确保应用运行过程中这两个服务器进程都在正常工作。

在正反双方的 Agent 服务器都正常运行后就可以开始运行辩论应用了，主流程的代码如下：

def run_main_process(parsed_args: argparse.Namespace) -> None:
    """Setup the main debate competition process"""
    pro_agent, con_agent, judge_agent = agentscope.init(
        model_configs="configs/model_configs.json",
        agent_configs="configs/debate_agent_configs.json",
    )
    pro_agent = pro_agent.to_dist(
        host=parsed_args.pro_host,
        port=parsed_args.pro_port,
        launch_server=False,
    )
    con_agent = con_agent.to_dist(
        host=parsed_args.con_host,
        port=parsed_args.con_port,
        launch_server=False,
    )
    # ...
    # 在此省略一些代码，完整版代码见GitHub仓库

其中分布式相关代码位于 7-16 行。不同于之前提到的对话以及搜索案例，这里需要在to_dist中填写已经启动的Agent服务器的host以及port，并且需要指定 launch_server为False。

可以在其中一台电脑（或是再创建一个命令行界面）上运行如下指令来执行主流程。请注意 --pro-host、--pro-port、--con-host、--con-port的值都要与上面启动 Agent 服务器时对应的参数相同。

python distributed_debate.py --role main \
    --pro-host localhost --pro-port 12011 \
    --con-host localhost --con-port 12012

在运行上述指令后，启动正方 Agent 服务器的命令行界面上应该已经出现了辩论赛的开场白，以及 User input 的提示，你可以在 User input 输入自己的辩词，输入回车后即视为本轮发言完毕，轮到反方发言，下面是一个简单的案例。

System: Welcome to the debate on whether Artificial General Intelligence (AGI) can be achieved
...
User input: Some of Pro arguments

在正方完成输入后，如果反方 Agent 服务器是以用户形式启动，其命令行界面上会打印出正方的辩词，同时还会出现 User input 的提示，你可以在 User input 输入自己的辩词来反驳正方的观点。

System: Welcome to the debate on whether Artificial General Intelligence (AGI) can be achieved
...
Pro: Some of Pro arguments
User input: Some of Con arguments

反方输入完成后，轮到裁判总结该轮，正反双方的命令行界面中都会收到来自裁判的评价信息，案例如下。

System: Welcome to the debate on whether Artificial General Intelligence (AGI) can be achieved
...
User input: Some of Pro arguments
Con: Some of Con arguments
Judge: Some reviews

接下来将重复上述流程展开第二轮的辩论。

5. 总结

至此，我们介绍了如何使用 AgentScope 完成了多进程甚至是分布式应用的搭建，实现多个智能体以及用户之间的高效协同。分布式作为 AgentScope 的一项独特功能，能够让 AgentScope 实现的多智能体应用更加高效，并简化涉及多用户的应用程序编写。AgentScope 分布式模式的更详细介绍可以参考。

延伸阅读和资源

比赛

看到这里，如果你有好玩的想法，不如实践起来，还可以顺便参加下面的比赛~

点击即可跳转~

Create @ AI 创客松第四季 (aliyun.com)

Multi-Agent实践第7期：智能体的高效协作

1. 前言

2. 转化现有的 Multi-Agent 应用

3. Multi-Agent 并行加速样例：大模型辅助搜索

第一步：准备工作

第二步：运行案例

4. Multi-Agent 分布式&多用户协同：辩论赛

第一步：准备工作

第二步：运行程序

5. 总结

相关文章

相关实验场景

相关课程

热门文章

最新文章

相关电子书