Group Chat 模式：多角色讨论
解决什么问题
旅行团要去云南，导游、预算员、摄影师各有主张，需要一轮一轮讨论才能达成共识。 Group Chat 让多个 Agent 在同一个对话空间里发言，支持两种轮次策略：
- Round-Robin：按固定顺序轮流说，简单可控
- Selector：由"主持人"根据上下文动态选下一个发言者，更灵活
复杂度
⭐⭐⭐ 中等。核心是消息广播 + 轮次控制逻辑。
和其他模式的关系
| 模式 | 谁决定下一步 | 什么时候用 |
|---|---|---|
| Group Chat | 主持人选发言者（或 Round-Robin） | 多角色需要协作达成共识 |
| Manager-Worker | Manager 分配子任务 | 子任务独立、不需要讨论 |
| Multi-Agent Debate | 裁判评分、正反方交替 | 需要对抗式找漏洞、不是协作 |
Group Chat 比 Manager-Worker 更民主（人人可发言），比 Debate 更协作（目标是共识不是对抗）。
角色关系
┌────────────┐
│ Moderator │ ← 控制轮次、判定终止
│ (主持人) │
└──┬───┬───┬──┘
│ │ │ 广播消息
▼ ▼ ▼
┌───┐┌───┐┌───┐
│ A1 ││ A2 ││ A3 │ ← 所有人看到同一个聊天记录
└───┘└───┘└───┘
- 共享聊天记录（所有人可见）
- Moderator 决定谁发言、何时结束
- 每轮只有一个 Agent 说话
完整 Python 实现
"""
Group Chat 云南旅行讨论示例
三个角色：导游(线路)、预算员(花费)、摄影师(拍照点)。
支持 round-robin 和 selector 两种轮次策略。
"""
from __future__ import annotations
import json
from dataclasses import dataclass, field
from enum import Enum
class TurnPolicy(Enum):
ROUND_ROBIN = "round_robin"
SELECTOR = "selector"
# ── MockLLM ──────────────────────────────────────────────
class MockLLM:
def __init__(self):
self.call_log: list[dict] = []
self._round = 0
def chat(self, role: str, prompt: str) -> str:
self.call_log.append({"role": role, "prompt": prompt[:80]})
self._round += 1
# 导游发言
if role == "guide":
if self._round <= 3:
return "建议路线: 昆明→大理→丽江→香格里拉，5天4晚，涵盖洱海、玉龙雪山。"
return "同意最终方案，路线确认。"
# 预算员发言
if role == "budget":
if self._round <= 4:
return "按这条线路估算: 交通¥1500+住宿¥1200+餐饮¥800+门票¥600=总计约¥4100/人。"
return "预算在5000以内，可行。"
# 摄影师发言
if role == "photographer":
if self._round <= 5:
return "推荐拍摄点: 洱海日出(6:00)、双廊渔船、玉龙雪山蓝月谷、丽江古城夜景。"
return "拍摄计划已对齐行程，没问题。"
# Selector 选下一个发言者
if role == "moderator":
if "选择" in prompt:
# 简单策略：依次选
agents = ["guide", "budget", "photographer"]
idx = (self._round - 1) % len(agents)
return agents[idx]
if "是否结束" in prompt:
return "yes" if self._round > 7 else "no"
return ""
# ── Agent ────────────────────────────────────────────────
@dataclass
class ChatAgent:
name: str
role: str
desc: str
llm: MockLLM = field(repr=False)
def speak(self, chat_history: list[dict]) -> str:
context = "\n".join(f"[{m['speaker']}] {m['content']}" for m in chat_history[-6:])
prompt = f"聊天记录:\n{context}\n\n请以{self.name}身份发言:"
return self.llm.chat(self.role, prompt)
# ── Group Chat ───────────────────────────────────────────
@dataclass
class GroupChat:
agents: list[ChatAgent]
moderator_llm: MockLLM = field(repr=False)
policy: TurnPolicy = TurnPolicy.ROUND_ROBIN
max_rounds: int = 6
chat_history: list[dict] = field(default_factory=list)
def _select_next_round_robin(self, round_idx: int) -> ChatAgent:
return self.agents[round_idx % len(self.agents)]
def _select_next_selector(self) -> ChatAgent:
context = "\n".join(f"[{m['speaker']}] {m['content']}" for m in self.chat_history[-4:])
prompt = f"聊天记录:\n{context}\n\n请选择下一个发言者:"
chosen_role = self.moderator_llm.chat("moderator", prompt)
for agent in self.agents:
if agent.role == chosen_role.strip():
return agent
return self.agents[0] # fallback
def _should_stop(self) -> bool:
prompt = f"已经进行了{len(self.chat_history)}轮讨论，是否结束？"
answer = self.moderator_llm.chat("moderator", prompt)
return "yes" in answer.lower()
def run(self, topic: str) -> list[dict]:
print(f"[GroupChat] 话题: {topic}")
print(f"[GroupChat] 轮次策略: {self.policy.value}")
self.chat_history.append({"speaker": "系统", "content": f"讨论话题: {topic}"})
for i in range(self.max_rounds):
# 选发言者
if self.policy == TurnPolicy.ROUND_ROBIN:
agent = self._select_next_round_robin(i)
else:
agent = self._select_next_selector()
# 发言
msg = agent.speak(self.chat_history)
self.chat_history.append({"speaker": agent.name, "content": msg})
print(f" [Round {i+1}] {agent.name}: {msg[:50]}...")
# 检查是否结束
if i >= 2 and self._should_stop():
print(f"[GroupChat] 主持人判定讨论结束")
break
print(f"[GroupChat] 共 {len(self.chat_history)-1} 轮发言")
return self.chat_history
# ── 运行入口 ─────────────────────────────────────────────
def main():
llm = MockLLM()
agents = [
ChatAgent("导游小王", "guide", "负责路线规划", llm),
ChatAgent("预算员小李", "budget", "负责费用核算", llm),
ChatAgent("摄影师小张", "photographer", "负责拍摄安排", llm),
]
# ── Round-Robin 模式 ──
print("=" * 50)
print("【Round-Robin 模式】")
gc_rr = GroupChat(agents=agents, moderator_llm=llm, policy=TurnPolicy.ROUND_ROBIN, max_rounds=6)
history_rr = gc_rr.run("云南5日团队游规划")
assert len(history_rr) > 3
assert any("洱海" in m["content"] for m in history_rr)
# ── Selector 模式 ──
print("\n" + "=" * 50)
print("【Selector 模式】")
llm2 = MockLLM()
agents2 = [
ChatAgent("导游小王", "guide", "负责路线规划", llm2),
ChatAgent("预算员小李", "budget", "负责费用核算", llm2),
ChatAgent("摄影师小张", "photographer", "负责拍摄安排", llm2),
]
gc_sel = GroupChat(agents=agents2, moderator_llm=llm2, policy=TurnPolicy.SELECTOR, max_rounds=6)
history_sel = gc_sel.run("云南5日团队游规划")
assert len(history_sel) > 3
print(f"\n✓ 两种策略均运行成功")
if __name__ == "__main__":
main()
运行记录
==================================================
【Round-Robin 模式】
[GroupChat] 话题: 云南5日团队游规划
[GroupChat] 轮次策略: round_robin
[Round 1] 导游小王: 建议路线: 昆明→大理→丽江→香格里拉，5天4晚，涵盖洱海、玉龙...
[Round 2] 预算员小李: 按这条线路估算: 交通¥1500+住宿¥1200+餐饮¥800+门票...
[Round 3] 摄影师小张: 推荐拍摄点: 洱海日出(6:00)、双廊渔船、玉龙雪山蓝月谷、丽...
[Round 4] 导游小王: 同意最终方案，路线确认。...
[GroupChat] 主持人判定讨论结束
[GroupChat] 共 4 轮发言
==================================================
【Selector 模式】
[GroupChat] 话题: 云南5日团队游规划
[GroupChat] 轮次策略: selector
[Round 1] 导游小王: 建议路线: 昆明→大理→丽江→香格里拉...
[Round 2] 预算员小李: 按这条线路估算...
[Round 3] 摄影师小张: 推荐拍摄点: 洱海日出...
[GroupChat] 主持人判定讨论结束
[GroupChat] 共 3 轮发言
✓ 两种策略均运行成功
踩坑记录
| 坑 | 现象 | trace 信号 | 修法 |
|---|---|---|---|
| 死循环 | 主持人永远不说结束 | round 数触达 max_rounds 仍未终止 | 设 max_rounds 硬上限 |
| 鹦鹉效应 | Agent 不断复述前人观点 | 连续发言内容文本相似度 > 0.8 | prompt 里加"请提出新观点或表示同意" |
| 上下文爆炸 | 10轮讨论 token 飙升 | 单次 chat 输入 token 数持续增长 | 只传最近 N 条消息，或做滚动摘要 |
| Selector 偏心 | 总选同一个人 | 发言者分布极度不均匀（某人占 70%+） | 加"不要连续选同一人"约束 |
工程备忘
- Round-Robin 适合角色对等的场景（头脑风暴）；Selector 适合需要动态跟进的场景（有人抛出新问题，需要特定专家回应）。
- 聊天记录管理：超过 N 轮做 rolling summary，避免 token 浪费。
- 终止条件：除了主持人判断，还可以设"连续 2 人说'同意'即结束"等规则。
- 发言质量：给每个 Agent 加 system_prompt 约束角色边界，避免越权发言。
- 与 Debate 的区别：Group Chat 是协作达成共识，Debate 是对抗式找漏洞。
- 生产建议：聊天记录持久化到数据库，支持断点续聊。
读完以后
如果讨论的目的不是共识而是找漏洞，看 Multi-Agent Debate。 如果角色之间不需要讨论、只需要各干各的再合并，看 Manager-Worker。 如果参与者完全对等、没有主持人，看 Swarm Blackboard。