跳到主要內容

AG Day 16 記憶：checkpointer 與 thread

AG Day 16 記憶：checkpointer 與 thread

執行需求：CPU 可跑。在昨天 AG Day 15（原文連結）中，我們把 research-agent 的工具執行邏輯拆開，改用手動組裝的 StateGraph 搭配 ToolNode，並且做好了暫時性錯誤與結構性錯誤的分流處理，讓代理面對外部服務不穩定時更有韌性。但不管昨天測試跑了幾次，每次呼叫 app.invoke(...) 都是一段全新的對話，代理完全不記得前一次研究進行到哪裡。今天要解決這個問題：讓 research-agent 具備跨輪、甚至跨行程重啟都能延續的記憶能力，靠的是 LangGraph 的檢查點（checkpointer）機制與 thread_id 概念。今天的範例全程可以在沒有任何 API 金鑰的情況下跑完。

引言

「記憶」這個詞在對話系統裡常被簡化成「把歷史訊息塞進 prompt」，但對一個要花好幾輪、甚至橫跨好幾天才能完成的研究任務來說，這樣還不夠。使用者可能今天問了一半，明天才想起來要接著問；系統也可能因為程式重新部署而重新啟動，這時候如果代理的所有狀態都只活在記憶體裡，一旦行程結束，使用者之前的研究進度就整個消失了。LangGraph 把「狀態怎麼被儲存與還原」這件事抽象成 checkpointer，它會在圖每執行完一個節點之後，把當下完整的狀態快照存下來；只要有這份快照，我們就能在任何時間點用同一個 thread_id 恢復執行，接著上一次停下的地方繼續。

thread_id 是理解這套機制的關鍵。你可以把它想成「一個對話串的身分證字號」：同一個 thread_id 底下的所有呼叫，會共用同一份不斷累積的狀態；換一個 thread_id，等於開了一個全新的、彼此獨立的對話。對 research-agent 而言，我們規劃讓每一個研究任務對應一個 thread_id，並讓它與我們在 AG Day 2（原文連結）建立的 runs 資料表裡的 run_id 一致，這樣就能把「LangGraph 內部的執行快照」與「我們自己記錄的任務中繼資料」串在同一把鑰匙下，方便日後查詢與除錯。

原理與觀念

checkpointer 的兩種常見實作：記憶體版與持久化版

LangGraph 提供了不只一種 checkpointer 實作。最簡單的是純記憶體版本（例如 InMemorySaver），狀態只存在目前這個 Python 行程的記憶體裡，行程一結束資料就消失，適合開發階段快速測試，但完全不適合正式環境。持久化版本則會把每一次的狀態快照寫進外部儲存，常見的選項包含 SQLite、Postgres 等關聯式資料庫的專用 saver。因為我們在 AG Day 2 已經選定 SQLite 作為 research-agent 的本機知識庫，今天延續同一個技術棧，改用 SQLite 版的 checkpointer，讓對話狀態與研究素材都落在同一類儲存機制上，維運起來更單純。

狀態快照裡到底存了什麼

每一次 checkpoint 存下來的，不只是訊息歷史，而是整個 StateGraph 當下的完整狀態物件（在我們的例子裡是 MessagesState，也就是累積的訊息串列），加上一些中繼資訊，例如目前執行到哪個節點、下一步預計要跑哪個節點。這代表如果代理在「呼叫工具」這一步被中斷（不管是程式當機、還是我們明天要介紹的人工核准中斷），下一次用同一個 thread_id 呼叫圖時，LangGraph 知道要從哪個節點接續，而不是傻傻地從頭重新問一次模型。

compile 時傳入 checkpointer，invoke 時傳入 thread_id

使用上分成兩個步驟：第一步，在 graph.compile(checkpointer=...) 時把 checkpointer 實例傳進去，讓編譯出來的圖知道要把每一步狀態存到哪裡；第二步，每次呼叫 app.invoke(...) 或 app.stream(...) 時，要在 config 參數裡帶上 {"configurable": {"thread_id": "..."}}，告訴這次呼叫屬於哪一條對話串。如果忘記帶 thread_id，多數 checkpointer 實作會丟出明確的錯誤或使用預設值，這是初學者最容易忽略、也最常在踩雷段落被提到的一步。

checkpoint 快照與 events 事件表的分工

看到這裡你可能會疑惑：我們在 AG Day 2 已經設計了 events 資料表，用來記錄每一次模型思考、工具呼叫與結果，這跟今天的 checkpoint 快照聽起來很像，是不是重工了？兩者的定位其實不同。checkpoint 快照是 LangGraph 框架內部的實作細節，格式與版本綁定，目的是「讓圖能被暫停、恢復、時間旅行」；events 資料表則是我們自己設計、格式穩定、可以直接下 SQL 查詢的觀測資料，目的是「讓開發者與使用者能看懂代理做了什麼」。實務上建議兩者並存：checkpoint 負責技術上的可恢復性，events 負責人類可讀的稽核紀錄，即使日後我們把底層框架從 LangGraph 換成別的方案，events 表的資料依然有效、依然能拿來做報表與除錯。這也是為什麼今天的 start_new_research 仍然呼叫 record_new_run 寫入 runs 表，而不是只依賴 checkpointer 就了事。

完整實作

今天我們把 graph.py 裡的 build_graph() 改成支援傳入 checkpointer，並在 storage.py 補上一個取得 checkpoint 用資料庫連線字串的輔助函式。先安裝需要的套件：

uv pip install langgraph-checkpoint-sqlite

第一步：修改 src/research_agent/graph.py，讓 build_graph 接受一個可選的 checkpointer 參數：

# research-agent/src/research_agent/graph.py（節錄修改處）
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode, tools_condition
from research_agent.tools import search_web, fetch_url
from research_agent.llm import call_model
from research_agent.graph_errors import handle_tool_error  # 沿用 AG Day 15 的錯誤處理函式

TOOLS = [search_web, fetch_url]
tool_node = ToolNode(TOOLS, handle_tool_errors=handle_tool_error)


def build_graph(checkpointer=None):
    graph = StateGraph(MessagesState)
    graph.add_node("agent", call_model)
    graph.add_node("tools", tool_node)
    graph.add_edge(START, "agent")
    graph.add_conditional_edges("agent", tools_condition, {"tools": "tools", END: END})
    graph.add_edge("tools", "agent")
    return graph.compile(checkpointer=checkpointer)

第二步：在 src/research_agent/memory.py 建立取得 SQLite checkpointer 的工廠函式，資料庫檔案獨立於 knowledge.db，放在 data/checkpoints.db，避免對話狀態與研究素材混在同一個檔案裡互相干擾：

# research-agent/src/research_agent/memory.py
import sqlite3
from pathlib import Path
from langgraph.checkpoint.sqlite import SqliteSaver

CHECKPOINT_DB = Path(__file__).resolve().parent.parent.parent / "data" / "checkpoints.db"


def get_checkpointer() -> SqliteSaver:
    """建立（或沿用）SQLite 版 checkpointer，第一次使用時自動建表。"""
    CHECKPOINT_DB.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(str(CHECKPOINT_DB), check_same_thread=False)
    saver = SqliteSaver(conn)
    saver.setup()  # 建立 checkpointer 所需的內部資料表，重複呼叫是安全的
    return saver

第三步：串起來，示範同一個 thread_id 呼叫兩次，第二次呼叫時模型能看到第一次的對話歷史：

# research-agent/verify_memory.py
from langchain_core.messages import HumanMessage
from research_agent.graph import build_graph
from research_agent.memory import get_checkpointer

checkpointer = get_checkpointer()
app = build_graph(checkpointer=checkpointer)

thread_id = "research-demo-001"
config = {"configurable": {"thread_id": thread_id}}

first = app.invoke({"messages": [HumanMessage(content="我想研究台灣的離岸風電發展")]}, config=config)
print("第一輪最終回覆：", first["messages"][-1].content[:60])

second = app.invoke({"messages": [HumanMessage(content="剛剛提到的主題，幫我列三個子題目")]}, config=config)
print("第二輪最終回覆：", second["messages"][-1].content[:60])
print("目前累積訊息數：", len(second["messages"]))

離線模式下的示範輸出：

第一輪最終回覆： （模型針對離岸風電主題的示範回應，實際內容視模型而定）
第二輪最終回覆： （模型記得上一輪主題，列出三個子題目的示範回應）
目前累積訊息數： 6

第四步：示範如何用 get_state 讀出目前某個 thread_id 的狀態快照，這在後面 Human-in-the-loop 與長任務章節都會反覆用到：

# research-agent/inspect_thread.py
from research_agent.graph import build_graph
from research_agent.memory import get_checkpointer

app = build_graph(checkpointer=get_checkpointer())
config = {"configurable": {"thread_id": "research-demo-001"}}

snapshot = app.get_state(config)
print("下一步預計執行的節點：", snapshot.next)
print("目前狀態中的訊息數：", len(snapshot.values["messages"]))
for msg in snapshot.values["messages"]:
    print(" -", type(msg).__name__, str(getattr(msg, "content", ""))[:40])

第五步：把 run_id 與 thread_id 對齊，讓 runs 資料表可以反查對應的檢查點資料庫。這裡示範在建立新任務時，直接拿 run_id 當 thread_id 使用：

# research-agent/src/research_agent/session.py
import uuid
from research_agent.storage import record_new_run
from research_agent.config import load_settings


def start_new_research(query: str) -> str:
    """建立一筆新的研究任務，回傳同時作為 run_id 與 thread_id 的識別碼。"""
    settings = load_settings()
    run_id = f"run_{uuid.uuid4().hex[:12]}"
    record_new_run(
        db_path=settings.database_path,
        run_id=run_id,
        query=query,
        model_name=settings.model_name,
    )
    return run_id

第六步：checkpoint 資料庫會隨著任務數量持續成長，我們在效能段落會提到清理策略，這裡先把「列出所有 thread 與最後活動時間」的查詢工具寫出來，讓你之後可以直接接上排程：

# research-agent/list_threads.py
import sqlite3
from research_agent.memory import CHECKPOINT_DB


def list_active_threads(limit: int = 20) -> list[tuple[str, str]]:
    """列出 checkpoints.db 裡的 thread_id 與最後一次寫入時間，供人工檢視或排程清理使用。"""
    conn = sqlite3.connect(str(CHECKPOINT_DB))
    try:
        rows = conn.execute(
            "SELECT thread_id, MAX(checkpoint_id) AS last_checkpoint "
            "FROM checkpoints GROUP BY thread_id "
            "ORDER BY last_checkpoint DESC LIMIT ?",
            (limit,),
        ).fetchall()
    finally:
        conn.close()
    return rows


if __name__ == "__main__":
    for thread_id, last_checkpoint in list_active_threads():
        print(f"{thread_id}｜最後檢查點：{last_checkpoint}")

這裡的資料表欄位名稱（checkpoints、checkpoint_id）是目前版本 SqliteSaver 內部使用的實作細節，實際欄位請以你安裝的 langgraph-checkpoint-sqlite 版本為準，不同版本可能有調整；重點是概念上「checkpointer 自己維護一份內部資料表，我們可以額外寫查詢工具去讀它」，而不是直接手動修改這份內部資料表的內容。

常見錯誤與踩雷

第一個常見錯誤是完全忘記傳 config 或漏帶 thread_id。有些 checkpointer 實作在沒有指定 thread_id 時會用固定的預設值（例如空字串），結果所有呼叫其實都共用同一條對話串，使用者以為自己開了新的研究，實際上卻讀到別的任務歷史。務必在每一個進入點（CLI、API）都明確產生並傳遞 thread_id，不要依賴預設值。

第二個是忘記呼叫 SqliteSaver 的 setup()，或是每次啟動都各自建立連線卻沒有共用同一個資料庫檔案路徑，導致每次執行都像是全新環境，狀態怎麼樣都串不起來。建議把 checkpointer 的初始化邏輯集中在一個工廠函式（就是我們今天寫的 get_checkpointer），確保整個應用程式只用同一份設定去開啟資料庫。

第三個是把 SQLite 連線用 check_same_thread=True（也就是預設值）建立，卻在多執行緒或非同步環境下共用同一個連線物件，這會直接觸發 sqlite3.ProgrammingError。我們在範例裡明確設了 check_same_thread=False，但這代表我們自己要負責確保不會有多個執行緒同時對同一個連線做寫入，正式環境建議搭配連線池或改用 Postgres 版的 checkpointer。

第四個雷是誤以為「有記憶」等於「上下文視窗不會爆」。checkpointer 只負責把狀態存起來、之後能還原，並不會自動幫你把訊息歷史做摘要或截斷。如果一個研究任務累積了幾十輪對話，訊息歷史會越疊越長，最終仍可能超出模型的上下文視窗限制。這個問題我們會在後面的長任務與效能最佳化篇章處理，今天先確保「記得住」這個基本能力到位。

第五個容易忽略的地方，是把 knowledge.db 與 checkpoints.db 混在同一份 SQLite 連線裡操作。雖然兩者都是 SQLite 檔案，但一個是我們自己寫的 schema、另一個是 checkpointer 框架內部維護的 schema，混在一起容易在交易（transaction）邊界上互相干擾，例如在同一個 with sqlite3.connect(...) 區塊裡同時寫 events 表跟觸發 checkpoint 寫入，若其中一邊拋出例外，另一邊可能已經寫入一半，造成兩份資料不同步。今天的設計刻意把兩個資料庫檔案分開（data/knowledge.db 與 data/checkpoints.db），讓兩套寫入各自獨立、互不牽連，這是有意的架構選擇，不是疏漏。

效能與實務提醒

持久化 checkpointer 的寫入頻率取決於圖裡節點的數量：圖每執行完一個節點就會寫一次快照，一輪「思考、呼叫工具、再思考」的完整迴圈，可能觸發三到四次資料庫寫入。對 SQLite 來說，這個寫入量在單機開發與中小型應用完全不是問題，但如果之後要支撐大量並行使用者，建議提早評估換成 Postgres 版的 checkpointer，或是把 PRAGMA journal_mode=WAL;（我們在 AG Day 2 已經替 knowledge.db 設定過同樣的模式）也套用在 checkpoints.db 上，降低讀寫互鎖的機率。

另一個實務考量是「檢查點要保留多久」。研究任務如果已經產出最終報告、使用者也確認滿意，繼續保留完整的逐輪狀態快照就只是佔用磁碟空間。建議在任務標記為完成時，另外把最終的訊息摘要寫進 runs.final_report 欄位（我們在 AG Day 2 已經預留這個欄位），詳細的逐輪快照則可以訂一個保留期限，定期清理，這部分會在後面的效能總檢章節具體示範清理腳本。

最後，開發階段建議先用記憶體版 checkpointer 快速迭代邏輯，等圖的行為穩定了再切換成 SQLite 版本做整合測試，因為持久化版本每次執行都會留下資料庫檔案，反覆手動刪除檔案來重置測試環境，容易忘記而汙染下一次的測試結果。建議在測試程式碼裡統一用 tmp_path（pytest 內建的暫存目錄夾具）產生獨立的 checkpoints.db，讓每個測試案例互不干擾，也不會不小心把測試資料寫進開發用的正式資料庫檔案。

還有一點常被低估：thread_id 的命名策略其實也是一種產品設計決策。如果你讓使用者可以自訂研究任務的名稱，並把這個名稱（或其雜湊值）當作 thread_id 的一部分，之後在 list_active_threads 這類工具裡就能顯示有意義的名稱，而不是一串隨機英數字。research-agent 目前用 run_{uuid} 的形式，優先確保唯一性；等到 AG Day 40 做使用者介面時，我們會在上層另外維護一個「顯示名稱」欄位，對應回這個內部識別碼，兩者分工，不必為了好記而犧牲唯一性。

小結

今天我們讓 research-agent 從「每次呼叫都失憶」進化成「能跨輪、甚至跨行程重啟延續對話」。關鍵是 LangGraph 的 checkpointer 機制：compile 時指定要用哪一種儲存實作，invoke 或 stream 時透過 thread_id 指定要接續哪一條對話串。我們選用 SQLite 版的 SqliteSaver，讓對話狀態與研究素材共用同一套技術棧，並把 thread_id 與 runs 資料表的 run_id 對齊，方便日後查詢與追溯。

今天新增的關鍵詞：檢查點（checkpoint）——圖每執行完一個節點後留存的完整狀態快照；thread_id——區分不同對話串、決定狀態要不要共用的識別碼；狀態快照（state snapshot）——透過 get_state 取得的某個時間點的完整狀態內容。這三個詞會在接下來的人工核准、子圖、長任務三個章節反覆出現，值得先花點時間確認自己真的分得清楚彼此的差異，而不只是背下名詞。

結語

有了記憶能力，代理現在可以放心地被中斷、之後再接續，這正是實作「人工核准」流程的前提：如果代理沒有記憶，中斷之後就沒有「接續」這件事可言，只能整個重來。

明天，我們會進入「AG Day 17 Human-in-the-loop：interrupt 與人工核准」，示範如何在代理準備執行高風險動作（例如呼叫會計費的外部服務、或寫入正式資料庫）之前主動中斷，等待人類明確核准後才繼續執行，而今天建立的 checkpointer 正是讓這個中斷、恢復流程得以運作的地基。

延伸資源

  • LangGraph 官方文件：Persistence／Checkpointer 概念與 SQLite、Postgres 版實作的參考頁面。
  • LangGraph 官方文件：thread 與 configurable 參數的使用說明。
  • Python 官方文件：sqlite3 模組的 check_same_thread 參數行為說明。
標籤：

留言

張貼留言

這個網誌中的熱門文章

Day 2 變數與資料型別

Day 2 變數與資料型別 引言 寫程式的過程中，變數與資料型別是處理資料的基礎。變數是存放資料的容器，資料型別則決定這筆資料有哪些特性、可以進行哪些操作。學會定義變數、認識各種資料型別，是學好 Python 的關鍵一步。 這篇文章會帶你了解 Python 中變數的觀念、如何定義變數，以及常見的資料型別，包括整數、浮點數、字串、布林值，還有串列、元組、字典與集合等容器型別。我們也會介紹變數的命名規則與撰寫風格建議，以及如何用 type() 檢查資料型別。 什麼是變數？如何在 Python 中定義變數 變數是在程式執行時用來存放資料的名稱。透過定義變數，我們可以給一筆資料一個名字，並在程式的其他地方用這個名字取用該筆資料。在 Python 中，變數不需要事先宣告型別，因為 Python 是動態型別語言，變數的型別由指定給它的值決定。 定義變數的基本語法 在 Python 中定義變數非常簡單，只要用賦值符號 = 把值指定給變數即可。例如： x = 5 # 定義變數 x，並把整數 5 賦值給它 name = "Alice" # 定義變數 name，並把字串 "Alice" 賦值給它 在這裡，x 是一個變數，被賦予整數 5；name 是另一個變數，被賦予字串 "Alice"。 變數的更新與覆寫 變數的值可以修改，也就是說，我們可以在程式的不同地方給同一個變數新的值。例如： x = 10 # x 最初被賦予 10 x = 15 # x 的值現在被更新為 15 這樣就能依照需求，在程式執行過程中靈活調整變數的值。 Python 的動態型別系統 Python 和某些靜態型別語言不同，定義變數時不需要宣告型別。賦值時，Python 會根據值自動判斷變數的型別。例如： x = 5 # x 是整數 x = 3.14 # x 變成浮點數 x = "Hi" # x 變成字串 同一個變數在程式執行過程中可以存放不同型別的值，這是 Python 的彈性之一。 常見資料型別 在 Python 中，資料型別決定我們可以對變數進行哪些操作...
張貼留言
閱讀完整內容

Day 1 Python 簡介與環境設定

Day 1 Python 簡介與環境設定 引言 在現在的科技環境裡，程式設計已經是一項重要技能。無論你是對資料科學有興趣、想成為開發者，或是想踏入人工智慧（AI）領域，學會寫程式都能明顯提升你的競爭力。在眾多程式語言中，Python 因為語法簡單、功能強大、應用範圍廣泛，成為許多人進入程式世界的第一選擇。這篇文章會帶你認識 Python 的背景與優勢，並一步步教你在不同系統上安裝與設定 Python 開發環境，最後寫出第一支 Python 程式。 為什麼選擇 Python？ Python 是一種高階程式語言，由 Guido van Rossum 在 1991 年發布。Python 的設計哲學強調程式碼的可讀性，並用縮排來定義程式區塊，這點和許多使用大括號的語言不同。簡潔的語法讓它成為初學者的理想選擇；就算是經驗豐富的開發者，也能用它完成複雜的專案。 Python 的優勢如下： 簡單易學 ：Python 的語法清楚、結構簡潔，初學者很快就能上手。和其他語言相比，學習曲線相對平緩，不需要先弄懂一堆複雜觀念，就能開始寫程式。 應用範圍廣泛 ：從資料科學、網頁開發、人工智慧、機器學習、自動化測試到網路爬蟲，Python 都有大量開源函式庫與工具支援，而且在這些領域都扮演關鍵角色。 豐富的函式庫與框架 ：Python 的函式庫生態系非常龐大。做資料分析有 NumPy、Pandas；開發網站有 Django、Flask；做深度學習有 TensorFlow、PyTorch。各種需求幾乎都能找到對應的套件，讓開發更有效率。 跨平台支援 ：Python 支援 Windows、macOS、Linux 等作業系統，程式通常不需要太多修改就能跨平台執行，讓開發與部署更有彈性。 活躍的社群 ：Python 擁有龐大的開發者社群。學習或開發上遇到問題，幾乎都能在社群與論壇（例如 Stack Overflow）找到答案，對初學者來說是很強的後盾，也能減少卡關時的挫折感。 Python 的應用領域 Python 的流行與強大功能，讓許多領域都開始大量使用它。以下是幾個常見的應用方向： 資料科學 ：隨著大數據與人工智慧興起，資料科學大量使用 Python。NumPy、Pandas 與 Matplotlib 等工具能處理和分析龐...
張貼留言
閱讀完整內容

Python 從入門到 PyTorch 深度學習：開啟 AI 世界的大門

Python 從入門到 PyTorch 深度學習：開啟 AI 世界的大門 隨著人工智慧（AI）與深度學習（Deep Learning）快速發展，越來越多人對這些技術產生興趣。不論你是想踏入 AI 領域的初學者，還是已經有程式基礎的開發者，學好 Python 與深度學習框架（例如 PyTorch），都能為你打開更多可能。 為什麼選擇 Python？ Python 已經是資料科學與人工智慧領域的首選語言。它的語法簡潔、容易上手，而且擁有龐大的生態系與大量開源函式庫。無論是資料處理、資料視覺化，還是建立機器學習與深度學習模型，Python 都能勝任。對想進入 AI 或資料科學領域的人來說，它幾乎是必備工具。 PyTorch 是什麼？ PyTorch 是由 Meta（原 Facebook）AI 研究團隊開發的開源深度學習框架，以易用、靈活和動態計算圖著稱，是許多 AI 研究人員與開發者的首選。相較於其他框架，PyTorch 的寫法更貼近原生 Python，對初學者相對友善。無論是簡單的實驗，還是複雜的深度學習模型，PyTorch 都能提供強大的支援。 這個系列能帶給你什麼？ 這個系列會從 Python 的基礎開始，帶你一步一步學習，最後能自己用 PyTorch 建立深度學習模型。即使你完全沒有寫過程式，也能跟著文章的節奏累積技能，理解 AI 與深度學習的核心觀念。 本系列涵蓋的主題 Python 基礎：從變數、條件判斷到函式與模組。 資料處理工具：用 NumPy 與 Pandas 有效率地操作資料。 資料視覺化：用 Matplotlib 與 Seaborn 把資料畫成圖表。 深度學習的數學基礎：線性代數、微積分與機率。 PyTorch 入門：理解張量、模型建構與 GPU 加速。 基礎深度學習模型：CNN 與 RNN 的實作應用。 深度學習專案實戰：從資料前處理到模型部署的端到端流程。 誰適合這個系列？ 程式初學者 ：如果你對 AI 充滿好奇，卻還沒寫過程式，系列的第一部分會帶你快速上手 Python，並幫助你理解深度學習的基本觀念。 資料科學愛好者 ：如果你已經熟悉一些資料處理方法，進階部分會教你如何用 PyTorch 建構深度學習模型。 開發者與研究人員 ：想更深入了...
張貼留言
閱讀完整內容