LangGraph 最核心、最强大 的能力：

1. 条件边（Conditional Edge）：实现 if/else 决策，走不同分支
2. 循环控制（Loop）：实现重复执行某段逻辑（反思、重试、多轮检索）
3. 反思工作流：LLM 先生成答案 → 自我检查 → 不满意就重写 → 直到合格

这是做 高级 RAG、智能体、推理模型 的必备技能！

一、先搞懂 2 个核心概念（超级重要）

1. 条件边（Conditional Edge）

作用：根据 State 里的值，决定下一步走哪个节点相当于代码里的：

if 合格:
    结束
else:
    重新生成

2. 循环（Loop）

作用：让工作流回到之前的节点，重复执行形成闭环：生成 → 检查 → 不合格 → 重新生成 → 再检查...

二、我们要做的：反思循环工作流（超级经典）

流程（完美演示 条件边 + 循环）

开始 → 生成答案 → 反思检查 →
        ↙        ↖
       不合格      合格
        ↓          ↓
      重新生成 → 结束

功能

1. 先生成一个答案
2. 反思：检查答案是否完整、正确
3. 条件判断：
   • 合格 → 结束
   • 不合格 → 回到生成步骤，循环重写
4. 最多循环 3 次（防止死循环）

三、完整可运行代码（带详细注释）

安装依赖

pip install -U langgraph langchain-openai python-dotenv

完整代码 langgraph_reflection_loop.py

import os
from sys import orig_argv
from typing import TypedDict, Literal

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langgraph.constants import START, END
from langgraph.graph import StateGraph

# 加载环境变量
load_dotenv()

# 定义State 全局共享数据
class GraphState(TypedDict):
    question: str # 用户问题
    answer: str # 当前答案
    reflection: str # 反思意见
    judgment: str  # 评判结果：合格/不合格（新增！）
    loop_count: int # 循环次数（防止死循环）

# 初始化 LLM
llm = ChatOpenAI(
    api_key=os.getenv("ZHIPU_API_KEY"),
    base_url=os.getenv("ZHIPU_BASE_URL"),
    model=os.getenv("LLM_MODEL"),
    temperature=0.3
)

# 定义 Node （每个步骤）
def generate_answer(state: GraphState):
    print("\n===================================")
    print("📝 执行节点：生成答案")
    question = state["question"]

    # 调用LLM生成答案
    prompt = f'请详细回答问题：{question}'
    answer = llm.invoke(prompt).content

    print('✅️答案生成完成')
    # 返回要更新的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]
    print(f"节点2result：{result}")
    opinion = result.replace(judgment, "").replace("意见", "").strip()

    print(f'评判结果：{judgment}')
    print(f'反思意见：{opinion}')

    return {
        "reflection":opinion,
        "judgment":judgment
    }

# 定义条件路由函数 核心
def should_continue(state: GraphState) -> Literal["continue","end"]:

    """
    条件判断： 继续循环 或 结束
    返回的字符串必须等于下一步节点名称
    :param state: 全局变量State
    :return:
    """
    judgment = state["judgment"]
    count = state["loop_count"]

    # 终止条件: 合格 或 超过三次循环
    if judgment == "合格" or count >= 3:
        return "end"
    else:
        return "continue"

# 构建流程图
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()


# 运行
if __name__ == "__main__":
    print("🚀 启动LangGraph 反思循环工作流")

    final_state = graph.invoke({
        "question": "请详细介绍一下LangGraph技术",
        "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 反思循环工作流

===================================
📝 执行节点：生成答案
✅️答案生成完成

===================================
🔍 执行节点：反思检查答案
节点2result：合格
意见：答案结构清晰，涵盖了背景、核心概念、关键特性、代码示例、对比及应用场景，内容准确且详实，符合“详细介绍”的要求。
评判结果：合格
反思意见：：答案结构清晰，涵盖了背景、核心概念、关键特性、代码示例、对比及应用场景，内容准确且详实，符合“详细介绍”的要求。


==============================
🎉 工作流完成！ 最终答案：
**LangGraph** 是由 LangChain 团队开发的一个开源库，专门用于构建有状态、多参与者的应用程序，特别是基于大语言模型（LLM）的智能体。

简单来说，如果说 LangChain 是为了方便开发者通过“链”的方式连接 LLM 和工具，那么 **LangGraph 则是为了解决更复杂的、包含循环、分支和长期记忆的“图”状逻辑而生的。**

以下是对 LangGraph 技术的详细介绍，包括其核心概念、架构原理、关键特性以及应用场景。

---

### 1. 背景与动机：为什么需要 LangGraph？

在 LangGraph 出现之前，构建 LLM 应用通常使用 **LangChain Expression Language (LCEL)** 或传统的链式结构。这些结构本质上是**有向无环图 (DAG)**，即数据单向流动，从输入流向输出，没有回路。

然而，高级的 AI 智能体往往需要更复杂的逻辑：
*   **循环：** 智能体可能需要多次思考、行动、观察结果，然后再思考（ReAct 模式）。这需要代码能够“回环”。
*   **状态管理：** 在多轮对话中，应用需要记住之前的交互、中间步骤的输出以及用户的偏好。
*   **可控性：** 开发者需要精确控制智能体的流程，而不是完全依赖 LLM 的“黑盒”推理。

LangGraph 正是为了解决这些问题，它允许开发者将应用定义为**循环图**，并提供了内置的状态管理机制。

---

### 2. 核心概念

LangGraph 的设计深受 Pregel（Google 的大规模图计算框架）的影响，其核心构建块包括：

#### 2.1 图
这是整个应用程序的结构。在 LangGraph 中，你定义的不再是线性的链，而是一个由节点和边组成的图。这个图可以被编译成一个可执行的应用。

#### 2.2 节点
节点是图中的处理单元。在代码中，一个节点通常就是一个 Python 函数。
*   **输入：** 节点接收当前的“状态”。
*   **处理：** 执行逻辑（例如调用 LLM、调用工具、简单的数学计算等）。
*   **输出：** 节点必须返回对“状态”的更新。

#### 2.3 边
边定义了节点之间的连接关系，决定了数据如何流动。LangGraph 支持几种类型的边：
*   **普通边：** 总是从节点 A 指向节点 B。
*   **条件边：** 根据当前状态或节点的输出，决定下一步走哪条路（例如：如果 LLM 说需要搜索，则去搜索节点；如果不需要，则结束）。
*   **入口边：** 定义图开始时首先进入哪个节点。

#### 2.4 状态
这是 LangGraph 最核心的概念。状态是一个在图的各个节点之间传递的共享数据结构。
*   通常使用 Python 的 `TypedDict` 或 Pydantic 模型来定义。
*   每个节点接收完整的状态，并返回需要更新的部分字段（或者覆盖整个状态）。
*   LangGraph 会自动处理状态的合并与传递。

---

### 3. 关键特性

#### 3.1 内置持久化
LangGraph 原生支持检查点机制。这意味着你可以在图的任何执行步骤保存状态。
*   **作用：** 支持“暂停并继续”执行。例如，智能体执行到一半需要人工介入，你可以暂停，人工操作后，智能体基于之前的记忆继续执行。
*   **时间旅行：** 可以回退到图的任意历史状态。

#### 3.2 人机协作
由于支持暂停和状态保存，LangGraph 非常适合构建需要人工审核的工作流。例如，在智能体发送邮件或执行高危操作前，可以设置一个节点暂停流程，等待人工批准（`interrupt`），然后继续。

#### 3.3 流式传输
LangGraph 提供了细粒度的流式输出支持。你不仅可以获得最终的 LLM Token，还可以实时看到每个节点的输出、每个边的执行情况，这对于调试和构建响应式 UI 非常有用。

#### 3.4 预构建的智能体
LangGraph 提供了一些高级封装，如 `prebuilt` 模块，允许开发者用几行代码快速创建常见的智能体模式（如 ReAct Agent）。

---

### 4. 工作原理与代码逻辑

一个典型的 LangGraph 应用构建流程如下：

1.  **定义状态：** 创建一个类或字典，规定在流程中需要共享哪些数据（如 `messages` 列表）。
2.  **定义节点：** 编写 Python 函数，定义每个步骤的逻辑（如 `call_llm`, `call_tool`）。
3.  **初始化图：** 创建一个 `StateGraph` 实例，传入状态定义。
4.  **添加节点和边：** 使用 `.add_node()`, `.add_edge()`, `.add_conditional_edges()` 方法构建逻辑流。
5.  **设置入口和出口：** 指定图的起点和终点。
6.  **编译图：** 调用 `.compile()` 生成可运行对象，此时可以传入检查点保存器以支持记忆。
7.  **运行图：** 调用 `.invoke()` 或 `.stream()` 启动应用。

**简单的伪代码示例：**

```python
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END

# 1. 定义状态
class AgentState(TypedDict):
    messages: Annotated[list, "这里存储对话历史"]

# 2. 定义节点逻辑
def llm_node(state: AgentState):
    # 调用 LLM 处理消息
    response = my_llm.invoke(state["messages"])
    return {"messages": [response]}

def tool_node(state: AgentState):
    # 调用工具
    tool_result = my_tool.invoke(state["messages"][-1])
    return {"messages": [tool_result]}

# 3. 定义条件边逻辑
def should_continue(state):
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return END

# 4. 构建图
workflow = StateGraph(AgentState)

workflow.add_node("agent", llm_node)
workflow.add_node("tools", tool_node)

workflow.set_entry_point("agent")

workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "tools": "tools",
        END: END
    }
)

workflow.add_edge("tools", "agent") # 工具执行完回到 Agent

# 5. 编译
app = workflow.compile()

# 6. 运行
result = app.invoke({"messages": ["你好"]})
```

---

### 5. LangChain vs. LangGraph

| 特性 | LangChain (LCEL) | LangGraph |
| :--- | :--- | :--- |
| **结构** | 线性链 / 有向无环图 (DAG) | 循环图 |
| **核心能力** | 简单的序列组合、并行处理 | 复杂的流程控制、循环、状态管理 |
| **状态管理** | 需要手动管理 (如 RunnableConfig) | 内置状态机，自动管理 |
| **适用场景** | 简单的 RAG、单次问答、简单的工具调用 | 复杂的智能体、多轮对话、人机交互工作流 |
| **记忆** | 依赖外部记忆对象 (如 History) | 原生支持通过 Checkpointer 实现持久化 |

---

### 6. 典型应用场景

1.  **自主智能体：**
    构建能够自主规划、执行、反思的智能体。例如，一个能够编写代码、运行测试、读取报错、修改代码、再次运行的编程助手。这需要循环结构，LangGraph 最擅长此类任务。

2.  **多智能体系统：**
    LangGraph 可以轻松管理多个节点，每个节点代表一个具有不同角色的 Agent（例如：研究员、作家、编辑）。图结构定义了它们之间的协作流程（例如：研究员写草稿 -> 编辑审核 -> 如果不通过则退回）。

3.  **需要人工介入的工作流：**
    在金融、医疗或企业级应用中，AI 的决策往往需要人工确认。LangGraph 的 `interrupt` 功能可以让 AI 在执行关键动作前暂停，等待人工输入，然后继续。

4.  **复杂的对话管理：**
    对于需要维护复杂上下文、分支对话路径的聊天机器人，LangGraph 的状态机比简单的链式结构更能清晰地表达业务逻辑。

### 7. 总结

**LangGraph 是 LLM 应用开发从“脚本化”向“工程化”迈进的重要一步。**

它不再将 LLM 应用视为简单的“输入-输出”函数，而是将其视为一个**状态机**。通过引入图结构和循环，LangGraph 赋予了开发者构建高度复杂、可控且具备长期记忆能力 AI 应用的能力。如果你正在开发简单的 RAG 应用，LangChain 足够；但如果你要构建复杂的智能体或多步骤自动化工作流，LangGraph 是目前业界最主流、最强大的选择之一。

进程已结束，退出代码为 0

---

六、掌握了 3 个硬核技能

✅ 条件边：add_conditional_edges

✅ 循环控制：节点跳转形成闭环

✅ 反思工作流：LLM 自我检查、自我优化

✅ 防死循环：通过计数限制次数