LangGraph 04：边与条件路由

发表于 更新于 分类于
讲解 add_edge 并行扇出、add_conditional_edges、path_map 与 defer 汇合，并用 Coding Plan 做 LLM 路由示例。

前言

状态与消息就绪之后，图的灵魂在 边：谁先执行、谁并行、按什么条件跳转、如何在汇合点等待。
本文讲普通边与 add_sequence、从 START 扇出并行、add_conditional_edges、path_map，以及节点上的 defer=True 延迟汇合。
最后给出一个用 Coding Plan 做内容类型路由的可运行示例。
LLM 示例继续用 OPENAI_BASE_URL 的 Coding Plan，与 《LangGraph 01：生态与环境》 一致。

依赖

1
uv add langgraph langchain-openai langchain-core python-dotenv rich

.env 仍使用：

1
2
OPENAI_API_KEY=你的火山方舟 API Key
OPENAI_BASE_URL=https://ark.cn-beijing.volces.com/api/coding/v3

实现

普通边与序列

add_edge(源, 目标) 声明固定跳转；START / END 是特殊端点。
多个节点串联时，可用 add_sequence([fn1, fn2, ...]) 按函数名自动建节点并顺序连边（langgraph 1.x）。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
from typing import TypedDict

from langgraph.graph import END, START, StateGraph
from rich import print as rprint


class OverAllState(TypedDict):
    username: str
    greeting: str
    output: str


def node_a(state: OverAllState) -> dict:
    return {"greeting": "Dear " + state["username"]}


def node_b(state: OverAllState) -> dict:
    return {"output": state["greeting"] + "，你好！"}


builder = StateGraph(OverAllState)
builder.add_edge(START, "node_a")
builder.add_sequence([node_a, node_b])
builder.add_edge("node_b", END)

graph = builder.compile()
rprint(graph.invoke({"username": "小黄"}))

等价于手写 START → node_a → node_b → END。
需要自定义节点名或插条件边时，仍建议显式 add_node + add_edge。

并行扇出

从同一源（常为 START）连出多条边时，目标节点会在同一超步 并行 执行。
各分支写不同字段最安全；写同一带 reducer 的字段则由 reducer 合并。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
from typing import TypedDict

from langgraph.graph import END, START, StateGraph
from rich import print as rprint


class OverAllState(TypedDict):
    topic: str
    poem_hint: str
    joke_hint: str


def node_a(state: OverAllState) -> dict:
    return {"poem_hint": f"关于「{state['topic']}」的诗"}


def node_b(state: OverAllState) -> dict:
    return {"joke_hint": f"关于「{state['topic']}」的笑话"}


builder = StateGraph(OverAllState)
builder.add_node("node_a", node_a)
builder.add_node("node_b", node_b)
builder.add_edge(START, "node_a")
builder.add_edge(START, "node_b")
builder.add_edge("node_a", END)
builder.add_edge("node_b", END)

graph = builder.compile()
rprint(graph.invoke({"topic": "猫咪"}))

图会等两条分支都完成后结束；接 LLM 时可缩短墙钟时间，但要注意限流。

条件边与 path_map

add_conditional_edges(源, path_fn) 让 path_fn(state) 返回下一节点名，或返回列表以条件并行扇出。
若希望函数返回业务标签（如 "poem"）而节点叫 node_a，用 path_map 映射。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
from typing import Literal, TypedDict

from langgraph.graph import END, START, StateGraph
from rich import print as rprint


class OverAllState(TypedDict):
    topic: str
    content_type: str
    poem: str
    joke: str


def node_a(state: OverAllState) -> dict:
    return {"poem": f"诗稿：{state['topic']}"}


def node_b(state: OverAllState) -> dict:
    return {"joke": f"笑话：{state['topic']}"}


def route(state: OverAllState) -> Literal["poem", "joke"]:
    return "poem" if "诗" in state["content_type"] else "joke"


builder = StateGraph(OverAllState)
builder.add_node("node_a", node_a)
builder.add_node("node_b", node_b)
builder.add_conditional_edges(
    START,
    route,
    path_map={"poem": "node_a", "joke": "node_b"},
)
builder.add_edge("node_a", END)
builder.add_edge("node_b", END)

graph = builder.compile()
rprint(graph.invoke({"topic": "猫咪", "content_type": "诗"}))
rprint(graph.invoke({"topic": "猫咪", "content_type": "笑话"}))

不写 path_map 时，route 应直接返回已注册的节点名（如 "node_a"）。
多标签列表同样可经 path_map 解析到多个节点。

defer 汇合

汇总节点若与分支一同挂在 START 上，默认可能过早执行、读不到分支结果。
给该节点加 defer=True，等当前挂起任务结束后再跑。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
from operator import add
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph
from rich import print as rprint


class OverAllState(TypedDict):
    topic: str
    poem: str
    joke: str
    logs: Annotated[list[str], add]


def node_a(state: OverAllState) -> dict:
    return {"poem": f"诗：{state['topic']}", "logs": ["a_done"]}


def node_b(state: OverAllState) -> dict:
    return {"joke": f"笑话：{state['topic']}", "logs": ["b_done"]}


def audit_node(state: OverAllState) -> dict:
    status = f"诗={'有' if state.get('poem') else '无'}，笑话={'有' if state.get('joke') else '无'}"
    return {"logs": [f"audit: {status}"]}


builder = StateGraph(OverAllState)
builder.add_node("node_a", node_a)
builder.add_node("node_b", node_b)
builder.add_node("audit_node", audit_node, defer=True)
builder.add_edge(START, "node_a")
builder.add_edge(START, "node_b")
builder.add_edge(START, "audit_node")
builder.add_edge("node_a", END)
builder.add_edge("node_b", END)
builder.add_edge("audit_node", END)

graph = builder.compile()
rprint(graph.invoke({"topic": "猫咪", "poem": "", "joke": "", "logs": []}))

audit 日志中应能看到诗与笑话都已写入。

LLM 路由示例

用 Coding Plan 根据用户意图写入 route，再经 path_map 进入生成节点。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
import os
from typing import Literal, TypedDict

from dotenv import load_dotenv
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import END, START, StateGraph
from rich import print as rprint

load_dotenv()

model = ChatOpenAI(
    model="ark-code-latest",
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_BASE_URL"],
    temperature=0,
)


class OverAllState(TypedDict):
    topic: str
    user_request: str
    route: str
    poem: str
    joke: str


def llm_router(state: OverAllState) -> dict:
    msg = model.invoke(
        [
            SystemMessage(content="只回答 poem 或 joke。想看诗答 poem，想看笑话答 joke。"),
            HumanMessage(content=state["user_request"]),
        ]
    )
    return {"route": "poem" if "poem" in msg.content.lower() else "joke"}


def gen_poem(state: OverAllState) -> dict:
    text = model.invoke([HumanMessage(content=f"写一首关于{state['topic']}的短诗。")]).content
    return {"poem": text}


def gen_joke(state: OverAllState) -> dict:
    text = model.invoke([HumanMessage(content=f"写一个关于{state['topic']}的短笑话。")]).content
    return {"joke": text}


def pick_branch(state: OverAllState) -> Literal["poem", "joke"]:
    return "poem" if state["route"] == "poem" else "joke"


builder = StateGraph(OverAllState)
builder.add_node("llm_router", llm_router)
builder.add_node("gen_poem", gen_poem)
builder.add_node("gen_joke", gen_joke)
builder.add_edge(START, "llm_router")
builder.add_conditional_edges(
    "llm_router",
    pick_branch,
    path_map={"poem": "gen_poem", "joke": "gen_joke"},
)
builder.add_edge("gen_poem", END)
builder.add_edge("gen_joke", END)

graph = builder.compile()
empty = {"route": "", "poem": "", "joke": ""}
rprint(graph.invoke({"topic": "猫咪", "user_request": "来一首诗", **empty}))
rprint(graph.invoke({"topic": "猫咪", "user_request": "讲个笑话", **empty}))

两次 invoke 应分别主要填充 poem 与 joke。
也可用 init_chat_model("openai:ark-code-latest", ...) 替换 ChatOpenAI。

验证

  1. 无 LLM：跑序列、并行、条件/path_map、defer 脚本，核对字段。
  2. 有 LLM：uv run python llm_route.py，对比两次结果的 route / poem / joke。
  3. 可选：print_ascii() 或 draw_mermaid() 检查拓扑。

总结

  1. 固定路径用 add_edge / add_sequence；同级多边即并行扇出。
  2. 动态路径用 add_conditional_edges；标签与节点名不一致时加 path_map。
  3. 与分支同挂上游又要读齐结果时，给汇合节点 defer=True。
  4. 下一篇见 《LangGraph 05：Command 与动态扇出》；消息基础见 《LangGraph 03：消息与多 Schema》。