RemoteGraph 與子圖巢狀:把編譯圖當節點用

源码版本1.2.9

職責

LangGraph 支援把一個編譯好的圖(本地的 Pregel 實例或遠端的 RemoteGraph)當成另一個圖的節點。RemoteGraph(RemoteGraph:118)是 PregelProtocol 的客戶端實作,透過 LangGraph Server API 呼叫遠端部署的圖;本地的 StateGraph.compile() 產物本身是 Pregel(也是 PregelProtocol),可以直接 parent_graph.add_node(child_graph) 嵌進去。

這一層負責把「巢狀圖」抽象掉:父圖排程到這個節點時,不管背後是本地 Pregel 還是遠端 HTTP API,都走統一的 invoke / stream 介面;PregelNode 透過 find_subgraph_pregel(find_subgraph_pregel:47)在節點 runnable 裡遞迴找出 Pregel 實例,把它掛到 node.subgraphs,這樣父圖的 get_subgraphs 能枚舉到子圖,stream(subgraphs=True) 能拿到子圖內部事件。

設計動機

• 分散式部署:大圖常被拆成多個服務(對話管理、檢索、工具呼叫),每個服務一個 LangGraph 部署,父圖透過 RemoteGraph 把它們串成工作流。每個服務可以獨立擴容、獨立升級,父圖不需要重新部署。
• 巢狀檢查點獨立:add_node(compiled_graph) 後,子圖可以配自己的 checkpointer(compile(checkpointer=...)),也可以從父圖繼承。子圖的 thread_id 由父圖透過 CONFIG_KEY_CHECKPOINT_NS 自動加 namespace 前綴(parent_node:<task_id>|child_node:<task_id>),所以父子圖的 checkpoint 不會串。
• PregelProtocol 統一介面:無論本地 Pregel 還是 RemoteGraph,都實作 PregelProtocol——invoke / stream / astream / get_state / aget_state / get_subgraphs 等方法簽名一致。父圖不需要區別對待。
• stream 透傳:開 subgraphs=True 後,父圖的 _output 在出隊時給事件打 namespace 前綴(get_subgraphs:1076),呼叫方拿到 (ns, mode, payload) 三元組,能知道這個事件來自哪個巢狀層級的哪個節點。
• Command.PARENT 跨層級:子圖節點可以 return Command(graph=Command.PARENT, update=...),把更新寫到父圖狀態——透過 pending writes 在父子圖之間傳遞,父圖的 _first 在收尾時處理子圖上報的 writes。

關鍵檔案

• RemoteGraph:118 — RemoteGraph(PregelProtocol) 客戶端,持 assistant_id / client / sync_client / name。
• RemoteGraph.__init__:132 — 接 url / api_key / headers / client / sync_client,預設用 get_client / get_sync_client 建立。
• RemoteGraph.stream:757 — 同步串流入口,把請求轉給 sync_client.run_stream。
• RemoteGraph.astream:912 — 非同步串流入口,轉給 client.run_stream 非同步迭代器。
• RemoteGraph.invoke:1133 — 同步收斂入口,內部走 stream 拿最後值。
• find_subgraph_pregel:47 — 在節點 bound runnable 裡遞迴找 PregelProtocol 實例,識別子圖。
• PregelNode:97 — PregelNode 類別,持有 subgraphs: Sequence[PregelProtocol] 欄位。
• PregelNode subgraphs init:180 — PregelNode.__init__ 裡呼叫 find_subgraph_pregel(self.bound) 自動識別子圖。
• get_subgraphs:1076 — 父圖枚舉直接子圖,recurse=True 時遞迴向下。
• attach_node:1431 — 編譯時把 add_node 註冊的 action 轉成 PregelNode(bound=node.runnable, ...),子圖識別在這一步發生。
• coerce_to_runnable:529 — 任何 Runnable 實例(包括 Pregel / RemoteGraph)直接 return,這就是「compiled graph as node」的入口。

資料流

add_node(compiled_child_graph) 走 coerce_to_runnable(coerce_to_runnable:529):

def coerce_to_runnable(
    thing: RunnableLike, *, name: str | None, trace: bool
) -> Runnable:
    """Coerce a runnable-like object into a Runnable."""
    if isinstance(thing, Runnable):
        return thing
    elif is_async_generator(thing) or inspect.isgeneratorfunction(thing):
        return RunnableLambda(thing, name=name)
    elif callable(thing):
        # ...
    elif isinstance(thing, dict):
        return RunnableParallel(thing)
    else:
        raise TypeError(...)

Pregel 和 RemoteGraph 都繼承 Runnable,所以走第一個分支直接回傳。PregelNode.__init__ 在 attach_node 階段呼叫 find_subgraph_pregel(self.bound)(find_subgraph_pregel:47):

def find_subgraph_pregel(candidate: Runnable) -> PregelProtocol | None:
    from langgraph.pregel import Pregel

    candidates: list[Runnable] = [candidate]

    for c in candidates:
        if (
            isinstance(c, PregelProtocol)
            # subgraphs that disabled checkpointing are not considered
            and (not isinstance(c, Pregel) or c.checkpointer is not False)
        ):
            return c
        elif isinstance(c, RunnableSequence) or isinstance(c, RunnableSeq):
            candidates.extend(c.steps)
        elif isinstance(c, RunnableLambda):
            candidates.extend(c.deps)
        elif isinstance(c, RunnableCallable):
            if c.func is not None:
                candidates.extend(
                    nl.__self__ if hasattr(nl, "__self__") else nl
                    for nl in get_function_nonlocals(c.func)
                )
            elif c.afunc is not None:
                candidates.extend(
                    nl.__self__ if hasattr(nl, "__self__") else nl
                    for nl in get_function_nonlocals(c.afunc)
                )

    return None

它在 RunnableSequence / RunnableLambda / RunnableCallable 裡遞迴查詢——如果一個節點包了一層 RunnableLambda 呼叫 Pregel,也能識別出來。識別到的子圖掛到 self.subgraphs = [subgraph],這樣父圖 get_subgraphs()(get_subgraphs:1076)能枚舉到。

執行時,父圖的 PregelRunner.tick 呼叫子圖 Pregel 的 stream / astream(或 RemoteGraph 的對應方法),子圖內部跑自己的 Pregel 迴圈,產生的事件透過父圖的 stream(subgraphs=True) 透傳——父圖在子圖呼叫時把 CONFIG_KEY_CHECKPOINT_NS 加上節點名前綴,子圖的事件帶上這個 namespace 寫回父圖的 stream 佇列,父圖 _output 出隊時把它和本地事件合併。

邊界與失敗

• 禁用 checkpointer 的子圖不被識別:find_subgraph_pregel 顯式跳過 c.checkpointer is False 的 Pregel(skip no-checkpointer:54)——這種「無狀態子圖」不當作子圖對待,父圖 get_subgraphs 看不到它,事件也不會帶 namespace。
• RemoteGraph 至少要配 url 或 client:__init__ 裡 client 和 sync_client 都是 None 時,如果沒傳 url,兩個 client 都不會被建立,後續呼叫 _validate_client() 會 raise ValueError(validate client:181)。
• RemoteGraph 不支援全部 stream_mode:_reject_v3_unsupported 會拒絕某些參數組合,遠端 API 的 stream 協議限制比本地 Pregel 多(reject v3:195)。
• 巢狀 namespace 用 NS_SEP:命名空間格式是 parent_node:<task_id> + NS_SEP + child_node:<task_id>,遞迴向下。add_node 時拒絕節點名裡含 NS_SEP(reject NS_SEP:794),避免命名空間歧義。
• 子圖 interrupt() 沿 namespace 傳播:子圖節點 interrupt() 拋 GraphInterrupt 時,父圖 tick 偵測到子圖的 pending interrupt,父圖也進入 interrupt 狀態,客戶端 resume 時 Command(resume=...) 透過 CONFIG_KEY_RESUME_MAP 按 namespace 路由到對應子圖(Command resume mapping:904)。
• Command.PARENT 邊界:子圖節點 return Command(graph=Command.PARENT, ...) 時,子圖 map_command 直接 raise InvalidUpdateError("There is no parent graph")(PARENT guard:58)——這其實是兜底,正常路徑是子圖 attach_node 時 _get_updates mapper 跳過 Command.PARENT,把它原樣寫到子圖 pending writes,父圖在 _first 裡識別並路由。

小結

RemoteGraph 讓 LangGraph 支援跨服務、跨程序的圖巢狀,本地 Pregel 巢狀則透過 find_subgraph_pregel 自動識別——兩者都走 PregelProtocol。父圖 stream(subgraphs=True) 透過 namespace 透傳子圖事件。要追子圖事件流細節看 /stream/run-stream,要追子圖節點排程看 /pregel/pregel。對照官方資料:LangGraph 文件 · README