logo
Loading...

Langchain 詳細記憶方法整理 - AI Agent 開發特訓營：短期實現智能自動化 - Cupoy

LangChain RAG Agent 記憶體管理完整教學指南 課程相關code我都上完到github空間中：請 git clone https://github.com/kevin801221/Ag...

LangChain RAG Agent 記憶體管理完整教學指南 課程相關code我都上完到github空間中：請 git clone https://github.com/kevin801221/AgenticU-The-Modular-Teaching-Hub-for-Modern-LLM-Agent-Frameworks.git 去下載相關程式碼。（可以幫忙點個星星唷！) 目錄 前言與重要性 記憶體管理基礎概念 ConversationBufferMemory - 基礎完整記憶 ConversationBufferWindowMemory - 滑動視窗記憶 ConversationTokenBufferMemory - Token 限制記憶 ConversationSummaryMemory - 摘要記憶 ConversationSummaryBufferMemory - 混合摘要記憶 ConversationEntityMemory - 實體記憶 ConversationKGMemory - 知識圖譜記憶 VectorStoreRetrieverMemory - 向量檢索記憶 SQL 資料庫持久化記憶 實際應用場景與選擇建議 效能最佳化與注意事項 前言與重要性 在現代 AI 應用開發中，特別是構建能夠進行多輪對話的 RAG (Retrieval-Augmented Generation) 系統時，記憶體管理 是決定系統品質的關鍵因素之一。正如人類對話需要記住之前談論的內容才能保持連貫性，AI Agent 也需要某種形式的「記憶」來理解對話上下文，提供相關且一致的回應。 為什麼記憶體管理如此重要？ 上下文連續性：確保 AI 能理解代詞指涉、話題延續等語言現象 個人化體驗：記住使用者偏好、歷史需求，提供客製化服務 效率最佳化：避免重複收集已知資訊，提升對話效率 一致性維持：確保 AI 在整個對話過程中保持一致的立場和知識狀態 LangChain 框架提供了豐富的記憶體管理機制，每種機制都針對不同的使用場景和技術限制進行了最佳化。本教學將深入探討這些機制的原理、實作和最佳實踐。 記憶體管理基礎概念 核心挑戰 在設計記憶體管理系統時，我們面臨幾個核心挑戰： Token 限制：大型語言模型都有上下文長度限制（如 GPT-4 的 8K 或 32K token） 計算成本：更長的上下文意味著更高的 API 調用成本 相關性：不是所有歷史對話都對當前問題有幫助 存儲效率：長期對話需要高效的存儲和檢索機制 LangChain 記憶體架構 LangChain 的記憶體系統基於以下核心組件： BaseMemory：所有記憶體類別的基礎抽象類別 ChatMessageHistory：負責實際的訊息存儲和檢索 Memory Variables：提供給 LLM 的上下文變數 # 基本記憶體介面 from abc import ABC, abstractmethod class BaseMemory(ABC): @abstractmethod def load_memory_variables(self, inputs: Dict[str, Any]) -> Dict[str, Any]: """從記憶體載入相關變數""" pass @abstractmethod def save_context(self, inputs: Dict[str, Any], outputs: Dict[str, str]) -> None: """將新的對話儲存到記憶體""" pass ConversationBufferMemory - 基礎完整記憶 概念說明 ConversationBufferMemory 是最直觀的記憶體實作，它將所有對話歷史完整地儲存在記憶體緩衝區中。這種方法確保了最完整的上下文保存，但可能導致 token 消耗過多的問題。 適用場景 短期對話：對話輪數較少（< 10 輪）的場景 開發測試：需要完整追蹤對話流程的開發階段 詳細分析：需要保留所有對話細節的應用 詳細實作 from langchain.memory import ConversationBufferMemory from langchain_openai import ChatOpenAI from langchain.chains import ConversationChain import logging # 設置日誌以追蹤記憶體使用情況 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # 初始化記憶體 memory = ConversationBufferMemory( return_messages=True, # 返回結構化的訊息對象 memory_key="history", # 在 prompt 中的變數名稱 human_prefix="Human", # 人類訊息的前綴 ai_prefix="Assistant", # AI 訊息的前綴 ) # 建立對話鏈 llm = ChatOpenAI( temperature=0, model_name="gpt-4o", max_tokens=1000 ) conversation = ConversationChain( llm=llm, memory=memory, verbose=True, # 顯示詳細的執行過程 ) # 模擬多輪對話 def simulate_conversation(): conversations = [ "你好，我想遠端開立一個銀行帳戶。該如何開始？", "我是台灣居民，有台灣身分證", "我需要準備哪些文件？", "開戶後多久可以使用網路銀行？" ] for i, user_input in enumerate(conversations, 1): logger.info(f"=== 對話輪次 {i} ===") response = conversation.predict(input=user_input) print(f"用戶: {user_input}") print(f"助手: {response}\n") # 檢查記憶體狀態 memory_vars = memory.load_memory_variables({}) logger.info(f"記憶體中的對話數: {len(memory_vars['history'])}") # 執行對話模擬 simulate_conversation() # 詳細檢查記憶體內容 print("=== 完整記憶體內容 ===") memory_content = memory.load_memory_variables({}) for i, message in enumerate(memory_content["history"]): print(f"{i+1}. {message.type}: {message.content}") 進階配置選項 # 自定義記憶體配置 custom_memory = ConversationBufferMemory( return_messages=True, memory_key="chat_history", input_key="question", # 指定輸入鍵 output_key="answer", # 指定輸出鍵 human_prefix="客戶", # 自定義前綴 ai_prefix="客服專員", ) # 手動操作記憶體 custom_memory.save_context( inputs={"question": "請問營業時間？"}, outputs={"answer": "我們的營業時間是週一到週五 9:00-17:00"} ) # 清空記憶體 custom_memory.clear() 優缺點分析 優點： 實作簡單，概念直觀 保留完整對話脈絡 無資訊遺失風險 缺點： Token 消耗可能快速增長 長對話可能超出模型限制 計算成本隨對話長度線性增長 ConversationBufferWindowMemory - 滑動視窗記憶 概念說明 滑動視窗記憶是對基礎緩衝記憶的改進，它只保留最近的 k 輪對話。這種方法在保持相關上下文的同時，有效控制了記憶體大小和 token 消耗。 核心原理 採用 FIFO (First In, First Out) 策略： 當對話輪數 ≤ k 時，保留所有對話 當對話輪數 > k 時，移除最舊的對話 始終維持記憶體中有最新的 k 輪對話 詳細實作 from langchain.memory import ConversationBufferWindowMemory from langchain_openai import ChatOpenAI from langchain.chains import ConversationChain class WindowMemoryAnalyzer: def __init__(self, window_size=3): self.memory = ConversationBufferWindowMemory( k=window_size, # 視窗大小 return_messages=True, memory_key="history" ) self.llm = ChatOpenAI(temperature=0, model_name="gpt-4o") self.conversation = ConversationChain( llm=self.llm, memory=self.memory, verbose=False ) self.conversation_count = 0 def chat(self, message): """進行對話並分析記憶體狀態""" self.conversation_count += 1 print(f"\n=== 對話 {self.conversation_count} ===") print(f"輸入: {message}") response = self.conversation.predict(input=message) print(f"回應: {response}") # 分析記憶體狀態 self._analyze_memory_state() return response def _analyze_memory_state(self): """分析並顯示當前記憶體狀態""" memory_vars = self.memory.load_memory_variables({}) history = memory_vars.get("history", []) print(f"記憶體狀態: 保留 {len(history)} 則訊息") for i, msg in enumerate(history): prefix = "🔵" if msg.type == "human" else "🤖" print(f" {prefix} {msg.content[:50]}...") def demonstrate_window_effect(self): """示範滑動視窗效果""" test_messages = [ "我叫 Alice，今年 25 歲", "我是軟體工程師", "我住在台北", "我喜歡看電影", "我最近在學習 AI", "請告訴我我的基本資訊" # 測試記憶體保留情況 ] for msg in test_messages: self.chat(msg) print("\n=== 最終記憶體內容 ===") final_memory = self.memory.load_memory_variables({}) for msg in final_memory["history"]: print(f"{msg.type}: {msg.content}") # 執行示範 analyzer = WindowMemoryAnalyzer(window_size=3) analyzer.demonstrate_window_effect() 動態視窗大小調整 class AdaptiveWindowMemory(ConversationBufferWindowMemory): """自適應視窗大小的記憶體""" def __init__(self, base_window_size=3, max_window_size=10): super().__init__(k=base_window_size, return_messages=True) self.base_window_size = base_window_size self.max_window_size = max_window_size self.importance_scores = [] def save_context(self, inputs, outputs): # 計算對話重要性（簡化範例） importance = self._calculate_importance(inputs, outputs) self.importance_scores.append(importance) # 根據重要性調整視窗大小 if importance > 0.8: # 高重要性對話 self.k = min(self.k + 1, self.max_window_size) elif importance < 0.3: # 低重要性對話 self.k = max(self.k - 1, self.base_window_size) super().save_context(inputs, outputs) def _calculate_importance(self, inputs, outputs): """計算對話重要性（示範用的簡化版本）""" # 實際應用中可以使用更複雜的評分機制 text = str(inputs) + str(outputs) keywords = ["重要", "關鍵", "問題", "解決", "幫助"] score = sum(1 for keyword in keywords if keyword in text) return min(score / len(keywords), 1.0) 最佳實踐 # 針對不同場景的視窗大小建議 WINDOW_SIZE_RECOMMENDATIONS = { "客服對話": 5, # 需要記住近期問題和回答 "教學輔導": 7, # 需要跟蹤學習進度 "閒聊機器人": 3, # 輕量級對話 "技術支援": 10, # 可能需要參考較多技術細節 "醫療諮詢": 8, # 需要記住症狀和病史資訊 } def create_optimized_window_memory(use_case, custom_size=None): """根據使用場景創建最佳化的視窗記憶體""" window_size = custom_size or WINDOW_SIZE_RECOMMENDATIONS.get(use_case, 5) return ConversationBufferWindowMemory( k=window_size, return_messages=True, memory_key="history" ) ConversationTokenBufferMemory - Token 限制記憶 概念說明 Token 緩衝記憶體是最精確的記憶體管理方式，它直接根據 token 數量而非對話輪數來管理記憶體。這種方法特別適合需要精確控制 API 成本和模型輸入長度的應用。 技術原理 Token 計算：使用 tiktoken 庫精確計算文本的 token 數量 動態截斷：當總 token 數超過限制時，從最舊的對話開始移除 實時監控：每次保存新對話時都會檢查 token 限制 詳細實作 from langchain.memory import ConversationTokenBufferMemory from langchain_openai import ChatOpenAI import tiktoken from typing import List, Dict class TokenMemoryManager: """Token 記憶體管理器""" def __init__(self, max_token_limit=1000, model_name="gpt-4o"): self.llm = ChatOpenAI(model_name=model_name, temperature=0) self.memory = ConversationTokenBufferMemory( llm=self.llm, max_token_limit=max_token_limit, return_messages=True, memory_key="history" ) # 初始化 tokenizer self.encoding = tiktoken.encoding_for_model(model_name) self.max_token_limit = max_token_limit def add_conversation(self, human_message: str, ai_message: str): """添加對話並分析 token 使用情況""" print(f"\n=== 添加新對話 ===") print(f"人類: {human_message}") print(f"AI: {ai_message}") # 計算新對話的 token 數 human_tokens = len(self.encoding.encode(human_message)) ai_tokens = len(self.encoding.encode(ai_message)) new_tokens = human_tokens + ai_tokens print(f"新對話 Token 數: {new_tokens} (人類: {human_tokens}, AI: {ai_tokens})") # 保存對話前的記憶體狀態 before_memory = self.memory.load_memory_variables({}) before_count = len(before_memory.get("history", [])) before_tokens = self._calculate_total_tokens(before_memory.get("history", [])) # 保存新對話 self.memory.save_context( inputs={"input": human_message}, outputs={"output": ai_message} ) # 保存對話後的記憶體狀態 after_memory = self.memory.load_memory_variables({}) after_count = len(after_memory.get("history", [])) after_tokens = self._calculate_total_tokens(after_memory.get("history", [])) # 分析變化 self._analyze_memory_change( before_count, after_count, before_tokens, after_tokens, new_tokens ) def _calculate_total_tokens(self, messages: List) -> int: """計算訊息列表的總 token 數""" total = 0 for message in messages: total += len(self.encoding.encode(message.content)) return total def _analyze_memory_change(self, before_count, after_count, before_tokens, after_tokens, new_tokens): """分析記憶體變化""" print(f"\n📊 記憶體分析:") print(f" 對話數量: {before_count} → {after_count}") print(f" Token 使用: {before_tokens} → {after_tokens}") print(f" Token 限制: {self.max_token_limit}") print(f" 使用率: {after_tokens/self.max_token_limit*100:.1f}%") if after_count < before_count + 1: removed_count = (before_count + 1) - after_count print(f" ⚠️ 移除了 {removed_count} 則舊對話以符合 Token 限制") if after_tokens + new_tokens > self.max_token_limit: print(f" ✂️ 觸發 Token 限制，進行記憶體截斷") def demonstrate_token_management(): """示範 Token 記憶體管理""" # 創建較小的 token 限制以便觀察截斷效果 manager = TokenMemoryManager(max_token_limit=500) # 模擬長對話 conversations = [ ("什麼是機器學習？", "機器學習是人工智慧的一個子領域，它使電腦能夠在沒有被明確程式設計的情況下學習和改進。"), ("深度學習和機器學習有什麼區別？", "深度學習是機器學習的一個子集，使用多層神經網路來模擬人腦的運作方式，能夠處理更複雜的模式識別任務。"), ("請解釋什麼是神經網路？", "神經網路是由互相連接的節點（神經元）組成的計算模型，這些節點模擬生物神經系統的工作方式。"), ("什麼是卷積神經網路？", "卷積神經網路（CNN）是一種特殊的神經網路架構，特別適合處理圖像資料，使用卷積層來提取特徵。"), ("請解釋反向傳播演算法", "反向傳播是訓練神經網路的核心演算法，它通過計算梯度來調整網路權重，從而最小化預測誤差。"), ("什麼是過擬合？", "過擬合是指模型在訓練資料上表現很好，但在新資料上表現不佳的現象，通常是因為模型過於複雜。") ] for human_msg, ai_msg in conversations: manager.add_conversation(human_msg, ai_msg) # 顯示最終記憶體狀態 print("\n" + "="*50) print("最終記憶體內容:") final_memory = manager.memory.load_memory_variables({}) for i, message in enumerate(final_memory["history"]): print(f"{i+1}. {message.type}: {message.content}") # 執行示範 demonstrate_token_management() 進階 Token 管理策略 class AdvancedTokenMemory(ConversationTokenBufferMemory): """進階 Token 記憶體管理""" def __init__(self, llm, max_token_limit, reserve_ratio=0.2): super().__init__(llm=llm, max_token_limit=max_token_limit, return_messages=True) self.reserve_ratio = reserve_ratio # 為新對話保留的 token 比例 self.priority_keywords = ["重要", "關鍵", "緊急", "問題"] def save_context(self, inputs, outputs): """智能保存上下文""" # 計算新對話的 token 數 new_content = str(inputs) + str(outputs) new_tokens = len(self.llm.get_num_tokens(new_content)) # 確保為新對話保留足夠空間 reserved_tokens = int(self.max_token_limit * self.reserve_ratio) available_tokens = self.max_token_limit - reserved_tokens # 如果需要，進行智能截斷 if self._get_current_token_count() + new_tokens > available_tokens: self._smart_truncate(available_tokens - new_tokens) super().save_context(inputs, outputs) def _smart_truncate(self, target_tokens): """智能截斷：優先保留重要對話""" messages = self.chat_memory.messages scored_messages = [] for i, msg in enumerate(messages): importance_score = self._calculate_importance(msg.content) scored_messages.append((i, msg, importance_score)) # 按重要性排序（降序） scored_messages.sort(key=lambda x: x[2], reverse=True) # 保留最重要的對話直到達到目標 token 數 kept_messages = [] current_tokens = 0 for _, msg, _ in scored_messages: msg_tokens = len(self.llm.get_num_tokens(msg.content)) if current_tokens + msg_tokens <= target_tokens: kept_messages.append(msg) current_tokens += msg_tokens # 按原始順序重新排列 kept_messages.sort(key=lambda x: messages.index(x)) self.chat_memory.messages = kept_messages def _calculate_importance(self, content): """計算內容重要性""" score = 0 content_lower = content.lower() # 關鍵字權重 for keyword in self.priority_keywords: score += content_lower.count(keyword) * 2 # 長度權重（較長的內容可能包含更多資訊） score += min(len(content) / 100, 3) # 問句權重（問題通常比較重要） if "?" in content or "？" in content: score += 1 return score ConversationSummaryMemory - 摘要記憶 概念說明 摘要記憶體使用 LLM 將冗長的對話歷史壓縮成簡潔的摘要，這種方法在保留關鍵資訊的同時大幅減少 token 消耗。特別適合需要長期記憶但對成本敏感的應用。 技術原理 逐步摘要：每次新增對話時，將所有歷史對話重新摘要 資訊壓縮：使用 LLM 的理解能力提取對話要點 上下文保持：摘要保留對話的核心脈絡和關鍵資訊 詳細實作 from langchain.memory import ConversationSummaryMemory from langchain_openai import ChatOpenAI from langchain.schema import BaseMessage import time class SummaryMemoryAnalyzer: """摘要記憶體分析器""" def __init__(self, model_name="gpt-4o"): self.llm = ChatOpenAI( model_name=model_name, temperature=0, max_tokens=500 ) # 自定義摘要提示 summary_prompt = """ 請將以下對話歷史總結成簡潔但完整的摘要，保留所有重要資訊和關鍵細節： 對話歷史： {history} 摘要應該： 1. 保留所有重要的事實和數據 2. 維持對話的邏輯脈絡 3. 突出關鍵決定和結論 4. 使用繁體中文 摘要： """ self.memory = ConversationSummaryMemory( llm=self.llm, return_messages=False, # 返回摘要文本而非訊息對象 memory_key="history", prompt_template=summary_prompt ) self.conversation_count = 0 self.token_usage = [] def add_conversation_turn(self, human_input: str, ai_response: str): """添加對話輪次並分析摘要效果""" self.conversation_count += 1 print(f"\n{'='*60}") print(f"對話輪次 {self.conversation_count}") print(f"{'='*60}") # 記錄添加前的狀態 before_summary = self.memory.load_memory_variables({}).get("history", "") before_tokens = len(self.llm.get_num_tokens(before_summary)) if before_summary else 0 print(f"🗣️ 人類: {human_input}") print(f"🤖 AI: {ai_response}") # 測量摘要生成時間 start_time = time.time() # 保存對話（觸發摘要生成） self.memory.save_context( inputs={"input": human_input}, outputs={"output": ai_response} ) end_time = time.time() summary_time = end_time - start_time # 記錄添加後的狀態 after_summary = self.memory.load_memory_variables({}).get("history", "") after_tokens = len(self.llm.get_num_tokens(after_summary)) # 分析摘要效果 self._analyze_summary_effect( before_summary, after_summary, before_tokens, after_tokens, summary_time ) # 記錄 token 使用情況 self.token_usage.append({ "turn": self.conversation_count, "tokens": after_tokens, "generation_time": summary_time }) def _analyze_summary_effect(self, before_summary, after_summary, before_tokens, after_tokens, summary_time): """分析摘要效果""" print(f"\n📋 摘要分析:") print(f" 摘要生成時間: {summary_time:.2f} 秒") print(f" Token 變化: {before_tokens} → {after_tokens}") if before_summary: compression_ratio = len(before_summary) / len(after_summary) if after_summary else 0 print(f" 文本壓縮比: {compression_ratio:.2f}x") print(f"\n📝 當前摘要:") print(f" {after_summary}") def demonstrate_summary_memory(self): """示範摘要記憶體的運作""" # 模擬一個詳細的客戶服務對話 conversations = [ ("我想申請信用卡，我的年收入是100萬台幣", "好的，年收入100萬符合我們的申請條件。請問您目前的職業是什麼？"), ("我是軟體工程師，在台積電工作已經3年了", "台積電是很好的公司。請問您目前有其他銀行的信用卡嗎？"), ("我有中國信託和玉山銀行的信用卡，都使用正常，沒有遲繳紀錄", "很好，良好的信用記錄對申請很有幫助。您希望申請哪種類型的信用卡？"), ("我經常出國出差，希望申請一張適合海外消費的信用卡", "我推薦我們的全球通信用卡，海外消費有2%現金回饋，而且免收海外交易手續費。"), ("聽起來不錯，申請流程需要多久？需要準備什麼文件？", "申請流程約7-10個工作天。您需要準備身分證、收入證明、工作證明，以及最近三個月的薪資單。"), ("好的，我現在可以線上申請嗎？", "可以的，我現在就為您開啟線上申請流程。請確認您的聯絡電話是否為09XX-XXX-XXX？") ] for human_msg, ai_msg in conversations: self.add_conversation_turn(human_msg, ai_msg) # 生成最終分析報告 self._generate_analysis_report() def _generate_analysis_report(self): """生成分析報告""" print(f"\n{'='*60}") print("摘要記憶體效能分析報告") print(f"{'='*60}") print(f"總對話輪次: {self.conversation_count}") if self.token_usage: avg_tokens = sum(usage["tokens"] for usage in self.token_usage) / len(self.token_usage) max_tokens = max(usage["tokens"] for usage in self.token_usage) min_tokens = min(usage["tokens"] for usage in self.token_usage) print(f"Token 使用統計:") print(f" 平均: {avg_tokens:.1f} tokens") print(f" 最大: {max_tokens} tokens") print(f" 最小: {min_tokens} tokens") avg_time = sum(usage["generation_time"] for usage in self.token_usage) / len(self.token_usage) print(f"平均摘要生成時間: {avg_time:.2f} 秒") # 估算成本效益 estimated_cost_without_summary = self.conversation_count * 50 * 0.00003 # 假設每輪50 tokens current_avg_tokens = avg_tokens if self.token_usage else 0 estimated_cost_with_summary = self.conversation_count * current_avg_tokens * 0.00003 print(f"\n💰 成本分析:") print(f" 不使用摘要預估成本: ${estimated_cost_without_summary:.6f}") print(f" 使用摘要實際成本: ${estimated_cost_with_summary:.6f}") print(f" 節省比例: {(1 - estimated_cost_with_summary/estimated_cost_without_summary)*100:.1f}%") # 執行摘要記憶體示範 analyzer = SummaryMemoryAnalyzer() analyzer.demonstrate_summary_memory() 自定義摘要策略 from langchain.prompts import PromptTemplate class CustomSummaryMemory(ConversationSummaryMemory): """自定義摘要記憶體""" def __init__(self, llm, summary_type="detailed"): # 定義不同類型的摘要模板 summary_templates = { "detailed": """ 請將以下對話歷史詳細總結，保留所有重要資訊： {history} 總結要求： 1. 保留所有數字、日期、名稱等具體資訊 2. 維持事件的時間順序 3. 突出重要決定和行動項目 4. 保留情感色彩和態度 詳細總結： """, "concise": """ 請將以下對話歷史簡潔總結，只保留核心要點： {history} 簡潔總結（50字內）： """, "structured": """ 請將以下對話歷史結構化總結： {history} 結構化總結： 【參與者】： 【討論主題】： 【關鍵決定】： 【行動項目】： 【未解決問題】： """ } prompt_template = summary_templates.get(summary_type, summary_templates["detailed"]) prompt = PromptTemplate.from_template(prompt_template) super().__init__( llm=llm, prompt=prompt, return_messages=False ) # 使用示例 def compare_summary_types(): """比較不同摘要類型的效果""" llm = ChatOpenAI(model_name="gpt-4o", temperature=0) summary_types = ["detailed", "concise", "structured"] memories = {stype: CustomSummaryMemory(llm, stype) for stype in summary_types} # 測試對話 test_conversation = [ ("我需要在下週一前完成專案報告", "了解，下週一是12月15日。這個專案報告的主題是什麼？"), ("是關於AI在金融業的應用分析", "好的，這是個很有趣的主題。您需要包含哪些具體方面？"), ("我想分析機器學習在風險評估中的應用", "機器學習在風險評估確實很重要。建議您可以從信用評分、欺詐檢測等角度分析。") ] # 為每種摘要類型添加相同的對話 for stype, memory in memories.items(): print(f"\n=== {stype.upper()} 摘要類型 ===") for human_input, ai_response in test_conversation: memory.save_context( inputs={"input": human_input}, outputs={"output": ai_response} ) summary = memory.load_memory_variables({})["history"] print(f"摘要結果:\n{summary}") print(f"Token 數: {len(memory.llm.get_num_tokens(summary))}") compare_summary_types() ConversationSummaryBufferMemory - 混合摘要記憶 概念說明 混合摘要記憶體是最平衡的解決方案，它結合了完整記憶和摘要記憶的優點。系統會保留最近的完整對話（在 token 限制內），並將更早的對話內容進行摘要。 技術架構 from langchain.memory import ConversationSummaryBufferMemory from langchain_openai import ChatOpenAI import tiktoken from typing import List, Tuple class HybridMemorySystem: """混合記憶體系統""" def __init__(self, max_token_limit=800, model_name="gpt-4o"): self.llm = ChatOpenAI(model_name=model_name, temperature=0) self.memory = ConversationSummaryBufferMemory( llm=self.llm, max_token_limit=max_token_limit, return_messages=True, memory_key="history" ) self.encoding = tiktoken.encoding_for_model(model_name) self.max_token_limit = max_token_limit self.conversation_log = [] # 用於追蹤對話歷史 def process_conversation(self, human_input: str, ai_response: str): """處理對話並展示混合記憶體的運作""" conversation_id = len(self.conversation_log) + 1 self.conversation_log.append((human_input, ai_response)) print(f"\n{'='*70}") print(f"對話 {conversation_id}: 混合記憶體分析") print(f"{'='*70}") # 分析對話前的記憶體狀態 before_state = self._analyze_memory_state("處理前") # 保存新對話 self.memory.save_context( inputs={"input": human_input}, outputs={"output": ai_response} ) # 分析對話後的記憶體狀態 after_state = self._analyze_memory_state("處理後") # 比較變化 self._compare_memory_states(before_state, after_state, conversation_id) print(f"\n📄 當前記憶體內容:") self._display_current_memory() def _analyze_memory_state(self, stage: str) -> dict: """分析記憶體狀態""" memory_vars = self.memory.load_memory_variables({}) messages = memory_vars.get("history", []) # 分析摘要和完整訊息 summary_content = "" full_messages = [] for msg in messages: if hasattr(msg, 'type') and msg.type == "system": summary_content = msg.content else: full_messages.append(msg) # 計算 token 數 summary_tokens = len(self.encoding.encode(summary_content)) if summary_content else 0 full_message_tokens = sum( len(self.encoding.encode(msg.content)) for msg in full_messages ) total_tokens = summary_tokens + full_message_tokens return { "stage": stage, "summary_content": summary_content, "summary_tokens": summary_tokens, "full_messages": full_messages, "full_message_tokens": full_message_tokens, "total_tokens": total_tokens, "message_count": len(full_messages) } def _compare_memory_states(self, before: dict, after: dict, conversation_id: int): """比較記憶體狀態變化""" print(f"\n📊 記憶體狀態比較:") print(f" 摘要部分:") print(f" Token 數: {before['summary_tokens']} → {after['summary_tokens']}") if after['summary_tokens'] > before['summary_tokens']: print(f" 🔄 摘要已更新（+{after['summary_tokens'] - before['summary_tokens']} tokens）") print(f" 完整訊息部分:") print(f" 訊息數: {before['message_count']} → {after['message_count']}") print(f" Token 數: {before['full_message_tokens']} → {after['full_message_tokens']}") print(f" 總計:") print(f" Token 數: {before['total_tokens']} → {after['total_tokens']}") print(f" Token 限制: {self.max_token_limit}") print(f" 使用率: {after['total_tokens']/self.max_token_limit*100:.1f}%") # 檢查是否觸發摘要 if after['summary_tokens'] > before['summary_tokens']: print(f" ⚡ 觸發摘要機制：部分歷史對話被摘要") if after['message_count'] < len(self.conversation_log): summarized_count = len(self.conversation_log) - after['message_count'] print(f" 📝 {summarized_count} 輪對話已被摘要") def _display_current_memory(self): """顯示當前記憶體內容""" memory_vars = self.memory.load_memory_variables({}) messages = memory_vars.get("history", []) for i, msg in enumerate(messages): if hasattr(msg, 'type'): if msg.type == "system": print(f" 📋 [摘要] {msg.content}") elif msg.type == "human": print(f" 👤 [完整] {msg.content}") elif msg.type == "ai": print(f" 🤖 [完整] {msg.content}") def demonstrate_hybrid_memory(self): """示範混合記憶體的完整運作流程""" # 設計一個漸進複雜的對話場景 conversations = [ # 第1-3輪：建立基礎資訊（應該保持完整） ("我是張小明，想諮詢房屋貸款", "張小明您好！我很樂意協助您房屋貸款的諮詢。請問您看中的房屋價格大約是多少？"), ("房屋總價是1500萬，我有300萬頭期款", "好的，1500萬房屋，您有300萬頭期款，需要貸款1200萬。請問您目前的月收入是多少？"), ("我月收入12萬，太太月收入8萬", "您們夫妻合計月收入20萬，貸款1200萬是可行的。請問您目前的工作年資多久了？"), # 第4-6輪：深入細節（可能開始摘要） ("我在科技公司工作5年了，職位是資深工程師", "5年的穩定工作經驗很好。根據您的條件，可以申請到1.8%的優惠利率。"), ("1.8%的利率聽起來不錯，還款期限可以多長？", "最長可以貸款30年，以1200萬30年計算，月付金約4.2萬元。"), ("月付金4.2萬可以接受。還需要準備什麼文件？", "需要準備身分證、收入證明、工作證明、以及房屋相關文件。"), # 第7-9輪：最終確認（應該保持完整） ("文件我都可以準備，大概多久可以核貸？", "文件齊全後約7-10個工作天。我們也提供線上申請服務。"), ("太好了，我想現在就開始申請流程", "沒問題！我現在為您安排專員聯繫，開始正式申請流程。"), ("謝謝您的協助，請問專員什麼時候會聯繫我？", "專員會在今天下午3點前聯繫您，請保持電話暢通。") ] print("🎯 混合記憶體示範：房屋貸款諮詢對話") print("觀察摘要機制如何在token限制下運作...\n") for human_msg, ai_msg in conversations: self.process_conversation(human_msg, ai_msg) input("\n按 Enter 繼續下一輪對話...") # 最終狀態分析 self._generate_final_analysis() def _generate_final_analysis(self): """生成最終分析報告""" print(f"\n{'='*70}") print("混合記憶體最終分析報告") print(f"{'='*70}") final_state = self._analyze_memory_state("最終") print(f"📈 效能指標:") print(f" 總對話輪次: {len(self.conversation_log)}") print(f" 保留完整對話: {final_state['message_count']} 輪") print(f" 摘要對話: {len(self.conversation_log) - final_state['message_count']} 輪") print(f" 最終 Token 使用: {final_state['total_tokens']}/{self.max_token_limit}") print(f" 記憶體使用率: {final_state['total_tokens']/self.max_token_limit*100:.1f}%") # 計算記憶體效率 if len(self.conversation_log) > 0: retention_rate = final_state['message_count'] / len(self.conversation_log) print(f" 對話保留率: {retention_rate*100:.1f}%") print(f"\n🎯 混合記憶體優勢:") print(f" ✅ 保留最新對話的完整細節") print(f" ✅ 通過摘要維持長期記憶") print(f" ✅ 自動平衡記憶體使用") print(f" ✅ 適應不同對話長度") # 執行混合記憶體示範 hybrid_system = HybridMemorySystem(max_token_limit=600) hybrid_system.demonstrate_hybrid_memory() ConversationEntityMemory - 實體記憶 概念說明 實體記憶體專注於從對話中提取和管理關鍵實體（如人名、地名、組織等）及其相關資訊。這種記憶體特別適合需要追蹤多個實體狀態的複雜對話場景。 詳細實作 from langchain.memory import ConversationEntityMemory from langchain_openai import ChatOpenAI from langchain.schema import BaseMessage import json import re class EntityMemoryManager: """實體記憶體管理器""" def __init__(self, model_name="gpt-4o"): self.llm = ChatOpenAI(model_name=model_name, temperature=0) self.memory = ConversationEntityMemory( llm=self.llm, return_messages=True, memory_key="history" ) self.conversation_count = 0 self.entity_evolution = {} # 追蹤實體資訊的演化 def process_conversation(self, human_input: str, ai_response: str): """處理對話並分析實體提取""" self.conversation_count += 1 print(f"\n{'='*70}") print(f"對話 {self.conversation_count}: 實體記憶體分析") print(f"{'='*70}") print(f"👤 人類: {human_input}") print(f"🤖 AI: {ai_response}") # 記錄對話前的實體狀態 before_entities = self._get_current_entities() # 保存對話（觸發實體提取） self.memory.save_context( inputs={"input": human_input}, outputs={"output": ai_response} ) # 記錄對話後的實體狀態 after_entities = self._get_current_entities() # 分析實體變化 self._analyze_entity_changes(before_entities, after_entities) # 更新實體演化記錄 self._update_entity_evolution(after_entities) def _get_current_entities(self) -> dict: """獲取當前記憶體中的實體資訊""" # 通過 memory.entity_store 獲取實體資訊 return dict(self.memory.entity_store.store) def _analyze_entity_changes(self, before: dict, after: dict): """分析實體變化""" print(f"\n🔍 實體分析:") # 新增的實體 new_entities = set(after.keys()) - set(before.keys()) if new_entities: print(f" ➕ 新識別實體: {', '.join(new_entities)}") for entity in new_entities: print(f" 📝 {entity}: {after[entity]}") # 更新的實體 updated_entities = [] for entity in set(after.keys()) & set(before.keys()): if before[entity] != after[entity]: updated_entities.append(entity) if updated_entities: print(f" 🔄 更新的實體: {', '.join(updated_entities)}") for entity in updated_entities: print(f" 📝 {entity}:") print(f" 舊: {before[entity]}") print(f" 新: {after[entity]}") # 實體統計 print(f" 📊 實體統計:") print(f" 總實體數: {len(before)} → {len(after)}") print(f" 新增: {len(new_entities)}") print(f" 更新: {len(updated_entities)}") def _update_entity_evolution(self, current_entities: dict): """更新實體演化記錄""" for entity, info in current_entities.items(): if entity not in self.entity_evolution: self.entity_evolution[entity] = [] # 檢查是否有變化 if not self.entity_evolution[entity] or self.entity_evolution[entity][-1]["info"] != info: self.entity_evolution[entity].append({ "conversation": self.conversation_count, "info": info }) def query_entity(self, entity_name: str): """查詢特定實體的資訊""" print(f"\n🔍 查詢實體: {entity_name}") # 從記憶體查詢 memory_vars = self.memory.load_memory_variables({"input": f"告訴我關於{entity_name}的資訊"}) print(f"📋 記憶體中的相關資訊:") if "history" in memory_vars: for msg in memory_vars["history"]: if entity_name.lower() in msg.content.lower(): print(f" 💬 {msg.content}") # 顯示實體演化過程 if entity_name in self.entity_evolution: print(f"\n📈 {entity_name} 資訊演化:") for record in self.entity_evolution[entity_name]: print(f" 對話 {record['conversation']}: {record['info']}") def demonstrate_entity_memory(self): """示範實體記憶體的運作""" # 設計一個包含多個實體的複雜對話 conversations = [ ("我是王大明，在台積電工作", "王大明您好！很高興認識您。台積電是台灣的知名企業。"), ("我住在新竹市，已經在台積電工作3年了", "新竹市是科技重鎮，台積電的總部也在那裡。3年的工作經驗很不錯！"), ("我太太李小美是護理師，在新竹馬偕醫院上班", "護理師是很有意義的工作。新竹馬偕醫院在當地有很好的聲譽。"), ("我們有一個5歲的兒子叫王小華，在新竹市立幼兒園就讀", "5歲正是活潑好動的年紀！新竹市立幼兒園的教育品質如何？"), ("王小華很喜歡畫畫，李小美希望培養他的藝術天份", "培養孩子的興趣很重要。王小華有藝術天份，李小美的想法很好。"), ("台積電最近有升遷機會，我可能會調到台南廠區", "恭喜您有升遷機會！台南廠區是台積電的重要據點，不過這意味著要搬家嗎？"), ("是的，如果升遷成功，全家就要搬到台南。李小美也在找台南的醫院工作", "搬家是大事，不過台南的生活品質也很好。相信李小美能在台南找到合適的護理師工作。") ] print("🎯 實體記憶體示範：家庭搬遷決策對話") print("觀察系統如何提取和追蹤多個實體資訊...\n") for human_msg, ai_msg in conversations: self.process_conversation(human_msg, ai_msg) input("\n按 Enter 繼續...") # 實體查詢示範 print(f"\n{'='*70}") print("實體查詢示範") print(f"{'='*70}") entities_to_query = ["王大明", "李小美", "王小華", "台積電", "新竹市"] for entity in entities_to_query: self.query_entity(entity) input(f"\n按 Enter 查詢下一個實體...") # 生成實體關係圖 self._generate_entity_relationships() def _generate_entity_relationships(self): """生成實體關係分析""" print(f"\n{'='*70}") print("實體關係分析") print(f"{'='*70}") current_entities = self._get_current_entities() # 按類型分類實體 entity_categories = { "人物": [], "地點": [], "組織": [], "其他": [] } for entity, info in current_entities.items(): info_lower = info.lower() if any(keyword in info_lower for keyword in ["工作", "住", "兒子", "太太", "護理師"]): entity_categories["人物"].append((entity, info)) elif any(keyword in info_lower for keyword in ["市", "醫院", "幼兒園", "廠區"]): entity_categories["地點"].append((entity, info)) elif any(keyword in info_lower for keyword in ["公司", "企業", "醫院"]): entity_categories["組織"].append((entity, info)) else: entity_categories["其他"].append((entity, info)) for category, entities in entity_categories.items(): if entities: print(f"\n📂 {category}:") for entity, info in entities: print(f" 🏷️ {entity}: {info}") print(f"\n📈 實體記憶體效能:") print(f" 總實體數: {len(current_entities)}") print(f" 對話輪次: {self.conversation_count}") print(f" 平均每輪提取實體: {len(current_entities)/self.conversation_count:.1f}") # 執行實體記憶體示範 entity_manager = EntityMemoryManager() entity_manager.demonstrate_entity_memory() ConversationKGMemory - 知識圖譜記憶 概念說明 知識圖譜記憶體是最先進的記憶體管理方式，它不僅提取實體，還建立實體之間的關係網絡。這種結構化的知識表示方式特別適合複雜的推理和查詢任務。 詳細實作 from langchain_community.memory.kg import ConversationKGMemory from langchain_openai import ChatOpenAI import networkx as nx import matplotlib.pyplot as plt from typing import List, Tuple, Dict class KnowledgeGraphMemorySystem: """知識圖譜記憶體系統""" def __init__(self, model_name="gpt-4o"): self.llm = ChatOpenAI(model_name=model_name, temperature=0) self.memory = ConversationKGMemory( llm=self.llm, return_messages=True, memory_key="history" ) self.conversation_count = 0 self.knowledge_graph = nx.DiGraph() # 用於視覺化的圖 self.triplet_history = [] # 記錄三元組生成歷史 def process_conversation(self, human_input: str, ai_response: str): """處理對話並構建知識圖譜""" self.conversation_count += 1 print(f"\n{'='*80}") print(f"對話 {self.conversation_count}: 知識圖譜構建") print(f"{'='*80}") print(f"👤 人類: {human_input}") print(f"🤖 AI: {ai_response}") # 記錄對話前的知識狀態 before_kg = dict(self.memory.kg.get_store()) # 保存對話（觸發知識圖譜更新） self.memory.save_context( inputs={"input": human_input}, outputs={"output": ai_response} ) # 記錄對話後的知識狀態 after_kg = dict(self.memory.kg.get_store()) # 分析知識圖譜變化 new_triplets = self._analyze_kg_changes(before_kg, after_kg) # 更新本地圖結構 self._update_local_graph(new_triplets) # 顯示當前知識狀態 self._display_current_knowledge() def _analyze_kg_changes(self, before: dict, after: dict) -> List[Tuple[str, str, str]]: """分析知識圖譜變化""" print(f"\n🧠 知識圖譜分析:") new_triplets = [] # 檢查每個實體的知識變化 for entity, relations in after.items(): if entity not in before: print(f" ➕ 新實體: {entity}") # 解析該實體的所有關係 triplets = self._parse_relations(entity, relations) new_triplets.extend(triplets) elif before[entity] != relations: print(f" 🔄 更新實體: {entity}") # 找出新增的關係 old_relations = before[entity] new_info = relations.replace(old_relations, "").strip() if new_info: triplets = self._parse_relations(entity, new_info) new_triplets.extend(triplets) if new_triplets: print(f" 🔗 新增關係三元組:") for subject, predicate, obj in new_triplets: print(f" ({subject}) --[{predicate}]--> ({obj})") self.triplet_history.append({ "conversation": self.conversation_count, "triplet": (subject, predicate, obj) }) print(f" 📊 圖譜統計:") print(f" 實體數: {len(before)} → {len(after)}") print(f" 新增三元組: {len(new_triplets)}") return new_triplets def _parse_relations(self, entity: str, relations_text: str) -> List[Tuple[str, str, str]]: """從關係文本中解析三元組""" triplets = [] # 簡化的關係解析（實際應用中可能需要更複雜的NLP處理） import re # 匹配各種關係模式 patterns = [ (r'(\w+)\s*lives in\s*(\w+)', lambda m: (m.group(1), "lives_in", m.group(2))), (r'(\w+)\s*works at\s*(\w+)', lambda m: (m.group(1), "works_at", m.group(2))), (r'(\w+)\s*is\s*(\w+)', lambda m: (m.group(1), "is", m.group(2))), (r'(\w+)\s*has\s*(\w+)', lambda m: (m.group(1), "has", m.group(2))), (r'(\w+)\s*married to\s*(\w+)', lambda m: (m.group(1), "married_to", m.group(2))), (r'(\w+)\s*child of\s*(\w+)', lambda m: (m.group(1), "child_of", m.group(2))), ] for pattern, extractor in patterns: matches = re.finditer(pattern, relations_text, re.IGNORECASE) for match in matches: triplets.append(extractor(match)) return triplets def _update_local_graph(self, new_triplets: List[Tuple[str, str, str]]): """更新本地圖結構用於視覺化""" for subject, predicate, obj in new_triplets: self.knowledge_graph.add_edge(subject, obj, relation=predicate) def _display_current_knowledge(self): """顯示當前知識狀態""" current_kg = dict(self.memory.kg.get_store()) print(f"\n📚 當前知識庫:") for entity, info in current_kg.items(): print(f" 🏷️ {entity}: {info}") def query_knowledge(self, query: str): """查詢知識圖譜""" print(f"\n🔍 知識查詢: {query}") # 使用記憶體的查詢功能 memory_vars = self.memory.load_memory_variables({"input": query}) print(f"📋 查詢結果:") if "history" in memory_vars: for msg in memory_vars["history"]: print(f" 💭 {msg.content}") # 基於圖結構的查詢 self._graph_based_query(query) def _graph_based_query(self, query: str): """基於圖結構的查詢""" query_lower = query.lower() print(f"\n🕸️ 圖結構查詢:") # 查找相關節點 relevant_nodes = [node for node in self.knowledge_graph.nodes() if node.lower() in query_lower] if relevant_nodes: for node in relevant_nodes: print(f" 🎯 {node} 的關係:") # 出邊（該節點作為主語） for target in self.knowledge_graph.successors(node): edge_data = self.knowledge_graph.get_edge_data(node, target) relation = edge_data.get('relation', 'related_to') print(f" → {relation}: {target}") # 入邊（該節點作為賓語） for source in self.knowledge_graph.predecessors(node): edge_data = self.knowledge_graph.get_edge_data(source, node) relation = edge_data.get('relation', 'related_to') print(f" ← {relation}: {source}") def visualize_knowledge_graph(self): """視覺化知識圖譜""" if len(self.knowledge_graph.nodes()) == 0: print("知識圖譜為空，無法視覺化") return plt.figure(figsize=(12, 8)) # 使用spring布局 pos = nx.spring_layout(self.knowledge_graph, k=2, iterations=50) # 繪製節點 nx.draw_networkx_nodes(self.knowledge_graph, pos, node_color='lightblue', node_size=2000, alpha=0.7) # 繪製邊 nx.draw_networkx_edges(self.knowledge_graph, pos, edge_color='gray', arrows=True, arrowsize=20, alpha=0.6) # 繪製標籤 nx.draw_networkx_labels(self.knowledge_graph, pos, font_size=10, font_weight='bold') # 繪製邊標籤（關係） edge_labels = nx.get_edge_attributes(self.knowledge_graph, 'relation') nx.draw_networkx_edge_labels(self.knowledge_graph, pos, edge_labels, font_size=8) plt.title("知識圖譜視覺化", fontsize=16, fontweight='bold') plt.axis('off') plt.tight_layout() plt.show() print(f"\n📊 圖譜統計:") print(f" 節點數: {len(self.knowledge_graph.nodes())}") print(f" 邊數: {len(self.knowledge_graph.edges())}") print(f" 密度: {nx.density(self.knowledge_graph):.3f}") def demonstrate_kg_memory(self): """示範知識圖譜記憶體的完整功能""" # 設計一個複雜的企業組織對話 conversations = [ ("Alice Chen 是我們公司的 CEO", "了解，Alice Chen 擔任 CEO 的職位。"), ("Bob Wang 是技術總監，他直接向 Alice 匯報", "好的，Bob Wang 是技術總監，向 Alice Chen 匯報。"), ("我們有三個部門：工程部、市場部和財務部", "公司組織結構很清楚，有工程、市場和財務三個主要部門。"), ("Bob Wang 管理工程部，工程部有 15 個工程師", "Bob Wang 管理工程部，部門規模是 15 人。"), ("Carol Liu 是市場部主管，市場部在台北辦公室", "Carol Liu 負責市場部，辦公地點在台北。"), ("David Kim 是財務部主管，他之前在 Google 工作", "David Kim 負責財務部，有 Google 的工作背景。"), ("Alice Chen 和 Carol Liu 都是台大 MBA 畢業", "Alice Chen 和 Carol Liu 都有台大 MBA 的教育背景。"), ("公司總部在新竹，但市場部在台北辦公室", "公司總部位於新竹，但市場部設在台北辦公室。"), ] print("🎯 知識圖譜記憶體示範：企業組織結構對話") print("觀察系統如何構建複雜的關係網絡...\n") for human_msg, ai_msg in conversations: self.process_conversation(human_msg, ai_msg) input("\n按 Enter 繼續...") # 知識查詢示範 print(f"\n{'='*80}") print("知識查詢示範") print(f"{'='*80}") queries = [ "Alice Chen 是誰？", "誰管理工程部？", "台大 MBA 畢業的有哪些人？", "公司有哪些辦公室？", "Bob Wang 的工作職責是什麼？" ] for query in queries: self.query_knowledge(query) input(f"\n按 Enter 執行下一個查詢...") # 生成關係分析報告 self._generate_relationship_analysis() # 視覺化圖譜（如果環境支援） try: self.visualize_knowledge_graph() except ImportError: print("matplotlib 未安裝，跳過圖譜視覺化") def _generate_relationship_analysis(self): """生成關係分析報告""" print(f"\n{'='*80}") print("知識圖譜關係分析報告") print(f"{'='*80}") current_kg = dict(self.memory.kg.get_store()) print(f"📈 統計資訊:") print(f" 對話輪次: {self.conversation_count}") print(f" 識別實體: {len(current_kg)}") print(f" 生成三元組: {len(self.triplet_history)}") print(f" 圖節點數: {len(self.knowledge_graph.nodes())}") print(f" 圖邊數: {len(self.knowledge_graph.edges())}") print(f"\n🔗 關係類型分析:") relation_counts = {} for record in self.triplet_history: _, predicate, _ = record["triplet"] relation_counts[predicate] = relation_counts.get(predicate, 0) + 1 for relation, count in sorted(relation_counts.items(), key=lambda x: x[1], reverse=True): print(f" {relation}: {count} 次") print(f"\n🏆 知識圖譜記憶體優勢:") print(f" ✅ 結構化知識表示") print(f" ✅ 支援複雜關係查詢") print(f" ✅ 便於推理和分析") print(f" ✅ 可視覺化知識網絡") # 執行知識圖譜記憶體示範 kg_system = KnowledgeGraphMemorySystem() kg_system.demonstrate_kg_memory() VectorStoreRetrieverMemory - 向量檢索記憶 概念說明 向量檢索記憶體使用語意相似性而非時間順序來檢索相關的歷史對話。這種方法特別適合需要從大量歷史對話中找到語意相關內容的場景。 詳細實作 import faiss from langchain_openai.embeddings import OpenAIEmbeddings from langchain_community.vectorstores.faiss import FAISS from langchain.memory import VectorStoreRetrieverMemory from langchain.docstore import InMemoryDocstore from langchain.schema import Document import numpy as np from typing import List, Dict, Tuple class VectorMemorySystem: """向量檢索記憶體系統""" def __init__(self, embedding_dim=1536, top_k=3): self.embeddings_model = OpenAIEmbeddings() self.embedding_dim = embedding_dim self.top_k = top_k # 初始化 FAISS 向量資料庫 index = faiss.IndexFlatL2(embedding_dim) self.vectorstore = FAISS( self.embeddings_model, index, InMemoryDocstore({}), {} ) # 創建檢索器 self.retriever = self.vectorstore.as_retriever( search_kwargs={"k": top_k} ) # 初始化記憶體 self.memory = VectorStoreRetrieverMemory( retriever=self.retriever, memory_key="history", input_key="input" ) self.conversation_count = 0 self.all_conversations = [] # 保存所有對話用於分析 self.similarity_scores = [] # 記錄相似度分數 def add_conversation(self, human_input: str, ai_response: str, metadata: Dict = None): """添加對話到向量記憶體""" self.conversation_count += 1 print(f"\n{'='*70}") print(f"添加對話 {self.conversation_count}") print(f"{'='*70}") print(f"👤 人類: {human_input}") print(f"🤖 AI: {ai_response}") # 保存完整對話記錄 conversation_record = { "id": self.conversation_count, "human": human_input, "ai": ai_response, "metadata": metadata or {} } self.all_conversations.append(conversation_record) # 將對話保存到向量記憶體 self.memory.save_context( inputs={"input": human_input}, outputs={"output": ai_response} ) print(f"✅ 對話已添加到向量記憶體") print(f"📊 當前記憶體大小: {len(self.all_conversations)} 條對話") def query_similar_conversations(self, query: str, detailed_analysis=True): """查詢語意相似的對話""" print(f"\n{'='*70}") print(f"向量檢索查詢") print(f"{'='*70}") print(f"🔍 查詢: {query}") # 使用記憶體進行檢索 memory_result = self.memory.load_memory_variables({"input": query}) print(f"\n📋 檢索結果:") retrieved_docs = memory_result.get("history", []) if not retrieved_docs: print(" ❌ 未找到相關對話") return [] # 詳細分析檢索結果 if detailed_analysis: self._analyze_retrieval_results(query, retrieved_docs) return retrieved_docs def _analyze_retrieval_results(self, query: str, retrieved_docs: List[Document]): """分析檢索結果""" print(f"\n🔬 檢索分析:") print(f" 檢索到 {len(retrieved_docs)} 條相關對話") # 計算相似度分數 query_embedding = self.embeddings_model.embed_query(query) for i, doc in enumerate(retrieved_docs): print(f"\n 📄 結果 {i+1}:") print(f" 內容: {doc.page_content}") # 計算相似度 doc_embedding = self.embeddings_model.embed_query(doc.page_content) similarity = self._cosine_similarity(query_embedding, doc_embedding) print(f" 相似度: {similarity:.4f}") # 查找原始對話 original_conv = self._find_original_conversation(doc.page_content) if original_conv: print(f" 原始對話 ID: {original_conv['id']}") print(f" AI 回應: {original_conv['ai']}") def _cosine_similarity(self, vec1: List[float], vec2: List[float]) -> float: """計算餘弦相似度""" vec1 = np.array(vec1) vec2 = np.array(vec2) return np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2)) def _find_original_conversation(self, content: str) -> Dict: """根據內容找到原始對話""" for conv in self.all_conversations: if conv["human"] in content or content in conv["human"]: return conv return None def compare_with_chronological_memory(self, query: str): """與時序記憶體比較""" print(f"\n{'='*70}") print("向量檢索 vs 時序檢索比較") print(f"{'='*70}") print(f"🔍 查詢: {query}") # 向量檢索結果 print(f"\n🧠 向量檢索結果 (語意相似):") vector_results = self.query_similar_conversations(query, detailed_analysis=False) # 時序檢索結果（最近的對話） print(f"\n⏰ 時序檢索結果 (最近對話):") recent_conversations = self.all_conversations[-self.top_k:] for i, conv in enumerate(recent_conversations): print(f" 📄 最近第 {i+1} 條:") print(f" 人類: {conv['human']}") print(f" AI: {conv['ai']}") # 分析兩種方法的差異 self._analyze_retrieval_differences(query, vector_results, recent_conversations) def _analyze_retrieval_differences(self, query: str, vector_results: List[Document], recent_conversations: List[Dict]): """分析檢索方法差異""" print(f"\n📊 檢索方法比較:") # 提取向量檢索的對話ID vector_conv_ids = [] for doc in vector_results: original = self._find_original_conversation(doc.page_content) if original: vector_conv_ids.append(original["id"]) recent_conv_ids = [conv["id"] for conv in recent_conversations] print(f" 向量檢索對話 ID: {vector_conv_ids}") print(f" 時序檢索對話 ID: {recent_conv_ids}") # 計算重疊 overlap = set(vector_conv_ids) & set(recent_conv_ids) print(f" 重疊對話數: {len(overlap)}") if len(overlap) < len(vector_conv_ids): print(f" 💡 向量檢索找到了時序方法錯過的相關對話") # 語意相關性分析 query_embedding = self.embeddings_model.embed_query(query) print(f"\n📈 語意相關性分析:") # 向量檢索結果的平均相似度 vector_similarities = [] for doc in vector_results: doc_embedding = self.embeddings_model.embed_query(doc.page_content) similarity = self._cosine_similarity(query_embedding, doc_embedding) vector_similarities.append(similarity) # 時序檢索結果的平均相似度 recent_similarities = [] for conv in recent_conversations: conv_embedding = self.embeddings_model.embed_query(conv["human"]) similarity = self._cosine_similarity(query_embedding, conv_embedding) recent_similarities.append(similarity) if vector_similarities: avg_vector_sim = sum(vector_similarities) / len(vector_similarities) print(f" 向量檢索平均相似度: {avg_vector_sim:.4f}") if recent_similarities: avg_recent_sim = sum(recent_similarities) / len(recent_similarities) print(f" 時序檢索平均相似度: {avg_recent_sim:.4f}") def demonstrate_vector_memory(self): """示範向量記憶體的完整功能""" # 準備多樣化的對話數據 conversations = [ # 技術問題 ("如何優化 Python 程式的效能？", "可以使用 profiling 工具找出瓶頸，考慮使用 Cython 或 NumPy。"), ("資料庫查詢太慢了，有什麼解決方案？", "可以考慮添加索引、優化查詢語句、或使用快取。"), ("Docker 容器如何最佳化？", "減少映像檔層數、使用多階段建構、選擇輕量基礎映像檔。"), # 商業問題 ("如何提升客戶滿意度？", "重點關注客戶服務品質、產品品質和售後支援。"), ("公司營收成長策略有哪些？", "可以考慮市場擴張、產品創新、客戶留存和定價策略。"), ("如何管理遠端團隊？", "建立清楚的溝通機制、使用協作工具、定期視訊會議。"), # 學習問題 ("如何學習機器學習？", "從數學基礎開始，學習 Python，實作小專案，閱讀論文。"), ("英文口說如何進步？", "多練習對話、看英文影片、參加語言交換活動。"), ("如何培養程式設計思維？", "多練習演算法、參與開源專案、學習設計模式。"), # 生活問題 ("如何保持健康的作息？", "規律睡眠、適度運動、均衡飲食、減少壓力。"), ("如何規劃退休生活？", "提早開始儲蓄、投資理財、培養興趣愛好。"), ] print("🎯 向量記憶體示範：多領域問答系統") print("建立包含技術、商業、學習、生活等多個領域的對話記憶體...\n") # 添加所有對話 for human_msg, ai_msg in conversations: self.add_conversation(human_msg, ai_msg) print(f"\n✅ 已添加 {len(conversations)} 條對話到向量記憶體") # 測試各種查詢 test_queries = [ "程式效能問題", # 應該找到技術相關對話 "團隊管理", # 應該找到管理相關對話 "學習方法", # 應該找到學習相關對話 "如何提升速度", # 可能涵蓋技術和生活 "Python 相關問題" # 特定技術查詢 ] print(f"\n{'='*70}") print("語意檢索測試") print(f"{'='*70}") for query in test_queries: self.query_similar_conversations(query) input(f"\n按 Enter 繼續下一個查詢...") # 比較檢索方法 print(f"\n{'='*70}") print("檢索方法比較測試") print(f"{'='*70}") comparison_queries = ["效能優化", "團隊協作"] for query in comparison_queries: self.compare_with_chronological_memory(query) input(f"\n按 Enter 繼續...") # 生成分析報告 self._generate_vector_analysis_report() def _generate_vector_analysis_report(self): """生成向量記憶體分析報告""" print(f"\n{'='*70}") print("向量記憶體效能分析報告") print(f"{'='*70}") print(f"📈 統計資訊:") print(f" 總對話數: {len(self.all_conversations)}") print(f" 向量維度: {self.embedding_dim}") print(f" 檢索數量: {self.top_k}") # 分析對話主題分佈 topics = { "技術": ["Python", "資料庫", "Docker", "程式", "演算法"], "商業": ["客戶", "營收", "團隊", "管理", "策略"], "學習": ["學習", "進步", "培養", "練習"], "生活": ["健康", "作息", "退休", "生活"] } topic_counts = {topic: 0 for topic in topics.keys()} for conv in self.all_conversations: text = conv["human"] + " " + conv["ai"] for topic, keywords in topics.items(): if any(keyword in text for keyword in keywords): topic_counts[topic] += 1 break print(f"\n📂 主題分佈:") for topic, count in topic_counts.items(): percentage = count / len(self.all_conversations) * 100 print(f" {topic}: {count} 條 ({percentage:.1f}%)") print(f"\n🎯 向量記憶體優勢:") print(f" ✅ 語意相似性檢索") print(f" ✅ 跨時間範圍找到相關對話") print(f" ✅ 支援大規模對話存儲") print(f" ✅ 自動主題聚類效果") # 執行向量記憶體示範 vector_system = VectorMemorySystem() vector_system.demonstrate_vector_memory() SQL 資料庫持久化記憶 概念說明 對於需要跨會話保存、多用戶共享、或大規模存儲的應用，SQL 資料庫提供了可靠的持久化記憶體解決方案。LangChain 提供了與各種 SQL 資料庫的整合支援。 詳細實作 from langchain_community.chat_message_histories import SQLChatMessageHistory from langchain.memory import ConversationBufferMemory from langchain.schema import BaseMessage, HumanMessage, AIMessage import sqlite3 import json from datetime import datetime from typing import List, Dict, Optional class PersistentMemoryManager: """持久化記憶體管理器""" def __init__(self, database_url="sqlite:///chat_memory.db"): self.database_url = database_url self.session_memories = {} # 快取不同會話的記憶體 # 初始化資料庫 self._initialize_database() def _initialize_database(self): """初始化資料庫結構""" if "sqlite:///" in self.database_url: db_path = self.database_url.replace("sqlite:///", "") conn = sqlite3.connect(db_path) cursor = conn.cursor() # 創建擴展表用於會話元數據 cursor.execute(""" CREATE TABLE IF NOT EXISTS session_metadata ( session_id TEXT PRIMARY KEY, user_id TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, last_activity TIMESTAMP DEFAULT CURRENT_TIMESTAMP, conversation_count INTEGER DEFAULT 0, metadata TEXT ) """) conn.commit() conn.close() def create_session_memory(self, session_id: str, user_id: str = None, metadata: Dict = None) -> ConversationBufferMemory: """為特定會話創建記憶體""" print(f"\n{'='*60}") print(f"創建會話記憶體: {session_id}") print(f"{'='*60}") # 創建 SQL 訊息歷史 chat_history = SQLChatMessageHistory( session_id=session_id, connection=self.database_url ) # 創建記憶體實例 memory = ConversationBufferMemory( chat_memory=chat_history, return_messages=True, memory_key="history" ) # 快取記憶體實例 self.session_memories[session_id] = memory # 更新會話元數據 self._update_session_metadata(session_id, user_id, metadata) print(f"✅ 會話記憶體已創建並連接到資料庫") # 載入歷史對話 existing_messages = chat_history.messages if existing_messages: print(f"📚 載入了 {len(existing_messages)} 條歷史訊息") return memory def _update_session_metadata(self, session_id: str, user_id: str = None, metadata: Dict = None): """更新會話元數據""" if "sqlite:///" in self.database_url: db_path = self.database_url.replace("sqlite:///", "") conn = sqlite3.connect(db_path) cursor = conn.cursor() cursor.execute(""" INSERT OR REPLACE INTO session_metadata (session_id, user_id, last_activity, metadata) VALUES (?, ?, ?, ?) """, ( session_id, user_id, datetime.now().isoformat(), json.dumps(metadata or {}) )) conn.commit() conn.close() def add_conversation(self, session_id: str, human_message: str, ai_message: str) -> None: """向會話添加對話""" if session_id not in self.session_memories: print(f"⚠️ 會話 {session_id} 不存在，自動創建...") self.create_session_memory(session_id) memory = self.session_memories[session_id] print(f"\n📝 添加對話到會話 {session_id}:") print(f" 👤 人類: {human_message}") print(f" 🤖 AI: {ai_message}") # 保存對話 memory.save_context( inputs={"input": human_message}, outputs={"output": ai_message} ) # 更新會話統計 self._increment_conversation_count(session_id) print(f"✅ 對話已保存到資料庫") def _increment_conversation_count(self, session_id: str): """增加對話計數""" if "sqlite:///" in self.database_url: db_path = self.database_url.replace("sqlite:///", "") conn = sqlite3.connect(db_path) cursor = conn.cursor() cursor.execute(""" UPDATE session_metadata SET conversation_count = conversation_count + 1, last_activity = ? WHERE session_id = ? """, (datetime.now().isoformat(), session_id)) conn.commit() conn.close() def get_session_history(self, session_id: str, limit: Optional[int] = None) -> List[BaseMessage]: """獲取會話歷史""" print(f"\n🔍 查詢會話歷史: {session_id}") if session_id not in self.session_memories: self.create_session_memory(session_id) memory = self.session_memories[session_id] memory_vars = memory.load_memory_variables({}) messages = memory_vars.get("history", []) if limit: messages = messages[-limit:] print(f"📚 找到 {len(messages)} 條歷史訊息") for i, msg in enumerate(messages): print(f" {i+1}. {msg.__class__.__name__}: {msg.content[:50]}...") return messages def search_conversations(self, keyword: str, session_id: Optional[str] = None) -> List[Dict]: """搜尋包含關鍵字的對話""" print(f"\n🔍 搜尋關鍵字: '{keyword}'") if session_id: print(f" 限制會話: {session_id}") results = [] if "sqlite:///" in self.database_url: db_path = self.database_url.replace("sqlite:///", "") conn = sqlite3.connect(db_path) cursor = conn.cursor() # 構建查詢 query = """ SELECT session_id, message, type, created_at FROM message_store WHERE message LIKE ? """ params = [f"%{keyword}%"] if session_id: query += " AND session_id = ?" params.append(session_id) query += " ORDER BY created_at DESC" cursor.execute(query, params) rows = cursor.fetchall() for row in rows: results.append({ "session_id": row[0], "message": row[1], "type": row[2], "created_at": row[3] }) conn.close() print(f"📋 找到 {len(results)} 條包含關鍵字的訊息") for result in results[:5]: # 只顯示前5條 print(f" 會話 {result['session_id']}: {result['type']} - {result['message'][:50]}...") return results def get_session_statistics(self) -> Dict: """獲取所有會話的統計資訊""" print(f"\n📊 會話統計分析") stats = { "total_sessions": 0, "total_conversations": 0, "active_sessions": 0, "sessions_by_user": {}, "recent_activity": [] } if "sqlite:///" in self.database_url: db_path = self.database_url.replace("sqlite:///", "") conn = sqlite3.connect(db_path) cursor = conn.cursor() # 基本統計 cursor.execute("SELECT COUNT(*) FROM session_metadata") stats["total_sessions"] = cursor.fetchone()[0] cursor.execute("SELECT SUM(conversation_count) FROM session_metadata") result = cursor.fetchone()[0] stats["total_conversations"] = result if result else 0 # 按用戶統計 cursor.execute(""" SELECT user_id, COUNT(*), SUM(conversation_count) FROM session_metadata WHERE user_id IS NOT NULL GROUP BY user_id """) for row in cursor.fetchall(): stats["sessions_by_user"][row[0]] = { "sessions": row[1], "conversations": row[2] } # 最近活動 cursor.execute(""" SELECT session_id, user_id, last_activity, conversation_count FROM session_metadata ORDER BY last_activity DESC LIMIT 10 """) for row in cursor.fetchall(): stats["recent_activity"].append({ "session_id": row[0], "user_id": row[1], "last_activity": row[2], "conversation_count": row[3] }) conn.close() # 顯示統計結果 print(f" 總會話數: {stats['total_sessions']}") print(f" 總對話數: {stats['total_conversations']}") print(f" 用戶數: {len(stats['sessions_by_user'])}") if stats["sessions_by_user"]: print(f"\n👥 用戶活動:") for user_id, user_stats in stats["sessions_by_user"].items(): print(f" {user_id}: {user_stats['sessions']} 會話, {user_stats['conversations']} 對話") if stats["recent_activity"]: print(f"\n⏰ 最近活動:") for activity in stats["recent_activity"][:3]: print(f" 會話 {activity['session_id']}: {activity['conversation_count']} 對話 ({activity['last_activity']})") return stats def cleanup_old_sessions(self, days_threshold: int = 30): """清理舊會話""" print(f"\n🧹 清理 {days_threshold} 天前的舊會話") if "sqlite:///" in self.database_url: db_path = self.database_url.replace("sqlite:///", "") conn = sqlite3.connect(db_path) cursor = conn.cursor() # 查找舊會話 cursor.execute(""" SELECT session_id FROM session_metadata WHERE last_activity < datetime('now', '-{} days') """.format(days_threshold)) old_sessions = [row[0] for row in cursor.fetchall()] if old_sessions: print(f"找到 {len(old_sessions)} 個舊會話") # 刪除訊息 for session_id in old_sessions: cursor.execute("DELETE FROM message_store WHERE session_id = ?", (session_id,)) # 刪除元數據 cursor.execute(""" DELETE FROM session_metadata WHERE last_activity < datetime('now', '-{} days') """.format(days_threshold)) conn.commit() print(f"✅ 已清理 {len(old_sessions)} 個舊會話") else: print("沒有需要清理的舊會話") conn.close() def demonstrate_persistent_memory(self): """示範持久化記憶體的完整功能""" print("🎯 持久化記憶體示範：多用戶客服系統") print("模擬多個用戶的跨會話對話...\n") # 模擬多個用戶的對話 users_conversations = { "user_alice": [ ("我想查詢我的帳戶餘額", "您好 Alice！您的帳戶餘額是 $1,250.00"), ("上次轉帳記錄在哪裡查看？", "您可以在交易記錄頁面查看所有轉帳記錄"), ("我想設定自動扣款", "好的，我為您開啟自動扣款設定頁面") ], "user_bob": [ ("我的信用卡被盜刷了", "非常抱歉！我立即為您凍結信用卡"), ("需要重新申請新卡嗎？", "是的，我已為您申請新卡，5-7個工作天內寄達"), ("臨時額度可以提高嗎？", "基於您的信用記錄，可以臨時提高到 $5,000") ], "user_charlie": [ ("想了解投資理財產品", "我們有定期存款、基金、股票等多種理財產品"), ("風險等級如何分類？", "分為保守型、穩健型、積極型三個等級"), ("我適合哪種投資？", "根據您的風險承受能力，建議從穩健型開始") ] } # 第一輪：創建會話並添加對話 print("=" * 70) print("第一輪：創建用戶會話") print("=" * 70) for user_id, conversations in users_conversations.items(): session_id = f"{user_id}_session_1" # 創建會話 self.create_session_memory( session_id=session_id, user_id=user_id, metadata={"service_type": "banking", "priority": "normal"} ) # 添加對話 for human_msg, ai_msg in conversations: self.add_conversation(session_id, human_msg, ai_msg) input(f"\n{user_id} 會話已創建，按 Enter 繼續...") # 第二輪：模擬會話重新連接 print("\n" + "=" * 70) print("第二輪：重新連接會話（模擬跨會話持久化）") print("=" * 70) for user_id in users_conversations.keys(): session_id = f"{user_id}_session_1" print(f"\n重新連接 {user_id} 的會話...") # 重新創建記憶體（應該載入歷史對話） memory = self.create_session_memory(session_id, user_id) # 添加新對話 if user_id == "user_alice": self.add_conversation(session_id, "我剛才問的問題解決了嗎？", "是的，您的自動扣款已經設定完成！") elif user_id == "user_bob": self.add_conversation(session_id, "新卡什麼時候會到？", "預計3個工作天內送達，我們會簡訊通知您") elif user_id == "user_charlie": self.add_conversation(session_id, "我決定先開始定期存款", "很好的選擇！我為您推薦1年期定存，利率2.1%") # 第三輪：搜尋和分析功能 print("\n" + "=" * 70) print("第三輪：搜尋和分析功能") print("=" * 70) # 關鍵字搜尋 search_keywords = ["信用卡", "帳戶", "投資"] for keyword in search_keywords: self.search_conversations(keyword) input(f"\n搜尋 '{keyword}' 完成，按 Enter 繼續...") # 統計分析 self.get_session_statistics() print("\n" + "=" * 70) print("持久化記憶體示範完成") print("=" * 70) print("🎯 持久化記憶體優勢:") print(" ✅ 跨會話保存對話歷史") print(" ✅ 支援多用戶隔離") print(" ✅ 可靠的資料持久化") print(" ✅ 豐富的查詢和分析功能") print(" ✅ 適合生產環境部署") # 執行持久化記憶體示範 persistent_manager = PersistentMemoryManager() persistent_manager.demonstrate_persistent_memory() 實際應用場景與選擇建議 記憶體類型選擇指南 根據不同的應用場景，選擇合適的記憶體管理策略至關重要： def recommend_memory_type(use_case_info: Dict) -> str: """根據使用場景推薦記憶體類型""" conversation_length = use_case_info.get("avg_conversation_length", 10) user_count = use_case_info.get("user_count", 1) budget_sensitive = use_case_info.get("budget_sensitive", False) need_persistence = use_case_info.get("need_persistence", False) complex_relationships = use_case_info.get("complex_relationships", False) semantic_search = use_case_info.get("semantic_search", False) recommendations = [] # 基於場景特徵的推薦邏輯 if conversation_length <= 5 and not budget_sensitive: recommendations.append(("ConversationBufferMemory", "適合短對話，實作簡單")) if 5 < conversation_length <= 20: recommendations.append(("ConversationBufferWindowMemory", "滑動視窗控制記憶體大小")) if budget_sensitive or conversation_length > 20: recommendations.append(("ConversationTokenBufferMemory", "精確控制 token 使用")) recommendations.append(("ConversationSummaryBufferMemory", "平衡詳細度與成本")) if complex_relationships: recommendations.append(("ConversationKGMemory", "建立實體關係網絡")) recommendations.append(("ConversationEntityMemory", "追蹤實體資訊")) if semantic_search or conversation_length > 50: recommendations.append(("VectorStoreRetrieverMemory", "語意相似性檢索")) if need_persistence or user_count > 1: recommendations.append(("SQL持久化記憶體", "跨會話保存，多用戶支援")) return recommendations # 應用場景範例 scenarios = { "客服聊天機器人": { "avg_conversation_length": 15, "user_count": 1000, "budget_sensitive": True, "need_persistence": True, "complex_relationships": False, "semantic_search": False }, "個人AI助手": { "avg_conversation_length": 30, "user_count": 1, "budget_sensitive": False, "need_persistence": True, "complex_relationships": True, "semantic_search": True }, "教育輔導系統": { "avg_conversation_length": 25, "user_count": 100, "budget_sensitive": True, "need_persistence": True, "complex_relationships": True, "semantic_search": True } } print("🎯 記憶體類型選擇建議") print("=" * 60) for scenario, info in scenarios.items(): print(f"\n📋 場景: {scenario}") recommendations = recommend_memory_type(info) print(" 推薦記憶體類型:") for memory_type, reason in recommendations: print(f" ✅ {memory_type}: {reason}") 效能最佳化與注意事項 記憶體效能最佳化策略 class MemoryOptimizer: """記憶體效能最佳化器""" @staticmethod def optimize_token_usage(memory_type: str, conversation_data: List) -> Dict: """最佳化 token 使用策略""" total_tokens = sum(len(conv["human"]) + len(conv["ai"]) for conv in conversation_data) avg_conversation_tokens = total_tokens / len(conversation_data) if conversation_data else 0 recommendations = { "current_usage": total_tokens, "avg_per_conversation": avg_conversation_tokens, "optimization_strategies": [] } if memory_type == "ConversationBufferMemory": if total_tokens > 2000: recommendations["optimization_strategies"].extend([ "考慮改用 ConversationBufferWindowMemory", "設定適當的視窗大小（建議 5-10 輪）", "實作對話重要性評分，保留關鍵對話" ]) elif memory_type == "ConversationSummaryMemory": recommendations["optimization_strategies"].extend([ "最佳化摘要提示詞，確保關鍵資訊不遺失", "定期檢查摘要品質", "考慮使用更便宜的模型進行摘要" ]) elif memory_type == "VectorStoreRetrieverMemory": recommendations["optimization_strategies"].extend([ "調整檢索數量（k值）平衡相關性與成本", "使用本地向量資料庫減少 API 調用", "實作向量快取機制" ]) return recommendations @staticmethod def monitor_memory_performance(memory_instance, operation_count: int = 100): """監控記憶體效能""" import time import psutil import os print(f"\n🔍 記憶體效能監控") print(f"操作次數: {operation_count}") # 記錄開始狀態 start_time = time.time() start_memory = psutil.Process(os.getpid()).memory_info().rss / 1024 / 1024 # MB # 模擬記憶體操作 for i in range(operation_count): memory_instance.save_context( inputs={"input": f"測試訊息 {i}"}, outputs={"output": f"回應 {i}"} ) # 每 10 次操作檢查一次 if (i + 1) % 10 == 0: current_memory = psutil.Process(os.getpid()).memory_info().rss / 1024 / 1024 print(f" 第 {i+1} 次操作 - 記憶體使用: {current_memory:.1f} MB") # 記錄結束狀態 end_time = time.time() end_memory = psutil.Process(os.getpid()).memory_info().rss / 1024 / 1024 # 分析結果 total_time = end_time - start_time memory_increase = end_memory - start_memory avg_time_per_op = total_time / operation_count print(f"\n📊 效能分析結果:") print(f" 總執行時間: {total_time:.2f} 秒") print(f" 平均每次操作: {avg_time_per_op:.4f} 秒") print(f" 記憶體增長: {memory_increase:.1f} MB") print(f" 每次操作記憶體增長: {memory_increase/operation_count:.4f} MB") return { "total_time": total_time, "avg_time_per_operation": avg_time_per_op, "memory_increase_mb": memory_increase, "memory_per_operation_mb": memory_increase / operation_count } # 記憶體最佳化實踐建議 print("\n🎯 記憶體管理最佳實踐") print("=" * 60) best_practices = [ "定期監控 token 使用量，避免超出模型限制", "根據對話重要性選擇不同的記憶體策略", "實作記憶體快取機制，減少重複計算", "使用批次處理最佳化向量操作效能", "定期清理不需要的歷史對話", "監控記憶體使用情況，防止記憶體洩漏", "為不同用戶實作記憶體隔離", "設定合理的記憶體大小限制", "使用異步操作提升並發效能", "實作記憶體備份與恢復機制" ] for i, practice in enumerate(best_practices, 1): print(f"{i:2d}. {practice}") print(f"\n💡 記憶體選擇決策樹:") print("短期對話 (< 10輪) → ConversationBufferMemory") print("中期對話 (10-30輪) → ConversationBufferWindowMemory") print("長期對話 + 成本敏感 → ConversationSummaryBufferMemory") print("複雜關係追蹤 → ConversationEntityMemory / ConversationKGMemory") print("語意搜尋需求 → VectorStoreRetrieverMemory") print("多用戶 + 持久化 → SQL 資料庫記憶體") 結論 本教學詳細介紹了 LangChain 中各種記憶體管理機制的原理、實作和應用場景。每種記憶體類型都有其獨特的優勢和適用場景： ConversationBufferMemory：最簡單直觀，適合短期對話 ConversationBufferWindowMemory：平衡記憶體使用與對話品質 ConversationTokenBufferMemory：精確控制成本和 token 使用 ConversationSummaryMemory：長期對話的成本最佳化解決方案 ConversationEntityMemory：實體追蹤和管理 ConversationKGMemory：複雜關係建模和推理 VectorStoreRetrieverMemory：語意相似性檢索 SQL 持久化記憶體：生產級多用戶支援 選擇合適的記憶體管理策略需要考慮對話長度、成本預算、功能需求、用戶規模等多個因素。在實際應用中，往往需要結合多種記憶體類型，構建混合式的記憶體管理系統，以達到最佳的效能和用戶體驗。 通過本教學的學習和實踐，您應該能夠： 理解各種記憶體管理機制的工作原理 根據具體需求選擇合適的記憶體類型 實作和最佳化記憶體管理系統 監控和調整記憶體效能 處理大規模部署中的記憶體管理挑戰 記憶體管理是構建高品質 AI 應用的基礎，掌握這些技能將大大提升您的 RAG 系統開發能力。