多轮对话管理实现

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