跳到主要內容

AG Day 18 子圖：把研究流程模組化

AG Day 18 子圖：把研究流程模組化

執行需求：CPU 可跑。在昨天 AG Day 17（原文連結）中，我們替 research-agent 加上了人工核准節點，讓代理在執行高成本工具呼叫前主動暫停、等待真人決定。到目前為止我們的圖只有 agent、approval_gate、tools 三個節點，還算容易一眼看懂。但從下週開始，我們會陸續加入向量檢索、報告產生、多代理協作等更多能力，如果全部塞進同一張扁平的圖裡，節點數量與邊的連法很快就會失控。今天要做的事，是把研究流程拆成幾個獨立的子圖（subgraph），讓每個子圖各自負責一件事、能單獨測試，再組裝回一張主圖。

引言

子圖聽起來像是個進階概念，但其實對應到軟體工程裡一個非常基本的道理：當一個函式變得太長、承擔太多責任時，我們會把它拆成幾個更小的函式。LangGraph 的子圖就是圖版本的「拆函式」——把一組彼此相關、通常會一起執行的節點，包裝成一個獨立編譯好的圖，這個圖本身可以像一個節點一樣，被放進另一張更大的圖裡使用。對 research-agent 而言，我們今天要拆出的第一個子圖是「檢索子圖」（retrieval subgraph），負責「決定要搜尋什麼、呼叫搜尋工具、視情況重新查詢」這一整組行為；主圖則只需要知道「呼叫檢索子圖，會得到一份整理好的搜尋結果」，不需要知道子圖內部是怎麼運作的。

這種拆分的好處有兩層。第一層是可讀性：主圖的節點數量變少，一眼就能看出整體流程的骨架，細節被封裝進各自的子圖裡。第二層、也是工程上更重要的一層，是可測試性：檢索子圖可以獨立編譯、獨立呼叫、獨立寫單元測試，不需要每次測試都要連同整個代理迴圈一起跑。今天我們會示範兩種子圖的組裝方式：一種是子圖與父圖共用同一份狀態結構（state schema），另一種是子圖有自己獨立的狀態、透過明確的轉接函式與父圖溝通，並說明什麼情況下該選哪一種。

原理與觀念

共享狀態子圖：最簡單的組裝方式

如果子圖使用的狀態結構（例如我們一直在用的 MessagesState）跟父圖完全相同，組裝起來非常直接：先用 StateGraph(MessagesState) 建好子圖、呼叫 compile() 拿到一個編譯好的圖物件，接著在父圖裡用 graph.add_node("子圖名稱", 編譯好的子圖物件)，把這個子圖當成一個節點加進去即可。LangGraph 在執行時會把父圖目前的狀態直接傳給子圖、子圖執行完的結果也會直接合併回父圖的狀態，因為兩邊用的是同一份 schema，資料格式完全相容，不需要額外轉換。

獨立狀態子圖：需要明確定義輸入與輸出的轉接

但並不是所有子圖都適合共用父圖的完整狀態。如果子圖只需要處理一部分資訊（例如檢索子圖其實不需要看到完整的對話歷史，只需要知道「目前要查什麼」），把整個 MessagesState 傳進去反而讓子圖的職責變得模糊、耦合了它不該關心的細節。這種情況下，子圖可以定義自己的、更精簡的狀態結構，父圖呼叫子圖前，需要寫一個轉接節點，把父圖狀態裡子圖需要的欄位取出來、包成子圖的狀態格式；子圖執行完後，也需要一個轉接節點，把子圖的輸出結果轉換回父圖能理解的格式。這種寫法多了幾行轉接程式碼，換來的是子圖介面的清晰：任何人只要看子圖的狀態定義，就知道它需要什麼輸入、會產生什麼輸出，不必去讀主圖的完整脈絡。

子圖與 checkpointer 的關係

子圖預設會沿用父圖傳下來的 checkpointer，不需要額外設定；也就是說，如果主圖已經在 compile(checkpointer=...) 時指定了 SQLite 版的儲存機制（我們在 AG Day 16 建立的那一套），子圖內部節點的執行狀態一樣會被記錄進同一份檢查點資料庫裡，中斷與恢復（我們在 AG Day 17 實作的人工核准）在子圖內部一樣能正常運作，這是子圖組裝方式對我們特別友善的地方：模組化不會犧牲掉前兩天建立的記憶與人工核准能力。

完整實作

我們先把「檢索」相關的邏輯獨立成 src/research_agent/subgraphs/retrieval_graph.py，這個子圖負責「呼叫搜尋工具、視需要重試一次不同的查詢字串」，並定義自己精簡的狀態結構：

# research-agent/src/research_agent/subgraphs/retrieval_graph.py
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from research_agent.tools import search_web


class RetrievalState(TypedDict):
    query: str
    attempts: int
    results: str
    done: bool


def run_search(state: RetrievalState) -> RetrievalState:
    """實際呼叫搜尋工具，並記錄已經嘗試的次數。"""
    results = search_web(state["query"])
    is_empty = "查無相關結果" in results or not results.strip()
    return {
        **state,
        "results": results,
        "attempts": state["attempts"] + 1,
        "done": not is_empty or state["attempts"] + 1 >= 2,
    }


def should_retry(state: RetrievalState) -> str:
    """查無結果且還沒重試過，就換一次查詢字串再試一次。"""
    return END if state["done"] else "retry_query"


def widen_query(state: RetrievalState) -> RetrievalState:
    """把查詢字串放寬（示範版：加上『概述』二字），模擬換句話說重新查詢。"""
    return {**state, "query": f"{state['query']} 概述"}


def build_retrieval_subgraph():
    graph = StateGraph(RetrievalState)
    graph.add_node("search", run_search)
    graph.add_node("retry_query", widen_query)
    graph.add_edge(START, "search")
    graph.add_conditional_edges("search", should_retry, {END: END, "retry_query": "retry_query"})
    graph.add_edge("retry_query", "search")
    return graph.compile()

因為 RetrievalState 跟主圖的 MessagesState 完全不同，我們需要在主圖裡寫兩個轉接函式：一個把主圖狀態轉成子圖需要的輸入，一個把子圖的輸出結果轉換回主圖狀態能理解的格式。轉接邏輯放在 src/research_agent/graph.py：

# research-agent/src/research_agent/graph.py（新增轉接與子圖節點）
from langchain_core.messages import ToolMessage, AIMessage
from research_agent.subgraphs.retrieval_graph import build_retrieval_subgraph

retrieval_subgraph = build_retrieval_subgraph()


def call_retrieval_subgraph(state):
    """轉接節點：從主圖狀態取出查詢字串，呼叫檢索子圖，再把結果轉回主圖狀態。"""
    last_message = state["messages"][-1]
    tool_call = last_message.tool_calls[0]
    query = tool_call["args"].get("query", "")

    sub_result = retrieval_subgraph.invoke({
        "query": query,
        "attempts": 0,
        "results": "",
        "done": False,
    })

    tool_message = ToolMessage(
        content=sub_result["results"],
        tool_call_id=tool_call["id"],
    )
    return {"messages": state["messages"] + [tool_message]}

把轉接節點接進主圖，取代原本直接呼叫 tool_node 的路徑（這裡示範把 search_web 的呼叫整條路由到子圖，fetch_url 仍走昨天的 ToolNode）：

# research-agent/src/research_agent/graph.py（調整路由邏輯）
def route_after_approval(state) -> str:
    """依工具名稱決定走檢索子圖還是原本的 ToolNode。"""
    last_message = state["messages"][-1]
    tool_call = last_message.tool_calls[0]
    if tool_call["name"] == "search_web":
        return "retrieval_subgraph"
    return "tools"


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

子圖拆出來後，最大的好處就是可以完全獨立測試，不需要啟動整個代理、也不需要任何模型呼叫：

# research-agent/tests/test_retrieval_subgraph.py
from research_agent.subgraphs.retrieval_graph import build_retrieval_subgraph

subgraph = build_retrieval_subgraph()


def test_search_returns_results_without_retry():
    result = subgraph.invoke({"query": "LangGraph 子圖", "attempts": 0, "results": "", "done": False})
    assert result["done"] is True
    assert result["attempts"] >= 1


def test_widen_query_appends_summary_keyword():
    from research_agent.subgraphs.retrieval_graph import widen_query
    widened = widen_query({"query": "冷門主題", "attempts": 1, "results": "", "done": False})
    assert widened["query"] == "冷門主題 概述"

最後驗證主圖與子圖組裝起來後，行為跟拆分之前一致：

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

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

result = app.invoke(
    {"messages": [HumanMessage(content="幫我搜尋台灣半導體供應鏈的最新概況")]},
    config=config,
)
print("最終回覆：", result["messages"][-1].content[:80])

為了對照「共享狀態子圖」與「獨立狀態子圖」的差異，我們再示範一個共用 MessagesState 的簡單子圖：一個把最後一則工具結果做簡短摘要的「摘要子圖」，因為它需要直接讀寫訊息歷史，用共享狀態反而比另外定義一份轉接邏輯更省事：

# research-agent/src/research_agent/subgraphs/summary_graph.py
from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_core.messages import AIMessage


def summarize_last_tool_result(state: MessagesState) -> MessagesState:
    """示範用：把最後一則 ToolMessage 摘要成一句話附加回訊息歷史。"""
    last_tool_message = next(
        (m for m in reversed(state["messages"]) if type(m).__name__ == "ToolMessage"),
        None,
    )
    if last_tool_message is None:
        return state

    summary = last_tool_message.content[:60] + ("..." if len(last_tool_message.content) > 60 else "")
    note = AIMessage(content=f"（內部摘要）搜尋結果重點：{summary}")
    return {"messages": state["messages"] + [note]}


def build_summary_subgraph():
    graph = StateGraph(MessagesState)
    graph.add_node("summarize", summarize_last_tool_result)
    graph.add_edge(START, "summarize")
    graph.add_edge("summarize", END)
    return graph.compile()

因為 build_summary_subgraph 用的也是 MessagesState，要把它插進主圖，只需要一行 graph.add_node("summarize", build_summary_subgraph())，不需要額外寫轉接函式；這跟檢索子圖需要兩個轉接函式形成鮮明對比，也印證了前面原理段落的判斷標準：狀態結構相不相容，決定了組裝時要付出多少額外的接線成本。實務上兩種寫法會同時存在於一個專案裡，重點是依每個子圖實際需要的輸入輸出範圍去選擇，而不是為了統一風格硬套同一種做法。

常見錯誤與踩雷

第一個常見錯誤是子圖與父圖用不同的狀態鍵名，卻忘記寫轉接邏輯，直接把子圖的編譯結果當成節點塞進共用狀態的父圖裡。這種情況下 LangGraph 可能會因為欄位對不上而丟出型別或鍵值錯誤，也可能悄悄地把不該有的欄位混進父圖狀態，事後很難排查。只要子圖的狀態結構跟父圖不同，就一定要寫明確的轉接函式，不要僥倖以為框架會自動處理。

第二個是子圖內部忘記正確處理迴圈終止條件。我們今天的 should_retry 用 attempts >= 2 當作硬性上限，避免子圖因為一直查不到結果而無限重試、無限迴圈。子圖跟主圖一樣，沒有明確的終止條件時，一樣可能卡死整條執行；子圖因為通常邏輯較單純，這種疏漏反而更容易被忽略。

第三個雷是把子圖設計得過度細碎，例如每個工具都拆一個子圖。子圖的價值在於封裝「一組有意義的行為」，如果每個子圖都只有一個節點，等於是繞了一大圈重新發明 ToolNode 已經做得很好的事，還額外增加了轉接程式碼的維護成本。拆分的判斷標準應該是「這組節點有沒有自己獨立的、值得單獨測試的邏輯」，而不是「圖看起來太大就拆」。

第四個是忘記子圖也會共用父圖傳下來的 checkpointer，導致子圖內部的中間狀態被記錄進檢查點資料庫，卻沒有意識到這件事，日後在查詢 list_active_threads（我們在 AG Day 16 寫的工具）看到大量子圖內部節點名稱時感到困惑。建議在子圖節點命名時保持清晰、有意義的名稱，方便日後對照。

第五個容易忽略的地方，是共享狀態子圖看似方便，卻也代表子圖可以任意讀寫父圖狀態裡的任何欄位，包括它其實不該碰的部分。今天的摘要子圖只讀取最後一則工具訊息、只新增一則訊息，沒有刪改既有內容；如果日後某個共享狀態子圖不小心覆寫了不該動的欄位（例如清空了訊息歷史），因為型別完全相容，這種錯誤不會在編譯或執行當下被抓到，只會在後續某個節點讀到不完整的資料時才爆出來，除錯起來格外費工。共享狀態帶來的方便，換來的代價就是失去了轉接函式原本能提供的邊界檢查，設計共享狀態子圖時要對這一點格外謹慎。

效能與實務提醒

子圖的模組化不只是為了程式碼整潔，它直接影響團隊協作的效率。當研究流程拆成檢索子圖、報告子圖等模組後，不同的人可以分別負責不同子圖的邏輯與測試，互不干擾，合併程式碼時衝突也會少很多。這對後面 AG Day 29 開始的多代理架構是重要的鋪墊：每個代理角色，某種程度上都可以看成是一個具備特定職責的子圖或子圖組合。

效能面上，子圖本身不會帶來額外的執行開銷；它在編譯後就是一張正常的圖，父圖呼叫它跟呼叫一個普通節點函式沒有本質差異。真正該注意的是轉接函式裡的資料複製：如果父圖狀態很龐大（例如訊息歷史已經累積了上百則），每次呼叫子圖都做一次不必要的深拷貝，會在高頻呼叫時逐漸累積成看得見的效能負擔。今天的轉接函式只取出必要欄位（query），這是刻意的設計，避免把整份訊息歷史複製進子圖狀態裡。

最後，建議把每個子圖都當成一個可以獨立發布版本的小模組來維護：清楚的輸入輸出定義、獨立的測試檔案、獨立的變更紀錄。當專案規模擴大到需要多人協作或未來考慮拆成獨立套件時，這種邊界清楚的子圖設計會讓遷移的成本小很多，也讓新加入專案的工程師可以先只讀懂一個子圖，就能開始貢獻程式碼，不必一口氣理解整張主圖的所有細節。

小結

今天我們把 research-agent 原本擠在單一節點裡的檢索邏輯，拆成一個獨立的子圖：它有自己的狀態結構、自己的重試終止條件、自己的單元測試，主圖只需要透過一個轉接節點呼叫它、把結果轉換回主圖狀態即可，主圖的節點清單也因此變得簡短許多。我們也確認了子圖會自動沿用父圖的 checkpointer，模組化不會犧牲掉前兩天建立的記憶與人工核准能力。

今天新增的關鍵詞：子圖（subgraph）——把一組相關節點封裝成可獨立編譯、獨立測試的圖模組；共享狀態子圖——與父圖使用同一份狀態結構、可直接當節點插入的子圖；轉接節點——負責在父圖狀態與子圖獨立狀態之間做欄位轉換的節點。判斷該用哪一種組裝方式時，記得回到今天強調的原則：狀態結構相容就直接共享，職責範圍不同就寫轉接函式劃清界線。

結語

把研究流程模組化之後，research-agent 的架構已經具備了應付更多功能的骨架。但到目前為止，我們一直是用 invoke() 等整條圖跑完才拿到結果，使用者在漫長的搜尋與思考過程中，畫面上完全沒有任何回饋。

明天，我們會進入「AG Day 19 串流輸出：stream 與事件流」，改用 LangGraph 的串流介面，讓 CLI 能即時顯示代理正在思考什麼、呼叫了哪個工具，大幅改善使用者在長時間任務中乾等畫面的體驗。

延伸資源

  • LangGraph 官方文件：Subgraphs 概念頁面，涵蓋共享狀態與獨立狀態兩種組裝方式的完整說明。
  • LangGraph 官方文件：State Schema 與 TypedDict 的使用慣例。
  • Python 官方文件：typing.TypedDict 的型別註解語法。
標籤：

留言

張貼留言

這個網誌中的熱門文章

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 建構深度學習模型。 開發者與研究人員 ：想更深入了...
張貼留言
閱讀完整內容