这次我们来看一套多智能体实战技术拆解LangGraph。如果你正在做 AI 大模型应用开发准备研究 agent、多智能体、工作流编排LangGraph 是目前绕不开的一个框架。很多人看 LangGraph 教程问题往往不是“看不懂概念”而是“概念看了一堆自己写一个多智能体却不知道怎么动手”。这篇博客会直接从架构到代码走一遍用一个小但完整的 Supervisor 多智能体示例把核心组件拆开讲清楚。LangGraph 是 LangChain 团队开源的多智能体编排框架它的核心思路不是继续堆“Agent 类”而是把 Agent 行为建模成一张可执行的状态图。状态在节点之间流动节点里可以放模型调用、工具调用、普通代码函数边上可以写“什么条件走哪个分支”。这种设计的价值在于当你有多个智能体角色、多个并行任务、多个分支决策时可以用图的方式表达并且天然支持人工确认、持久化、恢复和流式输出。这篇文章不是简单罗列概念而是拆成三层来讲第一层是硬件与环境门槛第二层是 LangGraph 的核心组件第三层是一套可以直接改的代码实战。所有代码示例都基于 LangGraph 的常见 API 编写模型调用部分你需要替换成自己的模型服务地址、API Key 和模型名称。看完之后你可以得到一套最小可运行的多智能体框架然后按业务需要改成自己的版本。1. LangGraph 多智能体核心能力速览能力项说明项目类型多智能体编排框架 / AI Agent 应用开发库开源来源LangChain 团队开源核心功能状态图编排、多节点协作、条件路由、并行分支、持久化检查点、流式输出、人机协同主要组件StateGraph、State、Node、Edge、Conditional Edge、Checkpointer、Send / Command、Tool Keeper硬件要求LangGraph 本身对硬件要求极低CPU 可完成编排测试资源占用主要取决于你接入的 LLM 是云端 API 还是本地模型显存占用取决于所接大模型本地部署模型由模型规格决定使用云端 API 时几乎没有显存压力支持平台Windows、macOS、Linux以 Python 环境为主启动方式库方式嵌入 Python 项目也可通过 LangGraph Server / CLI 启动服务是否支持 API支持LangGraph 部署服务会暴露接口具体路径以部署产出为准是否支持批量任务支持用 Python 循环或消息队列批量提交任务即可适合场景客服、数据分析、内容生产、自动化工作流、多角色协作、Agent 工具箱从表格能看出LangGraph 并不是一个“模型”而是一个流程编排层。真正消耗计算资源的是你要调用的 AI 大模型。如果你用的是 OpenAI、通义千问、DeepSeek、Kimi、智谱等 APILangGraph 本身跑在后台只占用内存和少量 CPU如果你用本地部署的模型则需要按模型的参数量准备 GPU 和显存。2. 适用场景与使用边界LangGraph 最适合解决的一类问题是多个智能体需要分工协作并且流程里有明确的状态流转。典型场景包括客户工单系统一个智能体负责意图识别一个负责信息检索一个负责生成回复最后一个人工审核节点确认再发出。数据分析 Agent一个智能体负责将自然语言转换为 SQL一个负责执行查询一个负责解释结果并生成图表代码。内容生产管线一个智能体做素材收集一个做大纲一个写初稿一个做改写润色一个做格式校验。自动化运维流程多节点顺序执行其中某个节点失败可以重试或转入人工处理。也有一些场景不一定适合 LangGraph。如果你的业务只是“单轮问答”直接用模型 API 就好了没必要引入状态图如果你的流程非常固定也不需要动态路由普通代码顺序调用反而更轻。LangGraph 的优势在“状态”和“分支”没有复杂状态流转时框架反而会成为额外负担。使用边界上必须强调三点第一凡是接入真实业务的工具调用都要做权限控制。尤其是让智能体写文件、发消息、调用支付或修改数据库的工具生产环境必须加白名单和人工确认节点。第二涉及用户隐私数据时要注意数据脱敏和存储合规。LangGraph 的 Checkpointer 会把对话状态持久化如果你的状态里包含用户手机号、地址等信息持久化存储本身要满足数据安全要求。第三涉及人脸、声音、肖像、版权素材生成类应用时必须确认素材授权和用户授权。多智能体可以在内容创作流程里大幅提效但不能替使用者解决授权问题。3. 环境准备与前置条件3.1 系统与语言环境LangGraph 是一个 Python 库首先需要准备 Python 环境。建议使用 Python 3.9 及以上版本更稳定的做法是在一个独立的虚拟环境中安装。Windows、macOS、Linux 都支持但部分 LangGraph CLI 功能在 Windows 上体验略弱建议在 Linux 或 WSL 环境中操作。检查环境常用命令python --version pip --version git --version如果没有创建虚拟环境建议先建一个python -m venv langgraph-demo source langgraph-demo/bin/activate # Windows 使用 langgraph-demo\Scripts\activate3.2 模型服务准备LangGraph 本身不直接提供模型能力你需要有一个可调用的 LLM 服务。常见选择有两类云端 APIOpenAI、DeepSeek、通义千问、Kimi、智谱、文心等准备好对应的 API Key 和接口地址。本地模型服务vLLM、Ollama、Xinference 等工具启动一个本地 OpenAI 兼容接口准备好模型名称和 base_url。如果只是测试代码流程可以先不调用真实模型用代码临时返回固定字符串然后再切换到模型 API。这样排查问题时会更容易定位是流程问题还是模型问题。3.3 网络与端口调用云端 API 需要模型服务商自身提供合法、合规的服务访问方式确保网络环境可以正常访问你选择的模型服务。如果后面要启动 LangGraph Server 预览功能注意检查端口是否被占用常见默认端口是 8123 或 2024以实际启动信息和官方文档为准。3.4 项目目录规划建议按下面的结构组织工程langgraph-demo/ ├── .env # API Key 等配置不要提交到代码仓库 ├── main.py # 多智能体主流程 ├── agents/ # 存放不同智能体节点 ├── tools/ # 存放工具函数 ├── inputs/ # 测试输入文件 ├── outputs/ # 输出结果目录 └── logs/ # 运行日志小项目可以直接平铺但养成目录分层的习惯后批量任务和多人协作会轻松很多。4. 安装部署与启动方式4.1 安装 LangGraph 核心依赖基础安装只需要两个包pip install langgraph langgraph-cli这里注意langgraph-cli是可选的它主要用于启动本地预览服务和管理 LangGraph 应用。如果只打算在 Python 代码里使用安装langgraph就够。按实际模型接口补充安装依赖。我用langchain-openai作为模型封装示例pip install langchain-openai python-dotenv具体依赖版本以安装时的官方信息为准不要盲目追求最新版本。项目稳定运行后建议把依赖版本固定到requirements.txtpip freeze requirements.txt4.2 验证安装是否成功python -c import langgraph; print(langgraph.__version__)能正常输出版本号说明环境基本就绪。如果提示缺少模块按报错信息补装。4.3 LangGraph Server 启动方式如果希望用 Web 界面查看图结构和对话记录可以使用 LangGraph CLI。以脚本方式定义好langgraph.json配置后执行langgraph dev也可以直接运行 Python 服务入口python main.py这两种方式各有侧重langgraph dev更偏向调试和可视化直接运行 Python 脚本更接近生产部署。按官方文档配置即可不需要两个都跑起来。5. LangGraph 多智能体核心组件拆解要读懂 LangGraph 代码必须先弄明白下面几个核心组件。我把它们按“从抽象到具体”的方式排列。5.1 State全局状态的唯一事实来源State 是 LangGraph 的“内存表”。你可以定义它为一个TypedDict或 Pydantic 模型所有节点都读取它、更新它。多智能体协作时每个节点只关注自己需要的字段但所有节点共享同一个 State 实例。5.2 StateGraph把节点和边组装成图StateGraph是 LangGraph 的核心图容器。先初始化一个图然后不断往里面add_node添加节点用add_edge或add_conditional_edges添加边最后调用compile()生成可执行对象。5.3 Node流程里的执行单元一个 Node 就是一个普通 Python 函数输入是 State输出是一个字典表示你想更新到全局状态里的字段。Node 内部可以是模型调用、工具调用、普通代码、甚至是另一个子图的入口。5.4 Edge 与 Conditional Edge路径控制普通边表示“无条件走到下一个节点”条件边则表示“根据某个函数返回值跳到不同节点”。这是多智能体路由的核心。5.5 Checkpointer状态的持久化与恢复Checkpointer 能把每一步状态保存下来。运行中断后可以恢复也可以在流程图中间加入人工确认节点。LangGraph 提供MemorySaver等实现适合单机调试。5.6 Send 与 Command并发扇出与动态分发Send用于把一个父节点拆分成多个并行子任务。比如列出 100 个任务然后用Send(task_node, task_state)让每个任务独立运行。Command则是一种更灵活的节点返回值形式可以在一个节点里同时更新状态、跳转路径或与外部交互。5.7 LangChain 与 LangGraph 的关系一个常见问题是 “LangChain 和 LangGraph 的区别”。简单说LangChain 是一套丰富的工具和组件库包含模型封装、提示词模板、输出解析器、文档加载器等LangGraph 是集中处理“流程编排”和“状态持久化”的框架层。你可以只用 LangChain 的模型封装配合 LangGraph 做流程也可以完全不依赖 LangChain直接用原生 OpenAI SDK 调用模型LangGraph 只负责编排。理解这一点后写代码时就不会被生态绑定住。6. 代码实战搭建一个可运行的 Supervisor 多智能体下面用一个“主管-研究员-写手”的小型多智能体来演示。主管节点负责接收任务并决定路由如果任务是查询类交给研究员节点如果任务是写作类交给写手节点研究员查完信息后写手再基于研究员的结果写最终回答。这个示例虽然简单但已经把状态定义、多节点、条件路由、并行扩展、持久化检查点都覆盖到了。先创建main.pyimport os from typing import TypedDict from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver from langchain_openai import ChatOpenAI # 替换为你的模型服务配置 llm ChatOpenAI( base_urlos.getenv(MODEL_BASE_URL, https://api.openai.com/v1), api_keyos.getenv(MODEL_API_KEY), modelos.getenv(MODEL_NAME, gpt-4o-mini), temperature0.2, ) class AgentState(TypedDict): task: str # 用户原始任务 resolved: str # 研究员查到的信息 written: str # 写手生成的正文 final_answer: str # 最终汇总输出 # 节点1主管判断任务类型 def supervisor_node(state: AgentState): task state[task] # 这里是简化的路由判断真实场景可以把 task 交给模型分类 if 查 in task or 数据 in task or 检索 in task: route researcher else: route writer print(f[supervisor] 路由到 - {route}) # 通过返回值中的 Command 字段跳转 return {final_answer: waiting, command: {goto: route}} # 节点2研究员执行检索类任务 def researcher_node(state: AgentState): task state[task] # 真实项目中这里会调用搜索工具或知识库工具 prompt f请简要说明{task}输出要点式回答。 response llm.invoke(prompt) return {resolved: response.content} # 节点3写手基于研究员结果或直接根据任务写作 def writer_node(state: AgentState): task state[task] resolved state.get(resolved, ) if resolved: prompt f根据下面的检索信息写一份结构化正文\n{resolved} else: prompt f写一篇结构化正文主题是{task} response llm.invoke(prompt) return {written: response.content} # 节点4汇总节点 def final_node(state: AgentState): written state.get(written, ) resolved state.get(resolved, ) final_answer f检索信息\n{resolved}\n\n最终正文\n{written} return {final_answer: final_answer} # 构建图 graph_builder StateGraph(AgentState) # 添加节点 graph_builder.add_node(supervisor, supervisor_node) graph_builder.add_node(researcher, researcher_node) graph_builder.add_node(writer, writer_node) graph_builder.add_node(final, final_node) # 添加边START - supervisor graph_builder.add_edge(START, supervisor) # 条件边supervisor 根据 command 字段跳转 graph_builder.add_conditional_edges( supervisor, lambda state: state.get(command, {}).get(goto), {researcher: researcher, writer: writer}, ) # 研究员和写手都汇总到 final graph_builder.add_edge(researcher, final) graph_builder.add_edge(writer, final) graph_builder.add_edge(final, END) # 加入检查点支持状态持久化和人工中断 checkpointer MemorySaver() app graph_builder.compile(checkpointercheckpointer) def run_task(task: str): config {configurable: {thread_id: demo-thread-1}} for event in app.stream({task: task}, config): print(事件:, event) if __name__ __main__: run_task(帮我查一下 LangGraph 多智能体的基本概念并整理成短文)这段代码里的command字段是一种简化写法实际使用中建议直接使用 LangGraph 的Command类型让状态更新和路由更规范。像command{goto: researcher}这种结构在早期版本里常用新版本里更推荐显式返回Command(goto...)。具体语法以你当前安装版本的官方文档为准。为了展示Command的标准用法这里补充一个改写片段from langgraph.types import Command def supervisor_node(state: AgentState): task state[task] if 查 in task or 数据 in task or 检索 in task: return Command(gotoresearcher, update{final_answer: waiting}) return Command(gotowriter, update{final_answer: waiting})然后启动服务# 先设置模型相关环境变量 export MODEL_BASE_URL你的模型服务地址 export MODEL_API_KEY你的APIKey export MODEL_NAME你的模型名称 python main.py如果配置正确运行后会在终端看到类似的事件流输出最终状态里会出现检索信息和最终正文。7. 功能测试与效果验证代码跑通只是第一步完整测试才是把多智能体接入业务的保障。建议按下面几条维度验证。7.1 多次连续对话是否保持状态测试项操作预期结果连续调用用同一个 thread_id 连续调用两次第二次能读到第一次写入的状态新会话隔离换一个 thread_id 调用两个会话状态互不干扰更新旧状态手工往 State 写一个字段再触发节点节点读取到的是最新值LangGraph 的 Checkpointer 在这里起关键作用。如果没有配置 Checkpointer连续调用不会记住上次状态配置后同一个 thread_id 就相当于一个持久化会话。7.2 路由是否正确给supervisor_node设计不同类型的测试任务“查一下最新的 AI 大模型开源项目”“写一篇关于多智能体架构的文章”从输出事件里观察路由分支是否分别走到 researcher 和 writer。判断标准是任务类型和路由结果匹配且最终输出没有被错误跳过。7.3 工具节点是否稳定真实项目中 researcher 节点通常会调用搜索引擎或知识库工具。建议先把工具函数写在tools/目录下单独写好单元测试再接入 LangGraph 图。工具节点最容易出问题的不是流程而是超时和异常返回。工具必须能够稳定返回一个字符串特殊情况下返回“未找到相关信息”而不是直接抛异常。7.4 外部系统对接验证多智能体通常会先经过外部系统再进入 LangGraph。建议顺序外部系统生成 JSON 请求。LangGraph 入口函数校验请求字段。完成图执行后返回 JSON 结果。外部系统读取结果。测试时至少覆盖正常输入、缺字段输入、空字符串输入、超长输入四种情况。多智能体的入口函数要像接口一样做参数校验不能假设外部系统传进来的数据一定干净。8. 接口 API 与批量任务LangGraph 的部署形态有两种一种是把图编译后嵌入自己的 Python 服务对外提供你自己的 API另一种是使用 LangGraph Server 自带的服务接口。无论用哪种客户端调用方式都非常接近。8.1 在 FastAPI 中暴露一个运行接口from fastapi import FastAPI from pydantic import BaseModel from main import app as graph_app api FastAPI() class RunRequest(BaseModel): thread_id: str task: str class RunResponse(BaseModel): final_answer: str api.post(/agent/run, response_modelRunResponse) async def run_agent(req: RunRequest): config {configurable: {thread_id: req.thread_id}} final_state await graph_app.ainvoke( {task: req.task}, config ) return RunResponse(final_answerfinal_state.get(final_answer, )) if __name__ __main__: import uvicorn uvicorn.run(api, host127.0.0.1, port8000)这里把 LangGraph 的可执行对象当成普通 Python 对象来用接口层完全按照你自己项目的规范来定义。好处是业务逻辑和接口逻辑解耦后续要换编排方案时接口不用大改。8.2 curl 调用示例curl -X POST http://127.0.0.1:8000/agent/run \ -H Content-Type: application/json \ -d {thread_id: thread-001, task: 写一份多智能体实战总结}预期返回一个 JSON 对象包含final_answer字段。8.3 Python 客户端调用示例import requests url http://127.0.0.1:8000/agent/run payload { thread_id: thread-001, task: 写一份多智能体实战总结, } resp requests.post(url, jsonpayload, timeout120) resp.raise_for_status() print(resp.json()[final_answer])8.4 批量任务设计多智能体批量任务的核心原则是每个任务一个独立thread_id任务之间互不影响。最简单的批量方式是循环提交但要考虑限流和容错。import time import requests tasks [ 任务1整理AI大模型发展关键节点, 任务2写一篇LangGraph入门文章, 任务3总结多智能体模式的优缺点, ] results {} for index, task in enumerate(tasks): thread_id fbatch-{index} payload { thread_id: thread_id, task: task, } try: resp requests.post(url, jsonpayload, timeout120) resp.raise_for_status() results[thread_id] resp.json().get(final_answer, ) print(f[完成] {thread_id}) except Exception as exc: print(f[失败] {thread_id}: {exc}) results[thread_id] None time.sleep(1) # 简单限流避免触发模型API的速率限制 print(批量任务执行结束)如果任务量达到几百上千个建议直接用消息队列。生产环境不要用一个裸循环处理大批量任务否则任何一个节点卡住会拖慢整批任务。正确做法是把任务写入队列用多个 Worker 并发消费。9. 资源占用与性能观察很多人关心 LangGraph 会不会吃资源。事实是LangGraph 本身占用的资源非常有限主要内存开销来自 State 数据和 Checkpointer。如果你用云端模型 APICPU 占用主要集中在 JSON 解析、路由判断和工具调用上如果你用本地模型那模型加载后的显存和 GPU 占用才是大头。9.1 观察方法本地模型场景下观察显存nvidia-smi -l 2观察 Python 进程的 CPU 和内存占用可以直接看任务管理器也可以用psutil或任务监控工具top9.2 影响性能的关键因素因素影响方式模型 API 响应时长占整个流程的大部分时间主要取决于模型服务端State 数据大小每次节点都会读写 State字段越多、数据越大耗时越长工具调用搜索、数据库查询、文件读写等外部 IO 是性能瓶颈并发任务数并发太多会遇到 API 限流Checkpointer每次状态保存有 IO 开销单机 MemorySaver 基本可忽略9.3 降低资源占用的建议State 中只保留必要字段不要把长文本结果反复堆在汇总字段里。不要为了“能用”而把整个历史对话放在 State 里长对话场景应压缩历史或按窗口截断。批量任务限制并发数模型 API 服务端一般会限流。如果不需要断点恢复可以不用 Checkpointer减少状态持久化开销。10. 常见问题与排查方法问题现象可能原因排查方式解决方案pip 安装失败网络问题或版本冲突查看完整报错日志换源安装、升级 pip、用虚拟环境隔离运行时报缺少 model 依赖只装了 langgraph没装模型封装库检查 import 报错按项目需求安装对应依赖调用模型 API 报认证错误API Key 错误或环境变量未加载打印环境变量检查是否为空检查 .env 文件确认 Key 有效路由结果不对条件边返回值和路由表不匹配打印路由函数返回值检查 return 字符串和路由表 key 一致状态没有更新节点函数返回了未定义的字段检查 State 类型定义和节点返回值确保返回字段在 State 类型中存在第二次运行看到上一次的数据Thread id 相同Checkpointer 恢复旧状态确认是否想要这样的会话恢复行为需要隔离时更换 thread_id端口被占用本地服务端口冲突使用 ss/lsof 查看端口占用更换端口或杀掉占用进程批量任务卡住某个任务 API 超时或工具死循环给任务加超时和日志给节点函数加超时处理使用队列和重试机制内存持续增长State 字段越来越大或 Checkpointer 无限累积观察内存曲线压缩状态、定期清理历史线程数据排查通用思路先把报错信息完整复制出来再定位是 LangGraph 框架的错误、模型 API 的错误、工具函数的错误还是外部系统的错误。最常见的浪费时间点是“模型 API 报错被当成 LangGraph 报错”看到 HTTP 401、429 之类的状态码时先去看模型服务端的日志。11. 最佳实践与使用建议11.1 先跑最小多智能体第一次接触 LangGraph 时不要上来就做 10 个节点、20 个工具的大工程。先写一个两三个节点的最小图跑通后再逐步加节点、加条件分支、加并发子任务。最小可运行配置要单独保存作为后续项目的模板。11.2 把工具函数和流程解耦所有工具函数应该独立于 LangGraph 节点。节点只负责从 State 读参数、调用工具函数、把结果写回 State。这样工具函数可以单独测试也可以在多个智能体之间复用。如果工具函数直接写在节点内部后面没法做单元测试也很难维护。11.3 日志与可观测性多智能体系统出问题时最难的是定位“在哪一步出了问题”。给每个节点入口和出口加结构化日志记录节点名称、输入摘要、输出摘要和耗时。批量任务场景下每条日志都要带上thread_id这样出现问题才能按任务维度回溯。推荐的日志思路import logging logger logging.getLogger(__name__) def researcher_node(state: AgentState): task state[task] logger.info([researcher] input task: %s, task[:200]) result llm.invoke(task) logger.info([researcher] output length: %d, len(result.content)) return {resolved: result.content}11.4 人工确认节点凡是“影响外部世界”的节点比如发送邮件、提交订单、修改数据库建议在命令路径中插入人工确认步骤。LangGraph 的 Checkpointer 和中断机制可以配合实现暂停恢复而不是让智能体全自动执行高风险操作。11.5 合规与授权检查使用多智能体处理真实业务前逐项确认模型服务是否已经获得合法访问权限。输入数据是否包含个人信息、商业机密。工具是否涉及写操作、支付、对外发布等高权限动作。输出内容是否涉及版权、肖像、声音等授权问题。批量任务是否会对下游系统产生压力。合规检查不是技术流程之外的事情它就是多智能体工程的一部分。12. 总结与下一步LangGraph 多智能体开发的学习曲线并不陡关键是把状态、节点、边、检查点这几个核心组件理解透。这套东西最值得尝试的点是你的业务流程可以被明确地画成一张图而不是散落在各种 if-else 里。相比直接用 LangChain 的 Agent 工具类LangGraph 在复杂流程、分支路由、并行任务、人工确认这些场景里明显更可控。如果你是从零开始下一步建议做三件事第一把文章里的最小多智能体代码跑通先不接真实模型用固定返回值测试图结构是否正确第二接上真实模型 API验证路由、状态更新和 Checkpointer第三加入一个真实工具节点比如搜索工具或本地知识库检索然后把入口封装成 FastAPI 接口。最容易踩的坑是路由表 key 和返回值不匹配以及 State 字段命名不一致。这两类错误通过打印日志就能快速定位。跑通最小框架之后你已经具备把 LangGraph 接到业务里的能力了剩下的就是按自己的业务场景去扩展节点和工具。建议直接收藏这篇文章等真正写代码的时候作为参考模板使用。
LangGraph多智能体实战：从架构到代码搭建Supervisor工作流
这次我们来看一套多智能体实战技术拆解LangGraph。如果你正在做 AI 大模型应用开发准备研究 agent、多智能体、工作流编排LangGraph 是目前绕不开的一个框架。很多人看 LangGraph 教程问题往往不是“看不懂概念”而是“概念看了一堆自己写一个多智能体却不知道怎么动手”。这篇博客会直接从架构到代码走一遍用一个小但完整的 Supervisor 多智能体示例把核心组件拆开讲清楚。LangGraph 是 LangChain 团队开源的多智能体编排框架它的核心思路不是继续堆“Agent 类”而是把 Agent 行为建模成一张可执行的状态图。状态在节点之间流动节点里可以放模型调用、工具调用、普通代码函数边上可以写“什么条件走哪个分支”。这种设计的价值在于当你有多个智能体角色、多个并行任务、多个分支决策时可以用图的方式表达并且天然支持人工确认、持久化、恢复和流式输出。这篇文章不是简单罗列概念而是拆成三层来讲第一层是硬件与环境门槛第二层是 LangGraph 的核心组件第三层是一套可以直接改的代码实战。所有代码示例都基于 LangGraph 的常见 API 编写模型调用部分你需要替换成自己的模型服务地址、API Key 和模型名称。看完之后你可以得到一套最小可运行的多智能体框架然后按业务需要改成自己的版本。1. LangGraph 多智能体核心能力速览能力项说明项目类型多智能体编排框架 / AI Agent 应用开发库开源来源LangChain 团队开源核心功能状态图编排、多节点协作、条件路由、并行分支、持久化检查点、流式输出、人机协同主要组件StateGraph、State、Node、Edge、Conditional Edge、Checkpointer、Send / Command、Tool Keeper硬件要求LangGraph 本身对硬件要求极低CPU 可完成编排测试资源占用主要取决于你接入的 LLM 是云端 API 还是本地模型显存占用取决于所接大模型本地部署模型由模型规格决定使用云端 API 时几乎没有显存压力支持平台Windows、macOS、Linux以 Python 环境为主启动方式库方式嵌入 Python 项目也可通过 LangGraph Server / CLI 启动服务是否支持 API支持LangGraph 部署服务会暴露接口具体路径以部署产出为准是否支持批量任务支持用 Python 循环或消息队列批量提交任务即可适合场景客服、数据分析、内容生产、自动化工作流、多角色协作、Agent 工具箱从表格能看出LangGraph 并不是一个“模型”而是一个流程编排层。真正消耗计算资源的是你要调用的 AI 大模型。如果你用的是 OpenAI、通义千问、DeepSeek、Kimi、智谱等 APILangGraph 本身跑在后台只占用内存和少量 CPU如果你用本地部署的模型则需要按模型的参数量准备 GPU 和显存。2. 适用场景与使用边界LangGraph 最适合解决的一类问题是多个智能体需要分工协作并且流程里有明确的状态流转。典型场景包括客户工单系统一个智能体负责意图识别一个负责信息检索一个负责生成回复最后一个人工审核节点确认再发出。数据分析 Agent一个智能体负责将自然语言转换为 SQL一个负责执行查询一个负责解释结果并生成图表代码。内容生产管线一个智能体做素材收集一个做大纲一个写初稿一个做改写润色一个做格式校验。自动化运维流程多节点顺序执行其中某个节点失败可以重试或转入人工处理。也有一些场景不一定适合 LangGraph。如果你的业务只是“单轮问答”直接用模型 API 就好了没必要引入状态图如果你的流程非常固定也不需要动态路由普通代码顺序调用反而更轻。LangGraph 的优势在“状态”和“分支”没有复杂状态流转时框架反而会成为额外负担。使用边界上必须强调三点第一凡是接入真实业务的工具调用都要做权限控制。尤其是让智能体写文件、发消息、调用支付或修改数据库的工具生产环境必须加白名单和人工确认节点。第二涉及用户隐私数据时要注意数据脱敏和存储合规。LangGraph 的 Checkpointer 会把对话状态持久化如果你的状态里包含用户手机号、地址等信息持久化存储本身要满足数据安全要求。第三涉及人脸、声音、肖像、版权素材生成类应用时必须确认素材授权和用户授权。多智能体可以在内容创作流程里大幅提效但不能替使用者解决授权问题。3. 环境准备与前置条件3.1 系统与语言环境LangGraph 是一个 Python 库首先需要准备 Python 环境。建议使用 Python 3.9 及以上版本更稳定的做法是在一个独立的虚拟环境中安装。Windows、macOS、Linux 都支持但部分 LangGraph CLI 功能在 Windows 上体验略弱建议在 Linux 或 WSL 环境中操作。检查环境常用命令python --version pip --version git --version如果没有创建虚拟环境建议先建一个python -m venv langgraph-demo source langgraph-demo/bin/activate # Windows 使用 langgraph-demo\Scripts\activate3.2 模型服务准备LangGraph 本身不直接提供模型能力你需要有一个可调用的 LLM 服务。常见选择有两类云端 APIOpenAI、DeepSeek、通义千问、Kimi、智谱、文心等准备好对应的 API Key 和接口地址。本地模型服务vLLM、Ollama、Xinference 等工具启动一个本地 OpenAI 兼容接口准备好模型名称和 base_url。如果只是测试代码流程可以先不调用真实模型用代码临时返回固定字符串然后再切换到模型 API。这样排查问题时会更容易定位是流程问题还是模型问题。3.3 网络与端口调用云端 API 需要模型服务商自身提供合法、合规的服务访问方式确保网络环境可以正常访问你选择的模型服务。如果后面要启动 LangGraph Server 预览功能注意检查端口是否被占用常见默认端口是 8123 或 2024以实际启动信息和官方文档为准。3.4 项目目录规划建议按下面的结构组织工程langgraph-demo/ ├── .env # API Key 等配置不要提交到代码仓库 ├── main.py # 多智能体主流程 ├── agents/ # 存放不同智能体节点 ├── tools/ # 存放工具函数 ├── inputs/ # 测试输入文件 ├── outputs/ # 输出结果目录 └── logs/ # 运行日志小项目可以直接平铺但养成目录分层的习惯后批量任务和多人协作会轻松很多。4. 安装部署与启动方式4.1 安装 LangGraph 核心依赖基础安装只需要两个包pip install langgraph langgraph-cli这里注意langgraph-cli是可选的它主要用于启动本地预览服务和管理 LangGraph 应用。如果只打算在 Python 代码里使用安装langgraph就够。按实际模型接口补充安装依赖。我用langchain-openai作为模型封装示例pip install langchain-openai python-dotenv具体依赖版本以安装时的官方信息为准不要盲目追求最新版本。项目稳定运行后建议把依赖版本固定到requirements.txtpip freeze requirements.txt4.2 验证安装是否成功python -c import langgraph; print(langgraph.__version__)能正常输出版本号说明环境基本就绪。如果提示缺少模块按报错信息补装。4.3 LangGraph Server 启动方式如果希望用 Web 界面查看图结构和对话记录可以使用 LangGraph CLI。以脚本方式定义好langgraph.json配置后执行langgraph dev也可以直接运行 Python 服务入口python main.py这两种方式各有侧重langgraph dev更偏向调试和可视化直接运行 Python 脚本更接近生产部署。按官方文档配置即可不需要两个都跑起来。5. LangGraph 多智能体核心组件拆解要读懂 LangGraph 代码必须先弄明白下面几个核心组件。我把它们按“从抽象到具体”的方式排列。5.1 State全局状态的唯一事实来源State 是 LangGraph 的“内存表”。你可以定义它为一个TypedDict或 Pydantic 模型所有节点都读取它、更新它。多智能体协作时每个节点只关注自己需要的字段但所有节点共享同一个 State 实例。5.2 StateGraph把节点和边组装成图StateGraph是 LangGraph 的核心图容器。先初始化一个图然后不断往里面add_node添加节点用add_edge或add_conditional_edges添加边最后调用compile()生成可执行对象。5.3 Node流程里的执行单元一个 Node 就是一个普通 Python 函数输入是 State输出是一个字典表示你想更新到全局状态里的字段。Node 内部可以是模型调用、工具调用、普通代码、甚至是另一个子图的入口。5.4 Edge 与 Conditional Edge路径控制普通边表示“无条件走到下一个节点”条件边则表示“根据某个函数返回值跳到不同节点”。这是多智能体路由的核心。5.5 Checkpointer状态的持久化与恢复Checkpointer 能把每一步状态保存下来。运行中断后可以恢复也可以在流程图中间加入人工确认节点。LangGraph 提供MemorySaver等实现适合单机调试。5.6 Send 与 Command并发扇出与动态分发Send用于把一个父节点拆分成多个并行子任务。比如列出 100 个任务然后用Send(task_node, task_state)让每个任务独立运行。Command则是一种更灵活的节点返回值形式可以在一个节点里同时更新状态、跳转路径或与外部交互。5.7 LangChain 与 LangGraph 的关系一个常见问题是 “LangChain 和 LangGraph 的区别”。简单说LangChain 是一套丰富的工具和组件库包含模型封装、提示词模板、输出解析器、文档加载器等LangGraph 是集中处理“流程编排”和“状态持久化”的框架层。你可以只用 LangChain 的模型封装配合 LangGraph 做流程也可以完全不依赖 LangChain直接用原生 OpenAI SDK 调用模型LangGraph 只负责编排。理解这一点后写代码时就不会被生态绑定住。6. 代码实战搭建一个可运行的 Supervisor 多智能体下面用一个“主管-研究员-写手”的小型多智能体来演示。主管节点负责接收任务并决定路由如果任务是查询类交给研究员节点如果任务是写作类交给写手节点研究员查完信息后写手再基于研究员的结果写最终回答。这个示例虽然简单但已经把状态定义、多节点、条件路由、并行扩展、持久化检查点都覆盖到了。先创建main.pyimport os from typing import TypedDict from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver from langchain_openai import ChatOpenAI # 替换为你的模型服务配置 llm ChatOpenAI( base_urlos.getenv(MODEL_BASE_URL, https://api.openai.com/v1), api_keyos.getenv(MODEL_API_KEY), modelos.getenv(MODEL_NAME, gpt-4o-mini), temperature0.2, ) class AgentState(TypedDict): task: str # 用户原始任务 resolved: str # 研究员查到的信息 written: str # 写手生成的正文 final_answer: str # 最终汇总输出 # 节点1主管判断任务类型 def supervisor_node(state: AgentState): task state[task] # 这里是简化的路由判断真实场景可以把 task 交给模型分类 if 查 in task or 数据 in task or 检索 in task: route researcher else: route writer print(f[supervisor] 路由到 - {route}) # 通过返回值中的 Command 字段跳转 return {final_answer: waiting, command: {goto: route}} # 节点2研究员执行检索类任务 def researcher_node(state: AgentState): task state[task] # 真实项目中这里会调用搜索工具或知识库工具 prompt f请简要说明{task}输出要点式回答。 response llm.invoke(prompt) return {resolved: response.content} # 节点3写手基于研究员结果或直接根据任务写作 def writer_node(state: AgentState): task state[task] resolved state.get(resolved, ) if resolved: prompt f根据下面的检索信息写一份结构化正文\n{resolved} else: prompt f写一篇结构化正文主题是{task} response llm.invoke(prompt) return {written: response.content} # 节点4汇总节点 def final_node(state: AgentState): written state.get(written, ) resolved state.get(resolved, ) final_answer f检索信息\n{resolved}\n\n最终正文\n{written} return {final_answer: final_answer} # 构建图 graph_builder StateGraph(AgentState) # 添加节点 graph_builder.add_node(supervisor, supervisor_node) graph_builder.add_node(researcher, researcher_node) graph_builder.add_node(writer, writer_node) graph_builder.add_node(final, final_node) # 添加边START - supervisor graph_builder.add_edge(START, supervisor) # 条件边supervisor 根据 command 字段跳转 graph_builder.add_conditional_edges( supervisor, lambda state: state.get(command, {}).get(goto), {researcher: researcher, writer: writer}, ) # 研究员和写手都汇总到 final graph_builder.add_edge(researcher, final) graph_builder.add_edge(writer, final) graph_builder.add_edge(final, END) # 加入检查点支持状态持久化和人工中断 checkpointer MemorySaver() app graph_builder.compile(checkpointercheckpointer) def run_task(task: str): config {configurable: {thread_id: demo-thread-1}} for event in app.stream({task: task}, config): print(事件:, event) if __name__ __main__: run_task(帮我查一下 LangGraph 多智能体的基本概念并整理成短文)这段代码里的command字段是一种简化写法实际使用中建议直接使用 LangGraph 的Command类型让状态更新和路由更规范。像command{goto: researcher}这种结构在早期版本里常用新版本里更推荐显式返回Command(goto...)。具体语法以你当前安装版本的官方文档为准。为了展示Command的标准用法这里补充一个改写片段from langgraph.types import Command def supervisor_node(state: AgentState): task state[task] if 查 in task or 数据 in task or 检索 in task: return Command(gotoresearcher, update{final_answer: waiting}) return Command(gotowriter, update{final_answer: waiting})然后启动服务# 先设置模型相关环境变量 export MODEL_BASE_URL你的模型服务地址 export MODEL_API_KEY你的APIKey export MODEL_NAME你的模型名称 python main.py如果配置正确运行后会在终端看到类似的事件流输出最终状态里会出现检索信息和最终正文。7. 功能测试与效果验证代码跑通只是第一步完整测试才是把多智能体接入业务的保障。建议按下面几条维度验证。7.1 多次连续对话是否保持状态测试项操作预期结果连续调用用同一个 thread_id 连续调用两次第二次能读到第一次写入的状态新会话隔离换一个 thread_id 调用两个会话状态互不干扰更新旧状态手工往 State 写一个字段再触发节点节点读取到的是最新值LangGraph 的 Checkpointer 在这里起关键作用。如果没有配置 Checkpointer连续调用不会记住上次状态配置后同一个 thread_id 就相当于一个持久化会话。7.2 路由是否正确给supervisor_node设计不同类型的测试任务“查一下最新的 AI 大模型开源项目”“写一篇关于多智能体架构的文章”从输出事件里观察路由分支是否分别走到 researcher 和 writer。判断标准是任务类型和路由结果匹配且最终输出没有被错误跳过。7.3 工具节点是否稳定真实项目中 researcher 节点通常会调用搜索引擎或知识库工具。建议先把工具函数写在tools/目录下单独写好单元测试再接入 LangGraph 图。工具节点最容易出问题的不是流程而是超时和异常返回。工具必须能够稳定返回一个字符串特殊情况下返回“未找到相关信息”而不是直接抛异常。7.4 外部系统对接验证多智能体通常会先经过外部系统再进入 LangGraph。建议顺序外部系统生成 JSON 请求。LangGraph 入口函数校验请求字段。完成图执行后返回 JSON 结果。外部系统读取结果。测试时至少覆盖正常输入、缺字段输入、空字符串输入、超长输入四种情况。多智能体的入口函数要像接口一样做参数校验不能假设外部系统传进来的数据一定干净。8. 接口 API 与批量任务LangGraph 的部署形态有两种一种是把图编译后嵌入自己的 Python 服务对外提供你自己的 API另一种是使用 LangGraph Server 自带的服务接口。无论用哪种客户端调用方式都非常接近。8.1 在 FastAPI 中暴露一个运行接口from fastapi import FastAPI from pydantic import BaseModel from main import app as graph_app api FastAPI() class RunRequest(BaseModel): thread_id: str task: str class RunResponse(BaseModel): final_answer: str api.post(/agent/run, response_modelRunResponse) async def run_agent(req: RunRequest): config {configurable: {thread_id: req.thread_id}} final_state await graph_app.ainvoke( {task: req.task}, config ) return RunResponse(final_answerfinal_state.get(final_answer, )) if __name__ __main__: import uvicorn uvicorn.run(api, host127.0.0.1, port8000)这里把 LangGraph 的可执行对象当成普通 Python 对象来用接口层完全按照你自己项目的规范来定义。好处是业务逻辑和接口逻辑解耦后续要换编排方案时接口不用大改。8.2 curl 调用示例curl -X POST http://127.0.0.1:8000/agent/run \ -H Content-Type: application/json \ -d {thread_id: thread-001, task: 写一份多智能体实战总结}预期返回一个 JSON 对象包含final_answer字段。8.3 Python 客户端调用示例import requests url http://127.0.0.1:8000/agent/run payload { thread_id: thread-001, task: 写一份多智能体实战总结, } resp requests.post(url, jsonpayload, timeout120) resp.raise_for_status() print(resp.json()[final_answer])8.4 批量任务设计多智能体批量任务的核心原则是每个任务一个独立thread_id任务之间互不影响。最简单的批量方式是循环提交但要考虑限流和容错。import time import requests tasks [ 任务1整理AI大模型发展关键节点, 任务2写一篇LangGraph入门文章, 任务3总结多智能体模式的优缺点, ] results {} for index, task in enumerate(tasks): thread_id fbatch-{index} payload { thread_id: thread_id, task: task, } try: resp requests.post(url, jsonpayload, timeout120) resp.raise_for_status() results[thread_id] resp.json().get(final_answer, ) print(f[完成] {thread_id}) except Exception as exc: print(f[失败] {thread_id}: {exc}) results[thread_id] None time.sleep(1) # 简单限流避免触发模型API的速率限制 print(批量任务执行结束)如果任务量达到几百上千个建议直接用消息队列。生产环境不要用一个裸循环处理大批量任务否则任何一个节点卡住会拖慢整批任务。正确做法是把任务写入队列用多个 Worker 并发消费。9. 资源占用与性能观察很多人关心 LangGraph 会不会吃资源。事实是LangGraph 本身占用的资源非常有限主要内存开销来自 State 数据和 Checkpointer。如果你用云端模型 APICPU 占用主要集中在 JSON 解析、路由判断和工具调用上如果你用本地模型那模型加载后的显存和 GPU 占用才是大头。9.1 观察方法本地模型场景下观察显存nvidia-smi -l 2观察 Python 进程的 CPU 和内存占用可以直接看任务管理器也可以用psutil或任务监控工具top9.2 影响性能的关键因素因素影响方式模型 API 响应时长占整个流程的大部分时间主要取决于模型服务端State 数据大小每次节点都会读写 State字段越多、数据越大耗时越长工具调用搜索、数据库查询、文件读写等外部 IO 是性能瓶颈并发任务数并发太多会遇到 API 限流Checkpointer每次状态保存有 IO 开销单机 MemorySaver 基本可忽略9.3 降低资源占用的建议State 中只保留必要字段不要把长文本结果反复堆在汇总字段里。不要为了“能用”而把整个历史对话放在 State 里长对话场景应压缩历史或按窗口截断。批量任务限制并发数模型 API 服务端一般会限流。如果不需要断点恢复可以不用 Checkpointer减少状态持久化开销。10. 常见问题与排查方法问题现象可能原因排查方式解决方案pip 安装失败网络问题或版本冲突查看完整报错日志换源安装、升级 pip、用虚拟环境隔离运行时报缺少 model 依赖只装了 langgraph没装模型封装库检查 import 报错按项目需求安装对应依赖调用模型 API 报认证错误API Key 错误或环境变量未加载打印环境变量检查是否为空检查 .env 文件确认 Key 有效路由结果不对条件边返回值和路由表不匹配打印路由函数返回值检查 return 字符串和路由表 key 一致状态没有更新节点函数返回了未定义的字段检查 State 类型定义和节点返回值确保返回字段在 State 类型中存在第二次运行看到上一次的数据Thread id 相同Checkpointer 恢复旧状态确认是否想要这样的会话恢复行为需要隔离时更换 thread_id端口被占用本地服务端口冲突使用 ss/lsof 查看端口占用更换端口或杀掉占用进程批量任务卡住某个任务 API 超时或工具死循环给任务加超时和日志给节点函数加超时处理使用队列和重试机制内存持续增长State 字段越来越大或 Checkpointer 无限累积观察内存曲线压缩状态、定期清理历史线程数据排查通用思路先把报错信息完整复制出来再定位是 LangGraph 框架的错误、模型 API 的错误、工具函数的错误还是外部系统的错误。最常见的浪费时间点是“模型 API 报错被当成 LangGraph 报错”看到 HTTP 401、429 之类的状态码时先去看模型服务端的日志。11. 最佳实践与使用建议11.1 先跑最小多智能体第一次接触 LangGraph 时不要上来就做 10 个节点、20 个工具的大工程。先写一个两三个节点的最小图跑通后再逐步加节点、加条件分支、加并发子任务。最小可运行配置要单独保存作为后续项目的模板。11.2 把工具函数和流程解耦所有工具函数应该独立于 LangGraph 节点。节点只负责从 State 读参数、调用工具函数、把结果写回 State。这样工具函数可以单独测试也可以在多个智能体之间复用。如果工具函数直接写在节点内部后面没法做单元测试也很难维护。11.3 日志与可观测性多智能体系统出问题时最难的是定位“在哪一步出了问题”。给每个节点入口和出口加结构化日志记录节点名称、输入摘要、输出摘要和耗时。批量任务场景下每条日志都要带上thread_id这样出现问题才能按任务维度回溯。推荐的日志思路import logging logger logging.getLogger(__name__) def researcher_node(state: AgentState): task state[task] logger.info([researcher] input task: %s, task[:200]) result llm.invoke(task) logger.info([researcher] output length: %d, len(result.content)) return {resolved: result.content}11.4 人工确认节点凡是“影响外部世界”的节点比如发送邮件、提交订单、修改数据库建议在命令路径中插入人工确认步骤。LangGraph 的 Checkpointer 和中断机制可以配合实现暂停恢复而不是让智能体全自动执行高风险操作。11.5 合规与授权检查使用多智能体处理真实业务前逐项确认模型服务是否已经获得合法访问权限。输入数据是否包含个人信息、商业机密。工具是否涉及写操作、支付、对外发布等高权限动作。输出内容是否涉及版权、肖像、声音等授权问题。批量任务是否会对下游系统产生压力。合规检查不是技术流程之外的事情它就是多智能体工程的一部分。12. 总结与下一步LangGraph 多智能体开发的学习曲线并不陡关键是把状态、节点、边、检查点这几个核心组件理解透。这套东西最值得尝试的点是你的业务流程可以被明确地画成一张图而不是散落在各种 if-else 里。相比直接用 LangChain 的 Agent 工具类LangGraph 在复杂流程、分支路由、并行任务、人工确认这些场景里明显更可控。如果你是从零开始下一步建议做三件事第一把文章里的最小多智能体代码跑通先不接真实模型用固定返回值测试图结构是否正确第二接上真实模型 API验证路由、状态更新和 Checkpointer第三加入一个真实工具节点比如搜索工具或本地知识库检索然后把入口封装成 FastAPI 接口。最容易踩的坑是路由表 key 和返回值不匹配以及 State 字段命名不一致。这两类错误通过打印日志就能快速定位。跑通最小框架之后你已经具备把 LangGraph 接到业务里的能力了剩下的就是按自己的业务场景去扩展节点和工具。建议直接收藏这篇文章等真正写代码的时候作为参考模板使用。
