在多智能体系统开发中，我们常常面临这样的挑战：如何让多个不同角色的智能体有序协作，共同完成一个复杂任务？比如在内容创作场景中，需要作家、编辑、插画师等多个角色协同工作。这时，群聊（Group Chat）设计模式就成为了理想的解决方案。今天，我们将通过一个生动的儿童故事书创作案例，深入探讨群聊模式的核心机制与实现方式，帮助大家在实际项目中应用这一强大的多智能体协作模式。

一、群聊模式的核心概念与工作原理

1.1 群聊模式的本质与特点

群聊模式是一种多智能体协作设计模式，其核心特点是：

  • 多个智能体共享同一个消息主题，形成一个 "对话线程"
  • 智能体按顺序轮流发言，每次只有一个智能体活跃
  • 由群聊管理器（Group Chat Manager）负责协调发言顺序
  • 每个智能体专注于特定任务，如写作、编辑、绘图等

这种模式就像一个小型的协作团队会议，每个成员按顺序发言，贡献自己的专业知识，最终共同完成一个复杂任务。在技术实现上，它通过主题订阅、消息路由和智能选择算法来确保协作的有序进行。

1.2 群聊模式的核心组件

一个完整的群聊系统包含以下关键组件：

  1. 参与智能体：具有明确角色的智能体，如作家、编辑、插画师
  2. 群聊管理器：负责协调发言顺序，选择下一个发言者
  3. 消息协议：定义智能体间通信的消息格式和流程
  4. 主题系统：智能体通过主题订阅实现消息共享

1.3 发言顺序控制算法

群聊管理器使用特定算法选择下一个发言者，常见策略包括：

  • 轮询算法（Round-Robin）：按固定顺序轮流选择
  • LLM 驱动选择：使用大语言模型根据对话内容动态选择
  • 优先级算法：根据任务优先级确定发言顺序

在我们的示例中，将使用 LLM 驱动的选择算法，让管理器根据对话历史和角色需求智能选择下一个发言者。

二、群聊模式的消息协议设计

2.1 核心消息类型

群聊模式定义了两种关键消息类型：

python

from pydantic import BaseModel
from autogen_core.models import UserMessage

class GroupChatMessage(BaseModel):
    body: UserMessage  # 群聊内容消息，包含实际对话内容

class RequestToSpeak(BaseModel):
    pass  # 请求发言消息，由管理器发送给下一个发言者

2.2 消息流转流程

群聊的消息流转遵循以下流程：

  1. 用户或外部智能体发布GroupChatMessage到公共主题
  2. 群聊管理器接收消息，选择下一个发言者
  3. 管理器向目标智能体发布RequestToSpeak消息
  4. 目标智能体接收请求，处理后发布新的GroupChatMessage
  5. 重复步骤 2-4，直到管理器判定任务完成

这个流程可以用以下图表表示：

plaintext

用户/外部智能体 → 公共主题(GroupChatMessage) → 群聊管理器
↓
管理器选择下一个发言者 → 目标主题(RequestToSpeak) → 目标智能体
↓
目标智能体处理 → 公共主题(GroupChatMessage) → 群聊管理器

三、群聊智能体的实现：儿童故事书创作团队

3.1 基础群聊智能体

首先定义一个基础智能体类，包含群聊所需的核心功能：

python

from autogen_core import RoutedAgent, MessageContext
from autogen_core.models import SystemMessage, UserMessage, AssistantMessage
from autogen_core import TypeSubscription, type_subscription

class BaseGroupChatAgent(RoutedAgent):
    """群聊中使用LLM的基础智能体"""
    def __init__(
        self,
        description: str,
        group_chat_topic_type: str,
        model_client: ChatCompletionClient,
        system_message: str,
    ) -> None:
        super().__init__(description=description)
        self._group_chat_topic_type = group_chat_topic_type
        self._model_client = model_client
        self._system_message = SystemMessage(content=system_message)
        self._chat_history: List[LLMMessage] = []
    
    @message_handler
    async def handle_message(self, message: GroupChatMessage, ctx: MessageContext) -> None:
        """处理群聊消息，更新聊天历史"""
        self._chat_history.extend([
            UserMessage(content=f"Transferred to {message.body.source}", source="system"),
            message.body,
        ])
    
    @message_handler
    async def handle_request_to_speak(self, message: RequestToSpeak, ctx: MessageContext) -> None:
        """处理发言请求，调用LLM生成响应并发布"""
        self._chat_history.append(
            UserMessage(content=f"Transferred to {self.id.type}, adopt the persona immediately.", source="system")
        )
        completion = await self._model_client.create([self._system_message] + self._chat_history)
        self._chat_history.append(AssistantMessage(content=completion.content, source=self.id.type))
        await self.publish_message(
            GroupChatMessage(body=UserMessage(content=completion.content, source=self.id.type)),
            topic_id=DefaultTopicId(type=self._group_chat_topic_type),
        )

这个基础类实现了群聊智能体的核心功能：消息处理、聊天历史管理和 LLM 调用。

后续作家（WriterAgent）、编辑（EditorAgent）、插画师（IllustratorAgent）等具体角色智能体均继承此基类，只需关注角色特有的系统提示（system_message），无需重复实现消息处理底层逻辑。

3.2 作家智能体（WriterAgent）

负责故事内容的创作：

python

class WriterAgent(BaseGroupChatAgent):
    def __init__(self, description: str, group_chat_topic_type: str, model_client: ChatCompletionClient) -> None:
        super().__init__(
            description=description,
            group_chat_topic_type=group_chat_topic_type,
            model_client=model_client,
            system_message="You are a Writer. You produce good work.",
        )

作家智能体使用简单的系统提示，告知 LLM 其角色是创作优质内容。

3.3 编辑智能体（EditorAgent）

负责审核和优化内容：

python

class EditorAgent(BaseGroupChatAgent):
    def __init__(self, description: str, group_chat_topic_type: str, model_client: ChatCompletionClient) -> None:
        super().__init__(
            description=description,
            group_chat_topic_type=group_chat_topic_type,
            model_client=model_client,
            system_message="You are an Editor. Plan and guide the task given by the user. Provide critical feedbacks to the draft and illustration produced by Writer and Illustrator. Approve if the task is completed and meets user's requirements.",
        )

编辑智能体的系统提示更复杂，要求其提供专业反馈，并在任务完成时批准。

3.4 插画师智能体（IllustratorAgent）

负责生成故事插图，集成 DALL-E 图像生成模型：

python

import openai
from autogen_core.tools import FunctionTool
from autogen_core import Image

class IllustratorAgent(BaseGroupChatAgent):
    def __init__(
        self,
        description: str,
        group_chat_topic_type: str,
        model_client: ChatCompletionClient,
        image_client: openai.AsyncClient,
    ) -> None:
        super().__init__(
            description=description,
            group_chat_topic_type=group_chat_topic_type,
            model_client=model_client,
            system_message="You are an Illustrator. You use the generate_image tool to create images given user's requirement. Make sure the images have consistent characters and style.",
        )
        self._image_client = image_client
        self._image_gen_tool = FunctionTool(
            self._image_gen, name="generate_image", description="Call this to generate an image."
        )
    
    async def _image_gen(
        self, character_appearence: str, style_attributes: str, worn_and_carried: str, scenario: str
    ) -> str:
        """调用DALL-E生成图像"""
        prompt = f"Digital painting of a {character_appearence} character with {style_attributes}. Wearing {worn_and_carried}, {scenario}."
        response = await self._image_client.images.generate(
            prompt=prompt, model="dall-e-3", response_format="b64_json", size="1024x1024"
        )
        return response.data[0].b64_json
    
    @message_handler
    async def handle_request_to_speak(self, message: RequestToSpeak, ctx: MessageContext) -> None:
        """处理发言请求，调用图像生成工具"""
        completion = await self._model_client.create(
            [self._system_message] + self._chat_history,
            tools=[self._image_gen_tool],
            extra_create_args={"tool_choice": "required"},
        )
        # 处理图像生成结果并发布
        ...

插画师智能体扩展了基础类，增加了图像生成功能，使用FunctionTool包装 DALL-E API 调用。

3.5 用户智能体（UserAgent）

代表人类用户，提供最终批准：

python

class UserAgent(RoutedAgent):
    def __init__(self, description: str, group_chat_topic_type: str) -> None:
        super().__init__(description=description)
        self._group_chat_topic_type = group_chat_topic_type
    
    @message_handler
    async def handle_request_to_speak(self, message: RequestToSpeak, ctx: MessageContext) -> None:
        """处理发言请求，获取用户输入"""
        user_input = input("Enter your message, type 'APPROVE' to conclude the task: ")
        await self.publish_message(
            GroupChatMessage(body=UserMessage(content=user_input, source=self.id.type)),
            DefaultTopicId(type=self._group_chat_topic_type),
        )

用户智能体通过控制台获取输入，在实际应用中可替换为前端交互。

四、群聊管理器：协作流程的指挥中心

4.1 管理器核心功能

群聊管理器负责：

  • 维护聊天历史
  • 分析对话内容
  • 选择下一个发言者
  • 判定任务完成条件

4.2 管理器实现代码

python

from typing import List

class GroupChatManager(RoutedAgent):
    def __init__(
        self,
        participant_topic_types: List[str],
        model_client: ChatCompletionClient,
        participant_descriptions: List[str],
    ) -> None:
        super().__init__("Group chat manager")
        self._participant_topic_types = participant_topic_types
        self._model_client = model_client
        self._chat_history: List[UserMessage] = []
        self._participant_descriptions = participant_descriptions
        self._previous_participant_topic_type: str | None = None
    
    @message_handler
    async def handle_message(self, message: GroupChatMessage, ctx: MessageContext) -> None:
        """处理群聊消息，选择下一个发言者"""
        self._chat_history.append(message.body)
        # 检查用户是否批准，若批准则结束群聊
        if message.body.source == "User" and message.body.content.lower().strip().endswith("approve"):
            return
        
        # 格式化聊天历史和角色信息，用于LLM选择
        messages = [f"{msg.source}: {msg.content}" for msg in self._chat_history]
        history = "\n".join(messages)
        roles = "\n".join([
            f"{topic_type}: {description}"
            for topic_type, description in zip(self._participant_topic_types, self._participant_descriptions)
            if topic_type != self._previous_participant_topic_type
        ])
        
        # 构建选择提示词，使用LLM选择下一个发言者
        selector_prompt = """You are in a role play game. The following roles are available:
{roles}.
Read the following conversation. Then select the next role from {participants} to play. Only return the role.
{history}
Read the above conversation. Then select the next role from {participants} to play. Only return the role.
"""
        system_message = SystemMessage(
            content=selector_prompt.format(
                roles=roles,
                history=history,
                participants=str([
                    topic_type for topic_type in self._participant_topic_types
                    if topic_type != self._previous_participant_topic_type
                ]),
            )
        )
        completion = await self._model_client.create([system_message])
        # 解析LLM输出，确定下一个发言者并发送请求
        ...

4.3 发言者选择算法

管理器使用 LLM 驱动的选择算法：

  1. 收集当前聊天历史和可用角色信息
  2. 构建提示词，要求 LLM 从可用角色中选择下一个发言者
  3. 解析 LLM 输出，确定目标角色
  4. 发送RequestToSpeak消息到目标角色主题

这种算法使选择过程更加智能，能够根据对话内容动态调整发言顺序。

五、群聊系统的搭建与运行

5.1 智能体注册与订阅

python

from autogen_core import SingleThreadedAgentRuntime, DefaultTopicId, TopicId

# 初始化运行时
runtime = SingleThreadedAgentRuntime()

# 定义主题类型
editor_topic_type = "Editor"
writer_topic_type = "Writer"
illustrator_topic_type = "Illustrator"
user_topic_type = "User"
group_chat_topic_type = "group_chat"

# 注册编辑智能体
editor_agent_type = await EditorAgent.register(
    runtime,
    editor_topic_type,
    lambda: EditorAgent(
        description="Editor for planning and reviewing the content.",
        group_chat_topic_type=group_chat_topic_type,
        model_client=model_client,
    ),
)
# 添加主题订阅
await runtime.add_subscription(TypeSubscription(topic_type=editor_topic_type, agent_type=editor_agent_type.type))
await runtime.add_subscription(TypeSubscription(topic_type=group_chat_topic_type, agent_type=editor_agent_type.type))

# 类似地注册作家、插画师和用户智能体
...

# 注册群聊管理器
group_chat_manager_type = await GroupChatManager.register(
    runtime,
    "group_chat_manager",
    lambda: GroupChatManager(
        participant_topic_types=[writer_topic_type, illustrator_topic_type, editor_topic_type, user_topic_type],
        model_client=model_client,
        participant_descriptions=[
            "Writer for creating any text content.",
            "An illustrator for creating images.",
            "Editor for planning and reviewing the content.",
            "User for providing final approval."
        ],
    ),
)
await runtime.add_subscription(
    TypeSubscription(topic_type=group_chat_topic_type, agent_type=group_chat_manager_type.type)
)

5.2 启动群聊并查看结果

python

# 启动运行时
runtime.start()

# 发布初始消息，触发群聊
session_id = str(uuid.uuid4())
await runtime.publish_message(
    GroupChatMessage(
        body=UserMessage(
            content="Please write a short story about the gingerbread man with up to 3 photo-realistic illustrations.",
            source="User",
        )
    ),
    TopicId(type=group_chat_topic_type, source=session_id),
)

# 等待群聊结束
await runtime.stop_when_idle()
await model_client.close()

5.3 执行结果展示

运行后，我们可以看到智能体按顺序协作的过程：

  1. 作家智能体首先生成故事草稿和插图描述
  2. 插画师智能体根据描述生成插图
  3. 编辑智能体审核内容并提出修改建议
  4. 作家和插画师根据反馈修改
  5. 最终用户批准完成的故事书

以下是部分执行结果示例：

plaintext

Writer:
Title: The Escape of the Gingerbread Man
Illustration 1: A Rustic Kitchen Scene...
Story: Once there was an old woman who lived alone...

Illustrator:
生成插图1的参数和图像...

Editor:
Thank you for submitting the draft...
Story Feedback:
1. Plot & Structure: The story follows the traditional...

Writer:
Certainly! Here’s the final version...

六、群聊模式的优势与实践建议

6.1 模式优势总结

  1. 角色明确：每个智能体专注于特定任务，符合单一职责原则
  2. 协作有序：通过管理器协调，确保任务按逻辑顺序推进
  3. 动态灵活：使用 LLM 驱动的选择算法，可根据任务需求动态调整
  4. 可扩展性：轻松添加新角色智能体，扩展协作能力

七、总结

通过本文的介绍，我们深入探讨了群聊模式的核心概念、实现方式和实际应用。从儿童故事书的创作案例中，我们看到了多个智能体如何通过有序协作，完成从故事构思、内容创作、插图生成到最终审核的完整流程。

如果本文对你有帮助，别忘了点赞收藏，关注我，一起探索更高效的开发方式～

# 人工智能
Logo
技术共进，成长同行——讯飞AI开发者社区

技术共进，成长同行——讯飞AI开发者社区

更多推荐

论文笔记：AlphaEdit: Null-Space Constrained Knowledge Editing for Language Models（AlphaEdit）

论文发表于人工智能顶会ICLR（基于定位和修改的模型编辑方法（针对和等）会破坏LLM中最初保存的知识，特别是在顺序编辑场景。为此，本文提出AlphaEdit：1、在将保留知识应用于参数之前，将扰动投影到保留知识的零空间上。2、从理论上证明，这种预测确保了在查询保留的知识时，编辑后的LLM的输出保持不变，从而减轻中断问题。3、对各种LLM（包括LLaMA3、GPT2XL和GPT-J）的广泛实验表明，

avatar 讯飞AI开发者社区

论文解读：MASS-EDITING MEMORY IN A TRANSFORMER（MEMIT）

论文发表于人工智能顶会ICLR（在模型编辑方法中，主要局限于更新单个事实。因此，基于ROME，本文开发了MEMIT，在大模型GPT-J（6B）和GPT-NeoX（20B）上实现了数千的批量编辑。阅读本文请同时参考原始论文图表。

avatar 讯飞AI开发者社区
cover

青柠AI（paperwx）专业的论文写作工具，堪称论文写作领域的实用神器

avatar 讯飞AI开发者社区