第 6 篇用 add_edge 搭出了第一条 StateGraph，但那条图里所有边都是焊死的——数据从 START 出发，沿固定管道流到 END，跑一万次路线不变。这对流水线够用，对 Agent 不够：模型可能直接回答、也可能要求调用工具；草稿可能一次过审、也可能被打回重写。下一步去哪，取决于运行时的状态，这正是控制流要解决的问题，也是第 6 篇结尾埋下伏笔、本篇要兑现的主题——循环不再靠 while 硬造，路由不再靠 if 层层嵌套，两者都成为图里显式声明、可渲染、可审查的一等结构。
本篇是《AI Agent 开发》系列第 8 篇，代码基于 langgraph 1.2.x（写作时最新 1.2.16），State 写法沿用第 7 篇的 TypedDict 与 reducer。全文按"条件边 → Command 路由 → 循环回边 → 递归上限 → 分支合并 → 实战"展开，看完你能画出 Agent 的标准环形骨架，并亲手搭一个会自我修订的写作循环。
1、静态边与条件边
add_edge("a", "b") 是静态边：编译期定死，a 执行完必到 b。确定性正是它的价值——整张图的骨架一屏读完，任何输入走同一条路，排查问题顺着边看就行。固定的处理流水线（切分 → 检索 → 生成）就该用它。
一旦"下一站"要看数据脸色，就换 add_conditional_edges：它挂在某个节点身后，节点执行完后调用你提供的路由函数，把返回值当作下一节点的名字。一个按问题类型分诊的最小例子：
from typing import Literal, TypedDict
from langgraph.graph import StateGraph, START, END
class State(TypedDict):
question: str
answer: str
def triage(state: State) -> Literal["search", "chat"]:
return "search" if "文档" in state["question"] else "chat"
g = StateGraph(State)
g.add_node("search", lambda s: {"answer": "检索文档中……"})
g.add_node("chat", lambda s: {"answer": "咱们聊点别的"})
g.add_conditional_edges(START, triage) # 条件入口：起点也动态
g.add_edge("search", END)
g.add_edge("chat", END)
print(g.compile().invoke({"question": "帮我查文档"})["answer"])
路由函数 triage 拿到当前 State，返回 "search" 或 "chat"；返回列表则多个目的地并行执行（第 5 节），返回 END 就地结束。两个细节值得注意：条件边也能挂在虚拟节点 START 上，让图的入口随输入变化；add_conditional_edges 还有第三个参数 path_map，当路由函数返回布尔值或业务枚举时，用 {True: "search", False: "chat"} 把返回值翻译成节点名——路由函数保持业务语义，翻译表留在图定义里，两处各司其职。
同一个节点不要同时挂静态出边和动态路由（条件边或
Command）：两条路径都会执行，行为很难排查。
还有一个工程细节：不带 path_map 时，路由函数返回的节点名要到运行时才校验，拼错不会在编译期暴露。给路由函数的返回值加 Literal 注解、或干脆用 path_map 把目的地显式声明出来，都能提前锁死合法去处。条件的更多形态见 Graph API 概览。
2、Command 节点内路由
条件边有个天生限制：路由函数只读。想"边判断边写状态"——比如审稿节点判断完合格与否，还要把结论写进 State 供下游使用——条件边就得把逻辑拆进两个节点。langgraph 1.x 的答案是 Command：节点函数直接返回它，路由与状态更新一步完成：
from typing import Literal
from langgraph.types import Command
from langgraph.graph import END
def critic(state: State) -> Command[Literal["writer", "__end__"]]:
if len(state["draft"]) >= 100:
return Command(goto=END, update={"feedback": "通过"})
return Command(
goto="writer",
update={"feedback": "篇幅太薄，扩写到 100 字以上"},
)
三处值得盯住。update 与普通节点返回字典完全同义，走同一套 reducer 合并规则（第 7 篇）；goto 收单个节点名，也收列表或 END；返回类型注解 Command[Literal["writer", "__end__"]] 不是装饰——LangGraph 靠它登记这个节点的全部合法目的地，图渲染与静态检查都以此为准，"__end__" 就是 END 的字面名。两种写法怎么选：
| 维度 | 条件边 | Command |
|---|---|---|
| 路由逻辑位置 | 图定义处，集中一屏可读 | 各节点函数内部 |
| 更新状态 | 不能，路由函数只读 | update 随路由一并提交 |
| 合法目的地 | 定义处显式可见 | 靠 Command[Literal[...] 注解声明 |
| 典型场景 | 纯粹的二分三分路由 | 算完才知道去哪、结果还要写回 |
经验法则：路由只依赖既有字段、逻辑一眼能看懂，用条件边，全图骨架集中在 builder 里可读性最好；路由依据是节点内刚算出来的结果，用 Command，少拆一个节点、少一次状态搬运。官方在 Thinking in LangGraph 一文里主推 Command 风格的图，多 Agent 的 handoff 也基于它（第 24 篇展开）；graph=Command.PARENT、resume 等完整参数见 Command API 参考。
3、循环与 Agent 回边
现在把第 5 篇手写的 while 循环请进图里。Agent 的标准形态是一个环：模型节点要么给出最终回答，要么发起工具调用；工具执行完把结果回填，模型再看一眼决定下一步——"模型 → 工具 → 模型"转圈，直到模型不再要工具。
START
│
▼
┌──────────┐ tool_calls 非空 ┌───────┐
│ model │──────────────────▶│ tools │
│ │◀──────────────────│ │
└────┬─────┘ 回填结果 └───────┘
│ tool_calls 为空
▼
END
接线只要六行：
from typing import Literal
from langgraph.graph import StateGraph, START, END
# State 为第 7 篇的消息型定义，messages 字段用 add_messages 合并
def has_tool_calls(state) -> Literal["tools", "__end__"]:
last = state["messages"][-1]
return "tools" if last.tool_calls else END
agent = StateGraph(State)
agent.add_node("model", call_model) # 节点实现第 9 篇展开
agent.add_node("tools", tool_node)
agent.add_edge(START, "model")
agent.add_conditional_edges("model", has_tool_calls)
agent.add_edge("tools", "model") # 回边：环的另一半
model 的出边是条件边：tool_calls 非空去 tools，为空直奔 END，调不调工具由模型在运行时决定。tools 的出边是静态回边：执行完必回 model，让模型看到工具结果再决定下一步。条件出边与静态回边咬合成环，这就是一切 ReAct Agent 的骨架，create_react_agent 生成的图核心也是它。
环是图模型对链模型的降维打击：链与 DAG 类编排天生表达不了"回到上游"，想在链式框架里循环，要么外面包一层 while、要么把循环伪装成递归，状态管理的复杂度全部回落给你；而图的执行模型（Pregel 超步）从设计上就允许环存在，循环成了一条普通的边。第 9 篇会给这个骨架填上真实模型与工具，从零复刻预构建 Agent。
4、递归上限与熔断
有环就有转不完的风险，LangGraph 给每个图配了硬上限：recursion_limit，默认 25。它数的是 superstep——每个节点执行一次记一步，不是"循环圈数"。以第 3 节的 Agent 为例，"模型 → 工具 → 模型 → 工具 → 模型（给出回答）"这样一次两轮工具的任务共耗 5 步；model 到 tools 再回 model 一圈耗 2 步，默认额度约够转 12 圈。
超限不是优雅收尾，而是直接抛异常：
from langgraph.errors import GraphRecursionError
try:
result = app.invoke(inputs, config={"recursion_limit": 25})
except GraphRecursionError:
print("触达递归上限：九成是回边的判断条件写反了")
为什么必须硬上限？环里每一步都可能是真金白银的模型调用。一个判断写反的条件边——比如打回条件永远成立——在裸 while 的世界里是无声跑一夜的进程和天亮后的账单；在图里，25 步后一记异常把你叫醒。上限是成本护栏，也是调试信号：撞线的图，先去检查条件边的返回值是不是永远停在同一个分支。
调整额度传 config 即可：config={"recursion_limit": 50}。它是每次调用独立的运行参数，同一个图可以对不同任务给不同额度。原则是按任务最坏路径估算而不是无脑调大——修订类流程 3 到 5 圈足够，deep research 型任务才需要 50 以上。预构建 Agent 吃同一个配置；报错的排查路径见官方 GRAPH_RECURSION_LIMIT 专页。第 5 篇 run_agent 里的 max_turns，就是同一思想的手写版。
5、分支与合并
环之外，第二类高频拓扑是扇出与汇合：一个问题拆成两路并行查，结果合并后统一作答。
START
│
▼
┌───────┐
│ split │
└───┬───┘
┌─────┴─────┐
▼ ▼
┌───────┐ ┌───────┐
│ web │ │ db │
└───┬───┘ └───┬───┘
└─────┬─────┘
▼
┌───────┐
│ merge │
└───────┘
扇出有两种写法：给一个节点挂多条静态出边，下一 superstep 所有目的地并行执行；或者让条件边的路由函数返回名字列表，按运行时数据动态扇出。前者结构在图定义里一目了然，后者连"分成几路"都由数据决定——两路的 split 用静态扇出足够，按文档数扇出就得靠列表或 Send。汇合则更简单——多路都连向同一个下游节点即可：
g = StateGraph(State)
g.add_node("split", split_query) # 拆出两路查询词
g.add_node("web", search_web)
g.add_node("db", search_db)
g.add_node("merge", merge_results)
g.add_edge(START, "split")
g.add_edge("split", "web") # 扇出：两条出边
g.add_edge("split", "db") # 下一超步并行执行
g.add_edge("web", "merge") # 汇合：两条入边
g.add_edge("db", "merge") # merge 收齐后只执行一次
g.add_edge("merge", END)
web 与 db 在同一个 superstep 里并行跑，merge 等两路消息到齐后执行一次，拿到的是合并后的 State——这正是 map-reduce 的 reduce 端。最大的坑在状态：并行节点同时写同一个字段，默认 reducer 会直接拒绝，LangGraph 抛 InvalidUpdateError，要求你显式声明合并策略。解法要么让两路各写各的字段、在 merge 里汇总，要么给字段配 Annotated[list, add_messages] 式的追加 reducer（第 7 篇）。分支数量运行时才定的场景——比如对 N 篇文档并行摘要——用 Send API 动态扇出，模型仍是 map-reduce，细节留给多 Agent 篇。
6、写作审稿修订循环
把前五节拼成一个能跑的例子：writer 写初稿，critic 审稿，不合格打回重写，合格收工。节点先用确定性函数模拟——不烧 token、输出可复现，重点全落在控制流上。先定义 State 与三个节点函数：
from typing import Literal, TypedDict
from langgraph.graph import StateGraph, START, END
class State(TypedDict):
topic: str
draft: str
feedback: str
round: int
def writer(state: State) -> dict:
if not state["draft"]: # 首轮：写初稿
draft = f"《{state['topic']}》讲条件边与循环。"
else: # 修订轮：按反馈补
draft = state["draft"] + "示例：审稿打回后的回边。"
return {"draft": draft, "round": state["round"] + 1}
def critic(state: State) -> dict:
if "示例" not in state["draft"]: # 审稿标准：必须有示例
return {"feedback": "缺少示例，请补充"}
return {"feedback": "通过"}
def revise_or_end(state: State) -> Literal["writer", "__end__"]:
return END if state["feedback"] == "通过" else "writer"
图的定义七行，环由 critic 身后的条件边闭合：
g = StateGraph(State)
g.add_node("writer", writer)
g.add_node("critic", critic)
g.add_edge(START, "writer")
g.add_edge("writer", "critic")
g.add_conditional_edges("critic", revise_or_end) # 环在这里闭合
app = g.compile()
init = {"topic": "控制流", "draft": "", "feedback": "", "round": 0}
for update in app.stream(init, stream_mode="updates"):
for node, delta in update.items():
print(f"{node:8}{delta}")
输出：
writer {'draft': '《控制流》讲条件边与循环。', 'round': 1}
critic {'feedback': '缺少示例，请补充'}
writer {'draft': '《控制流》讲条件边与循环。示例：审稿打回后的回边。', 'round': 2}
critic {'feedback': '通过'}
stream_mode="updates" 的每个块是"哪个节点写入了什么"，正好当循环的运行轨迹用（流式的完整能力第 13 篇展开）。轨迹是 writer → critic → writer → critic：初稿缺示例被打回，第二稿补上后条件边把流向切到 END。注意图定义里 writer → critic 只写了一次，实际转几圈完全由状态决定——声明一次、执行 N 次。接线完成后用 print(app.get_graph().draw_ascii()) 把图打印出来核对拓扑，是性价比最高的自查手段。
升级成真实 Agent 只动节点不动接线：writer 换成一次模型调用，把 feedback 组织进 prompt；critic 换成评审模型，或用第 4 篇的 with_structured_output 输出结构化的合格判定；revise_or_end 一字不改。这正是图模型的工程价值：骨架不关心节点里装的是假函数还是大模型。真实审稿往往两三轮才收敛，默认 25 步绰绰有余；要是把审稿标准写得苛刻到永远不通过，环闭合不了，就轮到第 4 节的 GraphRecursionError 兜底——而更体面的做法是给循环主动加熔断，这正是结尾练习的主题。
7、总结
控制流四件套一句话收束：条件边看状态选下一站，Command 把路由与写状态合一步，回边把图弯成环，recursion_limit 给环上保险。Agent 的能动性来自环，可控性来自这些显式声明、可渲染可审查的图结构——回边写在哪、条件是什么、上限是多少，打开图定义一目了然，这是裸 while 循环永远给不了的可审计性。
练习：给第 6 节的审稿循环加"最多修订 3 次"的强制熔断。
- 提示：决策点只有
revise_or_end，改成"通过或round >= 3即 END" - 进阶：熔断时往 State 写入"已达修订上限"，让下游能区分合格与超限截稿
- 验收：把 critic 的合格条件改成永假，确认第 3 稿后图正常终止而非撞递归上限
对于正准备把裸循环迁进 LangGraph 的场景，建议先用条件边把骨架画出来、跑通最小闭环，再逐个节点评估要不要换 Command。下一篇（第 9 篇）手写 ReAct Agent：给第 3 节的环装上真实模型、工具与消息历史，从零复刻 create_react_agent 的行为。
