当前位置：首页 > news >正文

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

news 2026/5/6 6:20:53

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.py`

import 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_key=os.getenv("DOUBAO_API_KEY"), base_url="https://ark.cn-beijing.volces.com/api/v3", model="doubao-pro-32k", temperature=0.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 反思循环工作流 ===================================== 📝 执行节点：生成答案 ✅ 答案生成完成 ===================================== 🔍 执行节点：反思检查答案 评判结果：不合格 反思意见：答案太简单，需要更详细解释原理... ===================================== 📝 执行节点：生成答案 ✅ 答案生成完成 ===================================== 🔍 执行节点：反思检查答案 评判结果：合格 反思意见：答案完整准确... 🎉 工作流完成！最终答案： ...