你现在要学的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 → 处理 → 返回更新后的 State3. Edge边 节点之间的 “执行顺序 / 连线”两种边普通边线性A → B按顺序走条件边决策A → 根据判断走 B 或 C三、用生活例子秒懂流程图开始→查天气→生成回答→结束State存「城市、天气结果、最终回答」Node查天气、生成回答每一步都是节点Edge开始 → 查天气 → 生成回答 → 结束四、实战用 LangGraph 实现一个线性工作流功能最简单、最适合入门输入问题步骤 1检索文档模拟步骤 2LLM 生成回答输出最终结果1. 安装依赖pip install -U langgraph langchain-openai python-dotenv2. 完整代码注释超详细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_keyos.getenv(ZHIPU_API_KEY), base_urlos.getenv(ZHIPU_BASE_URL), modelos.getenv(LLM_MODEL), temperature0 ) # 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} # 节点2LLM生成回答 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_keyos.getenv(ZHIPU_API_KEY), base_urlos.getenv(ZHIPU_BASE_URL), modelos.getenv(LLM_MODEL), temperature0 ) # 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} # 节点2LLM生成回答 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() 运行工作流