2026年AI Agent記憶架構設計完全指南
AI与大数据
2026年AI Agent記憶架構設計完全指南
如果你在2026年還在把AI Agent當成「無狀態的一問一答」，那你已經落後了。現實是：記憶系統才是AI Agent真正的瓶頸。大模型本身的能力已經足夠強大，但一個沒有記憶的Agent，就像一個失憶的天才——每次對話都要從零開始，永遠無法累積經驗、形成偏好、理解上下文。
2025年下半年開始，業界對Agent記憶架構的關注度急劇上升。LangGraph推出了原生的Memory模組，MemGPT專案正式併入LangChain生態，各大廠商紛紛發布自己的Agent記憶方案。本文將系統性地拆解AI Agent的四種記憶類型，給出完整的LangGraph實作程式碼，並分享生產環境中的實戰經驗。
為什麼記憶是Agent的核心瓶頸？
先看一組資料對比：
| 記憶類型 | 容量 | 持久性 | 檢索延遲 | 典型實作 |
|---|---|---|---|---|
| 感知記憶 | 極小（當前輸入） | 毫秒級 | ~1ms | 原始輸入緩衝 |
| 工作記憶 | 小（4-32K tokens） | 會話級 | ~10ms | 對話歷史視窗 |
| 情景記憶 | 中（事件片段） | 天-月級 | ~50ms | 向量資料庫+時間索引 |
| 長期記憶 | 大（知識+偏好） | 永久級 | ~100ms | 向量資料庫+知識圖譜 |
一個成熟的Agent系統必須同時管理這四種記憶，並在它們之間高效地流轉資訊。接下來我們逐一拆解。
一、感知記憶（Sensory Memory）
感知記憶是Agent接收原始輸入的第一層緩衝。它的作用類似於人類的感官記憶——短暫地保留原始訊號，供後續模組提取特徵。
核心特點：
- 生命週期極短，通常只在一個推理步驟內有效
- 保留原始輸入的完整資訊（文字、影像、音訊等）
- 不做任何壓縮或抽象
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, Optional
@dataclass
class SensoryMemory:
raw_input: Any
input_type: str # "text", "image", "audio", "multimodal"
timestamp: datetime = field(default_factory=datetime.now)
metadata: dict = field(default_factory=dict)
def extract_features(self) -> dict:
if self.input_type == "text":
return {
"length": len(self.raw_input),
"has_code": "```" in self.raw_input,
"language_hint": self._detect_language(),
}
return {}
def _detect_language(self) -> str:
chinese_chars = sum(1 for c in self.raw_input if '\u4e00' <= c <= '\u9fff')
if chinese_chars / max(len(self.raw_input), 1) > 0.3:
return "zh"
return "en"
感知記憶通常不需要持久化，但建議在除錯模式下保留原始輸入的雜湊值，方便問題追溯。
二、工作記憶（Working Memory）
工作記憶是Agent當前推理的「草稿紙」，對應對話上下文視窗。2026年的關鍵挑戰是：如何在有限的token預算內，保留最相關的資訊？
策略對比：
| 策略 | 原理 | 優點 | 缺點 |
|---|---|---|---|
| 滑動視窗 | 保留最近N輪對話 | 簡單高效 | 遺失早期重要資訊 |
| 摘要壓縮 | 對舊對話生成摘要 | 節省token | 摘要可能遺失細節 |
| 重要性評分 | 按重要性選擇性保留 | 精準 | 需要額外LLM呼叫 |
| 混合策略 | 摘要+滑動視窗 | 兼顧效率與品質 | 實作複雜度較高 |
推薦：混合策略——對最近3輪保持原文，3-10輪生成摘要，10輪以上只保留關鍵決策點。
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage
from langgraph.graph.message import MessagesState
class WorkingMemoryManager:
def __init__(self, llm, recent_window: int = 3, summary_window: int = 10):
self.llm = llm
self.recent_window = recent_window
self.summary_window = summary_window
self._summary_cache: str = ""
def compress(self, messages: list) -> list:
if len(messages) <= self.recent_window:
return messages
recent = messages[-self.recent_window:]
older = messages[:-self.recent_window]
if len(older) > self.summary_window:
key_decisions = [m for m in older if self._is_key_decision(m)]
to_summarize = older[-self.summary_window:]
else:
key_decisions = []
to_summarize = older
if to_summarize:
self._summary_cache = self._generate_summary(to_summarize)
result = []
if self._summary_cache:
result.append(SystemMessage(content=f"[對話摘要] {self._summary_cache}"))
result.extend(key_decisions)
result.extend(recent)
return result
def _is_key_decision(self, message) -> bool:
keywords = ["決定", "確認", "選擇", "decided", "confirmed", "chose"]
return any(kw in message.content for kw in keywords)
def _generate_summary(self, messages: list) -> str:
conversation = "\n".join(
f"{'使用者' if isinstance(m, HumanMessage) else '助手'}: {m.content}"
for m in messages
)
prompt = f"請用2-3句話總結以下對話的關鍵資訊：\n{conversation}"
return self.llm.invoke(prompt).content
三、情景記憶（Episodic Memory）
情景記憶記錄Agent的「經歷」——特定事件、互動結果和上下文片段。它讓Agent能回憶起「上次使用者遇到類似問題時，我是怎麼解決的」。
設計要點：
- 每條記錄包含：事件描述、時間戳、上下文、結果、相關度評分
- 檢索時同時考慮語意相似度和時間衰減
- 定期合併相似事件，避免冗餘
from datetime import datetime, timedelta
import math
@dataclass
class EpisodicRecord:
event: str
context: str
outcome: str
timestamp: datetime
importance: float = 0.5
embedding: list[float] = field(default_factory=list)
class EpisodicMemory:
def __init__(self, vector_store, time_decay_factor: float = 0.1):
self.vector_store = vector_store
self.time_decay_factor = time_decay_factor
async def store(self, record: EpisodicRecord):
record.embedding = await self.vector_store.embed(record.event)
await self.vector_store.add(
text=record.event,
embedding=record.embedding,
metadata={
"context": record.context,
"outcome": record.outcome,
"timestamp": record.timestamp.isoformat(),
"importance": record.importance,
}
)
async def recall(self, query: str, top_k: int = 5) -> list[EpisodicRecord]:
query_embedding = await self.vector_store.embed(query)
results = await self.vector_store.search(query_embedding, top_k=top_k * 2)
now = datetime.now()
scored = []
for r in results:
ts = datetime.fromisoformat(r.metadata["timestamp"])
days_ago = (now - ts).days
time_score = math.exp(-self.time_decay_factor * days_ago)
importance = r.metadata.get("importance", 0.5)
final_score = r.score * time_score * importance
scored.append((final_score, r))
scored.sort(key=lambda x: x[0], reverse=True)
return [r for _, r in scored[:top_k]]
四、長期記憶（Long-term Memory）
長期記憶是Agent的「知識庫+人格」，包含使用者偏好、領域知識和行為模式。這是最複雜也最有價值的記憶類型。
三種長期記憶子類型：
| 子類型 | 內容 | 更新頻率 | 範例 |
|---|---|---|---|
| 語意記憶 | 事實性知識 | 低 | "Python用縮排表示程式碼區塊" |
| 程序記憶 | 技能和流程 | 中 | "使用者喜歡先看範例再看解釋" |
| 偏好記憶 | 使用者個人化設定 | 高 | "使用者偏好繁體中文回覆" |
@dataclass
class LongTermMemoryEntry:
content: str
memory_type: str # "semantic", "procedural", "preference"
confidence: float
last_accessed: datetime
access_count: int = 0
source: str = "learned" # "learned", "explicit", "inferred"
class LongTermMemory:
def __init__(self, vector_store, knowledge_graph=None):
self.vector_store = vector_store
self.knowledge_graph = knowledge_graph
self._cache: dict[str, LongTermMemoryEntry] = {}
async def learn(self, entry: LongTermMemoryEntry):
embedding = await self.vector_store.embed(entry.content)
await self.vector_store.add(
text=entry.content,
embedding=embedding,
metadata={
"memory_type": entry.memory_type,
"confidence": entry.confidence,
"source": entry.source,
}
)
if self.knowledge_graph and entry.memory_type == "semantic":
await self.knowledge_graph.add_fact(entry.content)
async def retrieve(self, query: str, memory_types: list[str] = None) -> list:
filters = {}
if memory_types:
filters["memory_type"] = {"$in": memory_types}
results = await self.vector_store.search(
await self.vector_store.embed(query),
filters=filters,
top_k=10,
)
return results
async def consolidate(self):
entries = await self.vector_store.get_all()
groups = self._group_similar(entries, threshold=0.92)
for group in groups:
if len(group) > 1:
merged = self._merge_entries(group)
for old in group:
await self.vector_store.delete(old.id)
await self.learn(merged)
完整LangGraph實作
下面是一個整合四種記憶的完整Agent實作：
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import MessagesState
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
class AgentMemoryState(MessagesState):
sensory: SensoryMemory
episodic_results: list
long_term_results: list
user_preferences: dict
def perceive(state: AgentMemoryState) -> dict:
last_message = state["messages"][-1]
sensory = SensoryMemory(
raw_input=last_message.content,
input_type="text",
)
return {"sensory": sensory}
def recall_memories(state: AgentMemoryState, config: dict) -> dict:
query = state["sensory"].raw_input
user_id = config["configurable"]["user_id"]
episodic_memory = config["configurable"]["episodic_memory"]
long_term_memory = config["configurable"]["long_term_memory"]
episodic_results = await episodic_memory.recall(query, top_k=3)
long_term_results = await long_term_memory.retrieve(
query, memory_types=["preference", "procedural"]
)
user_prefs = await long_term_memory.retrieve(
"user preferences", memory_types=["preference"]
)
return {
"episodic_results": episodic_results,
"long_term_results": long_term_results,
"user_preferences": {r.text: r.metadata for r in user_prefs},
}
def generate_response(state: AgentMemoryState, config: dict) -> dict:
llm = config["configurable"]["llm"]
working_memory_mgr = config["configurable"]["working_memory_mgr"]
compressed = working_memory_mgr.compress(state["messages"])
memory_context = ""
if state.get("episodic_results"):
memory_context += "\n[相關歷史經驗]\n"
for r in state["episodic_results"][:3]:
memory_context += f"- {r.text} → {r.metadata.get('outcome', '')}\n"
if state.get("long_term_results"):
memory_context += "\n[相關知識]\n"
for r in state["long_term_results"][:3]:
memory_context += f"- {r.text}\n"
enhanced_messages = []
if memory_context:
enhanced_messages.append(SystemMessage(content=memory_context))
enhanced_messages.extend(compressed)
response = llm.invoke(enhanced_messages)
return {"messages": [response]}
def store_experience(state: AgentMemoryState, config: dict) -> dict:
episodic_memory = config["configurable"]["episodic_memory"]
query = state["messages"][-2].content if len(state["messages"]) >= 2 else ""
response = state["messages"][-1].content
record = EpisodicRecord(
event=query,
context=state.get("sensory", {}).raw_input if state.get("sensory") else "",
outcome=response,
timestamp=datetime.now(),
importance=0.5,
)
await episodic_memory.store(record)
return {}
def build_memory_agent():
graph = StateGraph(AgentMemoryState)
graph.add_node("perceive", perceive)
graph.add_node("recall", recall_memories)
graph.add_node("respond", generate_response)
graph.add_node("store", store_experience)
graph.add_edge(START, "perceive")
graph.add_edge("perceive", "recall")
graph.add_edge("recall", "respond")
graph.add_edge("respond", "store")
graph.add_edge("store", END)
checkpointer = MemorySaver()
return graph.compile(checkpointer=checkpointer)
記憶持久化與向量儲存
生產環境中，記憶必須持久化到外部儲存。以下是使用Chroma和PostgreSQL的實作方案：
import chromadb
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
import asyncpg
class PersistentMemoryStore:
def __init__(self, chroma_path: str, pg_dsn: str):
self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
self.chroma = Chroma(
persist_directory=chroma_path,
embedding_function=self.embeddings,
)
self.pg_dsn = pg_dsn
async def init_pg(self):
self.pg_pool = await asyncpg.create_pool(self.pg_dsn)
await self.pg_pool.execute("""
CREATE TABLE IF NOT EXISTS agent_memory (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
user_id TEXT NOT NULL,
memory_type TEXT NOT NULL,
content TEXT NOT NULL,
importance FLOAT DEFAULT 0.5,
created_at TIMESTAMPTZ DEFAULT NOW(),
accessed_at TIMESTAMPTZ DEFAULT NOW(),
access_count INT DEFAULT 0,
metadata JSONB DEFAULT '{}'
);
CREATE INDEX IF NOT EXISTS idx_memory_user_type
ON agent_memory(user_id, memory_type);
CREATE INDEX IF NOT EXISTS idx_memory_accessed
ON agent_memory(accessed_at DESC);
""")
async def save(self, user_id: str, memory_type: str, content: str,
importance: float = 0.5, metadata: dict = None):
await self.pg_pool.execute(
"""INSERT INTO agent_memory
(user_id, memory_type, content, importance, metadata)
VALUES ($1, $2, $3, $4, $5)""",
user_id, memory_type, content, importance,
json.dumps(metadata or {})
)
await self.chroma.aadd_texts(
texts=[content],
metadatas=[{"user_id": user_id, "type": memory_type}],
)
async def search(self, user_id: str, query: str, top_k: int = 5) -> list:
results = await self.chroma.asimilarity_search(
query, k=top_k,
filter={"user_id": user_id}
)
await self.pg_pool.execute(
"""UPDATE agent_memory SET accessed_at = NOW(), access_count = access_count + 1
WHERE user_id = $1 AND content = ANY($2)""",
user_id, [r.page_content for r in results]
)
return results
5個常見陷阱
| # | 陷阱 | 後果 | 解決方案 |
|---|---|---|---|
| 1 | 把所有對話歷史當工作記憶 | Token超限、成本飆升 | 使用混合壓縮策略 |
| 2 | 忽略記憶的時間衰減 | 檢索到過時資訊 | 加入指數時間衰減因子 |
| 3 | 長期記憶不做合併 | 冗餘記憶堆積 | 定期執行consolidate |
| 4 | 向量檢索不考慮使用者隔離 | 使用者A看到使用者B的記憶 | 所有查詢加user_id過濾 |
| 5 | 情景記憶和長期記憶混淆 | 該記住的沒記住 | 嚴格區分事件性vs知識性 |
10個常見錯誤排查
| # | 錯誤現象 | 可能原因 | 排查方法 |
|---|---|---|---|
| 1 | Agent「忘記」了之前的對話 | 工作記憶視窗太小 | 檢查recent_window和summary_window設定 |
| 2 | 檢索到不相關的記憶 | Embedding模型不匹配 | 確認儲存和檢索使用同一模型 |
| 3 | 記憶檢索延遲過高 | 向量庫索引未最佳化 | 檢查HNSW參數，考慮量化 |
| 4 | Chroma資料遺失 | 未呼叫persist() | 確認persist_directory設定正確 |
| 5 | PostgreSQL連線池耗盡 | 未限制並發數 | 設定asyncpg.create_pool(max_size=20) |
| 6 | 長期記憶越存越多 | 缺少consolidate | 設定排程任務執行合併 |
| 7 | 使用者偏好不生效 | 偏好檢索未加入prompt | 檢查generate_response中是否注入偏好 |
| 8 | 跨會話記憶遺失 | Checkpointer未持久化 | 使用SqliteSaver或PostgresSaver |
| 9 | 記憶內容包含敏感資訊 | 未做脫敏處理 | 儲存前執行PII偵測和脫敏 |
| 10 | LangGraph狀態序列化失敗 | 自訂物件不可pickle | 使用dataclass或Pydantic模型 |
進階最佳化技巧
1. 分層快取
對高頻存取的記憶加LRU快取，避免每次都查向量庫：
from functools import lru_cache
@lru_cache(maxsize=256)
def get_user_preferences(user_id: str) -> dict:
return await long_term_memory.retrieve(
"user preferences", memory_types=["preference"]
)
2. 非同步預載入
在使用者發起請求前，預載入其長期記憶到快取：
async def preload_user_memory(user_id: str):
prefs = await long_term_memory.retrieve("preferences", memory_types=["preference"])
cache.set(f"prefs:{user_id}", prefs, ttl=3600)
3. 記憶評分衰減
對長期未存取的記憶降低檢索權重，模擬人類遺忘曲線：
def compute_relevance(entry, current_time):
days_since_access = (current_time - entry.last_accessed).days
access_bonus = math.log1p(entry.access_count)
time_decay = math.exp(-0.05 * days_since_access)
return entry.importance * access_bonus * time_decay
4. 多模態記憶
2026年的Agent需要處理文字、影像、音訊等多種模態的記憶。使用統一的embedding空間進行跨模態檢索：
from langchain_openai import OpenAIEmbeddings
multimodal_embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
工具推薦
在建構Agent記憶系統的過程中，以下工具可以幫你處理資料格式和編碼問題：
- JSON格式化工具 — 處理記憶元資料的JSON序列化和除錯，確保儲存結構正確
- Base64編碼工具 — 對二進制記憶資料（如影像embedding）進行編碼傳輸
- 雜湊計算工具 — 為記憶條目生成唯一指紋，用於去重和變更偵測
總結：AI Agent的記憶架構不是「錦上添花」，而是「不可或缺」。四種記憶類型各有分工——感知記憶抓輸入，工作記憶管推理，情景記憶記經歷，長期記憶存知識。用LangGraph的StateGraph把它們串起來，配上向量庫做持久化，加上時間衰減和合併策略做最佳化，你就擁有了2026年最前沿的Agent記憶系統。記住：一個有記憶的Agent，才是一個真正的Agent。
本站提供瀏覽器本地工具，免註冊即可試用 →
#AI Agent#记忆架构#长期记忆#工作记忆#LangGraph#2026