多轮对话 Prompt 设计：保持上下文连贯的秘诀

┌─────────────────────────────────────────┐
│ System Prompt           ~500 tokens     │
│ 对话摘要（注入）        ~300 tokens     │
│ 用户画像/偏好           ~200 tokens     │
│ 历史对话（裁剪后）      ~2000 tokens    │
│ 当前用户输入            ~200 tokens     │
│ ──────────────────────────────────────  │
│ 总计                    ~3200 tokens    │
│ 模型上下文窗口          8K~128K tokens  │
│ 预留给输出              ~1000 tokens    │
└─────────────────────────────────────────┘

class ContextWindowManager:
    def __init__(self, model_context_size=8192, output_reserve=1024):
        self.total = model_context_size
        self.output = output_reserve
        self.input_budget = self.total - self.output
    
    def allocate(self, components):
        """分配 Token 预算"""
        budget = self.input_budget
        allocation = {}
        
        # 优先级分配
        priorities = [
            ("system", 0.15),      # 系统提示固定 15%
            ("summary", 0.10),     # 摘要 10%
            ("user_profile", 0.05), # 用户画像 5%
            ("history", 0.50),     # 历史对话 50%
            ("current", 0.20),     # 当前输入 20%
        ]
        
        for name, ratio in priorities:
            allocation[name] = min(
                components.get(name, ""),
                key=lambda x: self._truncate(x, int(budget * ratio))
            )
        
        return allocation
    
    def _truncate(self, text, max_tokens):
        """截断到指定 token 数"""
        tokens = tokenizer.encode(text)
        if len(tokens) <= max_tokens:
            return text
        return tokenizer.decode(tokens[:max_tokens])

class SummarizationTrigger:
    def __init__(self, threshold_ratio=0.7):
        self.threshold_ratio = threshold_ratio  # 上下文使用率达到70%时触发
    
    def should_summarize(self, messages, max_context):
        current_tokens = count_tokens(messages)
        return current_tokens > max_context * self.threshold_ratio
    
    def summarize_strategy(self, messages):
        """分层摘要策略"""
        total_tokens = count_tokens(messages)
        
        if total_tokens < 2000:
            return None  # 不需要摘要
        
        # 保留最近 2 轮，摘要其余
        recent = messages[-4:]  # 最近2轮（user+assistant各1）
        to_summarize = messages[:-4]
        
        return {
            "to_summarize": to_summarize,
            "keep_recent": recent,
            "target_summary_tokens": min(300, total_tokens // 4)
        }

SUMMARY_PROMPT = """请将以下对话浓缩为关键信息摘要。

要求：
1. 保留用户的核心需求和意图
2. 保留已确定的关键事实（人名、数值、决策等）
3. 保留未解决的问题
4. 用简洁的条目式表达
5. 省略寒暄和重复内容

对话内容：
{conversation}

输出格式：
- 用户意图：...
- 已知事实：...
- 待解决问题：...
- 上下文要点：...
"""

def generate_summary(conversation, llm):
    prompt = SUMMARY_PROMPT.format(
        conversation=format_conversation(conversation)
    )
    return llm.generate(prompt, temperature=0.0, max_tokens=300)

class IncrementalSummarizer:
    """增量更新摘要，避免每次全量重算"""
    
    def __init__(self, llm):
        self.llm = llm
        self.current_summary = ""
        self.summarized_until = 0  # 已摘要到的消息索引
    
    def update(self, messages):
        """增量更新摘要"""
        # 只处理新增的消息
        new_messages = messages[self.summarized_until:]
        if len(new_messages) < 4:  # 少于2轮不更新
            return self.current_summary
        
        # 保留最近2轮不摘要
        to_summarize = new_messages[:-4]
        recent = new_messages[-4:]
        
        if not to_summarize:
            return self.current_summary
        
        # 增量摘要
        if self.current_summary:
            prompt = f"""
            现有摘要：{self.current_summary}
            
            新增对话：{format_conversation(to_summarize)}
            
            请将新增对话信息整合到现有摘要中。保持简洁。
            """
        else:
            prompt = SUMMARY_PROMPT.format(
                conversation=format_conversation(to_summarize)
            )
        
        self.current_summary = self.llm.generate(prompt, max_tokens=300)
        self.summarized_until = len(messages) - len(recent)
        
        return self.current_summary

from dataclasses import dataclass, field
from typing import Dict, List, Optional
from enum import Enum

class ConversationState(Enum):
    GREETING = "greeting"
    GATHERING_INFO = "gathering_info"
    PROCESSING = "processing"
    AWAITING_CONFIRMATION = "awaiting_confirmation"
    COMPLETED = "completed"
    ERROR_RECOVERY = "error_recovery"

@dataclass
class DialogueState:
    state: ConversationState = ConversationState.GREETING
    topic: str = ""
    collected_info: Dict[str, str] = field(default_factory=dict)
    pending_questions: List[str] = field(default_factory=list)
    user_preferences: Dict[str, str] = field(default_factory=dict)
    turn_count: int = 0
    last_intent: str = ""
    
    def to_prompt_context(self):
        """将状态转为 Prompt 上下文"""
        return f"""
[对话状态]
当前阶段: {self.state.value}
话题: {self.topic}
已收集信息: {json.dumps(self.collected_info, ensure_ascii=False)}
待确认问题: {self.pending_questions}
用户偏好: {json.dumps(self.user_preferences, ensure_ascii=False)}
对话轮次: {self.turn_count}
"""

class StateTracker:
    def __init__(self, llm):
        self.llm = llm
        self.state = DialogueState()
    
    def update(self, user_input, assistant_output):
        """更新对话状态"""
        self.state.turn_count += 1
        
        # 意图检测
        intent = self.detect_intent(user_input)
        self.state.last_intent = intent
        
        # 状态转移
        transitions = {
            (ConversationState.GREETING, "ask_question"): ConversationState.GATHERING_INFO,
            (ConversationState.GATHERING_INFO, "provide_info"): ConversationState.GATHERING_INFO,
            (ConversationState.GATHERING_INFO, "request_action"): ConversationState.PROCESSING,
            (ConversationState.PROCESSING, "confirm"): ConversationState.AWAITING_CONFIRMATION,
            (ConversationState.AWAITING_CONFIRMATION, "yes"): ConversationState.COMPLETED,
            (ConversationState.AWAITING_CONFIRMATION, "no"): ConversationState.GATHERING_INFO,
            (ConversationState.COMPLETED, "new_topic"): ConversationState.GREETING,
        }
        
        new_state = transitions.get((self.state.state, intent))
        if new_state:
            self.state.state = new_state
        
        # 信息提取
        self.extract_info(user_input, assistant_output)
        
        # 话题切换检测
        if self.is_topic_change(user_input):
            self.state.collected_info.clear()
            self.state.topic = self.detect_topic(user_input)
    
    def detect_intent(self, text):
        prompt = f"""
        判断用户意图，从以下选项中选择：
        ask_question, provide_info, request_action, confirm, new_topic, other
        
        用户输入: {text}
        意图:
        """
        return self.llm.generate(prompt, temperature=0.0).strip()

class TopicDetector:
    def __init__(self, embedder, threshold=0.5):
        self.embedder = embedder
        self.threshold = threshold
        self.topic_history = []
    
    def detect(self, current_input):
        if not self.topic_history:
            self.topic_history.append(current_input)
            return False, None
        
        # 计算与最近话题的相似度
        current_emb = self.embedder.encode(current_input)
        recent_emb = self.embedder.encode(self.topic_history[-1])
        similarity = cosine_similarity(current_emb, recent_emb)[0][0]
        
        if similarity < self.threshold:
            # 话题切换
            self.topic_history.append(current_input)
            return True, self.extract_topic(current_input)
        
        return False, None
    
    def extract_topic(self, text):
        """用 LLM 提取话题关键词"""
        prompt = f"用一个短语概括以下内容的主题：\n{text}"
        return llm.generate(prompt, max_tokens=20)

def handle_topic_switch(new_topic, conversation_manager):
    """话题切换时保存当前上下文，开始新上下文"""
    # 1. 保存当前话题的对话摘要
    old_summary = conversation_manager.summarize_current()
    conversation_manager.saved_topics.append({
        "topic": conversation_manager.state.topic,
        "summary": old_summary,
        "turn_count": conversation_manager.state.turn_count
    })
    
    # 2. 重置状态
    conversation_manager.state = DialogueState()
    conversation_manager.state.topic = new_topic
    
    # 3. 注入话题切换提示
    context = f"""
[话题切换] 用户从"{conversation_manager.saved_topics[-1]['topic']}"切换到"{new_topic}"
之前的话题已归档，如有需要可以引用。
"""
    return context

class MemorySystem:
    def __init__(self):
        self.short_term = []  # 最近对话（原文）
        self.working_memory = {}  # 当前任务状态
        self.long_term = []  # 跨会话记忆
    
    def inject(self, user_input):
        """构造记忆注入上下文"""
        context_parts = []
        
        # 1. 长期记忆检索（基于相似度）
        relevant_memories = self.retrieve_long_term(user_input, top_k=3)
        if relevant_memories:
            context_parts.append(
                "[长期记忆]\n" + 
                "\n".join(f"- {m}" for m in relevant_memories)
            )
        
        # 2. 工作记忆
        if self.working_memory:
            context_parts.append(
                "[工作记忆]\n" + 
                json.dumps(self.working_memory, ensure_ascii=False, indent=2)
            )
        
        # 3. 短期记忆（最近对话摘要）
        if self.short_term:
            context_parts.append(
                "[近期对话摘要]\n" + self.short_term[-1]
            )
        
        return "\n\n".join(context_parts)
    
    def retrieve_long_term(self, query, top_k=3):
        """从长期记忆中检索相关内容"""
        if not self.long_term:
            return []
        
        query_emb = embedder.encode(query)
        memory_embs = [embedder.encode(m["content"]) for m in self.long_term]
        
        scores = cosine_similarity(query_emb.reshape(1,-1), memory_embs)[0]
        top_indices = np.argsort(scores)[-top_k:][::-1]
        
        return [self.long_term[i]["content"] for i in top_indices 
                if scores[i] > 0.3]  # 相似度阈值

class LongConversationStrategy:
    """长对话的分段处理策略"""
    
    def __init__(self, max_turns_per_segment=10):
        self.max_turns = max_turns_per_segment
        self.segments = []
    
    def process_turn(self, user_input, assistant_output, turn_count):
        # 检查是否需要分段
        if turn_count > 0 and turn_count % self.max_turns == 0:
            # 生成段摘要
            segment = self.get_current_segment()
            summary = self.summarize_segment(segment)
            self.segments.append({
                "turn_range": f"{turn_count - self.max_turns + 1}-{turn_count}",
                "summary": summary
            })
            self.reset_current_segment()
        
        self.add_to_current_segment(user_input, assistant_output)
    
    def get_context(self):
        """获取分段上下文"""
        if not self.segments:
            return ""
        
        context = "[历史分段摘要]\n"
        for seg in self.segments[-3:]:  # 只保留最近3个分段
            context += f"轮次{seg['turn_range']}: {seg['summary']}\n"
        return context

import redis
import json
from datetime import datetime, timedelta

class RedisContextStore:
    """Redis 存储对话上下文"""
    
    def __init__(self, redis_url="redis://localhost:6379"):
        self.redis = redis.from_url(redis_url)
        self.ttl = 86400 * 7  # 7天过期
    
    def save_context(self, session_id, context):
        """保存对话上下文"""
        key = f"chat:context:{session_id}"
        self.redis.setex(
            key,
            self.ttl,
            json.dumps({
                "context": context,
                "updated_at": datetime.now().isoformat()
            }, ensure_ascii=False)
        )
    
    def load_context(self, session_id):
        """加载对话上下文"""
        key = f"chat:context:{session_id}"
        data = self.redis.get(key)
        if data:
            return json.loads(data)["context"]
        return None
    
    def save_summary(self, session_id, summary):
        """保存对话摘要"""
        key = f"chat:summary:{session_id}"
        self.redis.setex(key, self.ttl, summary)
    
    def append_message(self, session_id, role, content):
        """追加消息到 Redis List"""
        key = f"chat:messages:{session_id}"
        message = json.dumps({
            "role": role,
            "content": content,
            "timestamp": datetime.now().isoformat()
        }, ensure_ascii=False)
        self.redis.rpush(key, message)
        self.redis.expire(key, self.ttl)
        
        # 控制列表长度
        if self.redis.llen(key) > 100:  # 最多保留100条
            self.redis.ltrim(key, -80, -1)  # 保留最近80条

class MultiTurnConversationEngine:
    def __init__(self, llm, redis_store, system_prompt):
        self.llm = llm
        self.redis = redis_store
        self.system_prompt = system_prompt
        self.summarizer = IncrementalSummarizer(llm)
        self.state_tracker = StateTracker(llm)
        self.memory = MemorySystem()
    
    def chat(self, session_id, user_input):
        # 1. 加载历史上下文
        context = self.redis.load_context(session_id) or []
        
        # 2. 更新摘要
        summary = self.summarizer.update(context)
        
        # 3. 更新对话状态
        self.state_tracker.update(user_input, "")
        
        # 4. 记忆注入
        memory_context = self.memory.inject(user_input)
        
        # 5. 构造完整 Prompt
        messages = self.build_messages(
            system_prompt=self.system_prompt,
            summary=summary,
            state=self.state_tracker.state,
            memory=memory_context,
            history=context[-6:],  # 最近3轮
            user_input=user_input
        )
        
        # 6. 生成回复
        response = self.llm.chat(messages)
        
        # 7. 更新状态
        self.state_tracker.update(user_input, response)
        
        # 8. 保存上下文
        context.append({"role": "user", "content": user_input})
        context.append({"role": "assistant", "content": response})
        self.redis.save_context(session_id, context)
        
        return response
    
    def build_messages(self, **kwargs):
        messages = [{"role": "system", "content": kwargs["system_prompt"]}]
        
        if kwargs.get("summary"):
            messages.append({
                "role": "system",
                "content": f"[对话摘要]\n{kwargs['summary']}"
            })
        
        if kwargs.get("state"):
            messages.append({
                "role": "system",
                "content": kwargs["state"].to_prompt_context()
            })
        
        if kwargs.get("memory"):
            messages.append({
                "role": "system",
                "content": kwargs["memory"]
            })
        
        for msg in kwargs.get("history", []):
            messages.append(msg)
        
        messages.append({"role": "user", "content": kwargs["user_input"]})
        
        return messages