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。