你的Agent流程还在写if-else？LangGraph焊死「图状态机」，从节点编排到循环控制一篇打通

你的Agent流程还在写if-else？LangGraph焊死「图状态机」，从节点编排到循环控制一篇打通

⚠️ 文中代码仅用于原理演示，省略异常捕获、鉴权、重试等工程逻辑，请勿直接复制用于生产环境；SDK 接口以官方最新文档为准。

我写了个Agent，用LangChain的Chain串起来。先搜索，再总结，再翻译。看起来挺顺，一上线全乱套：搜索结果质量不行想重试？Chain说"我只会往前走，不能回头"。翻译完发现信息不够要回头搜？Chain说"我是条单行道，掉不了头"。我开始写if-else判断各种分支，200行代码里150行在写路由逻辑，剩下50行在修bug。代码变成意大利面条，改一个条件断三处。问题不在我的逻辑，问题在于我用错了工具——Chain天生就不是干这个的。

注：开篇为模拟业务场景，用于辅助理解技术原理。

一、LangGraph是什么？一句话说清楚

LangGraph = 把Agent流程画成有向图，节点是动作，边是路由，支持循环和条件分支。

打个比方你就懂了：

• Chain是单行道：从A到B到C，只能往前开，不能掉头，不能绕路
• LangGraph是立交桥：可以掉头、可以分流、可以绕环岛、可以多车道并行

Chain能做到的（线性流程），LangGraph都能做。LangGraph能做的（循环、分支、并行），Chain做不了。

来看一张典型的StateGraph长什么样：

┌─────────────────────────────────────────────────────┐
│ StateGraph 状态图 │
│ │
│ ┌─────────┐ │
│ │ START │ │
│ └────┬────┘ │
│ │ │
│ ▼ │
│ ┌───────────┐ │
│ │ 搜索节点 │ ← Node：一个函数/Agent │
│ └─────┬─────┘ │
│ │ │
│ ▼ │
│ ┌───────────┐ │
│ │ 质量判断 │ ← Conditional Edge │
│ └──┬────┬───┘ 根据State决定下一步 │
│ │ │ │
│ 不够好 │ │ 够了 │
│ ▼ ▼ │
│ ┌────────┐ ┌──────────┐ │
│ │ 换关键词 │ │ 总结节点 │ │
│ │ 重搜 │ └────┬─────┘ │
│ └───┬────┘ │ │
│ │ ▼ │
│ │ ┌──────────┐ │
│ │ │ 翻译节点 │ │
│ │ └────┬─────┘ │
│ │ │ │
│ └────→──────┘ ← 循环：重搜完回到搜索节点 │
│ ▼ │
│ ┌──────┐ │
│ │ END │ │
│ └──────┘ │
└─────────────────────────────────────────────────────┘

这张图里包含了LangGraph的4个核心概念：

理解这4个概念，LangGraph你就懂了一大半。后面所有的高级用法，都是在这4样东西上做组合。

二、3个场景，让你秒懂LangGraph多香

代码说明：文中代码片段仅演示原理，已做精简（省略完整异常处理与工程配置），请勿直接复制上生产；生产用法以官方SDK文档为准。

场景1：自适应搜索Agent

搜索→判断结果质量→不够好换关键词重搜→够了才总结。

用Chain写？没法回头。用if-else写？路由逻辑写到崩溃。用LangGraph？几行搞定：

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, List

# 1. 定义共享状态 —— 所有节点都能读写这个字典
class SearchState(TypedDict):
query: str # 搜索关键词
results: List[str] # 搜索结果
quality: float # 结果质量分（0-1）
summary: str # 最终总结

# 2. 定义节点函数 —— 每个节点接收state，返回要更新的字段
def search_node(state: SearchState) –> dict:
results = search(state["query"]) # 实际接入搜索API
return {"results": results, "quality": score(results)}

def improve_query_node(state: SearchState) –> dict:
new_query = refine(state["query"]) # 搜索质量不行，换个关键词
return {"query": new_query} # 只返回要更新的字段

def summarize_node(state: SearchState) –> dict:
return {"summary": summarize(state["results"])}

# 3. 条件路由函数 —— 返回的是"下一个节点的名字"
def should_retry(state: SearchState) –> str:
if state["quality"] < 0.7:
return "improve_query" # 质量不够，换关键词
return "summarize" # 质量够了，去总结

# 4. 拼装图
graph = StateGraph(SearchState)
graph.add_node("search", search_node)
graph.add_node("improve_query", improve_query_node)
graph.add_node("summarize", summarize_node)

graph.add_edge(START, "search")
graph.add_conditional_edges("search", should_retry) # 条件路由
graph.add_edge("improve_query", "search") # 重搜 → 回到search，形成循环
graph.add_edge("summarize", END)

app = graph.compile()

# 5. 跑起来
result = app.invoke({"query": "LangGraph怎么用", "results": [], "quality": 0.0, "summary": ""})
print(result["summary"])

improve_query执行完回到search——这就是循环。Chain做不到的事，这里一个add_edge搞定。

场景2：多轮代码审查循环

写代码→Review→有问题→修改→再Review→通过→结束。这是典型的"直到满足条件才退出"的循环。

class ReviewState(TypedDict):
code: str
review_feedback: str
review_round: int
passed: bool

def write_code_node(state: ReviewState) –> dict:
return {"code": generate_code(state.get("review_feedback", ""))}

def review_node(state: ReviewState) –> dict:
feedback = code_review(state["code"])
return {
"review_feedback": feedback,
"review_round": state["review_round"] + 1,
"passed": "no issues" in feedback.lower(),
}

def fix_code_node(state: ReviewState) –> dict:
return {"code": fix_code(state["code"], state["review_feedback"])}

# 条件路由：通过就结束，没过就继续修
def review_router(state: ReviewState) –> str:
if state["passed"]:
return END
# ⚠️ 循环一定要有终止条件，否则死循环
if state["review_round"] >= 5: # 最多审5轮
return END
return "fix_code"

graph = StateGraph(ReviewState)
graph.add_node("write_code", write_code_node)
graph.add_node("review", review_node)
graph.add_node("fix_code", fix_code_node)

graph.add_edge(START, "write_code")
graph.add_edge("write_code", "review")
graph.add_conditional_edges("review", review_router)
graph.add_edge("fix_code", "review") # 修完回去再审 → 循环

app = graph.compile()
result = app.invoke({"code": "", "review_feedback": "", "review_round": 0, "passed": False})

注意那个review_round >= 5——循环不设上限，就是给生产环境埋定时炸弹。

场景3：Human-in-the-loop审批流程

Agent生成方案→暂停等人审批→通过则执行→拒绝则重新生成。企业里最常见的需求，也是LangGraph的杀手锏。

from langgraph.checkpoint.memory import MemorySaver

class ApprovalState(TypedDict):
plan: str
approved: bool
result: str

def generate_plan_node(state: ApprovalState) –> dict:
return {"plan": generate_plan()}

def execute_node(state: ApprovalState) –> dict:
return {"result": execute_plan(state["plan"])}

# 条件路由：通过则执行，拒绝则回到生成
def approval_router(state: ApprovalState) –> str:
if state["approved"]:
return "execute"
return "generate_plan" # 被拒了，重新生成

graph = StateGraph(ApprovalState)
graph.add_node("generate_plan", generate_plan_node)
graph.add_node("execute", execute_node)

graph.add_edge(START, "generate_plan")
graph.add_conditional_edges("generate_plan", approval_router)
graph.add_edge("execute", END)

# ⚠️ Human-in-the-loop 必须配合 checkpointer
# 没有checkpointer，中断后无法恢复状态
app = graph.compile(
checkpointer=MemorySaver(),
interrupt_before=["execute"] # 执行前暂停，等人审批
)

# 第一轮：生成方案，到execute前暂停
config = {"configurable": {"thread_id": "thread-1"}}
result = app.invoke({"plan": "", "approved": False, "result": ""}, config)
# 此时停在execute前，plan已生成，等人拍板

# 模拟人工审批
user_approved = True # 人看了方案，决定通过
app.update_state(config, {"approved": user_approved})

# 继续执行 —— 传None表示从断点继续
result = app.invoke(None, config)
print(result["result"])

interrupt_before就是断点。Agent跑到这个节点前面停下来，等你审批完再继续。任何涉及"人拍板"的流程都靠它。

三、LangGraph的4种高级用法

用法1：子图嵌套（Subgraph）

大流程拆成小模块，每个子图独立测试，主图负责调度。

# ===== 子图：搜索+总结是一个完整子流程 =====
class SubState(TypedDict):
query: str
results: str

sub_graph = StateGraph(SubState)
sub_graph.add_node("search", lambda s: {"results": "找到了：" + s["query"]})
sub_graph.add_node("summarize", lambda s: {"results": "总结：" + s["results"]})
sub_graph.add_edge(START, "search")
sub_graph.add_edge("search", "summarize")
sub_graph.add_edge("summarize", END)
sub_app = sub_graph.compile()

# ===== 主图：调用子图 =====
class MainState(TypedDict):
task: str
sub_result: str
final_output: str

def call_subgraph(state: MainState) –> dict:
# ⚠️ 子图的State类型和主图不一定一样
# 需要手动做字段映射
result = sub_app.invoke({"query": state["task"], "results": ""})
return {"sub_result": result["results"]}

def format_output(state: MainState) –> dict:
return {"final_output": f"最终结果：{state['sub_result']}"}

main_graph = StateGraph(MainState)
main_graph.add_node("sub_process", call_subgraph)
main_graph.add_node("format", format_output)
main_graph.add_edge(START, "sub_process")
main_graph.add_edge("sub_process", "format")
main_graph.add_edge("format", END)
main_app = main_graph.compile()

子图的好处是封装和复用——搜索+总结这套流程可以在多个主图里复用，改了只改一处。

用法2：Checkpointer持久化

跑到一半崩溃了？Checkpointer帮你存中间状态，重启后从断点继续。

from langgraph.checkpoint.memory import MemorySaver

# 开发测试用 MemorySaver（存内存）
app = graph.compile(checkpointer=MemorySaver())

# 每次调用用 thread_id 区分不同会话
config = {"configurable": {"thread_id": "user-123-session-1"}}
result = app.invoke(initial_state, config)

# 查看历史状态快照 —— 每个节点执行后都有快照
for snapshot in app.get_state_history(config):
print(f"下一个节点：{snapshot.next}")
print(f"当时状态：{snapshot.values}")
print("—")

# ⚠️ MemorySaver 存内存里，重启就没了
# 生产环境换 SqliteSaver 或 PostgresSaver
# from langgraph.checkpoint.sqlite import SqliteSaver
# app = graph.compile(checkpointer=SqliteSaver.from_conn_string("checkpoints.db"))

Checkpointer存的是每个节点执行后的State快照。作用三个：断点恢复、崩溃重跑、历史回溯调试。

用法3：并行节点执行

多个独立任务同时跑，跑完再合并结果。fan-out分流，fan-in汇聚。

class ParallelState(TypedDict):
query: str
web_results: str
db_results: str
cache_results: str
merged: str

def search_web(state: ParallelState) –> dict:
return {"web_results": "网页搜索结果"}

def search_db(state: ParallelState) –> dict:
return {"db_results": "数据库查询结果"}

def search_cache(state: ParallelState) –> dict:
return {"cache_results": "缓存命中结果"}

def merge_results(state: ParallelState) –> dict:
combined = f"{state['web_results']} | {state['db_results']} | {state['cache_results']}"
return {"merged": combined}

graph = StateGraph(ParallelState)
graph.add_node("web", search_web)
graph.add_node("db", search_db)
graph.add_node("cache", search_cache)
graph.add_node("merge", merge_results)

# fan-out：START同时连到3个节点 → 并行执行
graph.add_edge(START, "web")
graph.add_edge(START, "db")
graph.add_edge(START, "cache")

# fan-in：3个节点都连到merge → 全部完成后才执行merge
graph.add_edge("web", "merge")
graph.add_edge("db", "merge")
graph.add_edge("cache", "merge")
graph.add_edge("merge", END)

app = graph.compile()
# 三条搜索同时跑，不用排队等
# ⚠️ 并行节点如果写同一个State字段，后执行的会覆盖先执行的

关键点：并行节点各写各的字段，合并时再处理。写同一个字段就是写冲突。

用法4：动态路由

不写死路由规则，根据State内容实时决定下一步去哪。

class DynamicState(TypedDict):
user_input: str
intent: str
response: str

def classify_intent(state: DynamicState) –> dict:
# 用LLM判断用户意图
intent = llm_classify(state["user_input"])
return {"intent": intent}

def route_by_intent(state: DynamicState) –> str:
intent = state["intent"]
if intent == "qa":
return "answer_qa"
elif intent == "chitchat":
return "answer_chitchat"
elif intent == "search":
return "answer_search"
else:
return "answer_default" # ⚠️ 必须有默认路由，否则报错

def answer_qa(state: DynamicState) –> dict:
return {"response": "这是知识问答结果"}

# … 其他回答节点省略

graph = StateGraph(DynamicState)
graph.add_node("classify", classify_intent)
graph.add_node("answer_qa", answer_qa)
# … 添加其他节点

graph.add_edge(START, "classify")
graph.add_conditional_edges("classify", route_by_intent)
graph.add_edge("answer_qa", END)

app = graph.compile()
# 同一个图，不同输入走不同路径

动态路由的精髓：路由逻辑不写死在图结构里，而是运行时根据状态决定。用户问知识走QA线，闲聊走闲聊线，搜东西走搜索线。

四、避坑清单：8个暗坑

什么情况不该用LangGraph？

如果你的流程是纯线性的——A→B→C→D，没有分支没有循环——直接用LangChain的LCEL（prompt | model | parser）就够了。LangGraph不是银弹，简单流程上LangGraph等于大炮打蚊子，增加复杂度但不增加价值。图结构只有在需要循环、分支、并行时才值得引入。

五、面试速查表：8道必考题

Q1：LangGraph和LangChain的区别是什么？

LangChain的Chain是线性链式调用，只能A→B→C往前走，不能回头不能分支。LangGraph用StateGraph（有向图）替代链，支持循环、条件分支、并行执行和断点恢复。一句话：Chain适合无脑线性流程，LangGraph适合复杂带分支的Agent流程。

Q2：StateGraph的核心组件有哪些？

四个：State（TypedDict定义的共享状态）、Node（接收state返回部分更新的函数）、Edge（固定连接）、Conditional Edge（根据state动态路由）。图由这四样拼出来，没有别的魔法。

Q3：LangGraph怎么实现循环？

让Edge指向已经执行过的Node。比如graph.add_edge("improve_query", "search")，improve_query执行完又回到search，就形成了循环。退出循环靠Conditional Edge——条件满足时路由到下游节点或END。

Q4：什么是Conditional Edge？和普通Edge有什么区别？

普通Edge是"A执行完一定去B"，固定的。Conditional Edge是"A执行完，根据当前State内容决定去B还是去C"，动态的。用add_conditional_edges("node", router_function)添加，router_function返回下一个节点的名字（字符串）。

Q5：Human-in-the-loop怎么实现的？

两个东西配合：interrupt_before=["node_name"]设置断点，checkpointer保存状态。Agent跑到断点前暂停，状态被checkpointer持久化。人工审批后调用update_state写入决策，再调用invoke(None, config)从断点继续。必须配合checkpointer，否则中断后无法恢复。

Q6：Checkpointer的作用是什么？

在图执行过程中自动保存每个节点的状态快照。作用三个：一是Human-in-the-loop的断点恢复依赖它；二是崩溃后可以从最近快照重跑；三是可以回溯历史状态用于调试。开发用MemorySaver（内存），生产用SqliteSaver或PostgresSaver（持久化）。

Q7：LangGraph的并行执行是怎么做的？

fan-out + fan-in模式。从一个节点（或START）引多条Edge到不同节点，这些节点并行执行。然后把多个并行节点的Edge都连到同一个汇聚节点，LangGraph会等所有并行节点完成后才执行汇聚节点。注意并行节点不要写同一个State字段，避免覆盖。

Q8：什么场景该用LangGraph，什么场景不该用？

该用：流程有循环（多轮迭代）、有条件分支（动态路由）、需要并行、需要人工审批断点、需要状态持久化恢复。不该用：纯线性流程（A→B→C），没有分支没有循环——这种情况LCEL一行链式表达式更简洁。用LangGraph写线性流程，等于杀鸡用牛刀。

六、总结

记住3句话：

万能上手模板（复制即用）：

from langgraph.graph import StateGraph, START, END
from typing import TypedDict

# 1. 定义State
class MyState(TypedDict):
input_data: str
output_data: str

# 2. 定义节点函数 —— 接收state，返回部分更新
def my_node(state: MyState) –> dict:
return {"output_data": process(state["input_data"])}

# 3. 定义条件路由（可选，需要分支时才加）
def my_router(state: MyState) –> str:
if need_retry(state):
return "my_node" # 循环回去
return END # 结束

# 4. 拼图
graph = StateGraph(MyState)
graph.add_node("my_node", my_node)
graph.add_edge(START, "my_node")
graph.add_conditional_edges("my_node", my_router)

# 5. 编译 —— 简单场景不加checkpointer
app = graph.compile()
# 需要持久化时：
# from langgraph.checkpoint.memory import MemorySaver
# app = graph.compile(checkpointer=MemorySaver())

# 6. 执行
result = app.invoke({"input_data": "hello", "output_data": ""})
print(result["output_data"])

把图状态机焊死在你的Agent工作流里，让流程编排从面条代码变成清晰的节点图。

模型能力和SDK API更新很快，实际使用时请以官方最新文档为准。

参考资料

封面动物：蝠鲼 —— 在数据流中优雅滑翔，像LangGraph的状态图流转。