代理記憶系統
記憶是智能代理的基石。沒有記憶，每次互動都從零開始。本技能涵蓋代理記憶的架構：短期記憶（上下文窗口）、長期記憶（向量資料庫），以及組織它們的認知架構。
關鍵洞察：記憶不僅是儲存，更是檢索。一百萬個儲存的事實，如果找不到正確的那個，就毫無意義。區塊切割、嵌入和檢索策略決定了你的代理是記住還是忘記。
這個領域非常分散，術語不一致。我們採用 CoALA 認知架構框架：語意記憶（事實）、情節記憶（經驗）和程序記憶（如何做的知識）。
原則
- 記憶品質 = 檢作品質，而非儲存數量
- 為檢索而切割，非為儲存而切割
- 上下文隔離是記憶的敵人
- 對的資訊要用對的記憶類型
- 讓舊記憶衰減——並非所有記憶都該永久保存
- 在上線前測試檢索準確度
- 背景記憶形成優於即時處理
能力
- agent-memory
- long-term-memory
- short-term-memory
- working-memory
- episodic-memory
- semantic-memory
- procedural-memory
- memory-retrieval
- memory-formation
- memory-decay
範圍
- vector-database-operations → data-engineer
- rag-pipeline-architecture → llm-architect
- embedding-model-selection → ml-engineer
- knowledge-graph-design → knowledge-engineer
工具
記憶框架
- LangMem (LangChain) - 適用時機：需要持久記憶的 LangGraph 代理 備註：支援語意、情節、程序記憶類型
- MemGPT / Letta - 適用時機：虛擬上下文管理、作業系統風格的記憶 備註：分層記憶層級、自動分頁
- Mem0 - 適用時機：用於個人化的使用者記憶層 備註：專為使用者偏好和歷史設計
向量資料庫
- Pinecone - 適用時機：受管、企業級規模（數十億個向量） 備註：最佳查詢效能，成本最高
- Qdrant - 適用時機：複雜中繼資料過濾、開源 備註：基於 Rust，過濾能力出色
- Weaviate - 適用時機：混合搜尋、知識圖譜功能 備註：GraphQL 介面，適合處理關係
- ChromaDB - 適用時機：原型開發、小型/中型應用 備註：開發者友善，10 萬向量時 p50 約 20 毫秒
- pgvector - 適用時機：已在使用 PostgreSQL，設定更簡單 備註：適合小於 100 萬向量，熟悉的工具
嵌入模型
- OpenAI text-embedding-3-large - 適用時機：最佳品質，3072 維度 備註：每 100 萬 tokens $0.13
- OpenAI text-embedding-3-small - 適用時機：良好平衡，1536 維度 備註：每 100 萬 tokens $0.02，便宜 5 倍
- nomic-embed-text-v1.5 - 適用時機：開源，本地部署 備註：768 維度，品質良好
- all-MiniLM-L6-v2 - 適用時機：輕量級，快速本地嵌入 備註：384 維度，最低延遲
模式
記憶類型架構
為不同資訊選擇正確的記憶類型
使用時機: 設計代理記憶系統
記憶類型架構（CoALA 框架）：
""" 三種記憶類型用於不同目的：
-
語意記憶：事實和知識
- 你對世界的認知
- 使用者偏好、領域知識
- 儲存在設定檔（結構化）或集合（非結構化）中
-
情節記憶：經驗和事件
- 發生過什麼事（附時間戳記的事件）
- 過去的對話、任務結果
- 用於從經驗中學習
-
程序記憶：如何做事
- 規則、技能、工作流程
- 通常實作為少樣本範例
- "我以前是怎麼解決這個的？" """
LangMem 實作
""" from langmem import MemoryStore from langgraph.graph import StateGraph
初始化記憶儲存
memory = MemoryStore( connection_string=os.environ["POSTGRES_URL"] )
語意記憶：使用者設定檔
await memory.semantic.upsert( namespace="user_profile", key=user_id, content={ "name": "Alice", "preferences": ["dark mode", "concise responses"], "expertise_level": "developer", } )
情節記憶：過去的互動
await memory.episodic.add( namespace="conversations", content={ "timestamp": datetime.now(), "summary": "Helped debug authentication issue", "outcome": "resolved", "key_insights": ["Token expiry was root cause"], }, metadata={"user_id": user_id, "topic": "debugging"} )
程序記憶：學到的模式
await memory.procedural.add( namespace="skills", content={ "task_type": "debug_auth", "steps": ["Check token expiry", "Verify refresh flow"], "example_interaction": few_shot_example, } ) """
執行時期記憶檢索
""" async def prepare_context(user_id, query): # 取得使用者設定檔（語意） profile = await memory.semantic.get( namespace="user_profile", key=user_id )
# 尋找相關的過去經驗（情節）
similar_experiences = await memory.episodic.search(
namespace="conversations",
query=query,
filter={"user_id": user_id},
limit=3
)
# 尋找相關技能（程序）
relevant_skills = await memory.procedural.search(
namespace="skills",
query=query,
limit=2
)
return {
"profile": profile,
"past_experiences": similar_experiences,
"relevant_skills": relevant_skills,
}
"""
向量資料庫選擇模式
為你的使用案例選擇正確的向量資料庫
使用時機: 設定持久記憶儲存
向量資料庫選擇：
""" 決策矩陣：
| Pinecone | Qdrant | Weaviate | ChromaDB | pgvector | |
|---|---|---|---|---|---|
| 規模 | 數十億 | 1億+ | 1億+ | 100萬 | 100萬 |
| 受管 | 是 | 兩者 | 兩者 | 自行 | 自行 |
| 過濾 | 基本 | 最佳 | 良好 | 基本 | SQL |
| 混合 | 否 | 是 | 最佳 | 否 | 是 |
| 成本 | 高 | 中 | 中 | 免費 | 免費 |
| 延遲 | 5毫秒 | 7毫秒 | 10毫秒 | 20毫秒 | 15毫秒 |
| """ |
Pinecone (企業級規模)
""" from pinecone import Pinecone
pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"]) index = pc.Index("agent-memory")
使用中繼資料寫入
index.upsert( vectors=[ { "id": f"memory-{uuid4()}", "values": embedding, "metadata": { "user_id": user_id, "timestamp": datetime.now().isoformat(), "type": "episodic", "content": memory_text, } } ], namespace=namespace )
使用過濾查詢
results = index.query( vector=query_embedding, filter={"user_id": user_id, "type": "episodic"}, top_k=5, include_metadata=True ) """
Qdrant (複雜過濾)
""" from qdrant_client import QdrantClient from qdrant_client.models import PointStruct, Filter, FieldCondition
client = QdrantClient(url="http://localhost:6333")
使用 Qdrant 進行複雜過濾
results = client.search( collection_name="agent_memory", query_vector=query_embedding, query_filter=Filter( must=[ FieldCondition(key="user_id", match={"value": user_id}), FieldCondition(key="type", match={"value": "semantic"}), ], should=[ FieldCondition(key="topic", match={"any": ["auth", "security"]}), ] ), limit=5 ) """
ChromaDB (原型開發)
""" import chromadb
client = chromadb.PersistentClient(path="./memory_db") collection = client.get_or_create_collection("agent_memory")
簡單快速，適合原型
collection.add( ids=[str(uuid4())], embeddings=[embedding], documents=[memory_text], metadatas=[{"user_id": user_id, "type": "episodic"}] )
results = collection.query( query_embeddings=[query_embedding], n_results=5, where={"user_id": user_id} ) """
區塊切割策略模式
將文件切割成可檢索的區塊
使用時機: 處理文件以儲存到記憶中
區塊切割策略：
""" 區塊切割的兩難：
- 太大：向量失去特異性
- 太小：失去上下文
最佳區塊大小取決於：
- 文件類型（程式碼 vs 散文 vs 資料）
- 查詢模式（事實型 vs 探索型）
- 嵌入模型（每個模型都有最佳區間）
一般建議：大多數用例為 256-512 個 tokens """
固定大小切割（基準）
""" from langchain.text_splitter import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter( chunk_size=500, # 字元數 chunk_overlap=50, # 重疊可防止切斷句子 separators=["\n\n", "\n", ". ", " ", ""] # 優先順序 )
chunks = splitter.split_text(document) """
語意切割（更佳品質）
""" from langchain_experimental.text_splitter import SemanticChunker from langchain_openai import OpenAIEmbeddings
基於語意相似度進行分割
splitter = SemanticChunker( embeddings=OpenAIEmbeddings(), breakpoint_threshold_type="percentile", breakpoint_threshold_amount=95 )
chunks = splitter.split_text(document) """
結構感知切割（具層級結構的文件）
""" from langchain.text_splitter import MarkdownHeaderTextSplitter
尊重文件結構
splitter = MarkdownHeaderTextSplitter( headers_to_split_on=[ ("#", "Header 1"), ("##", "Header 2"), ("###", "Header 3"), ] )
chunks = splitter.split_text(markdown_doc)
每個區塊都帶有標題中繼資料，以提供上下文
"""
上下文感知切割（Anthropic 的方法）
"""
在嵌入前為每個區塊添加上下文
可減少 35% 的檢索失敗
def add_context_to_chunk(chunk, document_summary): context_prompt = f''' 文件摘要：{document_summary}
以下是此文件中的一個區塊：
{chunk}
'''
return context_prompt
嵌入已添加上下文的區塊，而非原始區塊
for chunk in chunks: contextualized = add_context_to_chunk(chunk, summary) embedding = embed(contextualized) store(chunk, embedding) # 儲存原始區塊，但嵌入已添加上下文的版本 """
程式碼專用切割
""" from langchain.text_splitter import Language, RecursiveCharacterTextSplitter
語言感知分割
python_splitter = RecursiveCharacterTextSplitter.from_language( language=Language.PYTHON, chunk_size=1000, chunk_overlap=200 )
尊重函式/類別邊界
chunks = python_splitter.split_text(python_code) """
背景記憶形成
非同步處理記憶以獲得更好的品質
使用時機: 你想要更高的召回率，但又不想拖慢互動
背景記憶形成：
""" 即時記憶提取會拖慢對話，並增加代理工具呼叫的複雜度。在對話結束後進行背景處理，可以產生更高品質的記憶。
模式：潛意識記憶形成 """
LangGraph 背景處理
""" from langgraph.graph import StateGraph from langgraph.checkpoint.postgres import PostgresSaver
async def background_memory_processor(thread_id: str): # 在對話結束或閒置後執行 conversation = await load_conversation(thread_id)
# 在沒有時間壓力的情況下提取洞察
insights = await llm.invoke('''
分析這段對話並提取：
1. 學到關於使用者的關鍵事實
2. 使用者透露的偏好
3. 已完成或待辦的任務
4. 使用者行為模式
請詳盡分析——這在背景執行。
對話內容：
{conversation}
''')
# 儲存到長期記憶
for insight in insights:
await memory.semantic.upsert(
namespace="user_insights",
key=generate_key(insight),
content=insight,
metadata={"source_thread": thread_id}
)
在對話結束或閒置超時時觸發
@on_conversation_idle(timeout_minutes=5) async def process_conversation(thread_id): await background_memory_processor(thread_id) """
記憶鞏固（如同睡眠）
"""
定期鞏固並去重記憶
async def consolidate_memories(user_id: str): # 取得使用者的所有記憶 memories = await memory.semantic.list( namespace="user_insights", filter={"user_id": user_id} )
# 尋找相似的記憶（潛在重複）
clusters = cluster_by_similarity(memories, threshold=0.9)
# 合併相似的記憶
for cluster in clusters:
if len(cluster) > 1:
merged = await llm.invoke(f'''
將這些相關的記憶合併為一個：
{cluster}
保留所有重要資訊。
''')
await memory.semantic.upsert(
namespace="user_insights",
key=generate_key(merged),
content=merged
)
# 刪除原始記憶
for old in cluster:
await memory.semantic.delete(old.id)
"""
記憶衰減模式
忘記舊的、不相關的記憶
使用時機: 記憶變得龐大，檢索速度變慢
記憶衰減：
""" 並非所有記憶都該永恆存在：
- 舊的偏好可能已過時
- 任務細節失去關聯性
- 衝突的記憶會混淆檢索
根據以下因素實作智慧型衰減：
- 新近性（何時建立/存取？）
- 頻率（多久被檢索一次？）
- 重要性（是核心事實還是細節？） """
基於時間的衰減
""" from datetime import datetime, timedelta
async def decay_old_memories(namespace: str, max_age_days: int): cutoff = datetime.now() - timedelta(days=max_age_days)
old_memories = await memory.episodic.list(
namespace=namespace,
filter={"last_accessed": {"$lt": cutoff.isoformat()}}
)
for mem in old_memories:
# 軟刪除（標記為已歸檔）
await memory.episodic.update(
id=mem.id,
metadata={"archived": True, "archived_at": datetime.now().isoformat()}
)
每週執行一次
async def scheduled_decay(): await decay_old_memories("conversations", max_age_days=90) await decay_old_memories("user_insights", max_age_days=365) """