使用AgentScope构建多智能体群聊系统

本文将介绍如何使用AgentScope框架构建一个简单的多智能体群聊系统，并解释其背后的实现逻辑。

首先写好设置文件。

agent_config.json

[
    {
        "class": "DialogAgent",
        "args":{
            "name": "Lingfeng",
            "sys_prompt": "You are Lingfeng, a noble in the kingdom of Xi'an. You are a wise and powerful wizard.",
            "model_config_name": "qwen_config",
            "use_memory": true
        }
    },
    {
        "class": "DialogAgent",
        "args":{
            "name": "Boyu",
            "sys_prompt": "You are Boyu, a friend of Lingfeng. Your speech is moderately humorous.",
            "model_config_name": "qwen_config",
            "use_memory": true
        }
    },
    {
        "class": "DialogAgent",
        "args":{
            "name": "Haotian",
            "sys_prompt": "You are Haotian, Lingfeng's cousin, a young man who is studying at the university.",
            "model_config_name": "qwen_config",
            "use_memory": true
        }
    }
]

model_config.json

{
    "config_name": "qwen_config",
    "model_type": "dashscope_chat",
    "model_name": "qwen-turbo",
    "api_key": "xxxxx",
    "generate_kwargs": {
        "temperature": 0.5
    }

}

实现代码

主逻辑 main.py

import agentscope
from agentscope.agents import DialogAgent, UserAgent
from agentscope.message import Msg
from agentscope.msghub import msghub
import logging

from groupchat_utils import (
    select_next_one,
    filter_agents,
)

# 定义用户发言超时时间
USER_TIME_TO_SPEAK = 30
# 默认聊天话题
DEFAULT_TOPIC = """
This is a chat room and you can speak freely and briefly.
"""
SYS_PROMPT = """
You can designate a member to reply to your message, you can use the @ symbol.
This means including the @ symbol in your message, followed by
that person's name, and leaving a space after the name.
All participants are: {agent_names}
"""

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def main() -> None:
    # 初始化NPC代理
    npc_agents = agentscope.init(
        model_configs="./model_config.json",
        agent_configs="./agent_config.json",
        project="Conversation with Mentions",
    )

    user = UserAgent()

    agents = list(npc_agents) + [user]
	# 创建初始提示消息
    hint = Msg(
        name="Host",
        content=DEFAULT_TOPIC
        + SYS_PROMPT.format(
            agent_names=[agent.name for agent in agents],
        ),
        role="assistant",
    )

    rnd = 0
    speak_list = []
	# 使用消息中心管理代理之间的通信
    with msghub(agents, announcement=hint):
        while True:
            try:
                x = user(timeout=USER_TIME_TO_SPEAK)
                if x.content == "exit":
                    break
            except TimeoutError:
                x = {"content": ""}
                logger.info(
                    f"User has not typed text for "
                    f"{USER_TIME_TO_SPEAK} seconds, skip."
                )
			# 根据用户输入过滤出需要发言的代理
            speak_list += filter_agents(x.content, npc_agents)
            # print("speak_list_before:", speak_list)
            # for agent in speak_list:
                # print(f"Name: {agent.name}")
			# 如果有需要发言的代理，则选择第一个
            if len(speak_list) > 0:
                next_agent = speak_list.pop(0)
                x = next_agent()
            else:
                # 否则根据轮次选择下一个代理
                next_agent = select_next_one(npc_agents, rnd)
                x = next_agent()

            speak_list += filter_agents(x.content, npc_agents)
            # print("speak_list_after:", speak_list)

            rnd += 1

if __name__ == "__main__":
    main()

辅助函数 groupchat_utils.py

import re
from typing import Sequence

def filter_agents(string: str, agents: Sequence) -> Sequence:
    """
    筛选出字符串中包含的代理名称，并返回找到的代理对象列表。
    """
    if len(agents) == 0:
        return []
    pattern = (
        r"@(" + "|".join(re.escape(agent.name) for agent in agents) + r")\b"
    )
    matches = re.findall(pattern, string)
    agent_dict = {agent.name: agent for agent in agents}
    ordered_agents = [
        agent_dict[name] for name in matches if name in agent_dict
    ]
    return ordered_agents

def select_next_one(agents: Sequence, rnd: int) -> Sequence:
    """
    根据轮次选择下一个发言的代理。
    """
    return agents[rnd % len(agents)]

代码解析

这段代码实现了多智能体之间的群聊功能。用户可以与多个智能体（NPC）进行互动，而这些智能体之间也能相互交流。关键在于：

• 初始化智能体：通过agentscope.init()函数加载配置文件，并创建智能体实例。
• 消息传递：使用msghub来管理智能体间的通讯。
• 指定回复者：通过在消息中包含@符号加上特定智能体的名字来指定回复的对象。
• 轮询机制：如果没有明确指定回复者，则按照轮询的方式选择下一个发言的智能体。

最终效果

运行上述代码后，用户可以参与到与多个智能体的对话中。用户输入的内容将被解析，如果含有@符号，则直接指定了下一个应该回应的智能体。如果没有指定，则按顺序选择智能体回应。

代码中的print部分为我加入的输出，可以看到目前在运行中还存在一些问题。 1.如果名字不打的完全准确是无法识别的

2.如果一次@了两个人，中间会需要用户进行输入

3.让NPC互相聊天会出现一个NPC把所有的NPC对话全部生成的情况。

有一些推想的解决方案，还没有实行，但是暂且列出来：

1.引入模糊匹配机制解决名字识别问题。通过使用正则表达式或其他字符串匹配算法，系统可以识别近似的名称输入，并将其映射到正确的NPC上。

2.可以通过调整代码逻辑解决。

3.目前还没有解决方案。