在 Part 1 中,我們建立了短期記憶和長期記憶的基礎架構。但光有記憶倉庫還不夠,我們需要一個智慧管理員來決定:
今天我們將完成這個記憶系統的最後一塊拼圖——採用模組化設計,打造智慧的記憶管理器,並整合所有組件實現完整的 AI 對話系統。
在深入實作之前,讓我們先了解整個系統的模組化設計:
📁 ai_memory_system/
├── config.py # 🔧 系統配置模組
├── utils.py # 🛠 文本處理工具模組
├── memory_manager.py # 🧠 記憶管理核心模組
├── chatbot.py # 🤖 AI對話整合模組
├── advanced_features.py # ⚡ 進階功能模組
└── main.py # 🚀 主程序模組
每個模組都有明確的職責分工,既獨立運作又相互協作。這樣的設計讓我們能夠:
配置模組是整個系統的控制中心,集中管理所有可調參數。
將系統行為參數化,讓不同的使用場景能夠快速調整系統特性:
from datetime import timedelta
class MemoryConfig:
"""記憶系統配置"""
# 重要性評估權重 - 決定什麼最值得記住
IMPORTANCE_WEIGHTS = {
'personal_info': 1.0, # 個人資訊權重最高
'preferences': 0.9, # 偏好設定
'goals_plans': 0.8, # 目標和計畫
'emotions': 0.7, # 情緒相關
'skills_knowledge': 0.6, # 技能和知識
'facts': 0.5, # 一般事實
'casual_chat': 0.2 # 閒聊內容
}
# 記憶整理觸發條件
CONSOLIDATION_CONVERSATION_THRESHOLD = 20 # 對話數量閾值
CONSOLIDATION_TIME_THRESHOLD = timedelta(hours=2) # 時間閾值
IMPORTANCE_THRESHOLD = 0.5 # 重要性閾值
# 相似度計算參數
SIMILARITY_THRESHOLDS = {
'profile': 0.3, # 用戶資料相關性
'facts': 0.3, # 事實相關性
'summaries': 0.2, # 摘要相關性
'progress': 0.4 # 進度相關性
}
# TF-IDF 向量化參數
TFIDF_CONFIG = {
'max_features': 1000,
'ngram_range': (1, 2),
'stop_words': None # 中文沒有內建停用詞
}
class DatabaseConfig:
"""資料庫配置"""
DB_PATH = "long_term_memory.db"
CACHE_EXPIRE_MINUTES = 5
MAX_SUMMARIES_LIMIT = 10
class AIConfig:
"""AI模型配置"""
DEFAULT_MODEL = "gemini-pro"
MAX_CONTEXT_TURNS = 5
SHORT_TERM_MAX_TURNS = 10
SHORT_TERM_MAX_AGE_MINUTES = 30
這樣的配置設計讓我們能夠輕鬆調整系統行為:
# 提高個人資訊的重要性
MemoryConfig.IMPORTANCE_WEIGHTS['personal_info'] = 1.2
# 更頻繁地觸發記憶整理
MemoryConfig.CONSOLIDATION_CONVERSATION_THRESHOLD = 10
# 調整相似度敏感度
MemoryConfig.SIMILARITY_THRESHOLDS['facts'] = 0.4
文本處理模組提供系統所需的核心算法,是記憶管理的技術基礎。
import re
import hashlib
import jieba
from typing import Dict, List, Any
from config import MemoryConfig
class MessageAnalyzer:
def analyze_message_importance(self, message: str, role: str) -> tuple:
if role == "assistant":
return 0.3, "assistant_response"
max_score = 0.0
message_type = "casual_chat"
for info_type, patterns in self.keyword_patterns.items():
for pattern in patterns:
if re.search(pattern, message):
score = MemoryConfig.IMPORTANCE_WEIGHTS[info_type]
if score > max_score:
max_score = score
message_type = info_type
# 其餘邏輯保持不變...
return max_score, message_type
記憶管理器是整個系統的「大腦皮層」,負責智慧的記憶處理。
from datetime import datetime, timedelta
from collections import Counter
from typing import List, Dict, Any, Tuple
from config import MemoryConfig, DatabaseConfig
from utils import EntityExtractor, SimilarityCalculator, MessageAnalyzer
class MemoryManager:
"""記憶管理器 - AI 的大腦皮層"""
def __init__(self):
# 初始化工具組件
self.entity_extractor = EntityExtractor()
self.similarity_calculator = SimilarityCalculator()
self.message_analyzer = MessageAnalyzer()
# 系統狀態追蹤
self.last_consolidation = None
def should_consolidate_to_longterm(self, conversation_length: int,
last_consolidation: datetime = None) -> bool:
"""判斷是否該將短期記憶整理到長期記憶"""
# 條件1:對話長度達到閾值
if conversation_length >= MemoryConfig.CONSOLIDATION_CONVERSATION_THRESHOLD:
return True
# 條件2:距離上次整理超過設定時間
if last_consolidation:
time_since_last = datetime.now() - last_consolidation
if time_since_last > MemoryConfig.CONSOLIDATION_TIME_THRESHOLD:
return True
# 條件3:記憶體使用量過高(簡化判斷)
if conversation_length >= 10:
return True
return False
核心功能:記憶整理機制
def consolidate_memories(self, short_term_memory, long_term_memory):
"""將短期記憶中的重要資訊整理到長期記憶"""
conversations = list(short_term_memory.conversations)
important_messages = []
topics = set()
# 第一階段:分析每個訊息的重要性
for msg in conversations:
importance, msg_type = self.message_analyzer.analyze_message_importance(
msg['content'], msg['role']
)
if importance > MemoryConfig.IMPORTANCE_THRESHOLD:
important_messages.append({
'message': msg,
'importance': importance,
'type': msg_type
})
if msg.get('topic'):
topics.add(msg['topic'])
# 第二階段:提取和儲存重要事實
self._process_important_messages(important_messages, long_term_memory)
# 第三階段:建立對話摘要
if important_messages and conversations:
summary = self._create_conversation_summary(important_messages)
date_range = f"{conversations[0]['timestamp'].strftime('%Y-%m-%d')} to {conversations[-1]['timestamp'].strftime('%Y-%m-%d')}"
long_term_memory.save_conversation_summary(
summary, list(topics), date_range, len(conversations)
)
def _process_important_messages(self, important_messages: List[Dict],
long_term_memory):
"""處理重要訊息,根據類型進行分類儲存"""
for item in important_messages:
msg = item['message']
msg_type = item['type']
# 提取實體資訊
entities = self.entity_extractor.extract_entities(msg['content'])
# 根據訊息類型進行不同處理
if msg_type == 'personal_info':
self._handle_personal_info(entities, long_term_memory)
elif msg_type == 'preferences':
long_term_memory.add_important_fact(
"preference", msg['content'], confidence=item['importance']
)
elif msg_type == 'goals_plans':
long_term_memory.add_important_fact(
"goal", msg['content'], confidence=item['importance']
)
elif msg_type == 'skills_knowledge':
self._handle_skills_knowledge(entities, msg['content'], long_term_memory)
def _handle_personal_info(self, entities: Dict, long_term_memory):
"""處理個人資訊"""
profile_updates = {}
if entities['names']:
profile_updates['name'] = entities['names'][0]
if entities['locations']:
profile_updates['location'] = entities['locations'][0]
if profile_updates:
long_term_memory.update_user_profile(profile_updates)
def _handle_skills_knowledge(self, entities: Dict, content: str, long_term_memory):
"""處理技能和知識相關資訊"""
if entities['skills']:
for skill in entities['skills']:
long_term_memory.update_learning_progress(
skill, "mentioned", {'mentioned_in': content}
)
def _create_conversation_summary(self, important_messages: List[Dict]) -> str:
"""建立對話摘要"""
if not important_messages:
return "無重要內容的一般對話"
# 統計主要話題
topics = Counter()
key_points = []
for item in important_messages:
msg_type = item['type']
content = item['message']['content']
# 簡化內容(取前50字)
simplified = content[:50] + "..." if len(content) > 50 else content
key_points.append(f"- {simplified}")
topics[msg_type] += 1
# 建立摘要
main_topic = topics.most_common(1)[0][0] if topics else "一般對話"
summary = f"主要討論{main_topic}相關話題。重點包括:\n"
summary += "\n".join(key_points[:5]) # 最多5個要點
return summary
核心功能:相關記憶搜尋
def find_relevant_memories(self, query: str, long_term_memory,
max_results: int = 5) -> List[Dict]:
"""根據查詢找到相關的記憶"""
relevant_memories = []
try:
# 1. 搜尋用戶資料
self._search_user_profile(query, long_term_memory, relevant_memories)
# 2. 搜尋重要事實
self._search_important_facts(query, long_term_memory, relevant_memories)
# 3. 搜尋對話摘要
self._search_conversation_summaries(query, long_term_memory, relevant_memories)
# 4. 搜尋學習進度
self._search_learning_progress(query, long_term_memory, relevant_memories)
except Exception as e:
print(f"搜尋相關記憶時發生錯誤: {e}")
return []
# 按相關性排序並返回前N個結果
relevant_memories.sort(key=lambda x: x['similarity'], reverse=True)
return relevant_memories[:max_results]
def _search_user_profile(self, query: str, long_term_memory, relevant_memories: List):
"""搜尋用戶資料"""
profile = long_term_memory.get_user_profile()
profile_text = " ".join([str(v) for v in profile.values() if v])
if profile_text:
similarity = self.similarity_calculator.calculate_similarity(query, profile_text)
if similarity > MemoryConfig.SIMILARITY_THRESHOLDS['profile']:
relevant_memories.append({
'type': 'profile',
'content': profile,
'similarity': similarity,
'source': 'user_profile'
})
def _search_important_facts(self, query: str, long_term_memory, relevant_memories: List):
"""搜尋重要事實"""
facts = long_term_memory.get_important_facts()
for fact in facts:
similarity = self.similarity_calculator.calculate_similarity(query, fact['content'])
if similarity > MemoryConfig.SIMILARITY_THRESHOLDS['facts']:
relevant_memories.append({
'type': 'fact',
'content': fact,
'similarity': similarity,
'source': 'important_facts'
})
def _search_conversation_summaries(self, query: str, long_term_memory, relevant_memories: List):
"""搜尋對話摘要"""
summaries = long_term_memory.get_recent_summaries(limit=DatabaseConfig.MAX_SUMMARIES_LIMIT)
for summary in summaries:
similarity = self.similarity_calculator.calculate_similarity(query, summary['summary'])
if similarity > MemoryConfig.SIMILARITY_THRESHOLDS['summaries']:
relevant_memories.append({
'type': 'summary',
'content': summary,
'similarity': similarity,
'source': 'conversation_summaries'
})
def _search_learning_progress(self, query: str, long_term_memory, relevant_memories: List):
"""搜尋學習進度"""
progress = long_term_memory.get_learning_progress()
for prog in progress:
prog_text = f"{prog['subject']} {prog['skill_level']}"
similarity = self.similarity_calculator.calculate_similarity(query, prog_text)
if similarity > MemoryConfig.SIMILARITY_THRESHOLDS['progress']:
relevant_memories.append({
'type': 'progress',
'content': prog,
'similarity': similarity,
'source': 'learning_progress'
})
聊天機器人模組是整個系統的指揮中心,協調所有模組的工作。
import google.generativeai as genai
from datetime import datetime, timedelta
from typing import List, Dict, Any, Optional
from config import MemoryConfig, AIConfig
from short_term_memory import ShortTermMemory
from long_term_memory import LongTermMemory
from memory_manager import MemoryManager
class SmartChatbot:
"""具有記憶功能的智慧聊天機器人"""
def __init__(self, user_id: str, api_key: str, model_name: str = None):
self.user_id = user_id
# 初始化記憶系統三大核心模組
self.short_term_memory = ShortTermMemory(
max_turns=AIConfig.SHORT_TERM_MAX_TURNS,
max_age_minutes=AIConfig.SHORT_TERM_MAX_AGE_MINUTES
)
self.long_term_memory = LongTermMemory(user_id)
self.memory_manager = MemoryManager()
# 初始化 AI 模型
genai.configure(api_key=api_key)
self.model = genai.GenerativeModel(model_name or AIConfig.DEFAULT_MODEL)
# 系統狀態追蹤
self.last_consolidation = None
self.conversation_count = 0
核心對話流程
def chat(self, user_message: str, topic: str = None) -> str:
"""
進行對話的主要方法 - 整合所有記憶功能
完整的對話流程:
1. 儲存用戶訊息到短期記憶
2. 搜尋相關的長期記憶
3. 建構豐富的AI上下文
4. 生成AI回應
5. 儲存AI回應到短期記憶
6. 檢查並執行記憶整理
"""
try:
# 步驟1:新增用戶訊息到短期記憶
self.short_term_memory.add_message("user", user_message, topic)
self.conversation_count += 1
# 步驟2:查找相關的長期記憶
relevant_memories = self.memory_manager.find_relevant_memories(
user_message, self.long_term_memory
)
# 步驟3:建構 AI 的豐富上下文
context = self._build_context(user_message, relevant_memories)
# 步驟4:生成 AI 回應
response = self.model.generate_content(context)
ai_response = response.text
# 步驟5:新增 AI 回應到短期記憶
self.short_term_memory.add_message("assistant", ai_response, topic)
# 步驟6:檢查是否需要記憶整理
if self.memory_manager.should_consolidate_to_longterm(
self.conversation_count, self.last_consolidation
):
self._consolidate_memories()
return ai_response
except Exception as e:
return f"抱歉,我遇到了一些技術問題:{str(e)}"
智慧上下文建構
def _build_context(self, current_message: str, relevant_memories: List[Dict]) -> str:
"""為 AI 建構豐富的對話上下文"""
# 基礎系統提示
system_prompt = """你是一個智慧的AI助手,具有記憶能力。請根據以下資訊來回應用戶:
1. 保持自然、友善的對話風格
2. 根據用戶的個人資訊和偏好來個性化回應
3. 參考歷史對話內容,保持一致性
4. 如果不確定某些資訊,可以詢問確認
"""
# 添加相關的歷史記憶
if relevant_memories:
system_prompt += "\n相關的歷史資訊:\n"
system_prompt += self._format_relevant_memories(relevant_memories)
# 添加近期對話上下文
recent_context = self.short_term_memory.get_recent_context(
turns=AIConfig.MAX_CONTEXT_TURNS
)
if recent_context:
conversation_history = self._format_recent_context(recent_context)
system_prompt += f"\n最近的對話:\n{conversation_history}"
# 添加當前用戶訊息
system_prompt += f"\n當前用戶訊息:{current_message}\n\n請回應:"
return system_prompt
def _format_relevant_memories(self, relevant_memories: List[Dict]) -> str:
"""格式化相關記憶為上下文"""
formatted_memories = ""
for memory in relevant_memories:
if memory['type'] == 'profile':
profile = memory['content']
if profile.get('name'):
formatted_memories += f"- 用戶姓名:{profile['name']}\n"
if profile.get('interests'):
formatted_memories += f"- 興趣:{', '.join(profile['interests'])}\n"
if profile.get('goals'):
formatted_memories += f"- 目標:{', '.join(profile['goals'])}\n"
elif memory['type'] == 'fact':
fact = memory['content']
formatted_memories += f"- {fact['type']}:{fact['content']}\n"
elif memory['type'] == 'progress':
prog = memory['content']
formatted_memories += f"- 學習進度:{prog['subject']} ({prog['skill_level']})\n"
return formatted_memories
def _format_recent_context(self, recent_context: List[Dict]) -> str:
"""格式化近期對話上下文"""
conversation_history = ""
for msg in recent_context[:-1]:
role = "用戶" if msg['role'] == 'user' else "AI"
content = msg['content']
conversation_history += f"{role}:{content}\n"
return conversation_history
記憶整理和狀態管理
def _consolidate_memories(self):
"""執行記憶整理"""
print("正在整理記憶...")
# 調用記憶管理器進行智慧整理
self.memory_manager.consolidate_memories(
self.short_term_memory,
self.long_term_memory
)
self.last_consolidation = datetime.now()
# 清理短期記憶中的過期內容
self._cleanup_short_term_memory()
print("記憶整理完成")
def _cleanup_short_term_memory(self):
"""清理短期記憶中的過期內容"""
cutoff_time = datetime.now() - timedelta(minutes=15)
while self.short_term_memory.conversations:
oldest_msg = self.short_term_memory.conversations[0]
if ('timestamp' in oldest_msg and
oldest_msg['timestamp'] < cutoff_time):
self.short_term_memory.conversations.popleft()
else:
break
# 提供外部查詢接口
def get_user_profile(self) -> Dict[str, Any]:
"""取得用戶資料"""
return self.long_term_memory.get_user_profile()
def get_conversation_summary(self) -> List[Dict]:
"""取得對話摘要"""
return self.long_term_memory.get_recent_summaries()
def get_learning_progress(self, subject: str = None) -> List[Dict]:
"""取得學習進度"""
return self.long_term_memory.get_learning_progress(subject)
def search_memories(self, query: str) -> List[Dict]:
"""搜尋相關記憶"""
return self.memory_manager.find_relevant_memories(
query, self.long_term_memory
)
進階功能模組為系統提供情緒記憶和上下文感知等高級能力。
import json
from datetime import datetime
from typing import Tuple, List
class EmotionalMemoryManager:
"""情緒記憶管理器"""
def __init__(self):
self.emotion_keywords = {
'positive': ['開心', '高興', '滿意', '喜歡', '興奮', '感謝'],
'negative': ['難過', '生氣', '失望', '擔心', '困惑', '挫折'],
'neutral': ['還好', '普通', '一般', '沒關係']
}
def analyze_emotion(self, message: str) -> Tuple[str, float]:
"""分析訊息的情緒傾向和強度"""
positive_count = sum(1 for keyword in self.emotion_keywords['positive']
if keyword in message)
negative_count = sum(1 for keyword in self.emotion_keywords['negative']
if keyword in message)
if positive_count > negative_count:
intensity = min(positive_count * 0.3, 1.0)
return 'positive', intensity
elif negative_count > positive_count:
intensity = min(negative_count * 0.3, 1.0)
return 'negative', intensity
else:
return 'neutral', 0.1
def update_emotion_history(self, long_term_memory, emotion: str,
intensity: float, context: str):
"""更新情緒歷史記錄"""
emotion_data = {
'emotion': emotion,
'intensity': intensity,
'context': context,
'timestamp': datetime.now().isoformat()
}
long_term_memory.add_important_fact(
"emotion_history",
json.dumps(emotion_data, ensure_ascii=False),
confidence=intensity
)
class ContextAwareManager:
"""上下文感知管理器"""
def __init__(self):
self.context_patterns = {
'time_sensitive': [
r'今天|明天|昨天|這週|下週|現在|剛才',
r'\d+點|\d+月|\d+日'
],
'location_based': [
r'在這裡|這邊|那邊|附近|當地'
],
'task_continuation': [
r'繼續|接下來|然後|下一步|剛才說的'
]
}
def analyze_context_needs(self, message: str) -> List[str]:
"""分析訊息需要的上下文類型"""
needed_contexts = []
for context_type, patterns in self.context_patterns.items():
for pattern in patterns:
if re.search(pattern, message):
needed_contexts.append(context_type)
break
return needed_contexts
def get_contextual_info(self, context_types: List[str], chatbot) -> Dict[str, Any]:
"""取得上下文資訊"""
context_info = {}
if 'time_sensitive' in context_types:
context_info['current_time'] = datetime.now().strftime('%Y-%m-%d %H:%M')
if 'task_continuation' in context_types:
recent_tasks = chatbot.short_term_memory.task_context
context_info['recent_tasks'] = recent_tasks
return context_info
進階聊天機器人
class AdvancedSmartChatbot(SmartChatbot):
"""具有情緒記憶和上下文感知的進階聊天機器人"""
def __init__(self, user_id: str, api_key: str, model_name: str = None):
super().__init__(user_id, api_key, model_name)
# 添加進階管理器
self.emotion_manager = EmotionalMemoryManager()
self.context_manager = ContextAwareManager()
def chat(self, user_message: str, topic: str = None) -> str:
"""進階對話功能,包含情緒和上下文感知"""
# 1. 情緒分析
emotion, intensity = self.emotion_manager.analyze_emotion(user_message)
# 2. 上下文分析
context_needs = self.context_manager.analyze_context_needs(user_message)
contextual_info = self.context_manager.get_contextual_info(context_needs, self)
# 3. 執行原有的對話流程
response = super().chat(user_message, topic)
# 4. 儲存情緒記憶(只儲存明顯的情緒)
if intensity > 0.3:
self.emotion_manager.update_emotion_history(
self.long_term_memory, emotion, intensity, user_message
)
return response
主程序模組提供完整的使用者介面和系統整合。
from chatbot import SmartChatbot
from advanced_features import AdvancedSmartChatbot
def main():
"""主程序入口"""
print("🤖 AI智慧聊天機器人已啟動!")
print("💡 輸入 'quit' 退出,輸入 '/help' 查看特殊指令")
print("=" * 50)
# 配置
API_KEY = "your-gemini-api-key" # 請替換成您的實際 API Key
USER_ID = "user_001"
try:
# 建立進階聊天機器人
chatbot = AdvancedSmartChatbot(USER_ID, API_KEY)
# 主對話循環
while True:
user_input = input("\n🧑 您: ").strip()
if user_input.lower() in ['quit', 'exit', '退出', 'q']:
print("👋 再見!感謝使用AI智慧助手")
break
if not user_input:
print("💬 請輸入您想說的話...")
continue
# 特殊指令處理
if user_input.startswith("/"):
handle_command(user_input, chatbot)
continue
# 正常對話
try:
response = chatbot.chat(user_input)
print(f"\n🤖 AI: {response}")
# 顯示記憶整理狀態
if chatbot.last_consolidation:
print(f"💭 [記憶整理時間:{chatbot.last_consolidation.strftime('%H:%M:%S')}]")
except Exception as e:
print(f"❌ 對話時發生錯誤: {e}")
except Exception as e:
print(f"❌ 系統初始化失敗: {e}")
def handle_command(command: str, chatbot):
"""處理特殊指令"""
if command == "/profile":
profile = chatbot.get_user_profile()
print("\n📋 === 用戶資料 ===")
for key, value in profile.items():
if value:
if isinstance(value, list):
print(f" {key}: {', '.join(map(str, value))}")
else:
print(f" {key}: {value}")
elif command == "/summary":
summaries = chatbot.get_conversation_summary()
print("\n📝 === 對話摘要 ===")
for i, summary in enumerate(summaries[:3], 1):
print(f" {i}. {summary['summary'][:100]}...")
elif command == "/progress":
progress = chatbot.get_learning_progress()
print("\n📚 === 學習進度 ===")
for prog in progress:
print(f" 📖 {prog['subject']}: {prog['skill_level']}")
elif command.startswith("/search "):
query = command[8:]
memories = chatbot.search_memories(query)
print(f"\n🔍 === 搜尋結果:'{query}' ===")
for i, memory in enumerate(memories, 1):
print(f" {i}. [{memory['type']}] 相關度: {memory['similarity']:.2f}")
elif command == "/help":
print("\n🆘 === 可用指令 ===")
print(" /profile - 查看用戶資料")
print(" /summary - 查看對話摘要")
print(" /progress - 查看學習進度")
print(" /search <關鍵字> - 搜尋相關記憶")
print(" /help - 顯示此幫助")
print(" quit - 退出程序")
else:
print(f"❌ 未知指令: {command}")
if __name__ == "__main__":
main()
演示和測試功能
def demo_conversation():
"""演示完整的對話流程"""
API_KEY = "your-gemini-api-key"
chatbot = AdvancedSmartChatbot("demo_user", API_KEY)
# 模擬完整的對話場景
conversations = [
("我叫小明,今年28歲,是一名軟體工程師", "自我介紹"),
("我想學習機器學習,希望能夠開發AI應用", "學習目標"),
("我比較喜歡實作導向的學習方式", "學習偏好"),
("你還記得我的年齡嗎?", "記憶測試"),
("可以給我一個Python機器學習的學習計畫嗎?", "學習規劃")
]
print("🎬 === 演示對話流程 ===")
for user_msg, topic in conversations:
print(f"\n👤 用戶:{user_msg}")
response = chatbot.chat(user_msg, topic)
print(f"🤖 AI:{response}")
if chatbot.last_consolidation:
print(f"💭 [記憶整理時間:{chatbot.last_consolidation.strftime('%H:%M:%S')}]")
# 展示學習到的資訊
print("\n📊 === 系統記憶狀況 ===")
print(f"👤 用戶資料: {chatbot.get_user_profile()}")
print(f"📚 學習進度: {chatbot.get_learning_progress()}")
print(f"📝 對話摘要: {len(chatbot.get_conversation_summary())} 條記錄")
# 選擇運行模式
if __name__ == "__main__":
import sys
if len(sys.argv) > 1 and sys.argv[1] == "demo":
demo_conversation()
else:
main()