内容纲要
AI群聊与协作系统（详解版）
目标：精通AI群聊设计、角色扮演、协作对话
目录
AI群聊概述
1. 什么是AI群聊？
【定义】
AI群聊是多个AI角色（Agent）在一个虚拟空间中进行
互动对话、协作解决问题的系统。
【特点】
1. 多角色互动
- 每个AI有独特性格
- 持久的角色记忆
- 一致的行为模式
2. 自然对话
- 模拟人类群聊
- 支持上下文理解
- 流畅的对话流
3. 协作解决问题
- 共同分析问题
- 分工合作
- 达成共识
【应用场景】
- 头脑风暴
- 团队协作模拟
- 角色扮演游戏
- 案例研讨
- 协作写作
- 代码审查2. 群聊类型
【自由讨论型】
特点：
- 无明确目标
- 自由发言
- 混合输出
适合：
- 创意激发
- 头脑风暴
- 开放讨论
【任务导向型】
特点：
- 有明确目标
- 结构化对话
- 结果导向
适合：
- 问题解决
- 决策制定
- 方案设计
【角色扮演型】
特点：
- 特定角色设定
- 角色互动
- 剧情发展
适合：
- 剧本创作
- 游戏NPC
- 情景模拟
【辩论型】
特点：
- 正反方对立
- 论证交锋
- 评判结果
适合：
- 政策辩论
- 方案对比
- 风险评估角色设计
1. 角色定义
from dataclasses import dataclass
from typing import List, Dict
@dataclass
class AICharacter:
"""AI角色定义"""
id: str # 角色ID
name: str # 角色名称
avatar: str # 头像URL
# 性格特征
personality: Dict[str, float] # 性格值（0-1）
# 开放性、尽责性、外向性、宜人性、神经质
# 角色设定
role: str # 职责
background: str # 背景
goals: List[str] # 目标
constraints: List[str] # 约束
# 对话风格
speaking_style: str # 说话风格
language_level: str # 语言水平
response_length: str # 回答长度
emotion_level: float # 情感程度
# 技能
skills: Dict[str, float] # 技能熟练度
tools: List[str] # 可用工具
# 记忆
memory: Dict # 角色记忆
knowledge_base: Dict # 知识库
# 示例角色定义
product_manager = AICharacter(
id="pm_001",
name="张产品",
avatar="/avatars/pm.jpg",
personality={
"openness": 0.8,
"conscientiousness": 0.9,
"extroversion": 0.7,
"agreeableness": 0.6,
"neuroticism": 0.3
},
role="产品经理",
background="10年互联网产品经验，擅长用户需求分析和产品规划",
goals=[
"理解用户需求",
"制定产品方案",
"推动产品落地"
],
constraints=[
"关注用户体验",
"考虑技术可行性",
"控制成本"
],
speaking_style="专业、逻辑清晰、善于总结",
language_level="专业",
response_length="中等",
emotion_level=0.5,
skills={
"需求分析": 0.9,
"产品规划": 0.85,
"项目管理": 0.8,
"沟通协调": 0.9
},
memory={},
knowledge_base={}
)
developer = AICharacter(
id="dev_001",
name="李开发",
avatar="/avatars/dev.jpg",
personality={
"openness": 0.7,
"conscientiousness": 0.95,
"extroversion": 0.5,
"agreeableness": 0.7,
"neuroticism": 0.4
},
role="高级开发工程师",
background="8年全栈开发经验，精通Java和Python",
goals=[
"编写高质量代码",
"优化系统性能",
"解决技术难题"
],
constraints=[
"代码规范",
"安全第一",
"可维护性"
],
speaking_style="技术性强、简洁、喜欢举例",
language_level="技术",
response_length="简洁",
emotion_level=0.3,
skills={
"编程": 0.95,
"系统设计": 0.9,
"问题解决": 0.9,
"代码审查": 0.85
},
memory={},
knowledge_base={}
)2. 角色模板
【专业角色模板】
产品经理：
- 关注用户价值
- 平衡多方需求
- 善于总结提炼
技术负责人：
- 关注技术可行性
- 架构设计能力
- 风险控制意识
设计师：
- 关注用户体验
- 视觉审美能力
- 细节把控
运营：
- 关注数据分析
- 策略制定能力
- 效果追踪
测试工程师：
- 关注质量保障
- 测试用例设计
- 问题发现能力
【创意角色模板】
创意总监：
- 思维活跃
- 善于创新
- 引领方向
文案策划：
- 文字功底
- 情感共鸣
- 品牌调性
数据分析师：
- 数据敏感
- 洞察力强
- 趋势判断
【对立角色模板】
乐观派：
- 积极正面
- 关注机遇
- 风险偏好高
悲观派：
- 审慎保守
- 关注风险
- 风险偏好低
激进派：
- 敢于突破
- 快速推进
- 接受失败
稳健派：
- 循序渐进
- 稳健行事
- 降低波动3. 角色交互关系
@dataclass
class Relationship:
"""角色间关系"""
from_character: str
to_character: str
# 关系类型
relationship_type: str # 同事、朋友、对手等
# 关系强度
closeness: float # 亲密程度（0-1）
trust: float # 信任程度（0-1）
influence: float # 影响力（0-1）
# 交互偏好
interaction_frequency: float # 互动频率
response_likelihood: float # 回应概率
# 定义角色关系
relationships = [
Relationship(
from_character="pm_001",
to_character="dev_001",
relationship_type="同事",
closeness=0.7,
trust=0.8,
influence=0.9,
interaction_frequency=0.8,
response_likelihood=0.9
),
Relationship(
from_character="dev_001",
to_character="pm_001",
relationship_type="同事",
closeness=0.6,
trust=0.75,
influence=0.7,
interaction_frequency=0.7,
response_likelihood=0.85
)
]对话管理
1. 消息模型
from datetime import datetime
from enum import Enum
class MessageType(Enum):
TEXT = "text"
ACTION = "action"
EMOTION = "emotion"
SYSTEM = "system"
@dataclass
class Message:
"""消息模型"""
id: str # 消息ID
conversation_id: str # 会话ID
character_id: str # 发送者ID
character_name: str # 发送者名称
# 消息内容
content: str # 消息文本
message_type: MessageType # 消息类型
metadata: Dict # 元数据
# 时间信息
timestamp: datetime # 发送时间
# 引用关系
reply_to: str = None # 回复的消息ID
mentions: List[str] = None # @的角色
# 情感标签
emotion: str = None # 情感标签
sentiment: float = None # 情感值（-1到1）
# 状态
is_edited: bool = False # 是否编辑
edit_history: List[Dict] = None # 编辑历史2. 对话上下文
@dataclass
class ConversationContext:
"""对话上下文"""
conversation_id: str
characters: List[AICharacter]
# 消息历史
messages: List[Message]
# 当前状态
current_topic: str # 当前话题
current_speaker: str # 当前发言者
turn_count: int # 轮次计数
# 协作状态
shared_goals: List[str] # 共同目标
shared_knowledge: Dict # 共享知识
decisions_made: List[Dict] # 已做决策
# 群体状态
group_mood: str # 群体氛围
engagement_level: float # 参与度
consensus_level: float # 共识程度
# 元数据
started_at: datetime
ended_at: datetime = None3. 对话轮次管理
class TurnManager:
"""对话轮次管理器"""
def __init__(self, characters: List[AICharacter]):
self.characters = characters
self.current_turn = 0
self.turn_history = []
def get_next_speaker(self, context: ConversationContext) -> AICharacter:
"""获取下一个发言者"""
strategies = [
self.round_robin_strategy,
self.relevance_strategy,
self.random_strategy,
self.influence_strategy
]
# 选择策略
strategy = self.select_strategy(context)
# 获取发言者
speaker = strategy(context)
# 记录轮次
self.turn_history.append({
"turn": self.current_turn,
"speaker": speaker.id,
"timestamp": datetime.now()
})
self.current_turn += 1
return speaker
def round_robin_strategy(self, context) -> AICharacter:
"""轮询策略"""
return self.characters[self.current_turn % len(self.characters)]
def relevance_strategy(self, context) -> AICharacter:
"""相关性策略：根据上一条消息的相关性选择"""
last_message = context.messages[-1]
# 找到最相关的角色
most_relevant = max(
self.characters,
key=lambda c: self.calculate_relevance(c, last_message)
)
return most_relevant
def random_strategy(self, context) -> AICharacter:
"""随机策略"""
import random
return random.choice(self.characters)
def influence_strategy(self, context) -> AICharacter:
"""影响力策略：优先让影响力高的角色发言"""
return max(
self.characters,
key=lambda c: sum(
rel.influence
for rel in context.relationships
if rel.to_character == c.id
)
)协作模式
1. 头脑风暴模式
class BrainstormingSession:
"""头脑风暴会话"""
def __init__(self, characters: List[AICharacter]):
self.characters = characters
self.ideas = []
self.feedback = []
def generate_ideas(self, topic: str, round_count: int = 3):
"""生成创意"""
for round_num in range(round_count):
print(f"\n=== 第 {round_num + 1} 轮头脑风暴 ===")
for character in self.characters:
# 生成创意
idea = self.generate_character_idea(character, topic, self.ideas)
self.ideas.append({
"character": character.name,
"idea": idea,
"round": round_num
})
print(f"{character.name}: {idea}")
# 收集反馈
self.collect_feedback()
# 创意整合
self.synthesize_ideas()
def generate_character_idea(self, character: AICharacter,
topic: str, existing_ideas: List[Dict]) -> str:
"""生成角色创意"""
existing_ideas_text = "\n".join([
f"{idea['character']}: {idea['idea']}"
for idea in existing_ideas
])
prompt = f"""
你是{character.name}，{character.role}。
{character.background}
你的说话风格：{character.speaking_style}
当前话题：{topic}
之前的创意：
{existing_ideas_text if existing_ideas else "暂无"}
请提供一个独特、有创意的想法，避免重复。
"""
return self.llm.generate(prompt)
def collect_feedback(self):
"""收集反馈"""
print("\n=== 反馈阶段 ===")
for character in self.characters:
feedback = self.generate_feedback(character, self.ideas)
self.feedback.append({
"character": character.name,
"feedback": feedback
})
print(f"{character.name}: {feedback}")2. 辩论模式
class DebateSession:
"""辩论会话"""
def __init__(self, proposition: str,
pro_side: AICharacter,
con_side: AICharacter,
judge: AICharacter):
self.proposition = proposition
self.pro_side = pro_side # 正方
self.con_side = con_side # 反方
self.judge = judge # 评判
self.rounds = []
def conduct_debate(self, max_rounds: int = 5) -> Dict:
"""进行辩论"""
print(f"\n=== 辩题：{self.proposition} ===\n")
for round_num in range(max_rounds):
print(f"\n=== 第 {round_num + 1} 轮 ===")
# 正方发言
pro_argument = self.generate_argument(
self.pro_side,
self.proposition,
self.get_opponent_arguments(self.con_side),
"支持"
)
print(f"\n【正方】{self.pro_side.name}：")
print(pro_argument)
# 反方发言
con_argument = self.generate_argument(
self.con_side,
self.proposition,
self.get_opponent_arguments(self.pro_side),
"反对"
)
print(f"\n【反方】{self.con_side.name}：")
print(con_argument)
# 记录轮次
self.rounds.append({
"round": round_num + 1,
"pro_argument": pro_argument,
"con_argument": con_argument
})
# 检查是否达成共识
if self.check_consensus():
break
# 评判
return self.judge_debate()
def judge_debate(self) -> Dict:
"""评判辩论"""
debate_history = "\n".join([
f"第{r['round']}轮\n正方：{r['pro_argument']}\n反方：{r['con_argument']}"
for r in self.rounds
])
prompt = f"""
作为评判，请对以下辩论进行评判：
辩题：{self.proposition}
辩论过程：
{debate_history}
请从以下角度评判：
1. 论点质量（满分10分）
2. 论据充分性（满分10分）
3. 逻辑清晰度（满分10分）
4. 反驳能力（满分10分）
最终判断：支持 / 反对 / 平局
理由：
"""
judgment = self.judge_llm.generate(prompt)
return {
"judgment": judgment,
"winner": self.determine_winner()
}3. 共识协商模式
class ConsensusSession:
"""共识协商会话"""
def __init__(self, characters: List[AICharacter]):
self.characters = characters
self.proposals = []
self.votes = {}
def reach_consensus(self, topic: str, max_iterations: int = 10) -> Dict:
"""达成共识"""
print(f"\n=== 协商主题：{topic} ===\n")
shared_knowledge = {}
for iteration in range(max_iterations):
print(f"\n=== 第 {iteration + 1} 轮协商 ===")
# 1. 各方提案
proposals = self.collect_proposals(topic, shared_knowledge)
# 2. 投票
votes = self.vote_on_proposals(proposals)
# 3. 检查共识
consensus = self.check_consensus(votes)
if consensus:
print(f"\n=== 达成共识！ ===")
return {
"consensus": consensus,
"iterations": iteration + 1
}
# 4. 共享观点
shared_knowledge = self.share_knowledge(proposals, votes)
# 5. 修正提案
self.revise_proposals(shared_knowledge)
return {
"status": "未达成共识",
"final_proposals": proposals
}
def check_consensus(self, votes: Dict[str, List[str]]) -> str:
"""检查是否达成共识"""
threshold = 0.8 # 80%同意
for proposal, votes_list in votes.items():
agree_count = sum(1 for v in votes_list if "同意" in v)
agree_ratio = agree_count / len(votes_list)
if agree_ratio >= threshold:
return proposal
return None群聊架构
1. 系统架构
【群聊系统架构】
┌─────────────────────────────────────┐
│ 用户界面层 │
│ - Web群聊界面 │
│ - 实时消息推送 │
│ - 角色状态展示 │
└─────────────────┬───────────────┘
│
┌─────────────────┴───────────────┐
│ 群聊管理层 │
│ - 会话管理器 │
│ - 消息路由器 │
│ - 轮次管理器 │
│ - 上下文维护器 │
└─────────────────┬───────────────┘
│
┌─────────────────┴───────────────┐
│ 角色管理层 │
│ - 角色注册中心 │
│ - 角色状态管理 │
│ - 关系管理器 │
└─────────────────┬───────────────┘
│
┌─────────────────┴───────────────┐
│ AI代理层 │
│ ┌──────┬──────┬──────┐ │
│ │Agent 1│Agent 2│Agent 3│ ... │
│ └──────┴──────┴──────┘ │
└─────────────────┬───────────────┘
│
┌─────────────────┴───────────────┐
│ 基础服务层 │
│ - LLM服务 │
│ - 向量数据库 │
│ - 消息队列 │
│ - 时序数据库 │
└───────────────────────────────────┘2. 核心组件
class ChatRoomManager:
"""群聊房间管理器"""
def __init__(self):
self.rooms = {} # conversation_id -> ChatRoom
def create_room(self, room_id: str,
characters: List[AICharacter],
mode: str = "free") -> ChatRoom:
"""创建群聊房间"""
room = ChatRoom(
room_id=room_id,
characters=characters,
mode=mode
)
self.rooms[room_id] = room
room.start()
return room
def get_room(self, room_id: str) -> ChatRoom:
"""获取房间"""
return self.rooms.get(room_id)
def close_room(self, room_id: str):
"""关闭房间"""
room = self.get_room(room_id)
if room:
room.stop()
del self.rooms[room_id]
class ChatRoom:
"""群聊房间"""
def __init__(self, room_id: str,
characters: List[AICharacter],
mode: str):
self.room_id = room_id
self.characters = {c.id: c for c in characters}
self.mode = mode
# 管理器
self.turn_manager = TurnManager(list(self.characters.values()))
self.context = ConversationContext(
conversation_id=room_id,
characters=characters,
messages=[],
started_at=datetime.now()
)
# 消息处理
self.message_handlers = {
MessageType.TEXT: self.handle_text_message,
MessageType.ACTION: self.handle_action_message,
MessageType.EMOTION: self.handle_emotion_message
}
self.running = False
def start(self):
"""启动群聊"""
self.running = True
print(f"群聊房间 {self.room_id} 已启动")
print(f"参与者：{', '.join(c.name for c in self.characters.values())}")
def stop(self):
"""停止群聊"""
self.running = False
self.context.ended_at = datetime.now()
print(f"群聊房间 {self.room_id} 已停止")
def add_message(self, message: Message):
"""添加消息"""
self.context.messages.append(message)
self.context.turn_count += 1
self.context.current_speaker = message.character_id
# 更新上下文
self.update_context(message)
# 触发下一步
if self.running:
self.next_turn()
def next_turn(self):
"""下一轮对话"""
# 检查是否结束
if self.should_end():
self.stop()
return
# 获取下一个发言者
next_speaker = self.turn_manager.get_next_speaker(self.context)
# 生成回复
response = self.generate_response(next_speaker)
# 添加消息
message = Message(
id=self.generate_message_id(),
conversation_id=self.room_id,
character_id=next_speaker.id,
character_name=next_speaker.name,
content=response["content"],
message_type=MessageType.TEXT,
metadata=response.get("metadata", {}),
timestamp=datetime.now()
)
self.add_message(message)
# 通知客户端
self.notify_clients(message)工程实现
1. 基于LangChain实现
from langchain.chat_models import ChatOpenAI
from langchain.schema import HumanMessage, AIMessage, SystemMessage
class MultiAgentChat:
"""基于LangChain的多Agent群聊"""
def __init__(self, characters: List[AICharacter]):
self.characters = {c.id: c for c in characters}
self.llm = ChatOpenAI(temperature=0.8)
self.messages = []
def build_character_prompt(self, character: AICharacter) -> str:
"""构建角色提示"""
return f"""
你是{character.name}，{character.role}。
背景介绍：
{character.background}
性格特点：
- 说话风格：{character.speaking_style}
- 语言水平：{character.language_level}
- 情感表达：{character.emotion_level}
目标：
{', '.join(character.goals)}
约束：
{', '.join(character.constraints)}
请保持角色一致性，用第一人称发言。
"""
def generate_message(self, character_id: str, context: str) -> str:
"""生成消息"""
character = self.characters[character_id]
# 构建系统提示
system_prompt = self.build.build_character_prompt(character)
# 构建对话历史
conversation_history = "\n".join([
f"{msg['name']}: {msg['content']}"
for msg in self.messages
])
# 完整提示
full_prompt = f"""
{system_prompt}
当前对话：
{conversation_history}
{character.name}，请继续对话：
"""
response = self.llm.predict(full_prompt)
# 记录消息
self.messages.append({
"name": character.name,
"content": response
})
return response
def run_chat(self, initial_message: str, max_turns: int = 10):
"""运行群聊"""
character_ids = list(self.characters.keys())
current_speaker = character_ids[0]
print(f"\n=== 群聊开始 ===")
print(f"参与者：{', '.join(c.name for c in self.characters.values())}\n")
# 初始消息
self.messages.append({
"name": "User",
"content": initial_message
})
print(f"User: {initial_message}\n")
# 对话轮次
for turn in range(max_turns):
# 轮换发言者
current_speaker = character_ids[turn % len(character_ids)]
character = self.characters[current_speaker]
# 生成回复
response = self.generate_message(current_speaker, "")
print(f"{character.name}: {response}\n")
print("\n=== 对话结束 ===")2. 基于Memory实现持久记忆
from langchain.memory import ConversationBufferMemory, ConversationSummaryMemory
from langchain.chains import ConversationChain
class PersistentCharacterAgent:
"""持久记忆的角色Agent"""
def __init__(self, character: AICharacter, memory_backend):
self.character = character
self.memory_backend = memory_backend
self.llm = ChatOpenAI(temperature=0.8)
# 加载历史记忆
self.memory = self.load_memory()
def load_memory(self) -> ConversationBufferMemory:
"""加载记忆"""
# 从后端加载历史对话
history = self.memory_backend.get_conversation_history(
self.character.id
)
memory = ConversationBufferMemory()
for msg in history:
if msg["role"] == "user":
memory.add_user_message(msg["content"])
else:
memory.add_ai_message(msg["content"])
return memory
def save_memory(self):
"""保存记忆"""
# 获取对话摘要
summary = self.memory.load_memory_variables()
# 保存到后端
self.memory_backend.save_conversation(
self.character.id,
summary
)
def chat(self, message: str) -> str:
"""对话"""
# 构建对话链
chain = ConversationChain(
llm=self.llm,
memory=self.memory,
prompt=self.build_prompt()
)
# 获取回复
response = chain.predict(input=message)
# 保存记忆
self.save_memory()
return response实战案例
1. 产品开发讨论
# 定义角色
pm = AICharacter(
id="pm",
name="张产品",
role="产品经理",
background="10年产品经验，擅长需求分析",
speaking_style="专业、逻辑清晰",
goals=["理解需求", "制定方案"]
)
tech_lead = AICharacter(
id="tech",
name="王技术",
role="技术负责人",
background="12年技术经验，架构专家",
speaking_style="技术性强、严谨",
goals=["评估可行性", "设计架构"]
)
designer = AICharacter(
id="design",
name="李设计",
role="UI/UX设计师",
background="8年设计经验，用户体验专家",
speaking_style="注重体验、表达生动",
goals=["优化体验", "美化界面"]
)
qa = AICharacter(
id="qa",
name="赵测试",
role="测试工程师",
background="6年测试经验，质量保障专家",
speaking_style="细致、关注质量",
goals=["设计测试", "保障质量"]
)
# 创建群聊
chat = MultiAgentChat([pm, tech_lead, designer, qa])
# 开始讨论
chat.run_chat(
initial_message="我们需要开发一个智能客服系统，各位有什么建议？",
max_turns=20
)2. 案例研讨
# 案例背景
case_study = """
某电商平台在大促期间出现订单处理延迟，
导致大量用户投诉。请分析原因并提出解决方案。
"""
# 定义研讨角色
performance_analyst = AICharacter(
id="perf",
name="性能专家",
role="系统性能分析师",
background="15年性能优化经验",
speaking_style="数据驱动、技术深入"
)
architect = AICharacter(
id="arch",
name="架构师",
role="系统架构师",
background="12年架构设计经验",
speaking_style="架构视角、全局思维"
)
ops = AICharacter(
id="ops",
name="运维专家",
role="运维工程师",
background="10年运维经验",
speaking_style="实战角度、关注细节"
)
business = AICharacter(
id="biz",
name="业务负责人",
role="业务分析师",
background="8年业务分析经验",
speaking_style="业务视角、用户导向"
)
# 创建研讨群聊
seminar = BrainstormingSession([performance_analyst, architect, ops, business])
# 进行研讨
seminar.generate_ideas(
topic=case_study,
round_count=4
)面试高频问法
Q1: 如何设计AI角色？
【标准回答】
角色设计要素：
1. 基础信息
- 名称、头像、身份
- 职责、背景
2. 性格特征
- 大五人格模型
- 说话风格
- 情感程度
3. 能力与约束
- 技能熟练度
- 可用工具
- 行为约束
4. 目标与动机
- 明确目标
- 隐性动机
5. 记忆与知识
- 个人记忆
- 领域知识
设计原则：
- 角色一致性
- 行为可预测
- 能力合理Q2: 如何实现角色互动？
【标准回答】
角色互动实现：
1. 上下文共享
- 共享对话历史
- 共享知识库
- 共享决策
2. 角色关系
- 定义角色间关系
- 关系强度量化
- 互动频率控制
3. 轮次管理
- 确定发言顺序
- 平衡发言机会
- 避免独占话语
4. 反馈机制
- 角色间互相评价
- 根据反馈调整
- 持续优化Q3: 如何控制对话流程？
【标准回答】
对话流程控制：
1. 轮次控制
- 轮询发言
- 随机发言
- 相关性发言
2. 话题引导
- 话题识别
- 话题切换
- 话题回归
3. 共识机制
- 提案投票
- 意见综合
- 决策记录
4. 结束条件
- 目标达成
- 时间限制
- 轮次限制记忆要点
【AI群聊】
多角色互动
自然对话
协作解决问题
【角色设计】
基础信息：名称、身份、职责
性格特征：大五人格、说话风格
能力约束：技能、工具、目标
记忆知识：个人记忆、领域知识
【协作模式】
头脑风暴：创意生成
辩论：正反对立
共识协商：达成一致
【对话管理】
消息模型：内容、类型、时间
上下文：历史、话题、状态
轮次管理：发言顺序、轮次记录
【架构】
用户界面：Web、实时推送
群聊管理：会话、路由、轮次
角色管理：注册、状态、关系
AI代理：多个角色Agent
基础服务：LLM、向量库、消息队列文档版本: 1.0
