这次我们直接聊 LangGraph。如果你已经进入 AI 大模型应用开发阶段或者正在做 Agent 项目大概率听过这个名字。它是目前构建可控、有状态 Agent 最主流的编排框架之一。这篇内容不绕弯直接拆解它是什么、和 LangChain 什么关系、怎么从零搭出第一个 Agent、怎么接 API 跑批量任务以及企业级落地时最容易被忽略的细节。LangGraph 相比纯 LangChain 或直接调模型 API最大的优势是把 Agent 的执行过程变成一张有向图每个节点是明确的处理逻辑每条边定义了流转条件。这意味着多轮对话的上下文、工具的调用顺序、状态的回滚和恢复都变得可管理。对于需要“确定性”的业务场景这个能力很关键。这篇文章会带你完成环境准备、构建第一个可运行的 LangGraph Agent、理解状态与条件路由、做一个 RAG 查询 Agent、把服务封装成 HTTP API并给出批量任务、调试和排查的完整思路。看完你不仅能跑通 demo还能知道企业级项目里哪些环节最容易踩坑。1. LangGraph 核心能力速览能力项说明项目定位基于 LangChain 生态的 Agent 编排框架核心是“图”结构驱动的状态机主要功能多节点工作流、条件路由、状态持久化、多智能体协作、人工介入、流式输出与 LangChain 关系LangGraph 由 LangChain 团队开发可以搭配 LangChain 的模型与工具生态也可以脱离 LangChain 单独使用运行环境Python 3.9 以上依赖包通过 pip 安装不需要特殊硬件显存要求框架本身不消耗显存显存占用取决于底层的 LLM本地模型或 API 模型支持模型OpenAI 兼容接口的模型均可接入也支持通过 LangChain 集成各类模型服务启动方式代码编写 命令行运行可输出到 Web UI 或封装为 FastAPI 服务是否支持 API支持可封装为 REST API也可以调用 LangGraph 官方提供的云服务可选是否支持批量任务支持可通过循环 异步方式批量执行也可配合消息队列做任务调度适合场景Agent 应用、RAG 问答系统、多步工具调用、需要人工审核的流程、企业级工作流这里要重点区分一个概念LangGraph 不是“替代” LangChain 的新框架而是解决 LangChain 在复杂 Agent 场景下“难控制、难恢复、难持久化”问题的一套编排层。它把 Chain 改造成了图把每一步的中间状态变成了可记录、可查询的对象。2. LangGraph 是什么解决什么问题2.1 它是用来解决什么痛点的先看一个真实场景你希望 Agent 能够先查询用户意图再决定调用哪一个工具然后根据工具返回结果生成最终答案。用传统 LangChain Chain 也能做但你会发现几个问题多步链路一旦在中间出错整个流程就得重头跑。每一步之间的状态数据散落在变量里不方便追踪和恢复。想人工介入审核某一步很难嵌入执行链。想调试“用户说这句话之后Agent 为什么调用这个工具”没有清晰的执行记录。LangGraph 直接改变了这个局面。它的核心玩法是把流程定义成一张图。每个节点是一个函数每个边是函数之间的流转条件。框架负责帮你维护整套状态State并且每一步都有 checkpoint 记录。出错了你不用从头跑直接从失败节点恢复。2.2 哪些场景适合用 LangGraph从实际项目看下面几类场景是 LangGraph 的主场复杂的多轮 Agent需要记忆对话历史并且在多轮中保持一致的工具调用逻辑。RAG 问答系统文档检索、上下文组装、答案生成分阶段进行中间还需要判断是否需要二次检索。多智能体协作比如一个 Agent 负责查数据库一个 Agent 负责写报告一个 Agent 负责质检最后汇总输出。人机协同流程AI 生成内容、人工审核修改、再放行输出。故障恢复要求高的任务例如批量处理数据单条失败了不能影响整体队列。反过来如果你的场景只是“调用一次模型 API 拿到 JSON”那完全不需要 LangGraph直接用 SDK 反而更轻量。3. LangGraph 与 LangChain 的区别“LangGraph 和 LangChain 的区别” 是搜索频率非常高的问题这里直接给出结论。LangChain 是一个工具集成框架它做的事情是把不同大模型、向量库、文档加载器、工具 API 封装成统一接口然后通过 Chain 串起来。你写 LLM 应用时LangChain 提供的是“积木块”。LangGraph 关注的是“积木块的组装规则”。它不关心你用哪个模型、哪个向量库它关心的是你的任务有几个阶段每个阶段前置条件是什么执行结果如何流转失败怎么处理。它是流程编排层。有一个容易混淆的点最新版本的 LangChain 已经把很多 Agent 相关能力演进到了 LangGraph 中官方推荐的新 Agent 架构也基于 LangGraph。也就是说你现在学 LangGraph本质上是在学 LangChain 生态里“往企业级走”的那条正规路线。从工程对比来看维度LangChainLangGraph核心抽象ChainGraphStateGraph状态管理靠变量传参链路复杂时难维护全局 State 对象显式定义更新规则流程控制线性为主分支需要额外逻辑节点 条件边天然支持分支和循环错误恢复需要自己写重试支持 checkpoint 恢复人工介入不好嵌入节点间可以设计中断点学习曲线较低中等需要理解图思维4. 环境准备与前置条件4.1 系统与 Python 环境LangGraph 是纯 Python 库跨平台支持很好。Windows、macOS、Linux 都可以运行不需要像本地大模型那样必须配独立显卡。如果你只是用 API 模型普通办公电脑就能跑。环境检查清单项目建议操作系统Windows 10/11、macOS、Ubuntu/CentOS 均可Python3.9 到 3.12 均可建议 3.10 或 3.11包管理工具pip 或 poetry / uv模型服务需要准备一个大模型 API 的 Key或本地部署一个 OpenAI 兼容服务网络要求能正常访问你使用的模型 API 服务即可推荐使用虚拟环境隔离依赖避免污染系统 Python。下面是通用创建流程python -m venv .venv source .venv/bin/activate # Windows 下执行 .venv\Scripts\activate pip install --upgrade pip4.2 安装 LangGraph 相关依赖核心需要安装的包有langgraph和langchain-core。如果你要用 LangChain 的模型封装工具还需要按需安装对应集成包。这里给一个相对完整的安装命令实际操作时按项目需要裁剪pip install langgraph langchain-core langchain-openai如果你的模型服务走 OpenAI 兼容协议langchain-openai就够了。如果要用其他模型服务需要对应安装官方集成包。安装完成后可以验证一下版本python -c import langgraph; print(langgraph.__version__)如果输出版本号说明安装成功。4.3 模型服务准备LangGraph 本身不绑定某个具体模型。你可以在节点函数里直接调用任意模型的 HTTP API也可以借助 LangChain 的ChatOpenAI封装。只要你的模型服务是 OpenAI 兼容协议配置方式都差不多。一个常见的做法是在环境变量里维护模型配置export OPENAI_API_KEY你的密钥 export OPENAI_API_BASEhttps://你的模型服务地址/v1 export OPENAI_MODEL_NAMEqwen-plus # 或者你实际使用的模型名如果使用本地模型只要能暴露一个 OpenAI 兼容的 HTTP 接口LangGraph 层的代码可以保持不变只需要替换base_url和模型名。这也是企业项目的常见姿势先接 API 模型验证逻辑再根据数据安全要求把底座切到私有化部署模型。5. 第一个 Agent从 StateGraph 开始环境准备好之后直接写第一个可运行的 Agent。这个例子不涉及复杂工具目的只有一个理解 StateGraph 的图构建逻辑。5.1 定义状态StateLangGraph 使用TypedDict或 Pydantic 模型来描述状态。状态是图的所有节点共享的数据结构节点的输出会写入状态后续节点从状态读取。from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages class AgentState(TypedDict): # 对话消息列表add_messages 表示新消息追加到已有消息后面 messages: Annotated[list, add_messages]这里的Annotated[list, add_messages]是 LangGraph 一个非常核心的概念它定义了状态字段的“归并方式”。不加add_messages时新值会覆盖旧值加上之后新旧内容合并成列表。这个机制在后面复杂流程中会反复用到。5.2 定义模型调用节点节点就是普通的 Python 函数参数是当前状态返回值是状态更新。from langchain_openai import ChatOpenAI model ChatOpenAI( modelgpt-4o-mini, temperature0.7, ) def chat_node(state: AgentState): response model.invoke(state[messages]) return {messages: [response]}如果上面这个例子报了连接错误通常是因为OPENAI_API_BASE或者密钥没有配置正确。换成你实际使用的 API 服务即可。需要注意的是不同服务的模型名、限流策略、超时时间不一样正式项目里建议把这些参数集中放在配置文件中不要散落在代码里。5.3 构建图并运行from langgraph.graph import StateGraph, START, END graph StateGraph(AgentState) graph.add_node(chat, chat_node) graph.add_edge(START, chat) graph.add_edge(chat, END) app graph.compile()然后调用result app.invoke({ messages: [{role: user, content: 你好用一句话介绍 LangGraph}] }) print(result[messages][-1].content)这段代码的运行逻辑是从START进入chat节点chat节点把用户消息传给模型返回结果后追加到messages最后走到END。整个流程非常直观。从这一步开始你已经能理解 LangGraph 最基本的运行模型图 节点 状态。后面所有复杂功能都是在这个模型上增加节点、增加条件分支、增加持久化。6. 图的核心机制状态、节点、边与条件路由6.1 节点Node与边EdgeLangGraph 的图由节点和边组成。节点是执行单元可以是模型调用、工具调用、数据处理函数等。边表示节点之间的流转。普通边从一个节点无条件流向另一个节点。条件边根据某个函数的返回值决定流向哪一个节点。条件边是 Agent 系统实现“思考后选择工具”的基础。你可以定义一个大模型节点输出“需要检索”或“直接回答”然后根据这个输出决定是进入检索节点还是直接进入回答节点。from typing import Literal def route_next(state: AgentState) - Literal[search, direct_answer]: last_message state[messages][-1].content if 需要检索 in last_message or 检索一下 in last_message: return search return direct_answer graph.add_conditional_edges( router, route_next, { search: search_node, direct_answer: answer_node, }, )很多做 Agent 开发的人一开始不习惯这种写法觉得多绕了一层。但正是因为这张图的存在你才能在开发和生产环境中清晰地看到用户的一个问题进入到哪个节点、执行了哪几步、哪一步消耗了多少 token、哪一步出了问题。6.2 工具调用节点一个真正的 Agent 必然要调用工具。LangGraph 中的工具调用本质上也是一个节点。通常的做法是把工具注册给模型模型根据用户问题决定是否调用工具、传什么参数然后把工具执行结果作为消息反馈给模型生成最终答案。下面是一个通用工具节点的写法from langchain_core.tools import tool tool def get_weather(city: str) - str: 查询指定城市的天气情况。 # 这里调用真实的天气 API示例直接返回固定数据 return f{city} 今天晴气温 22~28 摄氏度。 tools [get_weather] model_with_tools model.bind_tools(tools)在节点执行时先调用model_with_tools如果返回内容里有tool_calls就执行对应的工具再把执行结果放回消息列表最后让模型基于工具结果生成回答。这个“模型 → 工具 → 模型”的循环就是 ReAct 模式的核心。LangGraph 的优势在这里体现得最明显工具调用变成了图中可观察的节点每一步的执行输入输出都被记录在状态里。联调排错时能把日志完整拉出来而不是黑盒执行。6.3 状态归并与持久化状态归并规则决定了一个节点输出如何作用到全局状态。默认行为是覆盖通过Annotated[类型, 归并函数]可以自定义为追加、合并或取最大值。企业级项目里messages字段几乎必用add_messages归并因为你需要保留完整对话历史。持久化则依赖checkpointer。LangGraph 提供了内存和远端持久化方案生产环境建议用 Redis 等外部存储from langgraph.checkpoint.memory import InMemorySaver memory InMemorySaver() app graph.compile(checkpointermemory) config {configurable: {thread_id: user-123}} result app.invoke({messages: [{role: user, content: 记住我的名字是小张}]}, configconfig)给定了thread_id之后同一会话的多次调用之间状态会被自动保存和恢复。这个能力对多轮对话、断线恢复、人工审核流程都非常关键。7. 企业级实战RAG、多智能体与持久化7.1 RAG 查询 AgentRAG 场景是 LangGraph 在企业落地中最常见的形态。典型流程是用户提问 → 判断是否需要检索 → 向量库检索 → 组装 Prompt → 生成回答。用 LangGraph 实现这套流程比你手写 if-else 更清晰。from langchain_core.vectorstores import VectorStore def retrieve_node(state: AgentState): question state[messages][-1].content docs vector_store.similarity_search(question, k4) return {context: docs} def generate_node(state: AgentState): context_text \n\n.join([doc.page_content for doc in state[context]]) prompt f基于以下资料回答问题\n\n{context_text}\n\n问题{state[messages][-1].content} response model.invoke(prompt) return {messages: [response]}这里的context字段也要加入状态定义中。实际项目中还要考虑检索结果为空时怎么办、多路检索如何合并、引用来源如何标注。这些都可以设计成独立的节点在图上串联起来。7.2 多智能体协作模式LangGraph 支持在一个图中同时运行多个 Agent让其相互协作。最常用的两种模式是监督模式一个主 Agent 负责规划把任务分发给子 Agent子 Agent 返回结果后由主 Agent 汇总。流水线模式一个 Agent 的输出作为另一个 Agent 的输入例如先做摘要再做风险识别再生成报告。多智能体并不是越多越好。每个 Agent 节点都会增加一次模型调用也就增加了延迟和 token 成本。从项目经验看能用单 Agent 工具解决的场景不要强行拆成多智能体。多智能体真正适用的场景是子任务边界清晰、每个子任务需要不同的 Prompt 和工具配置、且任务之间可以异步或串行执行。7.3 人工审核与流程中断企业级 Agent 系统经常需要人工介入例如 AI 生成的内容要负责人确认后才能外发。LangGraph 的interrupt_before参数可以做到在进入某个节点前暂停执行config {configurable: {thread_id: order-001}} app.invoke(state, configconfig, interrupt_before[human_review_node])执行到human_review_node前图会暂停并保存当前状态。人工审核通过后从同一个thread_id继续执行即可new_result app.invoke(None, configconfig)这里读取的是上次保存的状态LangGraph 会自动从中断点继续。这种设计大大降低了“人工审核流程”的开发复杂度比起自己维护暂停状态、恢复指令要可靠得多。8. 接口 API 与批量任务8.1 封装为 FastAPI 接口LangGraph 能把图对象直接封装成 API。最简单的方式是用 FastAPI 包一层暴露/chat接口让外部系统调用。from fastapi import FastAPI, HTTPException from pydantic import BaseModel app_fastapi FastAPI() class ChatRequest(BaseModel): user_id: str message: str app_fastapi.post(/chat) def chat(req: ChatRequest): config {configurable: {thread_id: req.user_id}} try: result app.invoke( {messages: [{role: user, content: req.message}]}, configconfig, ) reply result[messages][-1].content return {code: 0, data: {reply: reply}} except Exception as e: raise HTTPException(status_code500, detailstr(e))启动命令uvicorn main:app_fastapi --host 0.0.0.0 --port 8000接口服务设计有一个细节要特别注意thread_id从哪个字段取。建议使用独立业务 ID比如订单号、工单号而不是用户 ID。因为一个用户会有多个会话每个会话的上下文要隔离。8.2 批量任务处理批量任务的核心在于每条任务独立、状态隔离、失败互不影响。在调用同一个 LangGraph 应用时给每条任务分配不同的thread_id即可。tasks [ {task_id: a-001, question: 公司报销制度是什么}, {task_id: a-002, question: 如何提交请假申请}, {task_id: a-003, question: 加班费怎么计算}, ] results [] for task in tasks: try: config {configurable: {thread_id: task[task_id]}} result app.invoke({messages: [{role: user, content: task[question]}]}, configconfig) results.append({task_id: task[task_id], status: success, answer: result[messages][-1].content}) except Exception as e: results.append({task_id: task[task_id], status: failed, error: str(e)}) for r in results: print(r)如果任务量很大建议引入并发和重试机制。Python 中可以使用ThreadPoolExecutor做简单并发也可以用 asyncio 异步调用。更规范的做法是把任务写入消息队列如 Redis 队列由 Worker 消费并执行同时记录执行状态。8.3 批量任务的失败重试模型 API 在调用时经常出现限流、超时、返回格式异常等问题。批量任务的失败重试策略一般遵循以下原则单次任务重试不超过 2 到 3 次。每次重试间隔递增避免加剧限流。对“模型返回了非法 JSON”这类错误重试效果不好应该直接标记失败人工复核。记录每次执行日志方便定位出问题的具体节点。LangGraph 的 checkpoint 让重试成本降到很低因为状态已经持久化重试时可以从失败节点继续走不需要从头调用模型。9. 性能观察与调试方法9.1 如何观察一次调用的耗时和流程LangGraph 提供了可视化调试工具和日志输出。简单场景下直接在节点函数里打印时间戳即可import time def chat_node(state: AgentState): start time.time() response model.invoke(state[messages]) print(f[chat_node] 耗时 {time.time() - start:.2f}s) return {messages: [response]}更标准化的做法是给每个节点加装饰器统一记录执行时间和 token 消耗。生产环境中还可以把日志收集到日志中心按thread_id检索完整调用链路。9.2 耗时的主要瓶颈从实际项目经验看Agent 流程耗时的大头集中在LLM 推理本身这是无法绕开的核心时间。多轮工具调用每增加一轮工具调用就增加一次 LLM 推理。文档解析与检索PDF 解析、向量检索在小数据量时不明显大数据量时差异很大。外部 API 延迟天气、订单查询等第三方接口的响应速度直接决定整条链路耗时。优化思路也对应明确减少不必要的工具调用轮数、用更快的检索方案、给外部 API 加超时和缓存、用较小模型做路由判断、把最终生成交给更强的模型。9.3 如何降低消费成本LLM 应用的企业成本主要是 token 消耗。LangGraph 让你可以精确统计每一步的输入输出从而发现成本黑洞。常见的降本方式有用轻量模型做意图识别和路由让重型模型只处理关键步骤。压缩注入到 Prompt 的上下文。RAG 场景中k值不是越大越好多余文档只会增加输入 token 和噪声。对工具的中间输出做摘要避免把大段原始内容反复传给模型。使用流式输出优化用户体感让首响应更快出现而不是整体耗时缩短。10. 常见问题与排查清单问题现象可能原因排查方式解决方案安装 langgraph 失败Python 版本过低或与现有包冲突执行python --version查看 pip 报错信息升级到 Python 3.10 以上使用干净虚拟环境调用模型报 401 错误API Key 配置错误或过期检查环境变量在代码里临时打印 Key 前几位看是否识别重新生成 Key确认配置正确调用模型报超时网络不通或模型服务响应慢单独测试模型 SDK 调用增加 timeout 参数更换网络环境使用代理不可行时切换服务商图执行后没有返回结果某个节点抛异常且未捕获查看完整堆栈日志定位到节点名在节点函数加 try-except打印上下文多轮对话丢失上下文没有配 checkpointer 或 thread_id 配置不对检查是否传了 config 参数给每次调用传入固定 thread_id条件边路由不符合预期路由函数判断逻辑有问题打印路由函数输入和输出在路由函数里加日志确认分支命中工具调用后模型没继续生成工具结果没有正确放回 messages检查代码是否把 tool result 加入了消息列表按 tool_calls 要求构造 ToolMessage批量任务偶发失败模型限流、外部服务抖动查看失败任务日志增加重试、退避、失败隔离接口并发高时服务崩溃线程池过小或同步阻塞压测时观察 CPU 和连接数改用异步 API限制单进程并发数必要时水平扩容状态字段被意外覆盖没有使用 Annotated 归并函数检查状态更新返回值对需要追加的字段使用 add_messages 或自定义归并11. 最佳实践与安全边界11.1 工程化建议先把固定 Prompt 和参数集中放到配置文件不要散落在代码里。模型名、温度、超时时间这些配置在不同环境开发、测试、生产往往不一样。每个节点保持单一职责。一个节点只做“检索”“调用工具”“生成回答”中的一个事情方便替换和测试。给每个节点加日志。日志记录输入摘要、输出摘要、耗时和成本对生产排错帮助很大。批量任务必须加幂等控制。同一个任务重复执行时不要产生重复的副作用例如重复发送通知。对模型输出做结构化校验。很多生产事故不是模型不回答而是模型回答了错误格式导致下游解析失败。11.2 数据安全与合规边界这一部分非常重要尤其在企业项目中密钥管理OPENAI_API_KEY等敏感信息严禁写进代码仓库统一使用环境变量或密钥管理系统。数据隔离多租户场景中每个租户的thread_id和检索范围必须严格隔离防止越权访问。内容审核Agent 生成结果在对外发布或发送用户之前需要有审核机制。人工审核节点可以先放在关键流程里。版权与授权RAG 场景中使用的文档、知识库内容必须有合法来源和授权。涉及人脸、声音、肖像等敏感信息时必须获得明确授权并遵守相关法律要求。使用边界不要用 Agent 处理超出其能力范围的决策。工具返回的数据要经过校验再放行避免“幻觉内容”被当成真实业务数据使用。12. 总结与下一步LangGraph 的价值已经不需要再证明。它把 LLM 应用从“脚本调用模型”带到了“工程化编排流程”的阶段让复杂 Agent 系统的开发、调试、恢复和部署变成了标准流程。如果你刚开始学我建议的路径是先不要急着追各种高级语法用 StateGraph 写一个最简单的对话节点跑通后再加工具、加条件路由、加载止、加 checkpointer。每一步都在现有图上加一个节点或边而不要上来就仿写一个复杂的多智能体架构。最容易踩的坑有三个第一状态归并规则理解不透彻导致字段被覆盖第二忽略 checkpointer 和 thread_id导致多轮会话丢上下文第三把核心业务逻辑写在不加日志的节点里出问题了完全无法定位。下一步你可以做这几件事把本文的 RAG Agent 例子改成自己业务的知识库问答然后把 FastAPI 接口接给前端应用最后给批量任务加上消息队列和重试机制。这个路径走完LangGraph 的企业级开发能力基本就掌握了。建议先把本文收藏实际动手部署时照着步骤操作能少走很多弯路。