LangGraph 超清晰入门指南：State、Node、Edge + 线性工作流实战

你现在要学的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()= 运行工作流