跳到主要内容

图引擎:一张 JSON 怎么变成可执行的 LangGraph

本章讲引擎的地基:GraphEngine.__init__ 启动后,如何把前端的 {nodes, edges} 一步步编译成一张能跑的 LangGraph 状态图。

1. 这一步要解决什么​

前端给的只是静态描述:一堆节点、一堆连线(JSON)。LangGraph 需要的是程序化构建:对每个节点 add_node(id, fn),对每条边 add_edge(a, b),分支处用 add_conditional_edges。

所以图引擎的核心任务,就是做这个翻译——并且翻译得正确,尤其是分支、汇聚、循环这些容易出错的结构。

2. 编译五步走​

构造函数末尾只有两行,却是整条编译流水线的起点:

# graph_engine.py:69-70
self.build_edges()
self.build_nodes()

build_nodes()(graph_engine.py:253)内部按顺序做五件事:

① init_nodes        每个 node JSON → 节点对象,add_node 进 LangGraph,
                    顺带标出 start / end / 要中断的节点
② add_edge(START, start_node) / add_edge(end_node, END)
③ build_node_level  给每个节点算「从 start 出发的最长路径深度」
④ add_node_edge     逐节点翻译出边(普通边 / 条件边 / output 的 fake 边)
⑤ build_more_fan_in_node  处理「多入边」节点要不要等(见第 02 章)
最后:graph_builder.compile(checkpointer=..., interrupt_before=...)

下面拆开看关键的三步。

3. ① 节点实例化:NodeFactory 按 type 造对象​

它要解决的小问题: JSON 里每个节点只有个 type 字符串("llm"、"condition"…),得变成真正能执行的对象。

思路: 经典工厂模式——一张 type → 类 的映射表,查表实例化。

# nodes/node_manage.py:16-30(节选)
NODE_CLASS_MAP = {
    NodeType.START.value: StartNode,
    NodeType.LLM.value: LLMNode,
    NodeType.CONDITION.value: ConditionNode,
    NodeType.OUTPUT.value: OutputNode,
    # ... 共 13 种
}

NodeFactory.instance_node(nodes/node_manage.py:38)查表,查不到就抛 Unknown node type。

实例化时,每个节点对象拿到三样关键依赖:它的入边/出边(从 EdgeManage 查)、共享的 GraphState、回调。同时引擎记下三张映射,供后面编译用:

# graph_engine.py:224-228(节选)
self.nodes_map[node_data.id] = node_instance
self.nodes_fan_in[node_instance.id] = self.edges.get_source_node(node_instance.id)  # 入边来源
self.nodes_next_nodes[node_instance.id] = self.edges.get_next_nodes(node_instance.id)  # 全部下游

nodes_fan_in(每个节点有哪些上游)是第 02 章「要不要等」判定的输入。

两个特殊处理(init_nodes,graph_engine.py:200):

  • note 类型直接跳过——它是前端的注释便签,不参与执行(graph_engine.py:209)。
  • 每个 output 节点会额外造一个 OutputFakeNode(graph_engine.py:246),用来承载「输出后要等用户交互」的中断。这是 HITL 的关键技巧,详见第 03 章。

4. ③ 节点层级:算「从 start 到我的最长路径」​

它要解决的小问题: 后面判断「多入边节点要不要等上游」时,需要知道两个节点谁在前、谁在后。node_level 就是这个先后的度量。

思路: 从 start 出发做 DFS,每到一个节点,把它的 level 取「已有值」和「当前深度」的较大者——所以叫最长路径。遇到已访问过的节点就回头,天然处理了循环。

# graph_engine.py:183-198(简化)
def mark_node_level(node_id, node_map, level):
    if node_id in node_map:      # 本条路径上已来过 → 是个环,停
        return
    self.node_level[node_id] = max(self.node_level.get(node_id, 0), level)
    node_map[node_id] = True
    for nxt in self.edges.get_target_node(node_id) or []:
        mark_node_level(nxt, node_map.copy(), level + 1)  # copy:每条路径独立

注意 node_map.copy():用「本条路径的已访问集」而非全局集来判环——这样菱形结构(A→B、A→C、B→D、C→D)里 D 不会被误判成环,只有真正回指祖先才算环。

5. ④ 出边翻译:三种边、三种译法​

add_node_edge(graph_engine.py:76)对每个节点决定怎么连下游。按节点类型分三类:

         ┌─ end / fake_output ───────────→ 不处理出边(end 已连 END;fake 由 output 统一接)
         │
节点类型 ─┼─ output ─────────→ output → output_fake → add_conditional_edges(按用户选择路由)
         │
         ├─ condition ─────→ add_conditional_edges(按条件判定路由)
         │
         └─ 普通节点 ───────→ add_edge(self, target);但若 target 是多入边节点,先跳过
                              (留给 build_more_fan_in_node 统一处理)

普通边最简单(graph_engine.py:111-117):对每个下游 add_edge。但有个关键跳过——如果下游是多入边节点,这里不连,留到第 02 章统一决定要不要等:

# graph_engine.py:114-117
if self.nodes_fan_in.get(node_id) and len(self.nodes_fan_in.get(node_id)) > 1:
    continue   # 多入边节点:延后处理
self.graph_builder.add_edge(node_instance.id, node_id)

条件边用 LangGraph 的 add_conditional_edges(graph_engine.py:103-107):传入节点的 route_node 函数和「可能去的所有目标」的映射。运行时 route_node 返回哪个 id,就走哪条边。ConditionNode.route_node 返回命中条件分支的目标(nodes/condition/condition.py:45)。

6. ⑤ 编译 + 递归上限​

所有边连完后,一次 compile 收尾:

# graph_engine.py:278-281
self.graph = self.graph_builder.compile(checkpointer=MemorySaver(),
                                        interrupt_before=interrupt_nodes)
self.graph_config['recursion_limit'] = max(
    (len(nodes) - len(end_nodes) - 1) * self.max_steps, 1) + len(end_nodes) + 1

两个细节值得记:

  • checkpointer=MemorySaver():用内存检查点保存执行现场,这是「能从断点续跑」的前提(配合 interrupt_before,见第 03 章)。
  • recursion_limit 按「节点数 × 单节点最大执行次数」动态算。因为支持循环,一个节点可能被反复执行;max_steps 是每个节点的执行次数上限(在 BaseNode.run 里强制,nodes/base.py:196),recursion_limit 则是整图的步数上限,防止死循环跑飞。

7. 巧妙之处​

  • 延后连「多入边」的边(graph_engine.py:114)。 普通边翻译时故意跳过多入边节点,把「要不要等」这件难事集中到一个地方(build_more_fan_in_node)决策,而不是散落在各处。关注点分离得干净。
  • fake 节点承载中断(graph_engine.py:246)。 output 本身要先执行(把消息发出去),又要在「等用户选择」处停——一个节点干不了两件事,于是拆成 output(执行)+ output_fake(被 interrupt_before 卡住)。
  • note 直接跳过(graph_engine.py:209)。 前端注释不污染执行图,编译期就过滤掉。

8. 代码地图​

主题文件符号
编译总入口graph/graph_engine.pyGraphEngine.build_nodes
节点实例化graph/graph_engine.pyGraphEngine.init_nodes
出边翻译graph/graph_engine.pyGraphEngine.add_node_edge
节点层级graph/graph_engine.pyGraphEngine.build_node_level
工厂表nodes/node_manage.pyNODE_CLASS_MAP, NodeFactory.instance_node
边查询edges/edges.pyEdgeManage.get_target_node, get_source_node, get_next_nodes