Day 8: 多輪對話與上下文管理

歡迎來到第二週的學習！經過第一週的基礎建立，今天我們將深入探討 AI 助理的核心能力之一：多輪對話與上下文管理。這是讓 AI 助理真正「智能」的關鍵技術，能夠讓對話更自然、更有連貫性。

🎯 為什麼需要上下文管理？

想像一下這樣的對話：

使用者：我想學習 Python
AI：好的！Python 是一門優秀的程式語言...
使用者：它難學嗎？
AI：什麼難學嗎？我需要更多資訊才能回答。

缺乏上下文管理的 AI 無法理解「它」指的是 Python，這樣的對話體驗很糟糕。今天我們要解決這個問題！

🧠 上下文管理的核心挑戰

1. 記憶容量限制

• 大多數 LLM 都有 token 限制
• 需要智能地選擇保留哪些上下文

2. 相關性判斷

• 不是所有歷史對話都與當前問題相關
• 需要篩選出最重要的上下文資訊

3. 對話主題轉換

• 使用者可能突然改變話題
• 需要識別主題轉換並適應

4. 長期記憶 vs 短期記憶

• 短期：當前對話的上下文
• 長期：使用者偏好、歷史互動模式

🏗 上下文管理系統架構

讓我們建立一個完整的上下文管理系統：

context_management/
├── main.py                     # 主程式
├── core/
│   ├── __init__.py
│   ├── context_manager.py      # 核心上下文管理
│   ├── memory_store.py         # 記憶儲存
│   └── relevance_scorer.py     # 相關性評分
├── strategies/
│   ├── __init__.py
│   ├── sliding_window.py       # 滑動視窗策略
│   ├── summarization.py        # 總結壓縮策略
│   └── topic_based.py          # 主題導向策略
├── integrations/
│   ├── __init__.py
│   ├── gemini_context.py       # Gemini 上下文整合
│   └── langgraph_context.py    # LangGraph 上下文整合
└── utils/
    ├── __init__.py
    ├── tokenizer.py            # Token 計算
    └── serializer.py           # 序列化工具

🔧 核心程式碼實作

1. 上下文資料結構 (core/context_manager.py)

from typing import List, Dict, Optional, Any, Tuple
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import json
import hashlib

class MessageType(Enum):
    USER = "user"
    ASSISTANT = "assistant"
    SYSTEM = "system"
    TOOL = "tool"

class ContextScope(Enum):
    IMMEDIATE = "immediate"      # 最近 3-5 輪對話
    SHORT_TERM = "short_term"   # 當前會話
    LONG_TERM = "long_term"     # 跨會話記憶

@dataclass
class ConversationMessage:
    role: MessageType
    content: str
    timestamp: datetime
    metadata: Dict[str, Any] = field(default_factory=dict)
    tokens: int = 0
    relevance_score: float = 1.0
    topic_tags: List[str] = field(default_factory=list)
    
    def __post_init__(self):
        if self.tokens == 0:
            self.tokens = self._estimate_tokens()
    
    def _estimate_tokens(self) -> int:
        """粗略估算 token 數量"""
        return len(self.content.split()) * 1.3  # 經驗值
    
    def to_dict(self) -> Dict:
        return {
            "role": self.role.value,
            "content": self.content,
            "timestamp": self.timestamp.isoformat(),
            "metadata": self.metadata,
            "tokens": self.tokens,
            "relevance_score": self.relevance_score,
            "topic_tags": self.topic_tags
        }

@dataclass
class ContextWindow:
    messages: List[ConversationMessage]
    max_tokens: int = 4000
    current_topic: str = ""
    topic_confidence: float = 0.0
    
    @property
    def total_tokens(self) -> int:
        return sum(msg.tokens for msg in self.messages)
    
    @property
    def message_count(self) -> int:
        return len(self.messages)

class AdvancedContextManager:
    def __init__(self, max_tokens: int = 4000, max_messages: int = 50):
        self.max_tokens = max_tokens
        self.max_messages = max_messages
        self.current_context = ContextWindow([], max_tokens)
        self.conversation_history: List[ConversationMessage] = []
        self.topic_history: List[Tuple[str, datetime, float]] = []
        self.user_profile: Dict[str, Any] = {}
        
    def add_message(self, role: MessageType, content: str, metadata: Dict = None) -> None:
        """添加新訊息到上下文"""
        message = ConversationMessage(
            role=role,
            content=content,
            timestamp=datetime.now(),
            metadata=metadata or {}
        )
        
        # 分析主題和相關性
        self._analyze_message(message)
        
        # 添加到歷史記錄
        self.conversation_history.append(message)
        
        # 更新當前上下文窗口
        self._update_context_window(message)
    
    def _analyze_message(self, message: ConversationMessage) -> None:
        """分析訊息的主題和相關性"""
        # 這裡可以使用 Gemini 進行更複雜的分析
        # 現在使用簡化版本
        content = message.content.lower()
        
        # 簡單的主題識別
        topics = []
        topic_keywords = {
            'python': ['python', 'py', '程式', '程式碼', 'code'],
            'ai': ['ai', '人工智慧', '機器學習', 'ml', 'llm'],
            'web': ['網站', 'html', 'css', 'javascript', 'web'],
            'data': ['資料', '數據', 'data', '分析', 'analysis']
        }
        
        for topic, keywords in topic_keywords.items():
            if any(keyword in content for keyword in keywords):
                topics.append(topic)
        
        message.topic_tags = topics
        
        # 計算與當前主題的相關性
        if self.current_context.current_topic:
            if self.current_context.current_topic in topics:
                message.relevance_score = 1.0
            else:
                message.relevance_score = 0.5
    
    def _update_context_window(self, new_message: ConversationMessage) -> None:
        """更新上下文窗口"""
        self.current_context.messages.append(new_message)
        
        # 如果超過限制，執行壓縮策略
        if (self.current_context.total_tokens > self.max_tokens or 
            self.current_context.message_count > self.max_messages):
            self._compress_context()
    
    def _compress_context(self) -> None:
        """壓縮上下文 - 使用混合策略"""
        # 策略1: 保留最重要的訊息
        important_messages = [
            msg for msg in self.current_context.messages 
            if msg.relevance_score > 0.8 or msg.role == MessageType.SYSTEM
        ]
        
        # 策略2: 保留最近的對話
        recent_messages = self.current_context.messages[-10:]  # 保留最近10條
        
        # 策略3: 總結中間部分
        if len(self.current_context.messages) > 20:
            middle_part = self.current_context.messages[5:-10]
            summary = self._summarize_messages(middle_part)
            if summary:
                summary_message = ConversationMessage(
                    role=MessageType.SYSTEM,
                    content=f"[對話摘要] {summary}",
                    timestamp=datetime.now(),
                    metadata={"is_summary": True}
                )
                important_messages.append(summary_message)
        
        # 合併並去重
        compressed_messages = []
        seen_content = set()
        
        for msg in important_messages + recent_messages:
            content_hash = hashlib.md5(msg.content.encode()).hexdigest()
            if content_hash not in seen_content:
                compressed_messages.append(msg)
                seen_content.add(content_hash)
        
        # 按時間排序
        compressed_messages.sort(key=lambda x: x.timestamp)
        
        self.current_context.messages = compressed_messages
    
    def _summarize_messages(self, messages: List[ConversationMessage]) -> str:
        """總結一組訊息"""
        if not messages:
            return ""
        
        # 這裡可以使用 Gemini 進行總結
        # 現在使用簡化版本
        topics = set()
        for msg in messages:
            topics.update(msg.topic_tags)
        
        return f"討論了關於 {', '.join(topics)} 的話題，共 {len(messages)} 輪對話。"
    
    def get_context_for_prompt(self, include_summary: bool = True) -> List[Dict]:
        """獲取用於提示詞的上下文"""
        context_messages = []
        
        # 添加使用者檔案摘要（如果有的話）
        if self.user_profile and include_summary:
            profile_summary = self._get_user_profile_summary()
            context_messages.append({
                "role": "system",
                "content": f"使用者背景資訊：{profile_summary}"
            })
        
        # 添加對話歷史
        for msg in self.current_context.messages:
            context_messages.append({
                "role": msg.role.value,
                "content": msg.content
            })
        
        return context_messages
    
    def _get_user_profile_summary(self) -> str:
        """獲取使用者檔案摘要"""
        if not self.user_profile:
            return "新使用者"
        
        summary_parts = []
        if "name" in self.user_profile:
            summary_parts.append(f"姓名：{self.user_profile['name']}")
        if "interests" in self.user_profile:
            summary_parts.append(f"興趣：{', '.join(self.user_profile['interests'])}")
        if "experience_level" in self.user_profile:
            summary_parts.append(f"經驗等級：{self.user_profile['experience_level']}")
        
        return "; ".join(summary_parts) if summary_parts else "一般使用者"
    
    def update_user_profile(self, updates: Dict[str, Any]) -> None:
        """更新使用者檔案"""
        self.user_profile.update(updates)
    
    def get_conversation_stats(self) -> Dict[str, Any]:
        """獲取對話統計資訊"""
        total_messages = len(self.conversation_history)
        total_tokens = sum(msg.tokens for msg in self.conversation_history)
        
        topic_counts = {}
        for msg in self.conversation_history:
            for topic in msg.topic_tags:
                topic_counts[topic] = topic_counts.get(topic, 0) + 1
        
        return {
            "total_messages": total_messages,
            "total_tokens": total_tokens,
            "current_context_size": self.current_context.message_count,
            "current_context_tokens": self.current_context.total_tokens,
            "top_topics": sorted(topic_counts.items(), key=lambda x: x[1], reverse=True)[:5],
            "conversation_duration": (
                self.conversation_history[-1].timestamp - self.conversation_history[0].timestamp
            ).total_seconds() / 60 if self.conversation_history else 0
        }

2. Gemini 上下文整合 (integrations/gemini_context.py)

import google.generativeai as genai
from core.context_manager import AdvancedContextManager, MessageType
from typing import List, Dict, Optional
import json

class GeminiContextualChat:
    def __init__(self, api_key: str, model_name: str = "gemini-pro"):
        genai.configure(api_key=api_key)
        self.model = genai.GenerativeModel(model_name)
        self.context_manager = AdvancedContextManager()
        self.system_prompt = """
        你是一個智能助理，具有以下特性：
        1. 能夠記住對話歷史並保持上下文一致性
        2. 會主動參考之前的對話內容
        3. 能識別話題轉換並適當回應
        4. 提供個性化的建議和回應
        """
    
    def chat(self, user_input: str, metadata: Dict = None) -> str:
        """進行上下文感知的對話"""
        # 記錄使用者輸入
        self.context_manager.add_message(
            MessageType.USER, 
            user_input, 
            metadata
        )
        
        # 構建上下文感知的提示詞
        context_messages = self.context_manager.get_context_for_prompt()
        
        # 建立完整的對話歷史
        full_prompt = self._build_contextual_prompt(context_messages, user_input)
        
        try:
            # 調用 Gemini API
            response = self.model.generate_content(full_prompt)
            assistant_response = response.text
            
            # 記錄助理回應
            self.context_manager.add_message(
                MessageType.ASSISTANT,
                assistant_response,
                {"confidence": getattr(response, 'confidence', 0.8)}
            )
            
            return assistant_response
            
        except Exception as e:
            error_response = f"抱歉，我遇到了一些技術問題：{str(e)}"
            self.context_manager.add_message(
                MessageType.ASSISTANT,
                error_response,
                {"error": True}
            )
            return error_response
    
    def _build_contextual_prompt(self, context_messages: List[Dict], current_input: str) -> str:
        """構建包含上下文的提示詞"""
        prompt_parts = [self.system_prompt]
        
        # 添加對話歷史
        if len(context_messages) > 1:  # 除了系統訊息還有其他內容
            prompt_parts.append("\n對話歷史：")
            for msg in context_messages[1:]:  # 跳過系統訊息
                role_name = {"user": "使用者", "assistant": "助理"}.get(msg["role"], msg["role"])
                prompt_parts.append(f"{role_name}：{msg['content']}")
        
        # 添加當前問題
        prompt_parts.append(f"\n當前問題：{current_input}")
        
        # 添加回應指引
        prompt_parts.append("""
        請基於以上對話歷史回應，注意：
        1. 保持與之前對話的一致性
        2. 適當引用之前提到的內容
        3. 如果話題發生轉換，請自然地過渡
        4. 提供個性化且有幫助的回應
        """)
        
        return "\n".join(prompt_parts)
    
    def analyze_context_relevance(self, query: str) -> Dict[str, float]:
        """分析查詢與當前上下文的相關性"""
        context_messages = self.context_manager.get_context_for_prompt()
        
        if len(context_messages) < 2:
            return {"relevance": 0.0, "topic_continuity": 0.0}
        
        # 使用 Gemini 分析相關性
        analysis_prompt = f"""
        分析以下查詢與對話上下文的相關性：
        
        對話上下文：
        {json.dumps(context_messages[-5:], ensure_ascii=False, indent=2)}
        
        新查詢：{query}
        
        請以 JSON 格式回應：
        {{
            "relevance": 0.0-1.0,  // 與上下文的相關性
            "topic_continuity": 0.0-1.0,  // 話題連續性
            "is_topic_change": true/false,  // 是否為話題轉換
            "suggested_context_weight": 0.0-1.0  // 建議的上下文權重
        }}
        """
        
        try:
            response = self.model.generate_content(analysis_prompt)
            return json.loads(response.text)
        except:
            return {
                "relevance": 0.5,
                "topic_continuity": 0.5,
                "is_topic_change": False,
                "suggested_context_weight": 0.5
            }
    
    def get_context_summary(self) -> str:
        """獲取上下文摘要"""
        stats = self.context_manager.get_conversation_stats()
        
        summary = f"""
        📊 對話統計：
        • 總訊息數：{stats['total_messages']}
        • 當前上下文：{stats['current_context_size']} 則訊息
        • Token 使用：{stats['current_context_tokens']}/{self.context_manager.max_tokens}
        • 對話時長：{stats['conversation_duration']:.1f} 分鐘
        """
        
        if stats['top_topics']:
            summary += f"\n• 主要話題：{', '.join([topic for topic, count in stats['top_topics']])}"
        
        return summary
    
    def reset_context(self, keep_user_profile: bool = True) -> None:
        """重置上下文"""
        user_profile = self.context_manager.user_profile.copy() if keep_user_profile else {}
        self.context_manager = AdvancedContextManager()
        if keep_user_profile:
            self.context_manager.user_profile = user_profile
    
    def export_conversation(self) -> List[Dict]:
        """匯出對話記錄"""
        return [msg.to_dict() for msg in self.context_manager.conversation_history]

3. LangGraph 上下文整合 (integrations/langgraph_context.py)

from langgraph.graph import StateGraph, END
from core.context_manager import AdvancedContextManager, MessageType
from typing import TypedDict, List, Dict, Any, Literal
from integrations.gemini_context import GeminiContextualChat

class ContextualWorkflowState(TypedDict):
    user_input: str
    context_analysis: Dict[str, Any]
    processing_mode: str  # "contextual", "fresh", "summarized"
    response: str
    context_updates: List[Dict]
    confidence: float

def analyze_context_needs(state: ContextualWorkflowState) -> ContextualWorkflowState:
    """分析上下文需求"""
    # 這裡使用 Gemini 分析是否需要上下文
    # 簡化實現
    user_input = state["user_input"].lower()
    
    context_keywords = ["它", "這個", "剛才", "之前", "繼續", "那個", "上面的"]
    needs_context = any(keyword in user_input for keyword in context_keywords)
    
    processing_mode = "contextual" if needs_context else "fresh"
    
    return {
        **state,
        "context_analysis": {
            "needs_context": needs_context,
            "confidence": 0.8 if needs_context else 0.3
        },
        "processing_mode": processing_mode
    }

def contextual_processing(state: ContextualWorkflowState) -> ContextualWorkflowState:
    """使用完整上下文處理"""
    # 這裡會使用 GeminiContextualChat 進行處理
    response = f"[基於上下文] 理解您指的是之前討論的內容。{state['user_input']}"
    
    return {
        **state,
        "response": response,
        "confidence": 0.9
    }

def fresh_processing(state: ContextualWorkflowState) -> ContextualWorkflowState:
    """獨立處理（不依賴上下文）"""
    response = f"[新對話] {state['user_input']}"
    
    return {
        **state,
        "response": response,
        "confidence": 0.7
    }

def route_processing_mode(state: ContextualWorkflowState) -> Literal["contextual", "fresh"]:
    """路由處理模式"""
    return state["processing_mode"]

def create_contextual_workflow():
    """建立上下文感知工作流程"""
    workflow = StateGraph(ContextualWorkflowState)
    
    # 添加節點
    workflow.add_node("analyze", analyze_context_needs)
    workflow.add_node("contextual", contextual_processing)
    workflow.add_node("fresh", fresh_processing)
    
    # 設定流程
    workflow.set_entry_point("analyze")
    
    workflow.add_conditional_edges(
        "analyze",
        route_processing_mode,
        {
            "contextual": "contextual",
            "fresh": "fresh"
        }
    )
    
    workflow.add_edge("contextual", END)
    workflow.add_edge("fresh", END)
    
    return workflow.compile()

4. 完整示例應用 (main.py)

import os
from dotenv import load_dotenv
from integrations.gemini_context import GeminiContextualChat

load_dotenv()

def main():
    """上下文感知聊天機器人示例"""
    print("🧠 智能上下文管理聊天機器人")
    print("💡 我能記住我們的對話並保持上下文一致性")
    print("📝 嘗試說：'它是什麼？'、'繼續說明'、'剛才提到的那個'")
    print("=" * 60)
    
    # 初始化聊天機器人
    chat_bot = GeminiContextualChat(os.getenv('GEMINI_API_KEY'))
    
    # 設定使用者檔案
    chat_bot.context_manager.update_user_profile({
        "name": "學習者",
        "interests": ["程式設計", "AI"],
        "experience_level": "中等"
    })
    
    print("🤖 助理：你好！我是你的智能助理，我會記住我們的對話內容。請問有什麼可以幫助你的？")
    
    while True:
        try:
            user_input = input("\n💬 你：").strip()
            
            if not user_input:
                continue
            
            if user_input.lower() in ['quit', 'exit', '退出']:
                print("\n📊 對話總結：")
                print(chat_bot.get_context_summary())
                print("\n👋 再見！")
                break
            
            # 特殊指令
            if user_input.startswith('/'):
                if user_input == '/stats':
                    print(chat_bot.get_context_summary())
                    continue
                elif user_input == '/reset':
                    chat_bot.reset_context()
                    print("🔄 對話上下文已重置")
                    continue
                elif user_input == '/export':
                    conversation = chat_bot.export_conversation()
                    print(f"📤 已匯出 {len(conversation)} 則訊息")
                    continue
            
            # 分析上下文相關性
            relevance = chat_bot.analyze_context_relevance(user_input)
            
            print(f"🔍 分析中... (相關性: {relevance.get('relevance', 0.5):.2f})")
            
            # 獲取回應
            response = chat_bot.chat(user_input)
            
            print(f"🤖 助理：{response}")
            
            # 顯示上下文提示
            if relevance.get('is_topic_change', False):
                print("💡 提示：檢測到話題轉換")
            
        except KeyboardInterrupt:
            print(f"\n📊 對話總結：")
            print(chat_bot.get_context_summary())
            print("\n👋 再見！")
            break
        except Exception as e:
            print(f"\n❌ 發生錯誤：{e}")
            continue

if __name__ == "__main__":
    main()

🎮 使用示範

💬 你：我想學習 Python 程式設計
🤖 助理：太好了！Python 是一門非常適合初學者的程式語言...

💬 你：它難學嗎？
🤖 助理：根據剛才我們討論的 Python，它其實相對容易學習...

💬 你：推薦一些學習資源
🤖 助理：基於你想學習 Python 的需求，我推薦以下資源...

💬 你：/stats
📊 對話統計：
• 總訊息數：6
• 當前上下文：6 則訊息  
• Token 使用：1250/4000
• 對話時長：3.2 分鐘
• 主要話題：python, learning

🚀 進階優化技巧

1. 動態上下文權重

def calculate_context_weight(message_age: int, relevance: float) -> float:
    """計算上下文權重"""
    time_decay = 0.9 ** message_age  # 時間衰減
    return relevance * time_decay

2. 主題聚類

def cluster_messages_by_topic(messages: List[ConversationMessage]) -> Dict[str, List]:
    """按主題聚類訊息"""
    topic_clusters = {}
    for msg in messages:
        for topic in msg.topic_tags:
            if topic not in topic_clusters:
                topic_clusters[topic] = []
            topic_clusters[topic].append(msg)
    return topic_clusters

3. 智能摘要策略

def smart_summarization(messages: List[ConversationMessage]) -> str:
    """智能摘要策略"""
    # 按重要性和相關性選擇關鍵訊息
    important_messages = [msg for msg in messages if msg.relevance_score > 0.7]
    return generate_summary(important_messages)

🎯 今日總結

今天我們建立了一個完整的上下文管理系統，讓 AI 助理能夠：

✅ 記憶對話歷史：保存和管理完整的對話記錄
✅ 智能上下文選擇：根據相關性篩選重要上下文
✅ 動態壓縮策略：在記憶容量限制下保留最重要的資訊
✅ 個性化體驗：記住使用者偏好和背景資訊
✅ 話題追蹤：識別和處理話題轉換

明天我們將進一步探索記憶功能的進階應用，包括長期記憶、使用者建模和個性化推薦系統！

💡 延伸思考

1. 如何處理多使用者的上下文隔離？
2. 怎樣設計更智能的記憶遺忘機制？
3. 如何在分散式系統中共享上下文？
4. 上下文管理對回應品質的影響如何量化？

今天的上下文管理系統為我們的 AI 助理增添了真正的「記憶」能力，明天我們將讓這個記憶變得更加智能和個性化！