AG Day 31 長任務:檢查點、恢復與背景執行
執行需求:CPU 可跑。AG Day 30(原文連結)把多代理系統的通訊方式升級成有明確契約的訊息協議,任務清單與交接封包都是可以序列化的結構化狀態。今天要把這個特性用起來,補上多代理系統長期執行時最重要的一塊:檢查點(checkpoint)、恢復(resume)與背景執行(background execution)。研究任務往往不是幾秒鐘就能跑完的事,可能牽涉多輪搜尋、多次模型呼叫,加總起來要跑上幾分鐘甚至更久;如果程式在跑到一半當掉、使用者關掉終端機,或只是想先問別的事、晚點再回來看結果,整條任務不該從頭重來。今天的實作全部使用 LangGraph 內建的檢查點機制搭配本機 SQLite,不需要外部服務或 API 金鑰就能完整驗證。
引言
檢查點這個概念在 AG Day 16 就出現過:我們用 checkpointer 搭配 thread 讓單一代理擁有跨輪對話的記憶。今天的差別在於,我們把同一套機制套用到「多步驟、多代理、可能耗時很久」的研究任務上,目標不只是記住對話歷史,而是記住整個任務進行到哪一步、哪些子任務已完成、哪些還在排隊。這讓 research-agent 具備三個新能力:程式意外中斷後可以從斷點接著跑,不必重新搜尋已經搜尋過的資料;使用者可以先送出一個研究問題,不必守在終端機前等待,稍後再回來查詢進度或結果;系統可以同時處理多個研究任務,用不同的 thread ID 互相隔離,不會彼此干擾。
今天會先解釋 LangGraph 檢查點在多代理場景下的運作方式,接著把 AG Day 29、30 建立的 supervisor-worker 圖接上 SQLite 檢查點,示範「跑一半模擬當機、重新啟動後接著跑完」的完整流程,最後再示範一個簡單的背景執行模式:把任務丟到背景執行緒,讓呼叫端可以立刻拿到一個任務 ID,之後再用這個 ID 查詢進度或結果。
原理/觀念
檢查點在多代理圖裡記的是什麼
LangGraph 的檢查點機制,本質上是在每個節點執行完之後,把當下完整的狀態(也就是我們的 TeamState)連同執行到哪一個節點的資訊,存進檢查點儲存後端。因為 AG Day 30 已經把任務進度明確表達成 task_queue 與 completed_tasks 這種可序列化的結構,檢查點存下來的內容就完整包含「哪些任務做完了、結果是什麼、哪些還在排隊」,不需要額外的邏輯去猜測任務進度——這正是昨天把協議設計得結構化的好處在今天兌現。每個獨立的任務執行都對應一個 thread_id,同一個 thread_id 底下的多次呼叫會共享同一份檢查點歷史,讓你可以隨時從某個 thread_id 讀出「這個任務目前的完整狀態」。
從記憶體檢查點換成 SQLite 檢查點
開發階段常用的 MemorySaver 把檢查點存在程式的記憶體裡,程式一結束資料就沒了,適合寫測試但不適合長任務。今天換成以檔案為後端的 SQLite 檢查點,讓狀態在程式重啟後依然存在,這跟 AG Day 2 建立 knowledge.db 是同樣的思路:用一個單一檔案資料庫承擔本機持久化的責任,不需要額外起一個資料庫伺服器。要注意 SQLite 檢查點資料庫跟 knowledge.db 是兩個獨立的檔案,各自負責不同的事——knowledge.db 存研究素材與分塊,檢查點資料庫存的是圖執行過程中每一步的完整狀態快照,兩者不要混在一起。
背景執行不是另一套系統,是同一張圖換一種呼叫方式
很多人會誤以為「背景執行」需要另外架一套任務佇列系統(例如 Celery),但對於研究助理這種單機運作的場景,用 Python 內建的執行緒或行程池就足夠了:把 graph.invoke(...) 丟進一個背景執行緒執行,立刻回傳一個 run_id 給呼叫端,執行緒跑完之後把結果寫回 runs 資料表,呼叫端隨時可以用 run_id 查詢狀態。這個模式很適合先驗證概念,等到 AG Day 38 把 research-agent 包成 FastAPI 服務時,同樣的想法會換成更正式的背景任務機制,但核心邏輯是一致的:呼叫端拿到的是一個可以之後查詢的識別碼,不是立刻等待的結果。
完整實作
先安裝 SQLite 檢查點所需的套件(需先 uv pip install langgraph-checkpoint-sqlite):
cd research-agent
uv pip install langgraph-checkpoint-sqlite
mkdir -p data
第一步:在 agents/graph.py 裡把圖接上 SQLite 檢查點。編譯圖時傳入 checkpointer,之後每次呼叫都要帶上 thread_id 設定,才能讓 LangGraph 知道這次執行要接續哪一段歷史:
# research-agent/src/research_agent/agents/graph.py(接上 SQLite 檢查點)
import sqlite3
from pathlib import Path
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.sqlite import SqliteSaver
from research_agent.agents.state import TeamState
from research_agent.agents.supervisor import supervisor_node, route_to_worker
from research_agent.agents.workers import search_worker, retrieval_worker, writer_worker
CHECKPOINT_DB = Path(__file__).resolve().parent.parent.parent.parent / "data" / "checkpoints.sqlite"
def build_team_graph():
"""組出多代理圖,並接上以檔案為後端的 SQLite 檢查點。"""
graph = StateGraph(TeamState)
graph.add_node("supervisor", supervisor_node)
graph.add_node("search_worker", search_worker)
graph.add_node("retrieval_worker", retrieval_worker)
graph.add_node("writer_worker", writer_worker)
graph.add_edge(START, "supervisor")
graph.add_conditional_edges("supervisor", route_to_worker, {
"search_worker": "search_worker",
"retrieval_worker": "retrieval_worker",
"writer_worker": "writer_worker",
"FINISH": END,
})
for worker in ("search_worker", "retrieval_worker", "writer_worker"):
graph.add_edge(worker, "supervisor")
CHECKPOINT_DB.parent.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(str(CHECKPOINT_DB), check_same_thread=False)
checkpointer = SqliteSaver(conn)
return graph.compile(checkpointer=checkpointer)
第二步:寫一個示範腳本,模擬「跑到一半當機」的情境——我們故意在 search_worker 完成後,用一個會拋例外的假 worker 中斷整個流程,再示範用同一個 thread_id 重新呼叫時,圖會從檢查點接續,不會重新執行已經完成的 search_worker:
# research-agent/scripts/simulate_crash.py
import uuid
from research_agent.agents.graph import build_team_graph
def main():
team = build_team_graph()
thread_id = str(uuid.uuid4())
config = {"configurable": {"thread_id": thread_id}}
initial_state = {
"messages": [], "topic": "檢查點與恢復機制",
"task_queue": [], "completed_tasks": [], "final_report": None, "step_count": 0,
}
print(f"=== 第一次執行(thread_id={thread_id[:8]}) ===")
for step in team.stream(initial_state, config=config):
node_name = next(iter(step))
print(f"執行到節點:{node_name}")
if node_name == "search_worker":
print("(模擬程式在這裡當機,直接中斷迴圈)")
break
print("\n=== 從檢查點讀出目前狀態 ===")
snapshot = team.get_state(config)
print(f"目前的 task_queue 狀態:{[(t.task_id, t.status.value) for t in snapshot.values['task_queue']]}")
print(f"\n=== 用同一個 thread_id 重新呼叫,從斷點接續 ===")
result = team.invoke(None, config=config)
print(f"最終報告:\n{result['final_report']}")
if __name__ == "__main__":
main()
這裡的關鍵是 team.invoke(None, config=config):傳入 None 而不是新的初始狀態,LangGraph 會讀取 thread_id 對應的最後一筆檢查點,從那裡繼續往下跑,不會重新從 START 開始,也就不會重複呼叫已經成功過的 search_worker。
第三步:實作簡單的背景執行輔助函式 agents/background.py,把 graph.invoke 丟進背景執行緒,立刻回傳 run_id,執行完畢後把結果寫回 runs 資料表:
# research-agent/src/research_agent/agents/background.py
import uuid
import threading
from research_agent.agents.graph import build_team_graph
from research_agent.storage import record_new_run, update_run_status
_team_graph = build_team_graph()
def start_research_job(db_path, topic: str) -> str:
"""立刻回傳 run_id,實際執行交給背景執行緒。"""
run_id = str(uuid.uuid4())
record_new_run(db_path, run_id, topic, model_name="team-graph")
def _worker():
config = {"configurable": {"thread_id": run_id}}
initial_state = {
"messages": [], "topic": topic,
"task_queue": [], "completed_tasks": [], "final_report": None, "step_count": 0,
}
try:
result = _team_graph.invoke(initial_state, config=config)
update_run_status(db_path, run_id, "DONE", final_report=result["final_report"])
except Exception as exc:
update_run_status(db_path, run_id, "FAILED", final_report=str(exc))
threading.Thread(target=_worker, daemon=True).start()
return run_id
第四步:補上 storage.py 裡的 update_run_status,讓背景執行緒跑完之後可以把結果寫回 runs 資料表,供之後查詢:
# research-agent/src/research_agent/storage.py(新增函式)
import sqlite3
def update_run_status(db_path, run_id: str, status: str, final_report: str | None = None) -> None:
"""更新 runs 資料表的執行狀態與最終報告,供背景任務完成後回寫。"""
with sqlite3.connect(db_path) as conn:
conn.execute(
"UPDATE runs SET status = ?, final_report = ?, finished_at = CURRENT_TIMESTAMP WHERE run_id = ?",
(status, final_report, run_id),
)
conn.commit()
def get_run(db_path, run_id: str) -> dict | None:
"""查詢單一任務目前的狀態,供背景執行時的輪詢使用。"""
with sqlite3.connect(db_path) as conn:
row = conn.execute(
"SELECT run_id, status, final_report FROM runs WHERE run_id = ?", (run_id,)
).fetchone()
if row is None:
return None
return {"run_id": row[0], "status": row[1], "final_report": row[2]}
第五步:寫一個查詢進度的腳本 scripts/query_progress.py,示範呼叫端不需要等待任務跑完,也能隨時用 get_state 讀出目前的檢查點快照,看看走到哪個節點、完成了哪些任務:
# research-agent/scripts/query_progress.py
from research_agent.agents.graph import build_team_graph
def show_progress(thread_id: str):
"""在任務還在背景執行時,隨時讀取目前的檢查點快照。"""
team = build_team_graph()
config = {"configurable": {"thread_id": thread_id}}
snapshot = team.get_state(config)
if snapshot.values is None:
print(f"找不到 thread_id={thread_id} 的任何檢查點紀錄")
return
queue = snapshot.values.get("task_queue", [])
done = [t for t in queue if t.status.value == "done"]
print(f"任務 {thread_id[:8]} 目前進度:{len(done)}/{len(queue)} 個子任務完成")
for t in queue:
print(f" - {t.task_id}({t.assigned_to}):{t.status.value}")
if __name__ == "__main__":
import sys
show_progress(sys.argv[1])
第六步:寫一個結合背景執行與輪詢查詢的完整示範 scripts/poll_background_job.py,把今天所有元件串起來:啟動一個背景任務、每隔一段時間查詢一次狀態,直到任務完成或失敗:
# research-agent/scripts/poll_background_job.py
import time
from pathlib import Path
from research_agent.agents.background import start_research_job
from research_agent.storage import get_run
DB_PATH = Path("data/knowledge.db")
def main():
run_id = start_research_job(DB_PATH, "背景執行與輪詢查詢")
print(f"任務已送出,run_id={run_id}")
while True:
run = get_run(DB_PATH, run_id)
print(f"目前狀態:{run['status']}")
if run["status"] in ("DONE", "FAILED"):
print(f"最終結果:\n{run['final_report']}")
break
time.sleep(1)
if __name__ == "__main__":
main()
這兩個腳本合起來展示了一個完整的非同步使用情境:使用者送出研究問題後立刻拿到 run_id,可以先去做別的事,之後再用同一個 run_id 查詢進度(透過檢查點快照)或最終結果(透過 runs 資料表),兩種查詢方式各有用途——檢查點快照能看到細部的任務進度,runs 資料表則是給使用者看的簡化摘要。
執行今天的當機模擬腳本:
uv run python scripts/simulate_crash.py
示範輸出(離線模擬模式,實際 thread_id 每次都不同):
=== 第一次執行(thread_id=1a2b3c4d) ===
執行到節點:supervisor
執行到節點:search_worker
(模擬程式在這裡當機,直接中斷迴圈)
=== 從檢查點讀出目前狀態 ===
目前的 task_queue 狀態:[('t1', 'done'), ('t2', 'pending'), ('t3', 'pending')]
=== 用同一個 thread_id 重新呼叫,從斷點接續 ===
最終報告:
# 檢查點與恢復機制(示範報告,離線模擬模式)
1. 示範搜尋結果,非真實網路資料。[來源:檢查點與恢復機制 示範資料]
可以看到重新呼叫後,t1 沒有被重跑,流程直接從 t2(retrieval_worker)接續下去,這就是檢查點機制在多代理長任務裡最直接的價值。
常見錯誤與踩雷
第一個常見錯誤是忘記幫每次獨立的任務產生獨立的 thread_id,結果所有使用者的研究任務都共用同一條 thread,狀態互相汙染。thread_id 必須是每個任務的唯一識別碼,通常直接沿用今天的 run_id 即可,這樣查詢 runs 資料表跟查詢檢查點狀態可以用同一個 ID 對應起來,不必額外維護一張對照表。
第二個常見錯誤是背景執行緒裡的例外被吞掉,主執行緒完全不知道背景任務失敗了。今天的 _worker 函式特別用 try/except 包住整段呼叫,失敗時把錯誤字串寫進 final_report 欄位並標記狀態為 FAILED,這樣呼叫端輪詢 get_run 時才能看到失敗原因,而不是任務永遠停在「執行中」卻查無結果。
第三個常見錯誤是把 SQLite 連線在多執行緒之間共用卻沒有處理並行存取。今天用 check_same_thread=False 允許連線跨執行緒使用,但這只解決了「不會直接丟例外」的問題,沒有解決「多個背景任務同時寫入同一個 SQLite 檔案」時的鎖定問題。跟 AG Day 2 一樣,實務上建議開啟 PRAGMA journal_mode=WAL;,讓讀寫可以並行,真正高並行的場景則應該考慮換成正式的資料庫伺服器。這個限制在單機示範階段不明顯,但只要背景任務數量一多,就容易在寫入尖峰時遇到短暫的鎖定延遲,務必在真正上線前用接近真實流量的並行壓力測試提前抓出來。
效能與實務提醒
檢查點機制每一步都要把完整狀態序列化寫進 SQLite,如果狀態裡塞了大量原始搜尋資料或超長對話歷史,每一步的寫入延遲會明顯增加。實務上建議只在狀態裡保留「重新執行需要的最小資訊」,龐大的原始素材另外存進 knowledge.db 的 documents 資料表,狀態裡只留參照用的 ID,這也呼應 AG Day 20-23 檢索設計裡「原始素材與檢索結果分開存放」的原則。
背景執行緒的方式對單機小規模場景很好用,但要注意 Python 執行緒仍然受 GIL 限制,如果背景任務裡有大量 CPU 密集運算(而不是等待網路 I/O),多個背景任務同時跑反而會互相搶佔。研究助理的工作負載主要是等待模型 API 與網路搜尋的回應,屬於 I/O 密集型,用執行緒是合理的選擇;如果之後有 CPU 密集的批次任務(例如大量檔案的向量化),會更適合用行程池而不是執行緒池。
最後,檢查點資料庫會隨著任務數量持續累積,正式環境裡需要定期清理已經完成很久、不再需要恢復的舊 thread,否則單一 SQLite 檔案會越長越大,拖慢查詢效能,這件事會在 AG Day 43 討論已知限制時再進一步展開。另外值得注意的是,今天的示範裡每個背景任務都用獨立的 thread_id,代表檢查點資料庫裡的紀錄不會互相覆寫,但也意味著儲存空間會隨任務數量線性成長,建議搭配 runs 資料表裡的 finished_at 時間戳記,定期把超過一定天數且狀態為 DONE 或 FAILED 的舊紀錄整批清除。
小結
今天我們把多代理圖接上以 SQLite 為後端的檢查點機制,讓每個任務的完整狀態(任務清單、交接封包、最終報告)都能在意外中斷後被完整恢復;也實作了一個簡單的背景執行輔助函式,讓呼叫端可以立刻拿到 run_id,不必守在原地等待任務跑完,稍後再用同一個 ID 查詢狀態或結果。
新增的術語:檢查點(checkpoint,某一時刻的完整狀態快照)、恢復(resume,從檢查點接續執行而不重新開始)、thread_id(區分不同任務的執行緒歷史識別碼)、背景執行(background execution,不阻塞呼叫端的非同步任務執行方式)、輪詢(polling,呼叫端定期查詢任務狀態直到完成的做法)。這些機制合起來,讓多代理系統從「一次性跑完就結束」變成「可以長時間運作、可觀察、可恢復」的服務雛形。
結語
有了檢查點與背景執行,research-agent 已經具備處理「跑很久、可能中斷、使用者不想一直等」這類真實研究任務的基礎能力,這是今天最重要的收穫。但目前我們還完全沒有回答一個更根本的問題:這個多代理系統做出來的報告到底好不好?搜尋得夠不夠全面?引用有沒有對到正確的來源?這些問題目前只能靠人工肉眼檢查。
明天,我們會進入「AG Day 32 評估基礎:Agent 為什麼難測」,先不急著寫評分程式,而是老實面對「為什麼代理系統的測試比一般軟體困難」這個根本問題:非決定性的輸出、多步驟的執行路徑、有副作用的工具呼叫,都讓傳統的單元測試思維不再夠用,我們需要一套全新的評估框架來補上這塊拼圖。
延伸資源
- LangGraph 官方文件的 Persistence/Checkpointing 章節:
https://langchain-ai.github.io/langgraph/。SqliteSaver的建構方式、thread_id設定與get_state用法請以官方文件當次版本為準。 - Python 官方文件的
threading模組:說明Thread、daemon參數與 GIL 對多執行緒效能的影響。 - Python 官方文件的
sqlite3模組:check_same_thread參數與跨執行緒使用連線的注意事項。
留言
張貼留言