在先前的對話程式中，是由呼叫端用 messages 變數手動累積歷史紀錄，每次呼叫 Graph 都要把完整的對話串列傳入。一旦同時有多個對話，呼叫端就必須自行維護每個對話的訊息變數，程式碼很快就會變得雜亂。

LangGraph 透過 Checkpointer 接手這些工作。呼叫端每次呼叫 Graph 時，只需提供對話識別碼（thread_id）與當前這輪的新訊息；LangGraph 會依 thread_id 自動載入上一輪的狀態快照、合併新訊息、執行節點，並在產生回覆後自動保存最新快照。

本篇對應的範例程式位於 langgraph-checkpointer。我們會先用記憶體儲存的 InMemorySaver 驗證 thread_id 如何自動延續與隔離對話，最後說明如何替換為 SQLite 儲存實體，讓狀態在程式重啟後依然能保留。

掛載 InMemorySaver 管理對話狀態

在 LangGraph 中啟用狀態管理，節點（Node）與連線（Edge）的寫法完全不變，只需要在 compile() 時傳入 Checkpointer 實體。Checkpointer 負責狀態的儲存與還原：執行前依 thread_id 自儲存層載入上一輪快照並合併新訊息，執行後再自動將包含回覆的最新狀態寫回儲存層。

1. 編譯時掛載 Checkpointer

我們在 build_graph 中引入 InMemorySaver：

from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage
from langgraph.checkpoint.base import BaseCheckpointSaver
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, MessagesState, StateGraph


def build_graph(model, checkpointer: BaseCheckpointSaver | None = None):
    def chatbot(state: MessagesState):
        response = model.invoke(state["messages"])
        return {"messages": [response]}

    builder = StateGraph(MessagesState)
    builder.add_node("chatbot", chatbot)
    builder.add_edge(START, "chatbot")
    builder.add_edge("chatbot", END)

    # 使用 InMemorySaver 將狀態保存在記憶體中
    saver = InMemorySaver() if checkpointer is None else checkpointer
    return builder.compile(checkpointer=saver)

builder.compile(checkpointer=saver) 會將 Checkpointer 綁定到圖的執行生命週期中。

2. 透過 thread_id 進行多輪對話

呼叫端執行 graph.invoke() 時，在 config 中指定 thread_id，且在 messages 中只需傳入使用者該輪最新的一則訊息，不需傳入過去的對話歷史：

model = ChatAnthropic(model="claude-haiku-4-5")
graph = build_graph(model)

config = {"configurable": {"thread_id": "session-101"}}

# 第一輪：只傳入第一則使用者訊息
result_1 = graph.invoke(
    {"messages": [HumanMessage(content="你好，我的客戶編號是 A-8899。")]},
    config=config,
)
print(result_1["messages"][-1].content)
# 輸出：你好！已為你記下客戶編號 A-8899。

# 第二輪：只傳入當前新提問（messages 串列只有 1 則訊息，完全不帶入 result_1 的歷史）
result_2 = graph.invoke(
    {"messages": [HumanMessage(content="請問我剛才說我的客戶編號是多少？")]},
    config=config,
)

# LangGraph 依據 session-101 自動載入前一輪紀錄並合併，State 累積為 4 則訊息
print(len(result_2["messages"]))
# 輸出：4

print(result_2["messages"][-1].content)
# 輸出：您的客戶編號是 A-8899。

第二輪呼叫時，傳給 invoke() 的 messages 串列只有長度 1（僅含最新提問）。但 LangGraph 依據 session-101 自動自 Checkpointer 載入第一輪對話（1 則 HumanMessage 與 1 則 AIMessage）並追加新提問，因此回傳的 State 訊息總數為 4 則，模型也能看見完整上下文並正確回答。

3. 切換 thread_id 隔離對話

當使用者開啟新對話時，呼叫端只要換一個全新的 thread_id，LangGraph 就會提供全新的空白狀態，不會混入其他 session 的紀錄：

new_config = {"configurable": {"thread_id": "session-102"}}

result_new = graph.invoke(
    {"messages": [HumanMessage(content="請問我剛才說我的客戶編號是多少？")]},
    config=new_config,
)
print(result_new["messages"][-1].content)
# 輸出：抱歉，我目前沒有您的客戶編號紀錄。請問您的編號是多少？

主動讀取狀態快照

如果不發送新訊息，只是需要查詢特定對話目前的完整狀態（例如網頁重新整理時載入歷史紀錄），可以直接呼叫 graph.get_state()：

config = {"configurable": {"thread_id": "session-101"}}
snapshot = graph.get_state(config)

for msg in snapshot.values["messages"]:
    print(f"[{msg.type}]: {msg.content}")

回傳的 snapshot.values 包含該對話當前 State 的所有欄位，snapshot.values["messages"] 即為已保存的歷史訊息串列。

清除特定 thread 的對話紀錄

LangGraph 的 MessagesState 預設使用追加型 Reducer。要清除特定 thread_id 的對話紀錄，不能直接覆蓋串列，必須透過 RemoveMessage 搭配 graph.update_state()：

from langchain_core.messages import RemoveMessage

config = {"configurable": {"thread_id": "session-101"}}
snapshot = graph.get_state(config)

# 為每則既有訊息產生 RemoveMessage
remove_ops = [RemoveMessage(id=msg.id) for msg in snapshot.values["messages"]]

# 更新狀態以清空訊息
graph.update_state(config, {"messages": remove_ops})

graph.update_state() 會將狀態更新視為一次新的快照寫入。執行後再次呼叫 graph.get_state(config).values["messages"]，回傳結果即為空串列。

若需求只是開啟一場全新的對話，直接換一個全新的 thread_id 會更直接；只有在必須保留同一個 thread_id（例如固定 Session 或固定頻道）且需要清空內容時，才需要使用 RemoveMessage。

對話歷史超出長度上限的處理方式

隨著對話輪次增加，messages 累積的訊息會逐漸消耗更多 Token，最終可能超出模型的上下文視窗上限（Context Window）。處理長度超出時有兩種策略：

1. 呼叫模型前裁剪上下文

如果 State 必須完整保留歷史紀錄（供後續查詢或稽核），但不能整個送進模型，可以在節點內呼叫模型前，使用 trim_messages 篩選送出的訊息：

from langchain_core.messages import trim_messages


def chatbot(state: MessagesState):
    # 只保留最後 10 則訊息傳給模型，且確保第一則是 HumanMessage
    selected_messages = trim_messages(
        state["messages"],
        max_tokens=10,
        token_counter=len,
        strategy="last",
        start_on="human",
    )
    response = model.invoke(selected_messages)
    return {"messages": [response]}

這項操作只改變傳給模型的上下文視窗，State 與 Checkpointer 中依然保有完整的對話歷史。

2. 在 State 中維持滑動視窗（以 RemoveMessage 剔除舊訊息）

如果連 State 都不需要永久保存過舊的對話，可以在節點回傳新回覆時，一併回傳最早訊息的 RemoveMessage：

def chatbot(state: MessagesState):
    response = model.invoke(state["messages"])

    # 超過 10 則時，自 State 剔除最舊的訊息
    max_messages = 10
    all_messages = state["messages"] + [response]
    overflow = len(all_messages) - max_messages

    removals = (
        [RemoveMessage(id=m.id) for m in all_messages[:overflow]]
        if overflow > 0
        else []
    )
    return {"messages": [response] + removals}

這樣每次 Checkpointer 保存狀態時，最早的訊息就會被自動刪除，避免儲存層資料量無限增長。

正式環境的狀態持久化：更換儲存實體

InMemorySaver 將檢查點保存在記憶體中，適合本機開發與測試。在正式環境中，伺服器重啟或多實例部署需要跨行程保存狀態。

由於 LangGraph 將狀態儲存抽象為 BaseCheckpointSaver 介面，Graph 的 Node 與 Edge 定義完全不需修改，只需在 compile() 時抽換 Checkpointer 實體：

import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver

# 本機持久化：寫入 SQLite 檔案
conn = sqlite3.connect("checkpoints.db", check_same_thread=False)
graph = build_graph(model, checkpointer=SqliteSaver(conn))

若部署在分散式架構，可改用 langgraph-checkpoint-postgres 套件提供的 PostgresSaver，將狀態寫入 PostgreSQL 資料庫。

最後，小結一下。透過 Checkpointer：

1. 呼叫介面更乾淨：呼叫端每次只需傳入當前這輪的新訊息與 thread_id，不需手動累積 messages。
2. 多 Session 隔離：更換 thread_id 即可切換獨立對話環境。
3. 支援狀態查詢、清空與長度控制：隨時可用 get_state() 取得歷史快照，並透過 RemoveMessage 或 trim_messages 管理對話長度與清空紀錄。
4. 儲存媒介可抽換：開發時用 InMemorySaver，正式上線可直接替換為資料庫儲存。

下一篇，我們會把固定流程拆成多個職責不同的 Node，並用自訂 State 傳遞表單資料與處理結果。