Agent 的記憶機制：短期、長期與向量資料庫

上一篇我們從零打造了一個最基礎的 Agent：LLM + 工具 + 一個迴圈。
它已經能查天氣、查時間，並根據工具結果生成回答。

但你有沒有發現一個問題：它每次對話都是全新的開始。

你昨天告訴它你住在台北，今天它又問你住哪裡。
你上次說你偏好簡短回答，這次它又給你長篇大論。

這一篇，我們要給 Agent 加上第一個重要能力：記憶。
我們會談短期記憶、長期記憶、向量資料庫，並用 Python 實作一個真正能「記住你」的 Agent。

一、為什麼 Agent 需要記憶？

先看看沒有記憶的 Agent 會發生什麼事。

情境一：重複問同樣的問題

使用者：我住在台北。
Agent：好的，我知道了。

（隔天）

使用者：今天天氣如何？
Agent：請問您在哪個城市？

Agent 昨天才被告知使用者住在台北，今天卻完全忘了。

情境二：無法累積偏好

使用者：請用條列式回答我。
Agent：（條列式回答）

（下一次對話）

使用者：幫我整理這篇文章。
Agent：（長篇大論的段落式回答）

使用者明明說過偏好條列式，Agent 卻沒有記住。

情境三：無法從經驗中學習

使用者：幫我查台北天氣。
Agent：呼叫 get_weather("台北")，成功。

（下一次）

使用者：幫我查台北天氣。
Agent：呼叫 get_weather("台北")，成功。

（再下一次）

使用者：幫我查台北天氣。
Agent：呼叫 get_weather("台北")，成功。

Agent 每次都從零開始，無法記住「使用者經常查台北天氣」這個模式。

這些問題的根源都一樣：Agent 沒有記憶。

人類之所以能進行連貫的對話、累積知識、從經驗中學習，是因為我們有記憶。
Agent 要真正實用，也必須要有記憶。

二、記憶的三個層次

在設計 Agent 的記憶機制之前，我們先區分三種不同類型的記憶：

1. 短期記憶（Short-term Memory）

• 範圍：當前這次對話。
• 內容：最近的幾輪對話、當前任務的上下文。
• 儲存方式：直接放在 LLM 的上下文窗口中。
• 生命週期：對話結束後就消失。

例如：

• 使用者剛才說「我住在台北」
• 上一步工具回傳的天氣結果
• 當前正在處理的任務描述

2. 長期記憶（Long-term Memory）

• 範圍：跨對話、跨 session。
• 內容：使用者的偏好、重要事實、過去的互動紀錄。
• 儲存方式：外部儲存（資料庫、檔案、向量資料庫）。
• 生命週期：持久保存，直到被主動刪除。

例如：

• 使用者住在台北
• 使用者偏好條列式回答
• 使用者上次問過東京旅遊

3. 工作記憶（Working Memory）

• 範圍：當前任務的暫存區。
• 內容：任務的中間結果、待辦事項、計畫步驟。
• 儲存方式：可以是上下文窗口，也可以是外部儲存。
• 生命週期：任務完成後就可以清除。

例如：

• 規劃中的步驟清單
• 已完成的步驟與結果
• 待處理的子任務

這三種記憶的關係可以這樣理解：

長期記憶（跨對話）
    ↓ 檢索相關資訊
短期記憶（當前對話）
    ↓ 當前任務
工作記憶（暫存區）

三、短期記憶：上下文窗口的管理

短期記憶最簡單的實作方式，就是把對話歷史直接放進 messages 陣列。

messages = [
    {"role": "system", "content": "你是一個樂於助人的助理。"},
    {"role": "user", "content": "我住在台北。"},
    {"role": "assistant", "content": "好的，我知道了。"},
    {"role": "user", "content": "今天天氣如何？"},
]

這樣 LLM 就能看到「使用者住在台北」這個資訊，並在回答時使用它。

但這種做法有個問題：上下文窗口有長度限制。

即使是支援 1M token 的模型，也不可能無限累積對話。
當對話越來越長，我們必須想辦法管理上下文。

策略一：滑動窗口（Sliding Window）

只保留最近的 N 輪對話，超出的部分直接丟棄。

def manage_context(messages: list, max_turns: int = 10) -> list:
    """只保留最近的 N 輪對話"""
    system_messages = [m for m in messages if m["role"] == "system"]
    conversation = [m for m in messages if m["role"] != "system"]

    # 保留最近的 max_turns * 2 則訊息（user + assistant）
    recent = conversation[-(max_turns * 2):]

    return system_messages + recent

優點：簡單、可控。
缺點：早期的重要資訊會被丟棄。

策略二：摘要壓縮（Summarization）

當對話太長時，用 LLM 把早期對話摘要成一段簡短文字。

SUMMARIZE_PROMPT = """請將以下對話摘要成一段簡短文字，保留重要資訊：

{dialogue}

摘要："""

def summarize_old_messages(messages: list, keep_recent: int = 4) -> list:
    """把早期對話摘要，只保留最近的幾則"""
    if len(messages) <= keep_recent + 1:
        return messages

    system = messages[0]
    old = messages[1:-keep_recent]
    recent = messages[-keep_recent:]

    # 把早期對話轉成文字
    dialogue_text = "\n".join(
        f"{m['role']}: {m['content']}" for m in old
    )

    # 用 LLM 摘要
    summary = call_llm([
        {"role": "user", "content": SUMMARIZE_PROMPT.format(dialogue=dialogue_text)}
    ])

    return [
        system,
        {"role": "system", "content": f"先前對話摘要：{summary}"},
        *recent,
    ]

優點：保留重要資訊，壓縮上下文長度。
缺點：摘要可能遺失細節，且需要額外的 LLM 呼叫。

策略三：重要資訊提取

主動從對話中提取重要事實，存入長期記憶。

EXTRACT_PROMPT = """請從以下對話中提取重要的事實或偏好，以 JSON 格式輸出：

對話：
{dialogue}

請輸出：
{{"facts": ["事實1", "事實2"], "preferences": ["偏好1"]}}

如果沒有重要資訊，輸出 {{"facts": [], "preferences": []}}。
"""

def extract_important_info(messages: list) -> dict:
    """從對話中提取重要資訊"""
    dialogue_text = "\n".join(
        f"{m['role']}: {m['content']}" for m in messages
        if m["role"] in ("user", "assistant")
    )

    result = call_llm([
        {"role": "user", "content": EXTRACT_PROMPT.format(dialogue=dialogue_text)}
    ])

    try:
        return json.loads(result)
    except json.JSONDecodeError:
        return {"facts": [], "preferences": []}

這是連接短期記憶與長期記憶的橋樑。

四、長期記憶：向量資料庫

短期記憶解決了「當前對話」的問題，但跨對話的記憶需要持久化儲存。

最直覺的做法是用關聯式資料庫：

# 用 SQLite 儲存使用者偏好
import sqlite3

conn = sqlite3.connect("agent_memory.db")
conn.execute("""
    CREATE TABLE IF NOT EXISTS user_facts (
        id INTEGER PRIMARY KEY,
        user_id TEXT,
        fact TEXT,
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )
""")

但這種做法有個問題：檢索不靈活。

如果使用者問「我適合去哪裡旅遊」，我們要如何從資料庫中找到相關的事實？
用關鍵字比對？用 SQL 查詢？這些都無法處理語意相似性。

這就是向量資料庫登場的地方。

什麼是向量資料庫？

向量資料庫的核心概念是：把文字轉成向量，用向量相似度來檢索。

具體流程如下：

1. 儲存階段：把每條記憶轉成嵌入向量，存入向量資料庫。
2. 檢索階段：把使用者的查詢也轉成向量，找出最相似的 N 條記憶。

記憶：「使用者住在台北」
    ↓ 嵌入模型
向量：[0.12, -0.45, 0.78, ...]

查詢：「我適合去哪裡旅遊？」
    ↓ 嵌入模型
向量：[0.11, -0.43, 0.76, ...]
    ↓ 相似度計算
找到最相關的記憶：「使用者住在台北」

常見的向量資料庫

名稱	特點	適用場景
Chroma	輕量、易用、可嵌入，與 AI 框架高度整合、開箱即用	本地開發、小型專案、AI 雛形驗證
FAISS	Meta 開發、極致效能、支援 GPU 加速的高效能檢索庫	研究、需要超高速向量檢索的底層核心
Pinecone	全託管、易擴展、Serverless 架構、免維護伺服器	生產環境、雲端應用、追求快速上線的企業
Weaviate	開源、支援混合搜尋、原生多模態、可結合知識圖譜	中大型專案、智慧客服、多模態 AI 應用
Qdrant	開源、Rust 實作、高併發、具備強大 Payload 過濾機制	高效能、生產環境、需複雜條件篩選的系統
pgvector	PostgreSQL 擴充、直接用 SQL 查詢、免額外架設資料庫	資料量中小型、系統以 PostgreSQL 為主的專案
Milvus	專為海量資料設計的分布式架構、支援十億級以上規模	超大規模企業級應用、大流量、私有化部署場景

對初學者來說，Chroma 是最容易上手的選擇。
它可以直接在 Python 中運行，不需要額外架設服務。

五、實作：帶有記憶的 Agent

現在讓我們把短期記憶與長期記憶整合進上一篇的 Agent。

步驟 1：安裝依賴

pip install chromadb openai

步驟 2：建立記憶模組

import json
import chromadb
from datetime import datetime

class AgentMemory:
    """Agent 的記憶模組，結合短期與長期記憶"""

    def __init__(self, user_id: str, persist_dir: str = "./memory_db"):
        self.user_id = user_id

        # 長期記憶：向量資料庫
        self.client = chromadb.PersistentClient(path=persist_dir)
        self.collection = self.client.get_or_create_collection(
            name=f"user_{user_id}_memory"
        )

        # 短期記憶：對話歷史
        self.conversation_history = []

    # ========== 短期記憶 ==========

    def add_message(self, role: str, content: str):
        """加入一則對話訊息"""
        self.conversation_history.append({
            "role": role,
            "content": content,
            "timestamp": datetime.now().isoformat(),
        })

    def get_recent_messages(self, n: int = 10) -> list:
        """取得最近的 N 則訊息"""
        return [
            {"role": m["role"], "content": m["content"]}
            for m in self.conversation_history[-n:]
        ]

    # ========== 長期記憶 ==========

    def save_fact(self, fact: str, metadata: dict = None):
        """儲存一條長期記憶"""
        fact_id = f"{self.user_id}_{len(self.collection.get()['ids'])}"
        self.collection.add(
            documents=[fact],
            ids=[fact_id],
            metadatas=[metadata or {"created_at": datetime.now().isoformat()}],
        )

    def retrieve_relevant_facts(self, query: str, n_results: int = 3) -> list:
        """檢索與查詢最相關的記憶"""
        if self.collection.count() == 0:
            return []

        results = self.collection.query(
            query_texts=[query],
            n_results=min(n_results, self.collection.count()),
        )

        return results["documents"][0] if results["documents"] else []

    def get_all_facts(self) -> list:
        """取得所有長期記憶"""
        result = self.collection.get()
        return result["documents"] if result["documents"] else []

這個記憶模組提供三種能力：

• 短期記憶：add_message 與 get_recent_messages。
• 長期記憶寫入：save_fact。
• 長期記憶檢索：retrieve_relevant_facts。

步驟 3：自動提取重要資訊

我們需要一個機制，自動從對話中提取值得記住的事實。

EXTRACT_PROMPT = """請從以下對話中提取值得長期記住的事實或偏好。

對話：
{dialogue}

請以 JSON 格式輸出：
{{"facts": ["事實1", "事實2"], "preferences": ["偏好1"]}}

只提取真正重要、跨對話仍然有用的資訊。
如果沒有，輸出 {{"facts": [], "preferences": []}}。
"""

def extract_and_save_facts(memory: AgentMemory, messages: list):
    """從對話中提取重要資訊並存入長期記憶"""
    dialogue_text = "\n".join(
        f"{m['role']}: {m['content']}" for m in messages
        if m["role"] in ("user", "assistant")
    )

    result = call_llm([
        {"role": "user", "content": EXTRACT_PROMPT.format(dialogue=dialogue_text)}
    ])

    try:
        extracted = json.loads(result)
    except json.JSONDecodeError:
        return

    for fact in extracted.get("facts", []):
        memory.save_fact(fact, metadata={"type": "fact"})

    for pref in extracted.get("preferences", []):
        memory.save_fact(pref, metadata={"type": "preference"})

步驟 4：整合進 Agent 主迴圈

現在把記憶模組整合進上一篇的 Agent。

def run_agent_with_memory(
    user_input: str,
    memory: AgentMemory,
    max_iterations: int = 5,
    verbose: bool = True,
) -> str:
    """帶有記憶的 Agent"""

    # 1. 把使用者輸入加入短期記憶
    memory.add_message("user", user_input)

    # 2. 檢索相關的長期記憶
    relevant_facts = memory.retrieve_relevant_facts(user_input)

    # 3. 建立系統提示，包含相關記憶
    system_content = (
        "你是一個樂於助人的助理，會使用工具來回答問題。"
        "如果需要即時資訊或精確計算，請呼叫對應的工具。"
    )

    if relevant_facts:
        system_content += "\n\n以下是關於使用者的相關資訊：\n"
        system_content += "\n".join(f"- {fact}" for fact in relevant_facts)

    if verbose and relevant_facts:
        print(f"\n[檢索到相關記憶]")
        for fact in relevant_facts:
            print(f"  - {fact}")

    # 4. 建立 messages
    messages = [
        {"role": "system", "content": system_content},
        *memory.get_recent_messages(n=10),
    ]

    # 5. Agent 主迴圈
    for iteration in range(max_iterations):
        if verbose:
            print(f"\n{'='*50}")
            print(f"Iteration {iteration + 1}")
            print(f"{'='*50}")

        response = call_llm(messages, tools=TOOLS_SCHEMA)
        message = response.choices[0].message
        messages.append(message)

        if not message.tool_calls:
            # 把最終回答加入短期記憶
            memory.add_message("assistant", message.content)

            # 從這次對話提取重要資訊
            extract_and_save_facts(memory, messages)

            if verbose:
                print(f"\n[最終回答]\n{message.content}")
            return message.content

        for tool_call in message.tool_calls:
            function_name = tool_call.function.name
            try:
                arguments = json.loads(tool_call.function.arguments)
            except json.JSONDecodeError:
                arguments = {}

            if verbose:
                print(f"\n[工具呼叫] {function_name}({arguments})")

            result = execute_tool(function_name, arguments)

            if verbose:
                print(f"[工具結果] {result}")

            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result,
            })

    return "已達到最大迭代次數，任務未完成。"

這裡的關鍵改動：

1. 檢索長期記憶：根據使用者輸入，檢索相關的事實。
2. 注入系統提示：把檢索到的記憶放入 system message。
3. 寫入短期記憶：把使用者輸入與最終回答加入對話歷史。
4. 提取長期記憶：對話結束後，自動提取重要資訊存入向量資料庫。

六、實際執行：跨對話的記憶

讓我們模擬兩次獨立的對話，看看 Agent 是否能記住資訊。

第一次對話

memory = AgentMemory(user_id="user_001")

answer = run_agent_with_memory(
    "你好，我住在台北，我喜歡簡短的回答。",
    memory,
)
print(answer)

輸出：

==================================================
Iteration 1
==================================================

[最終回答]
好的，我記住了。您住在台北，偏好簡短回答。
有什麼我可以幫您的嗎？

此時，extract_and_save_facts 會自動把「使用者住在台北」與「使用者偏好簡短回答」存入向量資料庫。

第二次對話（新的 session）

# 重新建立 memory 物件，模擬新的對話
memory2 = AgentMemory(user_id="user_001")

answer = run_agent_with_memory(
    "今天天氣如何？",
    memory2,
)
print(answer)

輸出：

[檢索到相關記憶]
  - 使用者住在台北
  - 使用者偏好簡短回答

==================================================
Iteration 1
==================================================

[工具呼叫] get_weather({'city': '台北', 'day': '今天'})
[工具結果] {"city": "台北", "day": "今天", "condition": "多雲", "rain": "60%", "temp": "22-28°C"}

==================================================
Iteration 2
==================================================

[最終回答]
台北今天多雲，降雨機率 60%，氣溫 22-28°C。
建議攜帶雨具。

你可以看到：

• Agent 從長期記憶中檢索到「使用者住在台北」。
• 它自動把「台北」填入工具參數，不需要使用者再說一次。
• 它遵循「偏好簡短回答」的偏好，給出簡潔的回應。

這就是記憶的威力。

七、記憶壓縮與遺忘

記憶不是越多越好。
如果什麼都記，會有三個問題：

1. 檢索變慢：向量資料庫太大，查詢延遲增加。
2. 雜訊變多：不相關的記憶會被檢索出來，干擾推理。
3. 隱私風險：長期保存過多個人資訊，增加洩漏風險。

所以我們需要遺忘策略。

策略一：時間衰減

為每條記憶加上時間戳，檢索時根據時間遠近給予不同權重。

def retrieve_with_recency(self, query: str, n_results: int = 3) -> list:
    """檢索時考慮時間衰減"""
    results = self.collection.query(
        query_texts=[query],
        n_results=n_results * 2,  # 多檢索一些，再篩選
    )

    # 根據時間排序，較新的優先
    facts_with_time = []
    for doc, meta in zip(results["documents"][0], results["metadatas"][0]):
        facts_with_time.append((doc, meta.get("created_at", "")))

    facts_with_time.sort(key=lambda x: x[1], reverse=True)
    return [f[0] for f in facts_with_time[:n_results]]

策略二：重要性評分

讓 LLM 為每條記憶評分，只保留高分的。

SCORE_PROMPT = """請為以下資訊的重要性評分（1-10 分）：

資訊：{fact}

評分標準：
- 10 分：使用者核心偏好或身分資訊
- 5 分：一般事實
- 1 分：臨時性或一次性的資訊

只輸出數字。
"""

def score_fact(fact: str) -> int:
    result = call_llm([
        {"role": "user", "content": SCORE_PROMPT.format(fact=fact)}
    ])
    try:
        return int(result.strip())
    except ValueError:
        return 5

策略三：定期清理

定期刪除過舊或低分的記憶。

def cleanup_memory(self, min_score: int = 3, max_age_days: int = 90):
    """清理低分或過舊的記憶"""
    from datetime import datetime, timedelta

    all_data = self.collection.get()
    cutoff = (datetime.now() - timedelta(days=max_age_days)).isoformat()

    to_delete = []
    for fact_id, doc, meta in zip(
        all_data["ids"], all_data["documents"], all_data["metadatas"]
    ):
        score = meta.get("score", 5)
        created = meta.get("created_at", "")

        if score < min_score or created < cutoff:
            to_delete.append(fact_id)

    if to_delete:
        self.collection.delete(ids=to_delete)

策略四：記憶合併

把相似的記憶合併成一條更精簡的記憶。

MERGE_PROMPT = """請將以下多條相關記憶合併成一條精簡的記憶：

{memories}

合併後的記憶："""

def merge_similar_facts(memory: AgentMemory, threshold: float = 0.9):
    """合併高度相似的記憶"""
    all_facts = memory.get_all_facts()
    # 實作略：計算兩兩相似度，超過 threshold 的用 LLM 合併
    pass

八、記憶的最佳實踐

1. 區分事實與偏好

• 事實：使用者住在台北、使用者有兩個小孩。
• 偏好：使用者喜歡簡短回答、使用者偏好條列式。

兩者的檢索時機不同，應該分開儲存。

2. 記憶要有來源

每條記憶都應該記錄：

• 什麼時候產生的
• 從哪次對話產生的
• 信心程度有多高

metadata = {
    "created_at": datetime.now().isoformat(),
    "source": "conversation_2024-01-15",
    "confidence": 0.9,
    "type": "fact",
}

3. 允許使用者查看與刪除記憶

使用者應該有權知道 Agent 記住了什麼，並能主動刪除。

def show_memory(memory: AgentMemory):
    """顯示所有記憶"""
    facts = memory.get_all_facts()
    print("Agent 記住的資訊：")
    for i, fact in enumerate(facts, 1):
        print(f"{i}. {fact}")

def clear_memory(memory: AgentMemory):
    """清空所有記憶"""
    memory.client.delete_collection(f"user_{memory.user_id}_memory")

4. 敏感資訊要加密或脫敏

不要明文儲存密碼、信用卡號等敏感資訊。

5. 記憶檢索要控制數量

一次檢索太多記憶會干擾 LLM。
通常 3 到 5 條就夠了。

九、總結：記憶讓 Agent 從工具變成夥伴

讓我們回顧這一篇的核心：

• 為什麼需要記憶：沒有記憶的 Agent 每次對話都從零開始，無法累積知識與偏好。
• 三種記憶層次：短期記憶（當前對話）、長期記憶（跨對話）、工作記憶（任務暫存）。
• 短期記憶管理：滑動窗口、摘要壓縮、重要資訊提取。
• 長期記憶儲存：向量資料庫，用嵌入向量做語意檢索。
• 實作方式：建立記憶模組，整合進 Agent 主迴圈。
• 遺忘策略：時間衰減、重要性評分、定期清理、記憶合併。
• 最佳實踐：區分事實與偏好、記錄來源、允許查看刪除、敏感資訊脫敏、控制檢索數量。

有了記憶，Agent 不再只是一個每次對話都失憶的工具，而是能累積經驗、理解你、陪你成長的夥伴。

不過，記憶只是 Agent 的其中一個能力。
當 Agent 需要處理大量專業知識時，我們還需要一個更強大的機制：RAG（檢索增強生成）。
它讓 Agent 能從外部知識庫中檢索相關資訊，再生成回答。

---

下一篇預告

《RAG 完整解析：檢索、重排、生成》

我們會解釋 RAG 為什麼能減少幻覺、Chunking 與 Embedding 的策略、Retriever 與 Reranker 的差異，以及進階 RAG 技術如 HyDE、Multi-Query 與 Graph RAG。