Appearance
29. 如何在 LangGraph 中实现条件分支和循环？
难度 P0 必背 · 岗位 应用 · 频率 ★★★ · 预计阅读 7 min
本题阅读地图
- 💡 简要回答
- 📝 详细解析
- 🎯 面试总结
👔面试官：如何在 LangGraph 中实现条件分支和循环？
🙋♂️我：条件分支用条件边，循环就是边连回前面的节点。
👔面试官：对，但要讲清楚：add_conditional_edges 的用法，路由函数怎么写，循环的退出条件怎么设。这些是实际编码的关键。
🙋♂️我：路由函数返回下一个节点名？
👔面试官：对，而且条件边可以返回多个可能的目标，在字典里映射。循环要配合退出条件（如最大步数、状态判断），否则可能无限循环。
TL;DR 速记
- 是什么：条件分支靠
add_conditional_edges，循环靠边连回前置节点。- 关键点：路由函数根据 State 返回路径名，再由映射表决定下一个节点。
- 怎么答：能循环不等于能安全运行，必须同时设计退出条件，比如任务完成、步数上限、空结果次数和失败转人工。
图解
💡 简要回答
条件分支：
python
workflow.add_conditional_edges(
"node_a", # 源节点
routing_function, # 路由函数，返回下一个节点名
{ # 目标节点映射
"path_1": "node_b",
"path_2": "node_c"
}
)循环：
- 边连回前面节点：
workflow.add_edge("node_c", "node_a") - 退出条件：路由函数判断（如步数超限、任务完成）
📝 详细解析
条件分支完整示例
python
from langgraph.graph import StateGraph, END
class State(TypedDict):
query: str
documents: list
answer: str
should_continue: bool
def retrieve(state: State):
docs = retriever.get_relevant_documents(state["query"])
return {"documents": docs}
def generate(state: State):
answer = llm.invoke(...)
return {"answer": answer}
def rewrite(state: State):
new_query = llm.invoke("改写查询：" + state["query"])
return {"query": new_query, "should_continue": True}
def route(state: State) -> str:
if len(state["documents"]) == 0 and state.get("should_continue", True):
return "rewrite"
elif len(state["documents"]) == 0:
return "end"
else:
return "generate"
workflow = StateGraph(State)
workflow.add_node("retrieve", retrieve)
workflow.add_node("generate", generate)
workflow.add_node("rewrite", rewrite)
workflow.set_entry_point("retrieve")
workflow.add_conditional_edges("retrieve", route, {
"rewrite": "rewrite",
"generate": "generate",
"end": END
})
workflow.add_edge("rewrite", "retrieve") # 循环
workflow.add_edge("generate", END)循环退出条件
方式一：步数限制
python
def route(state: State) -> str:
if state.get("step", 0) > 5:
return "end" # 超过5步，退出
# ...
return "continue"方式二：状态判断
python
def route(state: State) -> str:
if state.get("task_complete", False):
return "end"
# ...方式三：条件组合
python
def route(state: State) -> str:
step = state.get("step", 0)
docs_found = len(state.get("documents", [])) > 0
if step > 5 or docs_found:
return "generate"
return "rewrite"常见踩坑与反例
踩坑 1：路由函数返回节点名和映射混乱
错误描述：路由函数返回 "generate"，映射表里却写 "ok": "generate"。
正确做法：路由函数返回的是路径 key，必须和 add_conditional_edges 的映射 key 对齐；团队里最好用常量或枚举管理。
踩坑 2：循环没有退出条件
错误描述：rewrite -> retrieve 无限连回去，期待模型自己停。
正确做法：State 中记录 step、失败次数、是否已找到结果；路由函数显式判断完成、超限和失败终态。
踩坑 3：状态字段没有被节点正确更新
错误描述：路由函数判断 documents，但检索节点返回的字段名是 docs。
正确做法：先定义 State 契约，再写节点返回值；路由函数只依赖稳定字段，避免临时变量和隐式状态。
踩坑 4：把所有判断都交给 LLM
错误描述：「让模型判断要不要继续检索就行。」
正确做法：可让模型参与语义判断，但工程边界要用代码兜住，如最大步数、空结果、异常次数和超时。
面试官可能继续追问
追问 1：
add_edge和add_conditional_edges有什么区别？ 答题要点：add_edge是固定流转，源节点完成后一定去目标节点；add_conditional_edges会先执行路由函数，根据 State 选择目标。追问 2：路由函数应该返回什么？ 答题要点：返回路径标识，通常是字符串；这个标识再通过映射表转成具体节点或
END。追问 3：循环次数放在哪里记录？ 答题要点：放在 State 里，每次进入关键节点或每轮循环时更新，路由函数基于它判断是否继续。
追问 4：如何测试条件分支？ 答题要点：构造不同 State 快照单测路由函数；再用端到端用例验证无结果、成功、超限、异常分支都能走到预期终态。
🎯 面试总结
条件分支用 add_conditional_edges，循环是边连回+退出条件。退出条件可以是步数、状态判断或组合。