iT邦幫忙

2026 iThome 鐵人賽

DAY 17
0

昨天我們把 Memory 從 Python List 搬進 Chroma,讓 Vector Database 負責保存 Record 與執行 Semantic Search。

不過 Day 16 使用的是:

chroma_client = chromadb.EphemeralClient()

它只把資料放在目前的程式執行期間。只要輸入:

exit

關閉程式,重新啟動後 Collection 又會變成空的。

今天要做的改變很明確:

EphemeralClient

改成:

PersistentClient

並將 Day 16 分散在程式裡的 add()get()query()count(),整理成一個獨立的:

LongTermMemoryStore

讓 Memora 第一次真的能在程式重新啟動後,保留過去建立的 Memory。


一、什麼時候才算 Long-term Memory?

目前 Memora 已經有兩種不同生命週期的資料。

Conversation History 存在:

memory = ShortTermMemory(...)

當程式結束後就會消失,負責的是這一次對話。

今天的 Long-term Memory 則要保存在硬碟:

memora_db/

即使 Python Process 結束,資料仍然存在。

因此兩者最直接的差異是:

類型 保存內容 程式重啟後
Short-term Memory 目前對話與摘要 消失
Long-term Memory 抽取後的 Memory Records 保留

Long-term Memory 不代表所有 Conversation History 都要永久保存。我們要留下的是經過 Memory Extraction 判斷後,值得跨 Conversation 使用的 Memory。


二、替資料庫設定固定位置

先在原本的 Import 區加入:

from pathlib import Path

接著在設定區新增:

BASE_DIR = Path(__file__).resolve().parent

MEMORY_DB_PATH = BASE_DIR / "memora_db"

這裡使用程式檔案所在的位置作為基準。

假設目前專案是:

memora/
├── chatbot.py
└── memora_db/

無論從哪一個 Terminal 路徑執行 chatbot.py,Memora 都會使用同一個 memora_db

如果直接寫:

MEMORY_DB_PATH = "./memora_db"

資料庫位置會受到目前 Working Directory 影響。從不同資料夾執行程式時,可能會不小心建立另一個同名資料庫。

如果是在 Notebook 中測試,沒有 __file__,則可以改用:

MEMORY_DB_PATH = Path("./memora_db")

三、加入 Memory 建立時間

現在 Memory 會長期存在,因此除了內容和來源,也應該記錄它是什麼時候建立的。

在 Import 區加入:

from datetime import datetime, timezone

之後建立 Memory 時,可以取得 UTC 時間:

created_at = datetime.now(
    timezone.utc
).isoformat()

結果會類似:

2026-09-03T02:30:15.123456+00:00

目前只是把時間保存下來。Day 23 討論 Recency 與 Memory Decay 時,才會真正使用這個欄位計算記憶的新舊程度。


四、定義 Store 回傳的資料結構

Day 16 已經有:

MemorySearchResult

今天再新增一個表示已保存 Memory 的 Model:

class StoredMemory(BaseModel):
    memory_id: str
    content: str
    source: str
    created_at: str

同時替原本的 MemorySearchResult 加上 created_at

class MemorySearchResult(BaseModel):
    memory_id: str
    content: str
    source: str
    created_at: str
    distance: float
    score: float

兩個 Model 的用途不同:

StoredMemory

表示從 Database 讀出來的一筆完整 Memory。

MemorySearchResult

除了 Memory 資料,還包含它和目前 Query 的距離與相似度。


五、建立 LongTermMemoryStore

現在新增今天最重要的 Class:

class LongTermMemoryStore:
    def __init__(
        self,
        path: str,
        collection_name: str
    ):
        self.client = chromadb.PersistentClient(
            path=path
        )

        self.collection = (
            self.client.get_or_create_collection(
                name=collection_name,
                embedding_function=None,
                configuration={
                    "hnsw": {
                        "space": "cosine"
                    }
                },
                metadata={
                    "embedding_model": EMBEDDING_MODEL
                }
            )
        )

昨天使用:

chromadb.EphemeralClient()

今天改成:

chromadb.PersistentClient(
    path=path
)

PersistentClient 會把 Chroma 資料寫入指定目錄,而不是只存在記憶體。Chroma Python Client

接著使用:

get_or_create_collection()

代表:

  • 第一次啟動時,建立 memora_memories
  • 之後重新啟動時,取得原本已經存在的 Collection

因此不會每次執行程式都重新建立一個空的 Collection。


六、把新增 Memory 封裝進 Store

LongTermMemoryStore 裡加入:

def add(
    self,
    memories: list[EmbeddedMemoryCandidate],
    source: str
):
    if not memories:
        return []

    memory_ids = [
        str(uuid4())
        for _ in memories
    ]

    created_at = datetime.now(
        timezone.utc
    ).isoformat()

    self.collection.add(
        ids=memory_ids,
        documents=[
            memory.content
            for memory in memories
        ],
        embeddings=[
            memory.embedding
            for memory in memories
        ],
        metadatas=[
            {
                "source": source,
                "created_at": created_at
            }
            for _ in memories
        ]
    )

    return memory_ids

這段邏輯延續 Day 16 的:

add_memories_to_collection()

差別只是現在由:

LongTermMemoryStore.add()

負責執行。

每一筆 Memory 仍然包含:

ID
Document
Embedding
Metadata

Metadata 則從原本的:

{
    "source": source
}

增加成:

{
    "source": source,
    "created_at": created_at
}

七、加入列出所有 Memory 的方法

接著把 Day 16 的 memory_collection.get() 封裝起來:

def list_all(self):
    result = self.collection.get(
        include=[
            "documents",
            "metadatas"
        ]
    )

    stored_memories = []

    for memory_id, document, metadata in zip(
        result["ids"],
        result["documents"],
        result["metadatas"]
    ):
        metadata = metadata or {}

        stored_memories.append(
            StoredMemory(
                memory_id=memory_id,
                content=document,
                source=metadata.get(
                    "source",
                    "unknown"
                ),
                created_at=metadata.get(
                    "created_at",
                    "unknown"
                )
            )
        )

    stored_memories.sort(
        key=lambda item: item.created_at
    )

    return stored_memories

如此一來,主程式不需要知道 Chroma 的回傳格式,只需要呼叫:

long_term_memory.list_all()

就能取得:

list[StoredMemory]

八、把 Vector Search 也封裝起來

Day 16 的 semantic_search() 直接操作:

memory_collection.query()

今天將 Database Query 搬進 Store:

def search(
    self,
    query_embedding: list[float],
    top_k: int
):
    total_memories = self.collection.count()

    if total_memories == 0:
        return []

    if top_k <= 0:
        raise ValueError(
            "top_k must be greater than 0."
        )

    result = self.collection.query(
        query_embeddings=[
            query_embedding
        ],
        n_results=min(
            top_k,
            total_memories
        ),
        include=[
            "documents",
            "metadatas",
            "distances"
        ]
    )

    search_results = []

    for (
        memory_id,
        document,
        metadata,
        distance
    ) in zip(
        result["ids"][0],
        result["documents"][0],
        result["metadatas"][0],
        result["distances"][0]
    ):
        metadata = metadata or {}

        search_results.append(
            MemorySearchResult(
                memory_id=memory_id,
                content=document,
                source=metadata.get(
                    "source",
                    "unknown"
                ),
                created_at=metadata.get(
                    "created_at",
                    "unknown"
                ),
                distance=float(distance),
                score=1 - float(distance)
            )
        )

    return search_results

最後再加入簡單的計數方法:

def count(self):
    return self.collection.count()

到這裡,LongTermMemoryStore 已經負責:

新增 Memory
列出 Memory
搜尋 Memory
計算 Memory 數量

九、完整的 LongTermMemoryStore

將剛才的內容整理在一起:

class LongTermMemoryStore:
    def __init__(
        self,
        path: str,
        collection_name: str
    ):
        self.client = chromadb.PersistentClient(
            path=path
        )

        self.collection = (
            self.client.get_or_create_collection(
                name=collection_name,
                embedding_function=None,
                configuration={
                    "hnsw": {
                        "space": "cosine"
                    }
                },
                metadata={
                    "embedding_model": EMBEDDING_MODEL
                }
            )
        )

    def add(
        self,
        memories: list[EmbeddedMemoryCandidate],
        source: str
    ):
        if not memories:
            return []

        memory_ids = [
            str(uuid4())
            for _ in memories
        ]

        created_at = datetime.now(
            timezone.utc
        ).isoformat()

        self.collection.add(
            ids=memory_ids,
            documents=[
                memory.content
                for memory in memories
            ],
            embeddings=[
                memory.embedding
                for memory in memories
            ],
            metadatas=[
                {
                    "source": source,
                    "created_at": created_at
                }
                for _ in memories
            ]
        )

        return memory_ids

    def list_all(self):
        result = self.collection.get(
            include=[
                "documents",
                "metadatas"
            ]
        )

        stored_memories = []

        for (
            memory_id,
            document,
            metadata
        ) in zip(
            result["ids"],
            result["documents"],
            result["metadatas"]
        ):
            metadata = metadata or {}

            stored_memories.append(
                StoredMemory(
                    memory_id=memory_id,
                    content=document,
                    source=metadata.get(
                        "source",
                        "unknown"
                    ),
                    created_at=metadata.get(
                        "created_at",
                        "unknown"
                    )
                )
            )

        stored_memories.sort(
            key=lambda item: item.created_at
        )

        return stored_memories

    def search(
        self,
        query_embedding: list[float],
        top_k: int
    ):
        total_memories = self.collection.count()

        if total_memories == 0:
            return []

        if top_k <= 0:
            raise ValueError(
                "top_k must be greater than 0."
            )

        result = self.collection.query(
            query_embeddings=[
                query_embedding
            ],
            n_results=min(
                top_k,
                total_memories
            ),
            include=[
                "documents",
                "metadatas",
                "distances"
            ]
        )

        search_results = []

        for (
            memory_id,
            document,
            metadata,
            distance
        ) in zip(
            result["ids"][0],
            result["documents"][0],
            result["metadatas"][0],
            result["distances"][0]
        ):
            metadata = metadata or {}

            search_results.append(
                MemorySearchResult(
                    memory_id=memory_id,
                    content=document,
                    source=metadata.get(
                        "source",
                        "unknown"
                    ),
                    created_at=metadata.get(
                        "created_at",
                        "unknown"
                    ),
                    distance=float(distance),
                    score=1 - float(distance)
                )
            )

        return search_results

    def count(self):
        return self.collection.count()

這個 Class 要放在:

Pydantic Models

定義完成之後,以及建立 Store Instance 之前。


十、建立 Long-term Memory Store Instance

刪除 Day 16 的:

chroma_client = chromadb.EphemeralClient()

memory_collection = (
    chroma_client.get_or_create_collection(...)
)

改成:

long_term_memory = LongTermMemoryStore(
    path=str(MEMORY_DB_PATH),
    collection_name="memora_memories"
)

原本的 Short-term Memory 繼續保留:

memory = ShortTermMemory(...)

因此現在程式裡有兩個不同元件:

memory

負責目前 Conversation 的 Short-term Memory。

long_term_memory

負責跨程式保存的 Long-term Memory。


十一、修改 semantic_search()

Query Embedding 仍然由 Application 建立,Vector Search 則交給 Store。

將 Day 16 的 semantic_search() 改成:

def semantic_search(
    query: str,
    top_k: int = SEARCH_TOP_K
):
    query = query.strip()

    if not query:
        raise ValueError(
            "Search query cannot be empty."
        )

    if long_term_memory.count() == 0:
        return [], 0

    query_vectors, embedding_tokens = (
        create_embeddings([query])
    )

    search_results = long_term_memory.search(
        query_embedding=query_vectors[0],
        top_k=top_k
    )

    return search_results, embedding_tokens

職責現在分得更清楚:

semantic_search()
負責處理 Query 與建立 Query Embedding

LongTermMemoryStore.search()
負責向 Database 搜尋 Records

這也讓主程式不需要直接依賴 Chroma 的 query() 格式。


十二、修改手動與自動保存流程

Day 16 的手動保存原本使用:

add_memories_to_collection(
    memories=embedded_memories,
    source="manual"
)

現在改成:

memory_ids = long_term_memory.add(
    memories=embedded_memories,
    source="manual"
)

自動抽取的保存流程則改成:

long_term_memory.add(
    memories=embedded_memories,
    source="automatic"
)

其他步驟全部不變:

Conversation
↓
LLM 判斷是否值得記住
↓
Structured Memory Candidate
↓
建立 Embedding
↓
LongTermMemoryStore.add()

也就是說,我們沒有重新設計 Day 12 到 Day 16 的流程,只是把最後的保存位置換成能跨程式存在的 Store。


十三、修改 memories 指令

Day 16 的 memories 指令直接呼叫:

memory_collection.get()

現在改成:

if command == "memories":
    stored_memories = (
        long_term_memory.list_all()
    )

    print("\n--- Long-term Memories ---")

    if not stored_memories:
        print("(no memories)")
    else:
        for index, stored_memory in enumerate(
            stored_memories,
            start=1
        ):
            print(
                f"{index}. "
                f"{stored_memory.content}"
            )
            print(
                f"   ID: "
                f"{stored_memory.memory_id}"
            )
            print(
                f"   Source: "
                f"{stored_memory.source}"
            )
            print(
                f"   Created at: "
                f"{stored_memory.created_at}"
            )

    print("--------------------------")
    continue

也可以在原本的 status 指令中增加:

print(
    "Long-term memories:",
    long_term_memory.count()
)

如此一來,就能同時觀察 Short-term Memory 和 Long-term Memory 的狀態。


十四、第一次測試跨程式記憶

現在啟動程式:

python chatbot.py

手動加入一筆 Memory:

You:
remember 使用者的英文程度是 B1。

可能看到:

Memory added:
73b3d4c8-6b38-44b5-a987-900d04e438af

接著輸入:

You:
memories

確認資料已經存在:

--- Long-term Memories ---

1. 使用者的英文程度是 B1。
   ID: 73b3d4c8-...
   Source: manual
   Created at: 2026-09-03T02:30:15+00:00

--------------------------

然後關閉程式:

You:
exit

重新執行:

python chatbot.py

這次不要重新加入 Memory,直接輸入:

You:
memories

如果仍然看到:

使用者的英文程度是 B1。

就代表資料已經跨越前一個 Python Process,被保存在:

memora_db/

裡面。


十五、重新啟動後不需要重建 Memory Embedding

重新啟動程式時,我們只會重新連接:

chromadb.PersistentClient(
    path=str(MEMORY_DB_PATH)
)

原本保存的:

Document
Embedding
Metadata
ID

都可以直接讀回來。

因此不需要重新把全部 Memory 丟進 Embeddings API。

搜尋時只需要替新的 Query 建立 Embedding:

New Query
↓
Query Embedding
↓
和已保存的 Memory Embedding 搜尋

例如重新啟動後輸入:

You:
search 幫我準備符合目前程度的英文內容

仍然可能找回:

使用者的英文程度是 B1。

這就是預先保存 Memory Embedding 的價值。


十六、不要任意更換 Embedding Model

目前 Collection 裡的向量都是由:

EMBEDDING_MODEL = "text-embedding-3-small"

建立。

因為這些 Embedding 會長期保存,所以之後不能隨意把 Query 改用另一個 Model。

如果新舊 Model 的向量維度不同,Vector Database 可能直接拒絕搜尋;即使維度剛好相同,也不代表它們位於相同的向量空間。

如果未來真的要更換 Embedding Model,比較安全的做法是:

  1. 建立新的 Collection
  2. 使用新 Model 重新產生所有 Memory Embedding
  3. 將資料寫入新的 Collection
  4. 完成後再切換搜尋來源

因此我們在 Collection Metadata 中保存:

{
    "embedding_model": EMBEDDING_MODEL
}

提醒自己這一批向量是由哪一個 Model 建立的。


十七、不要把資料庫提交到Git

現在 memora_db 裡可能包含使用者的個人資料與對話記憶,因此不適合直接提交到公開Repository。

可以在 .gitignore 加入:

memora_db/

另外也不要因為資料存在本機,就直接認為它一定安全。真正的產品還需要考慮:

存取權限
資料加密
使用者刪除資料的能力
備份與還原
不同使用者之間的隔離

目前這一版只是在本機建立最小可行的 Long-term Memory Store,還不是 Production Database。


十八、它會保存,但還不會主動想起來

現在 Memora 已經可以:

保存 Memory
關閉程式
重新啟動
列出 Memory
搜尋 Memory

但一般聊天流程仍然是:

Current Context
↓
LLM
↓
Response

只有輸入:

search <query>

時,程式才會搜尋 Long-term Memory。

也就是說,如果直接問:

You:
請依照我的程度安排今天的英文學習。

Memora 還不會自動執行:

搜尋「使用者的英文程度」
↓
找回 B1
↓
加入 LLM Context
↓
產生符合 B1 的回答

目前完成的是:

Memory 可以跨程式保存。

還沒完成的是:

回答問題之前,自動找回並使用相關 Memory。

這正是下一篇要處理的 Memory Retrieval。


Day 17 小結

今天我們將 Day 16 的:

chromadb.EphemeralClient()

改成:

chromadb.PersistentClient(
    path=str(MEMORY_DB_PATH)
)

讓 Memory Record 第一次能被保存在硬碟。

同時建立:

LongTermMemoryStore

封裝以下操作:

add()
list_all()
search()
count()

因此主程式不需要到處直接操作 Chroma Collection。

目前 Memora 的 Memory 架構可以分成:

ShortTermMemory
負責目前 Conversation
程式關閉後消失

以及:

LongTermMemoryStore
負責值得跨 Conversation 保存的 Memory
程式關閉後仍然存在

Memora 的演進來到:

Day 14
Memory Embedding

Day 15
Semantic Search

Day 16
Vector Database

Day 17
Persistent Long-term Memory Store

不過「保存」只是 Long-term Memory 的一半。

如果過去的資料永遠只是躺在 Database 裡,卻沒有在適當的時候被放回 Context,模型仍然無法使用它。

Day 18|讓 AI 找回過去:Memory Retrieval

下一篇我我們來直接修改目前的聊天流程:

Current User Input
↓
建立 Query Embedding
↓
LongTermMemoryStore.search()
↓
取得 Relevant Memories
↓
放進 LLM Context
↓
產生 Response

到時候使用者不需要再手動輸入:

search <query>

Memora 會在回答之前,自動找回可能相關的 Long-term Memory。

今天我們讓 Memory 第一次真正留了下來;明天再讓 Memora 學會在需要的時候想起它!


上一篇
Day 16|Vector Database 到底在做什麼?
下一篇
Day 18|讓 AI 找回過去:Memory Retrieval
系列文
從 Stateless LLM 到 Agentic Memory:30 天打造會記憶的 AI Agent23
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言