LangGraph 最强进阶：循环控制 + 条件边（附反思循环工作流实战）

LangGraph 最核心、最强大的能力条件边Conditional Edge实现if/else决策走不同分支循环控制Loop实现重复执行某段逻辑反思、重试、多轮检索反思工作流LLM 先生成答案 → 自我检查 → 不满意就重写 → 直到合格这是做高级 RAG、智能体、推理模型的必备技能一、先搞懂 2 个核心概念超级重要1. 条件边Conditional Edge作用根据 State 里的值决定下一步走哪个节点相当于代码里的if 合格: 结束 else: 重新生成2. 循环Loop作用让工作流回到之前的节点重复执行形成闭环生成 → 检查 → 不合格 → 重新生成 → 再检查...二、我们要做的反思循环工作流超级经典流程完美演示 条件边 循环开始 → 生成答案 → 反思检查 → ↙ ↖ 不合格 合格 ↓ ↓ 重新生成 → 结束功能先生成一个答案反思检查答案是否完整、正确条件判断合格 → 结束不合格 → 回到生成步骤循环重写最多循环 3 次防止死循环三、完整可运行代码带详细注释安装依赖pip install -U langgraph langchain-openai python-dotenv完整代码langgraph_reflection_loop.pyimport os from dotenv import load_dotenv from typing import TypedDict, Literal from langgraph.graph import StateGraph, START, END from langchain_openai import ChatOpenAI # 加载配置 load_dotenv() # 1. 定义 State全局共享数据 class GraphState(TypedDict): question: str # 用户问题 answer: str # 当前答案 reflection: str # 反思意见 loop_count: int # 循环次数防死循环 # 2. 初始化 LLM llm ChatOpenAI( api_keyos.getenv(DOUBAO_API_KEY), base_urlhttps://ark.cn-beijing.volces.com/api/v3, modeldoubao-pro-32k, temperature0.3 ) # 2. 定义 Node每个步骤 # 节点1生成答案 def generate_answer(state: GraphState): print(\n) print( 执行节点生成答案) question state[question] # 调用LLM生成答案 prompt f请详细回答问题{question} answer llm.invoke(prompt).content print(f✅ 答案生成完成) # 返回要更新的State return {answer: answer, loop_count: state.get(loop_count, 0) 1} # 节点2反思检查判断答案是否合格 def reflect_answer(state: GraphState): print(\n) print( 执行节点反思检查答案) question state[question] answer state[answer] # 反思Prompt reflect_prompt f 你是严格检查官。 问题{question} 当前答案{answer} 请判断 1. 答案是否完整、准确 2. 如果不合格请给出修改意见。 输出格式必须严格如下 合格/不合格 意见xxx result llm.invoke(reflect_prompt).content.strip() judgment result.split(\n)[0] opinion result.replace(judgment, ).replace(意见, ).strip() print(f评判结果{judgment}) print(f反思意见{opinion}) return {reflection: opinion} # 3. 定义条件路由函数核心 def should_continue(state: GraphState) - Literal[continue, end]: 条件判断继续循环 或 结束 返回的字符串必须等于下一步的节点名称 judgment state[reflection] count state[loop_count] # 终止条件合格 或 超过3次循环 if 合格 in judgment or count 3: return end else: return continue # 4. 构建流程图 builder StateGraph(GraphState) # 添加节点 builder.add_node(generate, generate_answer) builder.add_node(reflect, reflect_answer) # 1. 开始 → 生成 builder.add_edge(START, generate) # 2. 生成 → 反思 builder.add_edge(generate, reflect) # 3. 反思 → 条件判断核心条件边 builder.add_conditional_edges( reflect, # 从哪个节点出发 should_continue, # 条件判断函数 { # 映射返回值 → 目标节点 continue: generate, # 回到生成节点循环 end: END # 结束 } ) # 编译工作流 graph builder.compile() # 5. 运行 if __name__ __main__: print( 启动 LangGraph 反思循环工作流) final_state graph.invoke({ question: 请详细介绍RAG技术, loop_count: 0 }) print(\n\n) print( 工作流完成最终答案) print(final_state[answer])四、逐行精讲你一定能看懂1. State 增加了循环控制loop_count: int # 记录循环次数防止死循环2. 条件判断函数最关键def should_continue(state): if 合格 or 循环次数 3: return end else: return continue它决定流程走向3. 条件边实现循环builder.add_conditional_edges( reflect, should_continue, { continue: generate, # 循环回去重写 end: END # 结束 } )这就是循环 决策的底层实现五、运行效果你能清晰看到循环 启动 LangGraph 反思循环工作流 执行节点生成答案 ✅ 答案生成完成 执行节点反思检查答案 评判结果不合格 反思意见答案太简单需要更详细解释原理... 执行节点生成答案 ✅ 答案生成完成 执行节点反思检查答案 评判结果合格 反思意见答案完整准确... 工作流完成最终答案 ...六、掌握了 3 个硬核技能✅条件边add_conditional_edges✅循环控制节点跳转形成闭环✅反思工作流LLM 自我检查、自我优化✅防死循环通过计数限制次数