LangGraph 这几年在大模型应用开发里刷屏的频率越来越高。很多刚入门的同学看到 LangGraph、MCP、RAG、Agent 这些词连在一起,第一反应是:这到底是一个体系,还是四个独立的东西?我的判断是,LangGraph 的价值恰恰在于把大模型应用里的状态、流程、工具和知识库串成一条可控的线。这篇文章不讲虚的,直接从“它解决什么问题”开始,然后带你从环境准备、最小案例、Agent 循环、RAG、MCP、多智能体、持久化到部署排查,整条路走一遍。适合两类人:一类是会用 LangChain 但总觉得链式结构不够灵活的开发者,另一类是刚接触 Agent 开发、想找一个能落地的编排框架的同学。
先说结论:LangGraph 不是“比 LangChain 更高级”的版本,而是一种不一样的应用组织方式。它把应用拆成节点、边和状态,用图来描述一次完整的任务流。这个设计带来的直接好处是:模型调用可以循环、可以分支、可以从任意节点恢复,而不是只能从上往下跑一条直线。后面你会看到,Agent 的“决定调用哪个工具、拿结果再回模型”这个循环,用 LangChain 的链写会很别扭,用 LangGraph 就是天然的主干结构。
1. 先搞清楚它到底解决什么问题
1.1 LangGraph 不是 LangChain 的升级版
很多教程把 LangGraph 放在 LangChain 后面讲,容易让人误会成“LangChain 的新版本”。其实两者定位不同。LangChain 更擅长把模型、提示词、检索器、输出解析器凑成一条链,适合流程相对固定、顺序比较明确的场景。LangGraph 更强调状态和图遍历,适合流程里有循环、分支、恢复和人工审核的 Agent 工作流。
换句话说,你写一个“用户提问 -> 拼接提示词 -> 调模型 -> 输出”的固定流程,LangChain 足够。但如果你希望模型判断“这个问题需不需要查知识库?要不要调用工具?调用完工具后要不要继续生成?”,流程就不再是直线了。这个时候用图来建模,维护成本会低很多。
我在实际项目里最明显的体感是:用 LangChain 写第一版 Demo 很快,但一旦增加条件分支、多轮工具调用、失败重试,代码会变成一大坨 if-else。LangGraph 把“下一步去哪”交给边和条件边,代码结构反而更清晰。
1.2 核心价值:图、状态、循环
LangGraph 的三个核心概念并不复杂。
图描述整个应用的执行流程。节点是具体的处理逻辑,比如“调用模型”“执行工具”“检查结果”。边表示从一个节点到另一个节点的路径。状态是节点之间共享的数据容器,每个节点可以读取和更新状态。循环则是通过把边指回前面的节点实现的,这是 Agent 最基本的运行方式。
这个设计带来的最大变化是:你不再用一个“调用函数”的思维写应用,而是把应用当成一张可以反复遍历的图。每次运行,LangGraph 都会维护当前状态,根据边和条件决定下一个节点是谁。这也是为什么 LangGraph 适合做 Agent:因为 Agent 本质上就是一个“模型决策 -> 工具执行 -> 再回到模型”的循环。
1.3 哪些场景才真正需要 LangGraph
不是所有大模型应用都需要上 LangGraph。我的建议是:
- 单轮问答、文本摘要、固定提示词,不需要 Graph,直接调模型即可。
- 一问一答但带知识库检索,可以用简单流程,也可以后期迁到 LangGraph。
- 多轮对话、需要持久化、需要条件分支、需要调用多个工具、需要多个模型或角色协同,LangGraph 就很合适。
- 你要做生产级 Agent,必须考虑状态、失败重试、长期记忆和任务恢复,LangGraph 的 Checkpointer 和 Store 能节省很多自研成本。
这个判断很重要。很多人一上来就搭 Graph,反而把简单问题复杂化。
2. 环境准备:从 Python 到第一个 LangGraph 应用
2.1 安装依赖
LangGraph 支持 Python 和 TypeScript。如果你是入门,优先用 Python,生态更全,教程也更多。先准备一个虚拟环境,不要直接装到系统 Python 里,否则依赖冲突会让你怀疑人生。
python -m venv .venv source .venv/bin/activate # Windows 下用 .venv\Scripts\activate pip install -U langgraph langchain-core langchain-openai如果你是做本地实验,还需要装文本切分、向量存储、文档加载相关的包。常见组合:
pip install langchain-community langchain-text-splitters chromadb这里的版本策略是:先统一升级到最新版,再逐步向下调整。因为 LangGraph 迭代很快,网上很多旧教程的 API 已经变了,安装时就锁定最新版,至少能保证示例代码的命中率高一点。
2.2 选模型:API 还是 Ollama 本地部署
LangGraph 本身不包含模型,它负责编排,真正执行“理解”的还是大模型。你可以用 OpenAI 兼容接口、国内厂商 API,也可以在本地用 Ollama 部署开源模型。
学习阶段我会优先推荐 Ollama 或兼容 API。原因有两个:第一,本地部署没有按次计费压力,你可以反复跑同一个图做实验;第二,不依赖远程商业接口,权限和网络问题少一截。很多平台也提供免费额度,够 Demo 用了。需要注意,本地模型对复杂指令的理解能力不如商业大模型,跑 Agent 场景时可能经常出现“不按指令调用工具”的情况。遇到这种情况,先别怀疑 LangGraph,大概率是模型能力不够。
如果你用 Ollama,先拉一个模型:
ollama pull qwen2.5然后通过 OpenAI 兼容地址访问:
from langchain_openai import ChatOpenAI llm = ChatOpenAI( base_url="http://localhost:11434/v1", api_key="ollama", model="qwen2.5", )这个写法在 LangChain 生态里非常通用。你后面换成真正的商业 API,只需要改 base_url 和 api_key。把模型接入层留好,工作流和模型解耦,是少走弯路的第一步。
注意:不要一上来就同时接入十几个工具和一堆插件。先让一个最简单的图跑通,再逐渐加东西。
2.3 最小可运行的 Graph
LangGraph 的最小案例非常简单。定义一个 State,里面放你想在节点之间传递的数据;定义若干节点函数,每个函数接收 state,返回要更新的字段;然后把节点和边连起来,编译后调用。
from typing import TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): input_text: str output_text: str def process(state: State): return {"output_text": f"收到:{state['input_text']}"} graph = StateGraph(State) graph.add_node("process", process) graph.add_edge(START, "process") graph.add_edge("process", END) app = graph.compile() result = app.invoke({"input_text": "你好 LangGraph"}) print(result)这段代码虽然简单,但已经把 LangGraph 的骨架体现出来了:State 定义数据,节点函数处理数据,图决定执行顺序。如果你能跑通这段代码,后续学习的核心就变成了“怎么往里加节点、加条件、加工具”。
这里最容易出的问题是:节点函数返回的字段名必须和 State 里的字段一致,否则状态更新不会生效。还有就是app.invoke({"input_text": "..."})传的是整个 State 的初始值,不是单独一个参数。
3. 从线性链到图状态机:先写一个能跑的 Agent
3.1 Agent 的本质是一个循环
LangChain 里常见的 Agent 是 AgentExecutor,运行时内部会循环调用模型和工具。LangGraph 的 Agent 就是把这种循环显式画出来。
一个最小 Agent 至少需要三个部分:
- 模型节点:决定下一步是调用工具还是直接回答。
- 工具节点:执行模型选择的工具,把结果写回状态。
- 条件边:根据模型输出决定是继续调工具,还是结束。
在 LangGraph 里没有内置的“固定 Agent 节点”,而是让你用这些基础砖块自己搭。好处是灵活,坏处是初始代码量比 LangChain 多。但只要搭一次,你就理解 Agent 到底是怎么转的。
3.2 一个带工具循环的示例
假设我们要让模型具备“查询当前时间”的能力。先定义工具,再通过bind_tools把工具描述传给模型:
from datetime import datetime from typing import TypedDict, Literal from langchain_core.messages import HumanMessage from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, START, END class AgentState(TypedDict): messages: list def get_current_time() -> str: return datetime.now().strftime("%Y-%m-%d %H:%M:%S") llm = ChatOpenAI( base_url="http://localhost:11434/v1", api_key="ollama", model="qwen2.5", ).bind_tools([get_current_time]) def call_model(state: AgentState): response = llm.invoke(state["messages"]) return {"messages": [response]} def call_tool(state: AgentState): last_message = state["messages"][-1] tool_calls = last_message.tool_calls results = [] for call in tool_calls: if call["name"] == "get_current_time": results.append({ "role": "tool", "content": get_current_time(), "tool_call_id": call["id"], }) return {"messages": results} def should_continue(state: AgentState) -> Literal["tools", "end"]: last_message = state["messages"][-1] if hasattr(last_message, "tool_calls") and last_message.tool_calls: return "tools" return "end" graph = StateGraph(AgentState) graph.add_node("model", call_model) graph.add_node("tools", call_tool) graph.add_edge(START, "model") graph.add_conditional_edges("model", should_continue, { "tools": "tools", "end": END, }) graph.add_edge("tools", "model") app = graph.compile() result = app.invoke({ "messages": [HumanMessage(content="现在几点?")] }) print(result["messages"][-1].content)这段代码展示了 LangGraph 最关键的设计:模型节点和工具节点之间通过条件边形成循环。只要模型还想调用工具,流程就回到模型;如果模型不再调用工具,就沿 END 边结束。这也是所有 Agent 框架背后的通用逻辑。
3.3 条件边:控制流的核心
条件边是 LangGraph 的灵魂。它读当前状态,返回一个字符串,决定下一步进入哪个节点。不要把它理解为 if-else 的替代品,而应该理解为“当前节点执行完后,交给谁”。
实际使用中,条件判断函数要尽量短,只做一件事:根据状态选择路径。复杂逻辑放到节点里做。比如上面should_continue只检查最后一条消息有没有工具调用。如果在这里塞入大量业务判断,图会很难调试。
另外要注意:工具调用后必须把工具结果写回消息列表,再回到模型节点。模型看不到工具结果,就无法继续生成。我在初学阶段经常犯的一个错误是:工具已经执行成功,但忘记把结果塞回 messages,结果模型一直在空转。
4. RAG 与知识库:LangGraph 里做检索增强
4.1 RAG 为什么需要图
RAG(Retrieval-Augmented Generation)的作用,是让模型在生成答案之前,先从外部知识库检索相关资料。传统 RAG 流程是线性的:问题 -> 向量化 -> 检索 -> 拼 Prompt -> 生成。这条链用 LangChain 也能实现,LangGraph 的优势在于可以在 RAG 里加入判断和循环。
比如一个 Agentic RAG 场景:模型先判断“用户的问题是否和知识库相关”。相关就检索,不相关就直接回答。如果第一轮检索结果明显不够,可以改写问题再检索一次。这些都不是单次线性流程,而是带分支和循环的图。用 LangGraph 把这些步骤显式编排,每个环节都能单独检查和恢复。
4.2 常用落地流程
RAG 的第一个难点不是 LangGraph,而是知识库本身。我建议的数据处理链路是:
- 文档加载。
- 文本切分。
- 向量化。
- 写入向量库。
- 建索引和检索器。
- 在图里接入检索节点和生成节点。
切分这一步最值得花时间。固定按 500 字切,对小文档还行,复杂文档很容易把语义切断。可以先用 RecursiveCharacterTextSplitter,按段落、句子两级切,不要一开始就用超大的 chunk 或带重叠的复杂策略。
from langchain_text_splitters import RecursiveCharacterTextSplitter splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=50, separators=["\n\n", "\n", "。", "!", "?", " "], ) chunks = splitter.split_text(document_text)这里的 500 是指单块字符数,不是绝对标准。文档类型不同,最优值差异很大。如果你处理的是代码或表格,切分策略又要变。不要盲信网上某个固定参数,最好抽样 20 条问题,看检索结果和最终答案,再回头调。
向量库可以用 Chroma。它本地运行,零配置文件,适合入门。
from langchain_community.vectorstores import Chroma vectorstore = Chroma.from_texts( texts=chunks, embedding=embedding_model, ) retriever = vectorstore.as_retriever(search_kwargs={"k": 4})如果业务里存在大量实体关联关系,可以再考虑知识图谱辅助召回,但入门阶段先从向量库开始,把基础链路跑通。知识图谱不是 RAG 的必需品,而是复杂度升级选项。
4.3 在 LangGraph 里编排 RAG
把 RAG 放进 LangGraph,不需要把整条链都变成图。更合理的做法是:把检索作为一个节点,生成作为一个节点,中间可以加一个判断节点。
def retrieve_node(state): docs = retriever.invoke(state["question"]) return {"context": docs} def judge_node(state): # 判断检索结果是否足够,不足则触发 rewrite 或直接回答 ... return {"decision": "generate"} def generate_node(state): context = "\n\n".join([d.page_content for d in state["context"]]) prompt = f"基于以下资料回答问题:\n{context}\n\n问题:{state['question']}" answer = llm.invoke(prompt) return {"answer": answer}这种结构的好处是:每个节点可以单独测试。我先给 retrieve_node 传入一个固定问题,看返回的 docs 是不是有用;再单独测 generate_node,看回答是否忠实于上下文。最后把它们连接起来成图。一步步拆,比整个流程跑完再看日志高效得多。
RAG 项目里最常见的失败场景不是图写错,而是检索质量差。如果答案翻来覆去不理想,先看召回的相关文档里到底有没有正确答案。不要把锅全部甩给 LangGraph。
5. 接入 MCP:让模型真正能使用外部工具
5.1 MCP 的本质
MCP 是 Model Context Protocol,模型上下文协议。它解决的是“大模型应用怎么标准化地连接外部数据源和工具”的问题。以前每个 Agent 接入数据库、文件系统、第三方服务,都要写一套专用适配代码;MCP 把工具和资源暴露成统一接口,Agent 只需要按协议调用。
你可以把 MCP Server 理解成一个提供工具的服务端,把 MCP Client 理解成 Agent 和 Server 之间的连接器。LangGraph 生态里已经有适配器,可以把 MCP 工具加载成普通工具,然后接入图里的工具节点。
我个人的看法是:MCP 对 Agent 开发是一次很明显的效率提升。大家不需要为每个工具重复开发接入逻辑,只需要按协议暴露和消费。但它不是万能的。如果你的场景只有一个内部 API,直接写工具函数可能比引一条 MCP 链路更简单。
5.2 在 LangGraph 中使用 MCP 工具
在 LangGraph 里接入 MCP 工具,常见流程是三步:
- 启动或配置一个 MCP Server。
- 用客户端建立连接。
- 把 Server 暴露的工具加载进来,交给 Agent 的工具节点调用。
代码形态很多,不同版本差异明显。给你的建议是:先看官方 Python SDK 示例,重点理解ClientSession、ListTools、CallTool这几个操作。在 LangGraph 集成里,优先使用官方适配器,把 MCP 工具转换成 LangChain Tool 格式,这样就能直接复用前面写的bind_tools流程。
我拿“让 Agent 查询数据库”举例。你可以让 MCP Server 暴露一个query_database工具,Agent 通过工具调用执行 SQL,再把查询结果写回消息。这里有一件事必须重视:数据库工具必须有权限边界。Agent 不应该拥有删除表、改 schema 这类高危权限。你在 MCP Server 层就把权限限制好,而不是依赖模型自觉。这是生产环境的基本意识。
5.3 MCP 的常见坑
MCP 相关坑,我遇到最多的有三类:
- 工具加载成功,但模型不会调用。多半是工具描述写得太简单,模型不知道什么时候用。把工具描述改成“当用户询问订单状态时,使用该工具查询订单表”这样带触发条件的描述。
- 连上了但调用超时。MCP Server 如果执行慢,模型会等得很久。要在服务端做超时控制,并让工具在执行前先返回“开始处理”的状态。
- 本地路径问题。Windows 下配置 Python 环境、可执行文件路径时,经常因为路径分隔符或环境变量导致启动失败。排查时先手动启动 MCP Server,确认能独立运行,再接入 LangGraph。
Windows 上创建 MCP Server,最稳的方法不是直接写在配置里,而是先建一个虚拟环境,在命令行手动python your_server.py跑通,再把这个 Python 解释器路径填进配置。路径里不要有空格和中文字符,能省掉一堆莫名其妙的报错。
6. 多智能体协同:Supervisor + Worker 模式
6.1 要不要拆多智能体
多智能体很容易被滥用。很多需求一个 Agent 加几个工具就能解决,硬拆成“规划者 + 执行者 + 审查者”,结果状态传递复杂、模型互相不信任,整体效果反而不如单 Agent。
什么时候才值得拆?我的判断标准是:
- 任务可以明确划分为不同职责域。
- 每个子任务需要不同的提示词、模型或工具集。
- 你需要让某个角色对结果做审查或纠偏。
- 状态量已经大到单 Graph 状态定义失控。
如果你只是想让一个模型同时用数据库和知识库,不要拆多智能体。一个 Agent 绑定多个工具就够了。拆多智能体是为了控制和隔离,不是为了炫技。
6.2 Supervisor 模式
LangGraph 里最经典的多智能体模式是 Supervisor + Worker。Supervisor 节点负责理解任务、决定把任务分配给哪个 Worker;Worker 执行完成后,把结果交回 Supervisor,由它判断是否还需要下一步。
实现上可以用一个图套一个图。每个 Worker 本身就是一个小型 LangGraph,或者至少是一个节点函数。Supervisor 读当前状态,通过条件边选择 Worker。
伪代码结构大致这样:
def supervisor(state): # 让模型基于 worker 候选列表选择执行者 # 返回 {"next": "worker_a"} 或 {"next": "finish"} ... def worker_a(state): # 执行 A 类任务 return {"results": ...} graph.add_conditional_edges("supervisor", decide_next, { "worker_a": "worker_a", "worker_b": "worker_b", "finish": END, })这种模式最大的好处是职责清晰。Supervisor 只做任务分派,不直接处理具体业务;Worker 只做自己的任务,不关心全局。某个 Worker 出现问题时,可以直接替换节点而不影响其他部分。
6.3 状态和任务分发细节
多智能体最怕的是状态没有约束。多个 Worker 往同一个 State 里写字段,很快会出现命名冲突或数据覆盖。
我的建议是:每个 Worker 负责的字段要独立命名,比如research_result、write_result、review_result。Worker 之间不要直接改对方的数据,需要协作时统一通过 Supervisor 汇总。这样在调试时,你能清楚知道每个结果来自哪个节点。
如果 Worker 数量多、任务长,还要考虑任务失败后的处理。常见做法是给 Worker 节点加一个单独的重试包装,或者让 Supervisor 在 Worker 返回异常时重新分配。不要把所有重试逻辑都堆在图外面,那样状态恢复会非常麻烦。
7. 记忆、持久化、批量任务和部署
7.1 长期记忆与 Checkpointer
Agent 的记忆可以分为短期和长期。短期记忆是当前对话中的消息列表;长期记忆是跨会话保存的用户偏好、历史事实、业务数据。
LangGraph 里的 Checkpointer 主要解决的是断点续跑和历史状态恢复。它会把每次图的执行状态保存下来。下次可以从某个节点恢复,而不是从头再跑。这对长任务、人工审核、失败重试非常有用。
使用方式上,你只需要在compile()时传入一个 checkpointer,并把config里的thread_id固定好。同一个 thread_id 的多次调用会被视为同一会话。
from langgraph.checkpoint.memory import InMemoryCheckpointSaver checkpointer = InMemoryCheckpointSaver() app = graph.compile(checkpointer=checkpointer) result = app.invoke( {"messages": [HumanMessage(content="你好")]}, config={"configurable": {"thread_id": "user-001"}}, )要注意,InMemory 只在进程内存里保存,重启就没了。生产环境需要切换到持久化存储。具体用哪个存储,以你项目的数据库选型为准。
长期记忆的另一个方案是 LangGraph Store。它适合保存跨会话的结构化记忆,比如“用户偏好返回简洁回答”“用户上次咨询过什么”。Store 的细节因版本而异,建议直接对照当前文档动手做一个小案例。
7.2 批量任务:从能跑到稳定跑
很多人把 LangGraph 的 demo 跑通后,直接拿去做批量离线任务,结果出现各种诡异问题:任务卡死、输出丢失、内存暴涨。这里面最核心的一点是:批量不是简单 for 循环。
你需要关注的三个层面:
- 输入准备:把所有任务的输入整理成结构化列表,包含唯一 ID。
- 执行方式:先串行跑 10 条,记录每条耗时和结果;再考虑并发,控制最大并发