# 2-3 上下文管理與持久化技術 回到白皮書首頁:[MCP 全方位技術白皮書](/@thc1006/mcp-whitepaper-home) --- ## 突破 AI 記憶限制:MCP 的智慧化上下文管理 如果說傳統 AI 就像患有「健忘症」的專家,每次對話都要重新開始,那麼 MCP 的上下文管理就是給 AI 裝上了「永久記憶」。這不只是技術升級,而是 AI 從「對話工具」邁向「工作夥伴」的關鍵一步。 ## 上下文管理的核心挑戰 ### 傳統 AI 的記憶困境 **Token 限制問題:** ``` Claude-3: 200K tokens ≈ 150,000 字 ChatGPT-4: 128K tokens ≈ 100,000 字 Gemini: 2M tokens ≈ 1,500,000 字 但複雜的企業對話往往需要: - 歷史決策記錄 - 專案背景資料 - 技術規格文件 - 工作流程狀態 = 輕易超過任何模型的上下文限制 ``` **狀態丟失問題:** ``` 對話前段:「我們決定使用 PostgreSQL 作為主資料庫」 對話中段:(討論API設計、前端開發...) 對話後段:「資料庫用什麼?」 AI回答:「建議考慮 MySQL 或 PostgreSQL...」 結果:AI 忘記了自己的決定! ``` ## MCP 的革命性解決方案 ### 智慧化上下文分層管理 MCP 建立了一個**三層記憶架構**,模擬人類的記憶模式: ``` 即時記憶層 (Session Memory) ├─ 當前對話內容 ├─ 近期操作記錄 └─ 暫時工作狀態 工作記憶層 (Working Memory) ├─ 專案相關上下文 ├─ 任務執行狀態 └─ 決策歷程記錄 長期記憶層 (Persistent Memory) ├─ 歷史對話摘要 ├─ 學習模式記錄 └─ 知識庫累積 ``` ### 會話管理機制 **會話生命週期:** ```python class MCPSessionManager: def __init__(self): self.sessions = {} self.context_store = PersistentContextStore() async def create_session(self, user_id: str, project_id: str = None): """建立新的工作會話""" session_id = str(uuid.uuid4()) session = MCPSession( id=session_id, user_id=user_id, project_id=project_id, created_at=datetime.now(), context_window=ContextWindow(max_tokens=150000), persistent_memory=await self._load_persistent_context(user_id, project_id), working_memory=WorkingMemory() ) self.sessions[session_id] = session return session async def restore_session(self, session_id: str): """恢復之前的會話狀態""" if session_id in self.sessions: return self.sessions[session_id] # 從持久化儲存恢復 session_data = await self.context_store.load_session(session_id) if session_data: session = MCPSession.from_dict(session_data) self.sessions[session_id] = session return session return None ``` ### 智慧化上下文壓縮 **重要性評分機制:** ```python class ContextImportanceEvaluator: def __init__(self): self.importance_factors = { 'decision_made': 1.0, 'error_encountered': 0.9, 'requirement_defined': 0.8, 'progress_milestone': 0.7, 'general_discussion': 0.3 } def evaluate_context_importance(self, context_item: dict) -> float: """評估上下文項目的重要性""" importance = 0.0 # 基於內容類型評分 content_type = context_item.get('type', 'general_discussion') importance += self.importance_factors.get(content_type, 0.3) # 基於引用頻率評分 reference_count = context_item.get('reference_count', 0) importance += min(reference_count * 0.1, 0.5) # 基於時間衰減 age_days = (datetime.now() - context_item['created_at']).days time_decay = max(0.1, 1.0 - (age_days * 0.02)) importance *= time_decay return min(importance, 1.0) async def compress_context(self, context_items: List[dict], target_tokens: int): """智慧化壓縮上下文到目標大小""" # 評估所有項目的重要性 scored_items = [] for item in context_items: score = self.evaluate_context_importance(item) scored_items.append((score, item)) # 按重要性排序 scored_items.sort(key=lambda x: x[0], reverse=True) # 選擇最重要的項目直到達到目標大小 selected_items = [] current_tokens = 0 for score, item in scored_items: item_tokens = self._estimate_tokens(item['content']) if current_tokens + item_tokens <= target_tokens: selected_items.append(item) current_tokens += item_tokens else: break return selected_items ``` ## 實際應用:MCP Memory Keeper ### 專案背景與功能 **MCP Memory Keeper** 是一個專門為 Claude Code 設計的上下文持久化服務,解決了 AI 編程助手的記憶問題。 **核心功能:** - **跨會話記憶**:保存工作歷程、決策和進度 - **專案上下文**:自動關聯 Git 分支和專案目錄 - **智慧化頻道**:基於主題的上下文組織 - **檔案變更追蹤**:監控重要檔案的修改 ### 技術實現 **資料模型設計:** ```sql -- 會話管理表 CREATE TABLE sessions ( id TEXT PRIMARY KEY, name TEXT NOT NULL, description TEXT, project_dir TEXT, default_channel TEXT, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, continued_from TEXT, FOREIGN KEY (continued_from) REFERENCES sessions(id) ); -- 上下文項目表 CREATE TABLE context_items ( id TEXT PRIMARY KEY, session_id TEXT NOT NULL, channel TEXT, key TEXT NOT NULL, value TEXT NOT NULL, category TEXT DEFAULT 'general', priority TEXT DEFAULT 'normal', created_at DATETIME DEFAULT CURRENT_TIMESTAMP, reference_count INTEGER DEFAULT 0, FOREIGN KEY (session_id) REFERENCES sessions(id) ); -- 檔案快取表 CREATE TABLE file_cache ( id TEXT PRIMARY KEY, session_id TEXT NOT NULL, file_path TEXT NOT NULL, content_hash TEXT NOT NULL, cached_at DATETIME DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (session_id) REFERENCES sessions(id) ); ``` **API 介面設計:** ```typescript interface MCPMemoryKeeper { // 會話管理 sessionStart(params: { name: string; description?: string; projectDir?: string; defaultChannel?: string; continueFrom?: string; }): Promise<SessionInfo>; sessionList(params: { limit?: number }): Promise<SessionInfo[]>; // 上下文操作 contextSave(params: { key: string; value: string; category?: 'task' | 'decision' | 'progress' | 'note'; priority?: 'high' | 'normal' | 'low'; channel?: string; }): Promise<void>; contextGet(params: { key?: string; category?: string; priority?: string; channel?: string; limit?: number; }): Promise<ContextItem[]>; // 檔案管理 cacheFile(params: { filePath: string; content: string; }): Promise<void>; fileChanged(params: { filePath: string; currentContent: string; }): Promise<boolean>; // 狀態查詢 status(): Promise<SessionStatus>; } ``` ### 使用案例 **開發工作流程:** ```typescript // 1. 開始新的開發會話 await mcp_context_session_start({ name: 'User Authentication Feature', description: 'Implementing OAuth 2.0 authentication system', projectDir: '/home/dev/myapp', defaultChannel: 'auth-feature' }); // 2. 記錄重要決策 await mcp_context_save({ key: 'auth_strategy', value: 'Using JWT tokens with 15-minute expiry and refresh tokens', category: 'decision', priority: 'high' }); // 3. 追蹤進度 await mcp_context_save({ key: 'current_progress', value: 'Completed user model, working on authentication middleware', category: 'progress', priority: 'normal' }); // 4. 快取重要檔案 await mcp_context_cache_file({ filePath: 'src/models/user.ts', content: userModelContent }); // 5. 會話恢復後取得上下文 const decisions = await mcp_context_get({ category: 'decision', priority: 'high' }); const progress = await mcp_context_get({ category: 'progress' }); ``` ## 企業級持久化策略 ### 分散式上下文架構 **多層快取設計:** ```python class EnterpriseContextManager: def __init__(self): self.l1_cache = InMemoryCache(ttl=300) # 5分鐘記憶體快取 self.l2_cache = RedisCache(ttl=3600) # 1小時 Redis 快取 self.l3_storage = PostgreSQLStorage() # 永久資料庫儲存 self.backup_storage = S3BackupStorage() # 雲端備份 async def save_context(self, session_id: str, context: dict): """多層儲存上下文""" # L1: 記憶體快取(最快存取) await self.l1_cache.set(f"ctx:{session_id}", context) # L2: Redis 快取(跨服務共享) await self.l2_cache.set(f"ctx:{session_id}", context) # L3: 資料庫儲存(持久化) await self.l3_storage.save_context(session_id, context) # 定期備份到雲端 if self._should_backup(context): await self.backup_storage.backup_context(session_id, context) async def load_context(self, session_id: str) -> dict: """智慧化載入上下文""" # 嘗試從最快的儲存層開始 context = await self.l1_cache.get(f"ctx:{session_id}") if context: return context context = await self.l2_cache.get(f"ctx:{session_id}") if context: # 回填到 L1 await self.l1_cache.set(f"ctx:{session_id}", context) return context context = await self.l3_storage.load_context(session_id) if context: # 回填到快取層 await self.l2_cache.set(f"ctx:{session_id}", context) await self.l1_cache.set(f"ctx:{session_id}", context) return context return {} ``` ### 智慧化上下文同步 **跨設備同步機制:** ```python class ContextSynchronizer: def __init__(self): self.sync_queue = asyncio.Queue() self.conflict_resolver = ConflictResolver() async def sync_context_across_devices(self, user_id: str): """跨設備同步上下文""" devices = await self.get_user_devices(user_id) for device in devices: local_context = await self.get_device_context(device.id) remote_context = await self.get_remote_context(user_id) if self._has_conflicts(local_context, remote_context): resolved_context = await self.conflict_resolver.resolve( local_context, remote_context ) else: resolved_context = self._merge_contexts( local_context, remote_context ) await self.update_device_context(device.id, resolved_context) await self.update_remote_context(user_id, resolved_context) ``` ## 台灣企業應用案例 ### 案例一:金融業客服系統 **背景:**某大型銀行需要 AI 客服能記住客戶的完整服務歷程。 **實現:** ```python class BankingContextManager: async def handle_customer_interaction(self, customer_id: str, message: str): # 載入客戶歷史上下文 customer_context = await self.load_customer_context(customer_id) # 包含: # - 過去 30 天的服務記錄 # - 產品使用狀況 # - 投訴和建議歷史 # - 個人化偏好設定 # 處理當前請求 response = await self.ai_service.process_with_context( message, customer_context ) # 更新上下文 await self.update_customer_context( customer_id, { 'latest_interaction': message, 'ai_response': response, 'satisfaction_score': await self.detect_satisfaction(response), 'timestamp': datetime.now() } ) return response ``` ### 案例二:製造業設備維護 **背景:**台積電等半導體廠需要 AI 記住每台設備的完整維護歷程。 **實現:** ```python class EquipmentMaintenanceContext: async def analyze_equipment_issue(self, equipment_id: str, sensor_data: dict): # 載入設備上下文 equipment_context = await self.load_equipment_history(equipment_id) # 包含: # - 歷史故障模式 # - 維護記錄 # - 效能趨勢 # - 相似設備經驗 # AI 分析 analysis = await self.ai_diagnostic.analyze_with_context( sensor_data, equipment_context ) # 更新設備學習記錄 await self.update_equipment_context( equipment_id, { 'latest_analysis': analysis, 'sensor_snapshot': sensor_data, 'maintenance_action': analysis.recommended_action, 'timestamp': datetime.now() } ) return analysis ``` ## 效能最佳化策略 ### 智慧化預載入 ```python class ContextPreloader: def __init__(self): self.usage_predictor = UsagePatternPredictor() async def predictive_preload(self, user_id: str): """基於使用模式預載入上下文""" # 分析使用者模式 patterns = await self.usage_predictor.analyze_user_patterns(user_id) # 預測可能需要的上下文 likely_contexts = await self.predict_needed_contexts(patterns) # 預載入到快取 for context_key in likely_contexts: await self.preload_to_cache(context_key) ``` ### 動態壓縮演算法 ```python class AdaptiveContextCompressor: def __init__(self): self.compression_strategies = { 'high_importance': SummaryCompressor(), 'medium_importance': KeyPointExtractor(), 'low_importance': TimestampOnlyCompressor() } async def adaptive_compress(self, context_items: List[dict], target_size: int): """動態選擇壓縮策略""" compressed_items = [] remaining_size = target_size # 按重要性分組 grouped_items = self._group_by_importance(context_items) for importance_level, items in grouped_items.items(): compressor = self.compression_strategies[importance_level] if remaining_size > 0: compressed = await compressor.compress(items, remaining_size) compressed_items.extend(compressed) remaining_size -= self._calculate_size(compressed) return compressed_items ``` ## 安全性與隱私保護 ### 上下文加密機制 ```python class SecureContextStorage: def __init__(self, encryption_key: bytes): self.cipher = Fernet(encryption_key) async def encrypt_context(self, context: dict) -> bytes: """加密上下文資料""" serialized = json.dumps(context).encode('utf-8') encrypted = self.cipher.encrypt(serialized) return encrypted async def decrypt_context(self, encrypted_data: bytes) -> dict: """解密上下文資料""" decrypted = self.cipher.decrypt(encrypted_data) context = json.loads(decrypted.decode('utf-8')) return context ``` ### 個人資料保護 ```python class PrivacyProtectedContext: def __init__(self): self.pii_detector = PIIDetector() self.anonymizer = DataAnonymizer() async def sanitize_context(self, context: dict) -> dict: """清理敏感資訊""" # 檢測個人敏感資訊 pii_items = await self.pii_detector.detect(context) # 匿名化處理 sanitized_context = await self.anonymizer.anonymize( context, pii_items ) return sanitized_context ``` ## 監控與分析 ### 上下文使用分析 ```python class ContextAnalytics: def __init__(self): self.metrics_collector = MetricsCollector() async def analyze_context_usage(self, session_id: str): """分析上下文使用效率""" metrics = { 'context_hit_rate': await self._calculate_hit_rate(session_id), 'compression_ratio': await self._calculate_compression_ratio(session_id), 'retrieval_latency': await self._calculate_retrieval_latency(session_id), 'storage_efficiency': await self._calculate_storage_efficiency(session_id) } return metrics ``` ## 小結:智慧化記憶的未來 MCP 的上下文管理與持久化技術不只是解決了 AI 的「健忘症」,更是為 AI 賦予了**學習和成長的能力**。 **核心價值:** 1. **連續性**:跨會話保持工作狀態和決策記錄 2. **智慧化性**:自動評估和管理上下文重要性 3. **可擴展性**:支援企業級的大規模部署 4. **安全性**:保護敏感資訊和個人隱私 **對台灣企業的意義:** - **提升效率**:AI 不用重複學習,直接延續之前的工作 - **降低成本**:減少重複說明和設定的時間成本 - **增強信任**:AI 能記住和遵循之前的決策 - **支援創新**:為複雜的長期專案提供 AI 支援 這項技術將 AI 從「一次性工具」升級為「長期夥伴」,這正是企業數位轉型所需要的 AI 能力。 --- **下一頁:** [2-4 安全機制與企業治理](/s/mcp-security-governance)