多轮对话管理实现
多轮对话的核心挑战 多轮对话不是简单的消息拼接——它需要管理对话状态、控制上下文窗口长度、处理话题切换、维护一致性。一个好的对话管理系统是LLM助手的"大脑"。 对话状态管理 from enum import Enum from dataclasses import dataclass, field class DialogState(Enum): GREETING = "greeting" INFORMATION_SEEKING = "information_seeking" PROBLEM_SOLVING = "problem_solving" CLARIFICATION = "clarification" CLOSING = "closing" @dataclass class ConversationContext: session_id: str user_id: str state: DialogState = DialogState.GREETING topic: str = "" entities: dict = field(default_factory=dict) # 提取的实体 history: list = field(default_factory=list) # 消息历史 user_preferences: dict = field(default_factory=dict) pending_clarification: str = "" class DialogManager: def __init__(self, llm): self.llm = llm self.sessions = {} # session_id -> ConversationContext def get_or_create_session(self, session_id, user_id): if session_id not in self.sessions: self.sessions[session_id] = ConversationContext( session_id=session_id, user_id=user_id ) return self.sessions[session_id] async def process_message(self, session_id, user_id, message): ctx = self.get_or_create_session(session_id, user_id) # 1. 状态检测 ctx.state = await self.detect_state(message, ctx) # 2. 实体提取 new_entities = await self.extract_entities(message) ctx.entities.update(new_entities) # 3. 构建上下文 context = self.build_context(ctx, message) # 4. 生成回复 response = await self.llm.generate(context) # 5. 更新历史 ctx.history.append({"role": "user", "content": message}) ctx.history.append({"role": "assistant", "content": response}) # 6. 上下文窗口管理 self.manage_window(ctx) return response 上下文窗口管理 class ContextWindowManager: def __init__(self, max_tokens=4096, reserve_tokens=1024): self.max_tokens = max_tokens self.reserve_tokens = reserve_tokens # 为生成预留 def manage(self, context): """管理上下文窗口大小""" available = self.max_tokens - self.reserve_tokens current_tokens = self.estimate_tokens(context.history) if current_tokens <= available: return # 在限制内，无需处理 # 策略1：摘要旧消息 while current_tokens > available and len(context.history) > 4: # 取最早的2条消息进行摘要 old_msgs = context.history[:2] summary = self.summarize(old_msgs) # 替换为摘要 context.history = [ {"role": "system", "content": f"[早期对话摘要]: {summary}"} ] + context.history[2:] current_tokens = self.estimate_tokens(context.history) # 策略2：如果仍然过长，截断 if current_tokens > available: # 保留system prompt和最近的消息 while current_tokens > available and len(context.history) > 2: context.history.pop(0 if context.history[0]["role"] != "system" else 1) current_tokens = self.estimate_tokens(context.history) def estimate_tokens(self, messages): return sum(len(m["content"]) // 4 for m in messages) def summarize(self, messages): """摘要旧消息""" text = "\n".join(f"{m['role']}: {m['content']}" for m in messages) # 使用LLM摘要 return f"[{len(messages)}条消息的摘要]" 话题管理 class TopicManager: def __init__(self, llm): self.llm = llm async def detect_topic_shift(self, current_message, history): """检测话题是否切换""" if len(history) < 2: return False, None recent_topic = await self.extract_topic(history[-2:]) current_topic = await self.extract_topic([{"content": current_message}]) similarity = await self.compute_similarity(recent_topic, current_topic) if similarity < 0.3: return True, current_topic # 话题切换 return False, recent_topic async def handle_topic_shift(self, ctx, new_topic): """处理话题切换""" # 保存当前话题的摘要 old_summary = await self.summarize_topic(ctx) ctx.topic_summaries = ctx.get("topic_summaries", []) ctx.topic_summaries.append({"topic": ctx.topic, "summary": old_summary}) # 切换到新话题 ctx.topic = new_topic ctx.state = DialogState.INFORMATION_SEEKING 意图跟踪 class IntentTracker: def __init__(self, llm): self.llm = llm self.intent_history = [] async def track(self, message, context): """跟踪用户意图""" prompt = f"""分析用户意图。 用户消息：{message} 对话历史摘要：{self.get_history_summary()} 当前状态：{context.state.value} 输出JSON： {{ "intent": "具体意图", "confidence": 0.0-1.0, "requires_clarification": true/false, "clarification_question": "如需澄清的问题" }}""" result = await self.llm.generate(prompt) intent = json.loads(result) self.intent_history.append(intent) return intent 会话持久化 import redis.asyncio as redis class SessionPersistence: def __init__(self, redis_url="redis://localhost:6379"): self.redis = redis.from_url(redis_url) async def save_session(self, session_id, context): """持久化会话""" await self.redis.setex( f"session:{session_id}", 86400, # 24小时过期 json.dumps({ "session_id": context.session_id, "user_id": context.user_id, "state": context.state.value, "topic": context.topic, "entities": context.entities, "history": context.history[-20:], # 只存最近20条 "user_preferences": context.user_preferences, }, ensure_ascii=False) ) async def load_session(self, session_id): """加载会话""" data = await self.redis.get(f"session:{session_id}") if data: d = json.loads(data) return ConversationContext( session_id=d["session_id"], user_id=d["user_id"], state=DialogState(d["state"]), topic=d["topic"], entities=d["entities"], history=d["history"], user_preferences=d["user_preferences"], ) return None 实践建议 会话超时：设置合理的会话过期时间（如30分钟无交互自动关闭） 上下文压缩：定期摘要旧对话，而非简单截断 意图确认：低置信度意图主动询问用户 多模态状态：不仅跟踪文本，还跟踪用户的情绪、满意度 A/B测试：对话策略的变更需要A/B测试验证效果 结语 多轮对话管理是LLM助手从"问答工具"升级为"对话伙伴"的关键。状态管理、上下文窗口控制、话题跟踪和会话持久化的协同工作，让Agent能够维持连贯、智能的多轮对话。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...
