呼叫大型語言模型(LLM)的 API 是無狀態(Stateless)的。在先前的 讓 LangGraph 接手對話狀態管理 中,我們已使用 Checkpointer 依據 thread_id 實現了單一對話 Session 的短期狀態管理。
除了單次對話的短期狀態,真實應用中的 Agent 同樣需要能跨越 Session、持久保存使用者偏好的長期記憶。兩者的生命週期與資料範圍不同,識別碼也不同:短期記憶以 thread_id 識別,換了新的 thread_id 就代表開啟新對話,對話紀錄重置為空白,避免舊話題干擾新任務;長期記憶以 user_id 識別,同一名使用者不論開啟多少次新對話,都能存取相同的設定。
以旅遊客服為例,旅客在對話中提到的「偏好靠窗座位」或「習慣家庭旅遊」,屬於長久有效的使用者屬性。長期記憶會將這些結論提取出來寫入掛在 user_id 底下的 Profile(可以是在使用者說「請記住」時透過 Tool 即時寫入,或在對話 Session 結束時由模型抽取存入)。當同一個旅客換了新的 thread_id 詢問新行程時,上一趟旅行的景點對話已被清空,但 Agent 依然能從長期記憶載入偏好,自動安排靠窗座位與適合家庭的行程。
在規劃 Agent 的長期記憶時,我們不需要將整段聊天歷史原封不動存進長期資料庫,否則既浪費空間,又會讓每次讀取的上下文膨脹。
因為這是專門處理旅遊諮詢的 Agent,業務上記錄的資訊範圍非常明確:旅客的座位偏好、旅遊類型(如家庭旅遊、獨旅)與回覆語言。對於這種特定領域的 Agent,我們會事先定義好強型別的 Schema,確保只有符合預期的屬性才能進入長期記憶庫。
範例中定義的 TravelerProfile 只允許三項明確的業務欄位進入長期記憶庫:
seat_preference):允許 window(靠窗)、aisle(靠走道)、no_preference。travel_style):允許 family(家庭旅遊)、solo(獨自旅遊)、couple(雙人旅遊)、business(商務)、no_preference。response_language):允許 zh-TW(繁中)、en(英文)、ja(日文)。使用 Pydantic 配合 Literal 枚舉型別與 extra="forbid",能防止 LLM 生成像「想坐舒服一點」這種非標準、無法對接 API 的自由文字,確保記憶庫欄位品質:
class TravelerProfile(BaseModel):
model_config = ConfigDict(extra="forbid")
seat_preference: Literal["window", "aisle", "no_preference"] | None = None
travel_style: (
Literal["family", "solo", "couple", "business", "no_preference"]
| None
) = None
response_language: Literal["zh-TW", "en", "ja"] | None = None
為避免不同使用者的偏好互相混淆,Store 採用命名空間(Namespace)建立隔離邊界:
def profile_namespace(user_id: str) -> tuple[str, ...]:
return ("travelers", user_id, "profile")
每輪對話執行時,chatbot 節點會先透過 user_id 從 Store 讀取該名使用者的 Profile,並將其轉化為 System Prompt 的一部分注入模型:
def chatbot(state: MessagesState, runtime: Runtime[Context]):
profile = read_profile(runtime.store, runtime.context.user_id)
profile_prompt = render_profile(profile)
system_prompt = BASE_SYSTEM_PROMPT
if profile_prompt:
system_prompt = f"{system_prompt}\n\n{profile_prompt}"
response = model_with_tools.invoke(
[SystemMessage(content=system_prompt), *state["messages"]]
)
return {"messages": [response]}
透過這種方式,無論開啟多少個新的 thread_id,只要 user_id 保持一致,模型都能掌握使用者的長遠偏好。
將偏好存入 Store 有兩條路徑:
當使用者說出「請記住」或「以後都要」等明確指令時,模型的最佳選擇是呼叫 Tool。定義寫入偏好的 Tool:
@tool
def save_traveler_preference(field: PreferenceField, value: str) -> str:
"""保存旅客明確要求跨旅行沿用的偏好。"""
return f"已保存旅客偏好:{field}={value}"
當模型判定需要記錄偏好時,會觸發 Tool 呼叫,Graph 路由至 save_preference 節點:驗證欄位是否合規、與既有 Profile 合併,並寫回 Store。當輪對話便能立即看到更新後的記憶生效。
使用者在聊天過程中未必每次都會說「請記住」。當輸入 /new 或 /exit 結束對話 Session 時,系統會取得該 Session 的完整對話紀錄,傳給設定了結構化輸出的 Extractor 模型:
snapshot = graph.get_state(config)
messages = snapshot.values["messages"]
result = extractor.invoke(
[
SystemMessage(content=EXTRACTION_SYSTEM_PROMPT),
HumanMessage(content=conversation_text(messages)),
]
)
update = TravelerProfileUpdate.model_validate(result)
Extractor Prompt 會過濾掉目的地、日期與預算等一次性需求,只擷取符合 TravelerProfile 的通用偏好(如家庭旅遊風格),並自動更新至 Store。這讓 Agent 不用在每輪對話頻繁呼叫工具,也能在背景補齊使用者畫像。
範例專案的完整實作位於 langgraph-traveler-memory。
定義好負責對話的 chatbot 節點、負責寫入偏好的 save_preference 節點與 Store 之後,最後一步是將它們組裝進 StateGraph 並完成編譯。
在 LangGraph 中,節點無法憑空存取外部儲存;我們必須在 compile() 時將 store 注入 Graph,節點的 runtime.store 才能正常運作。同時,為了維持多輪對話上下文,我們也會一併掛上負責短期記憶的 checkpointer。
在 main.py 的 build_graph() 函式中,我們宣告圖的拓撲結構,並在 compile() 時同時注入兩組儲存後端:
# 位於 main.py 的 build_graph()
def build_graph(model: Any, *, checkpointer: Any, store: BaseStore) -> Any:
...
builder = StateGraph(MessagesState, context_schema=Context)
builder.add_node("chatbot", chatbot)
builder.add_node("save_preference", save_preference)
builder.add_edge(START, "chatbot")
builder.add_conditional_edges("chatbot", route_after_model)
builder.add_edge("save_preference", "chatbot")
# 同時掛載 Checkpointer(短期)與 Store(長期)
return builder.compile(
checkpointer=checkpointer,
store=store,
)
main() 中呼叫 graph.invoke()在 main.py 的進入點函式 main() 中,實體化儲存物件並執行對話迴圈。呼叫 graph.invoke() 時,兩組識別鍵值必須透過不同通道傳入:
# 位於 main.py 的 main() 進入點
store = InMemoryStore()
graph = build_graph(
model,
checkpointer=InMemorySaver(),
store=store,
)
# 呼叫時雙通道分流:
result = graph.invoke(
{"messages": [HumanMessage(content=question)]},
{"configurable": {"thread_id": session_id}}, # 短期對話通道(給 Checkpointer)
context=Context(user_id=user_id), # 長期儲存通道(給 Store)
)
configurable={"thread_id": session_id}:短期對話通道。傳給 Checkpointer 用於定位該次 Session 的狀態快照。context=Context(user_id=user_id):長期儲存通道。傳入執行環境,供節點從 Store 存取該名使用者的 Profile。這套組合正是實現「清空舊對話、保留舊偏好」的關鍵:
thread_id 自動載入上一輪 Checkpoint 狀態,追加本輪傳入的 HumanMessage,執行節點並更新 State,執行完畢後自動將最新 State 寫回 Checkpoint。chatbot 節點執行時,依據傳入的 user_id 從 Store 讀取長久 Profile,將其動態渲染注入 System Prompt;若對話中觸發 Tool,則由 save_preference 節點更新至 Store。thread_id 時,Checkpointer 找不到過去紀錄,自然會以全新的空白 State 開始執行,上一趟對話細節完全不外流;但因為 user_id 不變,Agent 依然能精準載入長期偏好!在開發與測試階段,使用 InMemorySaver 與 InMemoryStore 可以省去資料庫架設與設定檔維護。然而,記憶體後端僅存在於 Python 程序執行期間,關閉 CLI 或重啟伺服器後資料即會消失。
正式上線時,不需要修改 Graph 的架構邏輯或節點程式碼,只需將編譯時的後端替換為持久化資料庫套件(例如 PostgreSQL Saver 與 Store):
# 生產環境替換範例
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.store.postgres import PostgresStore
# checkpointer 與 store 改為連線至實體資料庫
graph = builder.compile(
checkpointer=postgres_checkpointer,
store=postgres_store,
)
不論底層是記憶體陣列還是 PostgreSQL 資料庫,thread_id 控制對話生命週期、user_id 隔離使用者記憶的核心原則皆維持不變。