iT邦幫忙

2026 iThome 鐵人賽

DAY 16
0
自我挑戰組

AI Agent 從零開始系列 第 16 篇

自己打造有 Memory 的 Agent

  • 分享至 

  • xImage
  •  

要打造一個真正具備 Memory 的 Agent,核心在於建立一個 記憶管理管線(Memory Pipeline):將短期的原始對話(Conversation History)進行提煉與結構化,並寫入持久化儲存庫,最後在新的對話中透過動態檢索(Retrieval)注入模型的工作記憶中。

下面結合前面的 Tool System 與 ReAct Engine,實作一個具備 「跨會話長期記憶 (Cross-Session Memory)」 的 Agent。


1. 有記憶 Agent 的架構設計

系統包含三大核心模組:

  1. Short-Term Memory(短期記憶):當前對話 Session 的 messages 陣列(工作記憶)。
  2. Long-Term Memory Store(長期記憶庫):儲存提煉後的實體偏好與經驗事實(這裡使用 JSON/KV 結構進行精確檢索)。
  3. Memory Extractor & Refiner(記憶提煉器):在對話背景非同步觸發,將原始對話中的 key information 抽取並寫入長期記憶。
[使用者輸入] 
     │
     ├── 1. 檢索長期記憶 (Read Long-Term Memory) ──> 注入 Context
     └── 2. 載入當前對話 (Short-Term Memory)
     │
     ▼
[ Agent Loop (ReAct) ] ──> 產生回答並執行工具
     │
     └── 3. 背景提煉記憶 (Extract & Write Memory) ──> 寫回 Long-Term Store


2. 完整實作程式碼 (Python)

使用純 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


3. 實測:驗證 Agent 的「跨 Session 記憶」

模擬兩次完全獨立的對話 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)


4. 運行結果分析

① 背景記憶提煉日誌(Memory Consolidation)

當 Session 1 執行時,MemoryExtractor 自動提煉出以下結構並存檔至 agent_memory.json:

{
  "user_profile": {
    "name": "Marlin",
    "primary_languages": "C++, NASM",
    "preferred_style": "簡潔風格"
  },
  "past_facts": [
    "使用者目前正使用 C++ 與 GDI 開發 Win32 Tetris 遊戲"
  ]
}

② Session 2 的 Agent 回覆表現

在 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++;
    }
};


5. 生產環境升級路徑(Production Readiness)

當專案擴大時,可將上述架構做以下三點替換升級:

  1. 結構化 KV + 向量混合檢索 (Hybrid Storage):
  • 用 SQLite / Postgre 儲存強結構化的 User Profile(精確讀取)。
  • 用 ChromaDB / Qdrant 儲存非結構化的 Past Facts / Trajectories(語意 RAG 搜尋)。
  1. 記憶衰減與權重 (Memory Decay Rate):
  • 給每條記憶加上 timestamp 與 access_count。長時間未被使用的邊緣記憶降權或自動歸檔,避免 Context Window 再次爆滿。
  1. 記憶衝突修復 (Memory Conflict Resolution):
  • 當使用者說:「我最近改用 Rust 了」,提煉器需要觸發 update/overwrite 操作,而非盲目 append 舊的 C++ 記憶。

上一篇
Context Window——Agent 為什麼會「失憶」?
下一篇
Agent 到底需要 Planning 嗎?
系列文
AI Agent 從零開始 共 18 篇
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言