昨天我們把 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。
目前 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 會長期存在,因此除了內容和來源,也應該記錄它是什麼時候建立的。
在 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 時,才會真正使用這個欄位計算記憶的新舊程度。
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。
在 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
}
接著把 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]
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 之前。
刪除 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/
裡面。
重新啟動程式時,我們只會重新連接:
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 的價值。
目前 Collection 裡的向量都是由:
EMBEDDING_MODEL = "text-embedding-3-small"
建立。
因為這些 Embedding 會長期保存,所以之後不能隨意把 Query 改用另一個 Model。
如果新舊 Model 的向量維度不同,Vector Database 可能直接拒絕搜尋;即使維度剛好相同,也不代表它們位於相同的向量空間。
如果未來真的要更換 Embedding Model,比較安全的做法是:
因此我們在 Collection Metadata 中保存:
{
"embedding_model": EMBEDDING_MODEL
}
提醒自己這一批向量是由哪一個 Model 建立的。
現在 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 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,模型仍然無法使用它。
下一篇我我們來直接修改目前的聊天流程:
Current User Input
↓
建立 Query Embedding
↓
LongTermMemoryStore.search()
↓
取得 Relevant Memories
↓
放進 LLM Context
↓
產生 Response
到時候使用者不需要再手動輸入:
search <query>
Memora 會在回答之前,自動找回可能相關的 Long-term Memory。
今天我們讓 Memory 第一次真正留了下來;明天再讓 Memora 學會在需要的時候想起它!