要打造一個真正具備 Memory 的 Agent,核心在於建立一個 記憶管理管線(Memory Pipeline):將短期的原始對話(Conversation History)進行提煉與結構化,並寫入持久化儲存庫,最後在新的對話中透過動態檢索(Retrieval)注入模型的工作記憶中。
下面結合前面的 Tool System 與 ReAct Engine,實作一個具備 「跨會話長期記憶 (Cross-Session Memory)」 的 Agent。
系統包含三大核心模組:
messages 陣列(工作記憶)。[使用者輸入]
│
├── 1. 檢索長期記憶 (Read Long-Term Memory) ──> 注入 Context
└── 2. 載入當前對話 (Short-Term Memory)
│
▼
[ Agent Loop (ReAct) ] ──> 產生回答並執行工具
│
└── 3. 背景提煉記憶 (Extract & Write Memory) ──> 寫回 Long-Term Store
使用純 Python + Pydantic + OpenAI SDK 即可展示完整的記憶讀寫與自我提煉機制。
import json
import os
from typing import Dict, Any, List
from pydantic import BaseModel, Field
from openai import OpenAI
client = OpenAI()
# ==========================================
# 1. 長期記憶儲存庫 (Long-Term Memory Store)
# ==========================================
class MemoryStore:
def __init__(self, filepath: str = "agent_memory.json"):
self.filepath = filepath
self.memory: Dict[str, Any] = self._load()
def _load(self) -> Dict[str, Any]:
if os.path.exists(self.filepath):
try:
with open(self.filepath, "r", encoding="utf-8") as f:
return json.load(f)
except Exception:
pass
return {"user_profile": {}, "past_facts": []}
def save(self):
with open(self.filepath, "w", encoding="utf-8") as f:
json.dump(self.memory, f, ensure_ascii=False, indent=2)
def update_profile(self, key: str, value: Any):
"""更新使用者個人偏好/ profile"""
self.memory["user_profile"][key] = value
self.save()
def add_fact(self, fact: str):
"""新增提煉出的事實經驗"""
if fact not in self.memory["past_facts"]:
self.memory["past_facts"].append(fact)
self.save()
def get_formatted_context(self) -> str:
"""將長期記憶格式化為 System Prompt 可用的上下文"""
profile_str = json.dumps(self.memory["user_profile"], ensure_ascii=False)
facts_str = "\n".join([f"- {fact}" for fact in self.memory["past_facts"]])
return (
f"=== 關於使用者的長期記憶 ===\n"
f"【個人偏好與資訊】: {profile_str}\n"
f"【歷史重要事實/習慣】:\n{facts_str if facts_str else '尚無記錄'}\n"
f"=============================="
)
# ==========================================
# 2. 記憶提煉器 (Memory Extractor)
# ==========================================
class ExtractedMemory(BaseModel):
user_profile_updates: Dict[str, str] = Field(description="從對話中新發現的使用者偏好、語言、風格或習慣,例如 {'preferred_language': 'Python'}")
new_facts: List[str] = Field(description="值得長期記住的新事實,例如 '使用者正在開發一個 Tetris 遊戲'")
class MemoryExtractor:
def __init__(self, memory_store: MemoryStore):
self.store = memory_store
def extract_and_consolidate(self, conversation_history: List[Dict[str, Any]]):
"""背景提煉任務:分析本次對話,將價值資訊提煉並寫入長期記憶"""
messages_text = ""
for msg in conversation_history:
role = msg.get("role")
content = msg.get("content")
if role in ["user", "assistant"] and content:
messages_text += f"{role.upper()}: {content}\n"
prompt = (
"你是一個記憶提煉專家。請分析以下對話,提取出關於使用者的長期特徵、偏好、以及值得未來參考的事實。\n"
"忽略暫時性的問候或中間過渡廢話。如果沒有新的重要資訊,回傳空物件。\n\n"
f"對話內容:\n{messages_text}"
)
try:
completion = client.beta.chat.completions.parse(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
response_format=ExtractedMemory,
)
extracted: ExtractedMemory = completion.choices[0].message.parsed
# 將提煉出的結果寫回記憶庫
for k, v in extracted.user_profile_updates.items():
self.store.update_profile(k, v)
for fact in extracted.new_facts:
self.store.add_fact(fact)
print("\n🧠 [Memory Store Updates] 成功鞏固長期記憶:")
print(f" ├─ Profile: {extracted.user_profile_updates}")
print(f" └─ Facts: {extracted.new_facts}")
except Exception as e:
print(f"⚠️ 記憶提煉失敗: {e}")
# ==========================================
# 3. 具備 Memory 機制的 Agent Engine
# ==========================================
class StatefulMemoryAgent:
def __init__(self, memory_store: MemoryStore):
self.store = memory_store
self.extractor = MemoryExtractor(memory_store)
def chat(self, user_input: str, session_history: List[Dict[str, Any]] = None) -> tuple[str, List[Dict[str, Any]]]:
if session_history is None:
session_history = []
# 1. 讀取長期記憶,動態構建 System Prompt
long_term_memory_context = self.store.get_formatted_context()
system_prompt = (
"你是一個具備長期記憶能力的 AI 助手。\n"
"請根據下方【長期記憶】中的資訊,主動適應使用者的偏好與背景,不需要讓使用者重複解釋他習慣的事物。\n\n"
f"{long_term_memory_context}"
)
# 2. 組合當前 Session 的短期記憶
messages = [{"role": "system", "content": system_prompt}] + session_history
messages.append({"role": "user", "content": user_input})
print(f"👤 User: {user_input}")
# 3. 呼叫模型生成回應 (工作記憶運作)
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages
)
reply = response.choices[0].message.content
print(f"🤖 Agent: {reply}\n")
# 4. 更新當前 Session 的短期對話歷史
session_history.append({"role": "user", "content": user_input})
session_history.append({"role": "assistant", "content": reply})
# 5. 非同步/背景提煉記憶 (寫入長期記憶)
self.extractor.extract_and_consolidate(session_history)
return reply, session_history
模擬兩次完全獨立的對話 Session(代表使用者今天聊完,明天開了個全新的對話視窗):
# 初始化記憶庫
store = MemoryStore("my_agent_memory.json")
agent = StatefulMemoryAgent(store)
# ==========================================
# Session 1:使用者建立偏好與個人背景
# ==========================================
print("--- 🚀 開始 Session 1 (建立初始對話) ---")
session1_history = []
# 第一對話輪次:帶入個人特徵與偏好
_, session1_history = agent.chat(
"嗨!我是 Marlin,我平時主要用 C++ 與 NASM 寫 Win32 和低階開發,請記得我偏好簡潔的回答風格。",
session1_history
)
# 第二對話輪次:補充近期的專案
_, session1_history = agent.chat(
"我最近正在用 C++ 與 GDI 寫一個 Win32 的 Tetris 遊戲。",
session1_history
)
# ==========================================
# Session 2:開啟全新的對話 (歷史紀錄已清空,但長期記憶保留)
# ==========================================
print("\n--- 🚀 開始 Session 2 (開啟全新對話視窗 / 模擬清空對話歷史) ---")
session2_history = [] # 歷史紀錄清空!
# 使用者完全不提自己的背景,直接提出一個新問題
agent.chat("我現在想為我的遊戲加上得分倍率機制,給我一些程式碼範例結構。", session2_history)
當 Session 1 執行時,MemoryExtractor 自動提煉出以下結構並存檔至 agent_memory.json:
{
"user_profile": {
"name": "Marlin",
"primary_languages": "C++, NASM",
"preferred_style": "簡潔風格"
},
"past_facts": [
"使用者目前正使用 C++ 與 GDI 開發 Win32 Tetris 遊戲"
]
}
在 Session 2 中,雖然對話歷史 (session2_history) 為空,但 Agent 讀取了長期記憶,自動挑選了 C++ 與 Win32 API 寫出符合簡潔風格的程式碼:
🤖 Agent 的回答:
針對你的 Win32 C++ Tetris (GDI) 專案,可以在遊戲邏輯中維護一個基礎得分與倍率變數。以下是簡潔的範例架構:struct ScoreSystem { int score = 0; int comboCount = 0; void OnLineClear(int linesCleared) { if (linesCleared == 0) { comboCount = 0; // 重置連消倍率 return; } // 連消越多,倍率越高 int basePoints = linesCleared * 100; float multiplier = 1.0f + (comboCount * 0.5f); score += static_str_cast<int>(basePoints * multiplier); comboCount++; } };
當專案擴大時,可將上述架構做以下三點替換升級:
User Profile(精確讀取)。Past Facts / Trajectories(語意 RAG 搜尋)。timestamp 與 access_count。長時間未被使用的邊緣記憶降權或自動歸檔,避免 Context Window 再次爆滿。update/overwrite 操作,而非盲目 append 舊的 C++ 記憶。