LangGraph 从入门到实践：构建复杂AI工作流全指南

目录一、LangGraph 核心介绍1.1 基本概述1.2 核心概念1.3 主要特点1.4 使用场景1.5 与LangChain的关系二、LangGraph 快速开始实操落地2.1 环境准备2.2 创建ReAct Agent快速上手2.3 自定义工作流核心实操2.3.1 基础聊天机器人无工具2.3.2 添加工具调用增强能力2.3.3 添加记忆实现多轮对话连贯三、核心进阶状态持久化与记忆管理3.1 记忆的两种类型3.2 InMemoryStore 与 MemorySaver 的区别3.3 结合两种存储的实操示例四、实战案例LangGraph 实现RAG与人工介入4.1 实现RAG检索增强生成4.2 加入分支检索不到答案转人工处理五、总结与拓展在大模型应用开发中简单的线性任务链已无法满足复杂场景需求而LangGraph作为LangChain生态下的图结构工作流框架凭借其状态管理、循环逻辑和多主体协作能力成为构建智能代理Agent的核心工具。本文将从基础介绍到实操落地带你快速掌握LangGraph的核心用法避开冗余细节聚焦实用知识点。一、LangGraph 核心介绍1.1 基本概述LangGraph是LangChain团队开源的一款底层编排框架专为构建基于大语言模型LLM的复杂、有状态、多主体应用而设计。它通过图结构Graph建模工作流相比传统线性工作流具备更强的灵活性和控制能力尤其适合需要循环推理、状态管理和多主体协作的场景如智能代理、多步骤任务处理。核心定位以图结构为核心简化复杂AI工作流的构建、部署与管理支持LLM与外部工具的无缝集成。官方文档https://langchain-ai.github.io/langgraph/1.2 核心概念LangGraph的核心能力源于其图结构设计关键概念可总结为3点图结构Graph Structure工作流被抽象为有向图节点Nodes代表具体操作调用LLM、执行工具等边Edges定义节点执行顺序支持普通边和条件边动态选择下一步。状态管理State Management共享状态贯穿整个图执行过程记录上下文信息每个节点可读取状态、执行任务并更新状态确保多步骤交互的一致性。循环能力Cyclical Workflows区别于LangChain的线性任务链LCEL支持循环逻辑可实现反复推理、动态交互如Agent循环调用工具直到达成目标。1.3 主要特点灵活性精细控制工作流逻辑与状态更新适配复杂业务需求持久性内置状态保存与恢复机制支持错误恢复和长时间任务多主体协作支持多个Agent分工协作通过图结构协调交互工具集成轻松对接外部工具如搜索API和自定义函数人机交互支持“human-in-the-loop”关键步骤可由人类介入决策。1.4 使用场景LangGraph尤其适合以下3类场景覆盖大部分复杂AI应用需求对话代理构建具备上下文记忆、动态调整策略的智能聊天机器人多步骤任务处理需分解为多个阶段的复杂任务如研究、写作、数据分析多代理系统协调多个Agent分工如一个负责搜索、一个负责总结。1.5 与LangChain的关系两者同属LangChain生态但定位不同可互补使用LangGraph是独立模块可单独使用也可结合LangChain组件提示模板、工具接口LangChain擅长线性任务链DAGLangGraph专注于复杂循环和多主体场景。二、LangGraph 快速开始实操落地以下实操基于通义千问模型qwen-max/qwen-plus简化冗余运行日志保留核心代码与步骤可直接复用。2.1 环境准备先安装所需依赖集中整理避免重复!pip install -U langgraph langchain langchain-community dashscope langchain-tavily pymupdf faiss-cpu2.2 创建ReAct Agent快速上手LangGraph内置ReAct架构Agent可快速配置LLM、工具和记忆实现简单的工具调用能力from langchain_community.chat_models.tongyi import ChatTongyi from langgraph.prebuilt import create_react_agent from langgraph.checkpoint.memory import InMemorySaver # 自定义工具示例获取天气 def get_weather(city: str) - str: Get weather for a given city. return fIts always sunny in {city}! # 配置内存存储短期记忆、LLM、工具 checkpointer InMemorySaver() model ChatTongyi(modelqwen-max, temperature0) # 创建ReAct Agent agent create_react_agent( modelmodel, tools[get_weather], checkpointercheckpointer, promptYou are a helpful assistant ) # 运行Agent简化调用无需冗余输出 config {configurable: {thread_id: 1}} agent.invoke( {messages: [{role: user, content: 上海的天气怎样}]}, config )2.3 自定义工作流核心实操自定义工作流是LangGraph的核心能力分为3个关键步骤构建状态、定义节点、配置边以下分场景实现。2.3.1 基础聊天机器人无工具构建最简单的图工作流仅包含一个聊天节点实现基础对话from typing import Annotated from langchain_community.chat_models.tongyi import ChatTongyi from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages # 1. 定义状态存储对话消息 class State(TypedDict): messages: Annotated[list, add_messages] # add_messages自动追加消息 # 2. 创建图构建器定义节点 graph_builder StateGraph(State) llm ChatTongyi(modelqwen-plus, temperature0) # 聊天节点读取状态中的消息调用LLM并返回新消息 def chatbot(state: State): return {messages: [llm.invoke(state[messages])]} # 3. 添加节点和边START→chatbot→END graph_builder.add_node(chatbot, chatbot) graph_builder.add_edge(START, chatbot) graph_builder.add_edge(chatbot, END) # 编译图核心步骤 graph graph_builder.compile() # 运行简化交互逻辑 def stream_graph_updates(user_input: str): for event in graph.stream({messages: [{role: user, content: user_input}]}): for value in event.values(): print(Assistant:, value[messages][-1].content)2.3.2 添加工具调用增强能力给聊天机器人添加Tavily搜索引擎实现“提问→工具调用→回答”的闭环核心是添加工具节点和条件边from langchain_tavily import TavilySearch from langgraph.prebuilt import ToolNode # 1. 初始化工具 tavily_search TavilySearch(max_results5) tools [tavily_search] # 2. 给LLM绑定工具告知可调用的工具 llm_with_tools llm.bind_tools(tools) # 3. 重写聊天节点使用绑定工具的LLM def chatbot(state: State): return {messages: [llm_with_tools.invoke(state[messages])]} # 4. 添加工具节点配置条件边判断是否需要调用工具 graph_builder StateGraph(State) graph_builder.add_node(chatbot, chatbot) tool_node ToolNode([tavily_search]) graph_builder.add_node(tools, tool_node) # 条件边根据LLM输出判断是否调用工具 def route_tools(state: State): messages state.get(messages, []) if not messages: raise ValueError(No messages found in state) ai_message messages[-1] # 有工具调用则跳转到工具节点否则结束 if hasattr(ai_message, tool_calls) and len(ai_message.tool_calls) 0: return tools return END # 配置边START→chatbotchatbot→工具/END工具→chatbot形成闭环 graph_builder.add_conditional_edges(chatbot, route_tools, {tools: tools, END: END}) graph_builder.add_edge(tools, chatbot) graph_builder.add_edge(START, chatbot) # 编译并运行逻辑同前可复用stream_graph_updates函数 graph graph_builder.compile()2.3.3 添加记忆实现多轮对话连贯通过检查点checkpointer实现状态持久化让机器人记住对话上下文核心是编译图时添加记忆存储from langgraph.checkpoint.memory import InMemorySaver # 初始化内存记忆生产环境建议用Redis/Sqlite/Postgres存储 memory InMemorySaver() # 编译图时添加记忆关键步骤 graph graph_builder.compile(checkpointermemory) # 运行时指定thread_id确保状态复用 config {configurable: {thread_id: 1}} # 第一次交互 graph.stream({messages: [{role: user, content: Hi there! My name is Lutianyu.}]}, config) # 第二次交互可记住姓名 graph.stream({messages: [{role: user, content: Remember my name?}]}, config)三、核心进阶状态持久化与记忆管理3.1 记忆的两种类型LangGraph的记忆功能分为短期和长期适配不同场景短期记忆线程范围内的记忆用于单个对话线程通过checkpointer保存可随时恢复确保单轮对话连贯长期记忆跨对话线程共享可自定义命名空间如按用户ID划分用于保存用户偏好、历史信息等实现个性化交互。3.2 InMemoryStore 与 MemorySaver 的区别两者均用于存储数据但定位不同可结合使用InMemoryStore基于内存的临时存储侧重快速访问适合短期会话数据MemorySaver用于状态持久化可将数据保存到数据库适合长期存储、跨会话复用。3.3 结合两种存储的实操示例实现“记忆用户信息→跨线程复用”核心是结合InMemoryStore存储长期记忆和MemorySaver存储短期状态from langgraph.store.memory import InMemoryStore from langchain_community.embeddings import DashScopeEmbeddings import uuid from langgraph.graph import MessagesState # 1. 初始化长期记忆存储结合嵌入模型 in_memory_store InMemoryStore( index{ embed: DashScopeEmbeddings(modeltext-embedding-v1), dims: 1536, } ) # 2. 定义节点调用LLM检索/存储长期记忆 def call_model(state: MessagesState, config, *, store): user_id config[configurable][user_id] namespace (memories, user_id) # 按用户ID划分记忆空间 # 检索用户相关记忆 memories store.search(namespace, querystr(state[messages][-1].content)) info \n.join([d.value[data] for d in memories]) # 存储新记忆当用户要求remember时 last_message state[messages][-1] if remember in last_message.content.lower(): store.put(namespace, str(uuid.uuid4()), {data: User name is Lutianyu}) # 调用LLM传入记忆上下文 response model.invoke( [{role: system, content: fUser info: {info}}] state[messages] ) return {messages: response} # 3. 构建图并编译结合短期记忆和长期记忆 builder StateGraph(MessagesState) builder.add_node(call_model, call_model) builder.add_edge(START, call_model) graph builder.compile(checkpointerMemorySaver(), storein_memory_store)四、实战案例LangGraph 实现RAG与人工介入4.1 实现RAG检索增强生成结合FAISS向量库和LangGraph构建“文档检索→LLM回答”的工作流核心是添加检索节点from langchain_community.embeddings import DashScopeEmbeddings from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_community.vectorstores import FAISS from langchain_community.document_loaders import PyMuPDFLoader from langchain.prompts import ChatPromptTemplate, HumanMessagePromptTemplate # 1. 文档加载、切分、构建向量库 loader PyMuPDFLoader(./data/deepseek-v3-1-4.pdf) # 替换为你的文档路径 pages loader.load_and_split() text_splitter RecursiveCharacterTextSplitter(chunk_size512, chunk_overlap200) texts text_splitter.create_documents([page.page_content for page in pages[:2]]) embeddings DashScopeEmbeddings(modeltext-embedding-v1) db FAISS.from_documents(texts, embeddings) retriever db.as_retriever(search_kwargs{k: 5}) # 2. 定义检索节点根据用户问题检索文档 def retrieval(state: MessagesState): user_query state[messages][-1] if state[messages] else docs retriever.invoke(str(user_query)) # 构建提示模板传入检索结果 prompt ChatPromptTemplate.from_messages([ HumanMessagePromptTemplate.from_template(请根据对话历史和下面提供的信息回答用户问题:\n{query}) ]) messages prompt.invoke(\n.join([doc.page_content for doc in docs])).messages return {messages: messages} # 3. 构建RAG工作流START→检索→聊天→END graph_builder StateGraph(MessagesState) graph_builder.add_node(retrieval, retrieval) graph_builder.add_node(chatbot, chatbot) # 复用之前的chatbot节点 graph_builder.add_edge(START, retrieval) graph_builder.add_edge(retrieval, chatbot) graph_builder.add_edge(chatbot, END) graph graph_builder.compile()4.2 加入分支检索不到答案转人工处理通过条件边实现“检索验证→正常回答/人工介入”的分支逻辑提升应用健壮性from langchain.schema import HumanMessage, AIMessage from typing import Literal from langgraph.types import interrupt, Command # 1. 验证节点判断检索信息是否能回答问题 def verify(state: MessagesState) - Literal[chatbot, ask_human]: verify_msg HumanMessage(请判断已知信息是否能回答用户问题直接输出Y或N) ret llm.invoke(state[messages] [verify_msg]) return chatbot if Y in ret.content else ask_human # 2. 人工介入节点中断流程等待人工输入 def ask_human(state: MessagesState): user_query state[messages][-2].content human_response interrupt({question: user_query}) # 中断并获取人工输入 return {messages: [AIMessage(human_response)]} # 3. 重构工作流添加分支 graph_builder StateGraph(MessagesState) graph_builder.add_node(retrieval, retrieval) graph_builder.add_node(chatbot, chatbot) graph_builder.add_node(ask_human, ask_human) # 配置边START→检索检索→验证→chatbot/ask_humanask_human/chatbot→END graph_builder.add_edge(START, retrieval) graph_builder.add_conditional_edges(retrieval, verify) graph_builder.add_edge(ask_human, END) graph_builder.add_edge(chatbot, END) # 编译需添加记忆支持中断后恢复 graph graph_builder.compile(checkpointerMemorySaver())五、总结与拓展LangGraph的核心价值的是“用图结构解耦复杂工作流”通过节点、边和状态管理实现循环逻辑、多主体协作和状态持久化相比传统线性框架更适合构建复杂AI代理。除本文介绍的功能外LangGraph还支持并行处理、子图管理、历史动作回放调试、多智能体协作等高级特性更多细节可参考官方文档。拓展建议生产环境中将InMemorySaver替换为RedisSaver/PostgresSaver提升状态存储的稳定性结合LangChain的其他组件如提示模板、工具集进一步简化开发流程。