你现在要学的 LangGraph 是 LangChain 官方下一代工作流 / 智能体编排框架，比普通 Chain 强大太多！

它专门用来做：

• 多步骤工作流
• 决策分支（if/else）
• 循环（Loop）
• 多智能体协作
• 可视化执行流程

我会用最通俗、最不绕弯的方式，一次性讲透 3 大核心概念，然后带你写一个能跑、能看、能调试的线性工作流！

一、先搞懂：LangGraph 到底是什么？

大白话：LangGraph = 用 “图”（Graph）来编排 AI 执行步骤你可以把它理解成：AI 版的流程图 / 工作流引擎

它解决普通 Chain 做不到的事：

• 步骤之间可以跳转
• 可以判断走哪条路
• 可以循环执行
• 可以保存中间状态
• 可以可视化每一步干了什么

二、LangGraph 3 大核心概念（必须背下来）

1. State（状态）

= 工作流的 “共享记事本 / 全局变量”所有 Node 都能读写它，数据在步骤之间传递靠它！

就像：

• 存用户问题
• 存检索结果
• 存 LLM 回答
• 存是否需要继续执行
• 存错误信息

所有步骤共享同一个 State。

2. Node（节点）

= 工作流的 “一个步骤 / 一个函数”每个 Node 就是一个功能：

• 检索文档
• 调用 LLM
• 判断是否继续
• 计算、查询、爬取

每个 Node 接收 State → 处理 → 返回更新后的 State

3. Edge（边）

= 节点之间的 “执行顺序 / 连线”两种边：

• 普通边（线性）：A → B（按顺序走）
• 条件边（决策）：A → 根据判断走 B 或 C

三、用生活例子秒懂

流程图：

开始 → 查天气 → 生成回答 → 结束

• State：存「城市、天气结果、最终回答」
• Node：查天气、生成回答（每一步都是节点）
• Edge：开始 → 查天气 → 生成回答 → 结束

四、实战：用 LangGraph 实现一个线性工作流

功能（最简单、最适合入门）

1. 输入问题
2. 步骤 1：检索文档（模拟）
3. 步骤 2：LLM 生成回答
4. 输出最终结果

1. 安装依赖

pip install -U langgraph langchain-openai python-dotenv

2. 完整代码（注释超详细）

from dotenv import load_dotenv
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
import os

# ===================== 加载环境 =====================
load_dotenv()


# ===================== 1. 定义 State（核心！）=====================
# 这是整个工作流共享的数据结构
class WorkflowState(TypedDict):
    # 用户问题
    question: str
    # 检索到的文档片段
    retrieved_docs: str
    # 最终AI回答
    answer: str


# ===================== 2. 定义 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（每个步骤=一个函数）=====================

# 节点1：模拟文档检索
def retrieve_documents(state: WorkflowState):
    print("🔍 执行节点：文档检索")

    # 从State里拿用户问题
    question = state["question"]

    # 模拟检索到的文档（真实场景这里调用向量库）
    fake_docs = f"模拟检索结果：与问题「{question}」相关的文档内容..."

    # 返回新的State（更新retrieved_docs字段）
    return {"retrieved_docs": fake_docs}


# 节点2：LLM生成回答
def generate_answer(state: WorkflowState):
    print("🤖 执行节点：生成回答")

    # 从State拿数据
    question = state["question"]
    docs = state["retrieved_docs"]

    # 调用LLM
    prompt = f"根据文档回答问题：\n文档：{docs}\n问题：{question}"
    answer = llm.invoke(prompt).content

    # 返回更新后的State
    return {"answer": answer}


# ===================== 4. 构建图（Node + Edge）=====================
builder = StateGraph(WorkflowState)

# 添加节点
builder.add_node("retrieve", retrieve_documents)
builder.add_node("generate", generate_answer)

# 添加边（定义执行顺序：线性执行）
builder.add_edge(START, "retrieve")  # 开始 → 检索
builder.add_edge("retrieve", "generate")  # 检索 → 生成
builder.add_edge("generate", END)  # 生成 → 结束

# 编译工作流
graph = builder.compile()

# ===================== 5. 运行工作流 =====================
if __name__ == "__main__":
    print("🚀 启动 LangGraph 线性工作流")

    # 输入初始状态
    result = graph.invoke({
        "question": "什么是RAG？"
    })

    # 输出最终结果
    print("\n==================== 最终结果 ====================")
    print("问题：", result["question"])
    print("检索内容：", result["retrieved_docs"])
    print("AI回答：", result["answer"])from dotenv import load_dotenv
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
import os

# ===================== 加载环境 =====================
load_dotenv()


# ===================== 1. 定义 State（核心！）=====================
# 这是整个工作流共享的数据结构
class WorkflowState(TypedDict):
    # 用户问题
    question: str
    # 检索到的文档片段
    retrieved_docs: str
    # 最终AI回答
    answer: str


# ===================== 2. 定义 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（每个步骤=一个函数）=====================

# 节点1：模拟文档检索
def retrieve_documents(state: WorkflowState):
    print("🔍 执行节点：文档检索")

    # 从State里拿用户问题
    question = state["question"]

    # 模拟检索到的文档（真实场景这里调用向量库）
    fake_docs = f"模拟检索结果：与问题「{question}」相关的文档内容..."

    # 返回新的State（更新retrieved_docs字段）
    return {"retrieved_docs": fake_docs}


# 节点2：LLM生成回答
def generate_answer(state: WorkflowState):
    print("🤖 执行节点：生成回答")

    # 从State拿数据
    question = state["question"]
    docs = state["retrieved_docs"]

    # 调用LLM
    prompt = f"根据文档回答问题：\n文档：{docs}\n问题：{question}"
    answer = llm.invoke(prompt).content

    # 返回更新后的State
    return {"answer": answer}


# ===================== 4. 构建图（Node + Edge）=====================
builder = StateGraph(WorkflowState)

# 添加节点
builder.add_node("retrieve", retrieve_documents)
builder.add_node("generate", generate_answer)

# 添加边（定义执行顺序：线性执行）
builder.add_edge(START, "retrieve")  # 开始 → 检索
builder.add_edge("retrieve", "generate")  # 检索 → 生成
builder.add_edge("generate", END)  # 生成 → 结束

# 编译工作流
graph = builder.compile()

# ===================== 5. 运行工作流 =====================
if __name__ == "__main__":
    print("🚀 启动 LangGraph 线性工作流")

    # 输入初始状态
    result = graph.invoke({
        "question": "什么是RAG？"
    })

    # 输出最终结果
    print("\n==================== 最终结果 ====================")
    print("问题：", result["question"])
    print("检索内容：", result["retrieved_docs"])
    print("AI回答：", result["answer"])

---

五、逐行精讲（你一定能看懂）

1. State 定义（最重要）

class WorkflowState(TypedDict):
    question: str       # 问题
    retrieved_docs: str # 检索结果
    answer: str         # 最终回答

• 这是全局共享数据
• 所有 Node 都能读写
• 步骤之间传数据靠它

2. Node 就是普通函数

def retrieve_documents(state: WorkflowState):
    return {"retrieved_docs": "结果"}

• 输入：State
• 输出：要更新的字段
• 自动合并到 State 里

3. Edge 定义顺序

builder.add_edge(START, "retrieve")
builder.add_edge("retrieve", "generate")
builder.add_edge("generate", END)

这就是线性工作流：开始 → 检索 → 生成 → 结束

六、运行效果（清晰看到每一步执行）plaintext

🚀 启动LangGraph线性工作流
🔍 执行节点：文档搜索
🤖 执行节点：生成回答

=====================最终结果====================
问题：什么是RAG？
检索内容：模拟检索结果： 与问题「什么是RAG？」相关的文档内容...
AI回答：根据提供的文档内容，无法回答该问题。文档仅显示了“模拟检索结果”的标题，未包含关于“什么是RAG”的具体解释。

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

七、你学会了什么？（核心总结）

✅ State = 工作流共享内存

✅ Node = 一个步骤 / 函数

✅ Edge = 步骤之间的连线

✅ StateGraph = 构建工作流

✅ graph.invoke() = 运行工作流