LangGraph 是 LangChain 生态中用于构建复杂、有状态多智能体应用的核心框架。它不是一个独立的模型而是一个编排工具让你能用“图”的思维来设计和运行由多个 AI 智能体Agent和工具Tool协同工作的系统。简单说它解决了单个 Agent 能力有限、任务流程复杂的问题让你能像搭积木一样组合多个 AI 智能体来完成写代码、查资料、审核、决策等一系列连贯动作。这篇文章不讲虚的概念直接带你上手。我们会拆解 LangGraph 的核心架构用代码实战一个能自动分析需求、写代码、运行测试的多智能体系统。整个过程不依赖付费 API完全基于开源模型如 Ollama在本地运行重点在于理解其工作流Workflow设计、状态State管理和监督Supervision机制。无论你是想构建自动化开发助手、智能客服链还是数据分析流水线这套方法都能直接复用。1. 核心能力速览能力项说明项目类型多智能体Multi-Agent工作流编排框架核心价值将复杂的 AI 应用拆解为由多个智能体协同完成的任务图实现有状态、可循环、带条件判断的复杂流程。硬件门槛无特定要求。框架本身是 Python 库计算负载取决于背后连接的 AI 模型如 OpenAI API、本地 Ollama 模型。本地运行大模型则需要相应 GPU/CPU 资源。启动方式通过 Python 脚本启动工作流。提供 Web 界面LangGraph Studio进行可视化开发和调试。主要功能1.定义工作流图将智能体节点和逻辑边组织成图。2.状态管理在整个工作流中持久化和传递共享数据。3.条件路由根据智能体输出决定下一步走向哪个节点。4.循环与中断支持循环执行直到满足条件或人工干预Human-in-the-loop。5.并行与监督可并行执行任务并设置监督节点协调或复核。是否支持 API是。可将定义好的工作流封装为 FastAPI 等 Web 服务提供 HTTP 接口。是否支持批量任务是。工作流本身可被多次调用通过初始化不同状态来处理批量任务。适合场景自动化开发流水线、复杂决策支持系统、多步骤内容生成与审核、交互式 AI 应用原型开发。2. 适用场景与使用边界适合谁用AI 应用开发者希望超越简单的单轮问答构建具有复杂逻辑链的 AI 应用。技术团队需要将 AI 能力嵌入到现有业务系统实现自动化流程如自动生成报告、代码审查、客户工单处理。研究者与学习者希望深入理解多智能体系统MAS的工程化实现。能解决什么问题任务分解与协作将一个复杂任务如“开发一个网页爬虫”自动分解为需求分析、架构设计、代码编写、单元测试等子任务并由不同特长的智能体分工完成。有状态会话管理在长时间的交互中如客服对话、设计迭代保持上下文连贯记住之前所有的决策和输出。流程控制与纠错当某个智能体输出不符合要求时能自动路由到复核或重试节点甚至引入人工审核。工具集成与调度灵活调用搜索引擎、数据库、代码执行环境、外部 API 等多种工具。不适合什么场景极其简单的单次问答用 LangChain 的简单 Chain 或直接调用模型 API 更轻量。对延迟极其敏感多智能体间的多次调用和状态传递会增加整体响应时间。完全静态、无分支的线性流程可能过于复杂。合规与安全边界责任归属由多智能体自动生成的代码、文本等内容开发者需承担最终审核责任确保其合规、安全、无偏见。数据隐私工作流中传递的数据可能包含敏感信息需注意加密存储和传输避免泄露。工具使用智能体调用的工具如代码执行、网络访问具有潜在风险必须在沙箱或严格权限控制下运行。3. 环境准备与前置条件我们将构建一个本地运行的多智能体系统核心框架是 LangGraph智能体大脑使用本地部署的 Ollama 模型。基础环境清单操作系统Windows 10/11, macOS, 或 Linux (推荐 Ubuntu 20.04)Python版本 3.10 或 3.11。避免使用 3.12 以防某些包兼容性问题。包管理工具pip或conda。本地大模型服务Ollama。用于提供本地 LLM 能力我们将使用llama3.2或qwen2.5等轻量模型。网络能正常访问 PyPI 和 GitHub 以下载 Python 包。可选但推荐的工具代码编辑器VS Code 或 PyCharm。虚拟环境使用venv或conda创建独立环境避免包冲突。LangGraph StudioLangGraph 的可视化开发工具方便调试。4. 安装部署与启动方式4.1 创建并激活虚拟环境# 创建虚拟环境 python -m venv langgraph_env # 激活环境 (Windows) langgraph_env\Scripts\activate # 激活环境 (macOS/Linux) source langgraph_env/bin/activate4.2 安装核心依赖pip install langgraph langchain langchain-community ollamalanggraph: 核心工作流编排框架。langchain: 提供智能体、链、提示词模板等基础组件。langchain-community: 包含大量第三方工具和模型集成。ollama: LangChain 对 Ollama 的官方集成包。4.3 安装并启动 Ollama 服务如果你还没有安装 Ollama请先安装并拉取一个模型。# 前往 https://ollama.com 下载并安装 Ollama # 拉取一个模型例如 llama3.2 ollama pull llama3.2 # 启动 Ollama 服务它默认会在 11434 端口提供 API # Ollama 会作为后台服务运行4.4 验证 Ollama 服务运行一个简单的 Python 脚本来测试 Ollama 连接是否正常。# test_ollama.py from langchain_community.llms import Ollama llm Ollama(modelllama3.2) response llm.invoke(Hello, what is LangGraph?) print(response)如果能看到模型返回的关于 LangGraph 的介绍文本说明环境配置成功。5. 功能测试与效果验证构建一个代码生成与审查多智能体我们将构建一个包含三个智能体的工作流需求分析师 (Analyst)理解用户模糊的需求输出清晰、可执行的任务规格。程序员 (Coder)根据任务规格编写 Python 代码。审查员 (Reviewer)审查代码检查语法错误、逻辑问题并提出修改建议。5.1 定义共享状态 (State)工作流中所有智能体都能读写这个状态。# state.py from typing import TypedDict, List, Annotated import operator class State(TypedDict): # 用户原始输入 original_request: str # 需求分析师输出的清晰任务描述 clarified_spec: str # 程序员生成的代码 generated_code: str # 审查员提出的修改建议 review_comments: List[str] # 记录工作流执行到了哪一步 step: str5.2 实现智能体节点 (Nodes)每个智能体是一个函数接收当前State更新它并返回更新后的State。# nodes.py from langchain_core.prompts import ChatPromptTemplate from langchain_community.llms import Ollama from .state import State llm Ollama(modelllama3.2) def analyst_node(state: State) - State: 需求分析智能体 prompt ChatPromptTemplate.from_messages([ (system, 你是一个资深产品经理擅长将模糊的需求转化为清晰、无歧义的技术任务描述。), (human, 用户的需求是{request}\n\n请输出一份清晰的技术任务描述包含输入、处理过程、输出示例。) ]) chain prompt | llm clarified_spec chain.invoke({request: state[original_request]}) # 更新状态 state[clarified_spec] clarified_spec state[step] 需求分析完成 print(f[Analyst] 需求已澄清{clarified_spec[:100]}...) return state def coder_node(state: State) - State: 程序员智能体 prompt ChatPromptTemplate.from_messages([ (system, 你是一个专业的 Python 程序员严格按照给定的任务描述编写简洁、高效、可运行的代码。只输出代码块不要解释。), (human, 任务描述{spec}\n\n请编写完整的 Python 代码。) ]) chain prompt | llm generated_code chain.invoke({spec: state[clarified_spec]}) state[generated_code] generated_code state[step] 代码生成完成 print(f[Coder] 代码已生成长度{len(generated_code)} 字符) return state def reviewer_node(state: State) - State: 代码审查智能体 prompt ChatPromptTemplate.from_messages([ (system, 你是一个严格的代码审查员。检查代码的语法、逻辑、风格和潜在错误。如果代码优秀说‘通过’否则列出具体的修改建议。), (human, 任务描述{spec}\n\n待审查代码\npython\n{code}\n) ]) chain prompt | llm review_result chain.invoke({spec: state[clarified_spec], code: state[generated_code]}) # 简单判断审查结果 comments [review_result] if review_result.strip() ! 通过 else [] state[review_comments] comments state[step] 代码审查完成 if comments: print(f[Reviewer] 审查未通过建议{comments}) else: print([Reviewer] 审查通过) return state5.3 构建工作流图 (Graph) 并设置路由逻辑这是 LangGraph 的核心定义节点如何连接。# graph.py from langgraph.graph import StateGraph, END from .state import State from .nodes import analyst_node, coder_node, reviewer_node # 1. 创建图并指定状态类型 workflow StateGraph(State) # 2. 添加节点 workflow.add_node(analyst, analyst_node) workflow.add_node(coder, coder_node) workflow.add_node(reviewer, reviewer_node) # 3. 设置入口点 workflow.set_entry_point(analyst) # 4. 定义边连接逻辑 workflow.add_edge(analyst, coder) # 分析完直接去写代码 workflow.add_edge(coder, reviewer) # 写完代码去审查 # 5. 定义条件边根据审查结果决定是否结束 def decide_after_review(state: State) - str: 审查后决策如果有修改建议则重新分析需求模拟迭代否则结束。 if state[review_comments]: print(审查未通过返回需求分析节点进行迭代...) return analyst # 返回分析师节点开始新一轮迭代实际中可能返回coder else: return END # 结束工作流 workflow.add_conditional_edges( reviewer, # 从哪个节点出发 decide_after_review, # 决策函数 { analyst: analyst, # 如果返回 analyst则跳转到 analyst 节点 END: END # 如果返回 END则结束 } ) # 6. 编译图得到可执行对象 app workflow.compile()5.4 运行工作流进行测试现在我们可以用这个图来处理一个用户请求了。# main.py from graph import app from state import State # 初始化状态 initial_state: State { original_request: 帮我写一个程序能从网上抓取今日天气并保存到文件。, clarified_spec: , generated_code: , review_comments: [], step: 开始 } print( 开始执行多智能体工作流 ) # 运行工作流 try: final_state app.invoke(initial_state) print(\n 工作流执行完毕 ) print(f最终状态: {final_state[step]}) print(f\n生成的代码:\n{final_state[generated_code]}) if final_state[review_comments]: print(f\n审查意见迭代后: {final_state[review_comments]}) except Exception as e: print(f工作流执行出错: {e})执行与观察在终端运行python main.py。你会看到控制台依次打印[Analyst]、[Coder]、[Reviewer]的日志清晰地展示了工作流的执行顺序。如果审查员第一次给出了修改建议 (review_comments非空)根据我们设定的条件路由工作流会跳回analyst节点开始新一轮的“分析-编码-审查”循环直到审查通过。最终控制台会输出生成的 Python 代码。效果验证点顺序执行是否严格按照分析师 - 程序员 - 审查员的顺序执行状态传递clarified_spec是否从分析师正确传递给了程序员和审查员条件路由当review_comments不为空时工作流是否真的回到了analyst节点可以在reviewer_node中故意返回一条评论来测试。输出质量生成的代码是否基本符合“抓取天气并保存”的需求虽然本地小模型能力有限但流程是通的。6. 接口 API 与批量任务6.1 将工作流封装为 FastAPI 服务将上述编译好的app暴露为 HTTP API方便集成。# api_server.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from graph import app from state import State app_fastapi FastAPI(title多智能体代码生成服务) class CodeRequest(BaseModel): original_request: str class CodeResponse(BaseModel): clarified_spec: str generated_code: str review_comments: list step: str app_fastapi.post(/generate_code, response_modelCodeResponse) async def generate_code(request: CodeRequest): 接收用户需求返回多智能体协作生成的代码 initial_state: State { original_request: request.original_request, clarified_spec: , generated_code: , review_comments: [], step: 开始 } try: final_state app.invoke(initial_state) return CodeResponse(**final_state) except Exception as e: raise HTTPException(status_code500, detailf工作流执行失败: {str(e)}) if __name__ __main__: import uvicorn uvicorn.run(app_fastapi, host0.0.0.0, port8000)运行python api_server.py服务将在http://127.0.0.1:8000启动。6.2 调用 API 示例使用curl或 Python 客户端进行测试。# 使用 curl 调用 curl -X POST http://127.0.0.1:8000/generate_code \ -H Content-Type: application/json \ -d {original_request: 写一个函数计算斐波那契数列的前N项。}# 使用 Python requests 调用 import requests import json url http://127.0.0.1:8000/generate_code payload {original_request: 写一个函数计算斐波那契数列的前N项。} headers {Content-Type: application/json} response requests.post(url, datajson.dumps(payload), headersheaders) if response.status_code 200: result response.json() print(f任务状态: {result[step]}) print(f生成代码:\n{result[generated_code]}) else: print(f请求失败: {response.status_code}, {response.text})6.3 批量任务处理工作流本身是无状态的状态由每次调用初始化。因此处理批量任务非常简单循环调用app.invoke()或并发调用 API。# batch_processor.py from graph import app from state import State import concurrent.futures requests [ 写一个Python脚本重命名当前目录下所有.txt文件。, 写一个函数验证电子邮件地址格式是否正确。, 写一个程序从JSON文件中读取数据并生成柱状图。, ] def process_one_request(user_request: str): 处理单个请求 initial_state: State { original_request: user_request, clarified_spec: , generated_code: , review_comments: [], step: 开始 } result app.invoke(initial_state) return { request: user_request, code: result[generated_code], status: result[step] } # 顺序处理 print( 顺序处理 ) for req in requests: output process_one_request(req) print(f需求: {output[request][:30]}... | 状态: {output[status]}) # 使用线程池并发处理注意如果后端是CPU/GPU密集型需控制并发数 print(\n 并发处理 (4线程) ) with concurrent.futures.ThreadPoolExecutor(max_workers4) as executor: future_to_req {executor.submit(process_one_request, req): req for req in requests} for future in concurrent.futures.as_completed(future_to_req): req future_to_req[future] try: output future.result() print(f需求: {output[request][:30]}... | 状态: {output[status]}) except Exception as exc: print(f{req} 生成时发生异常: {exc})关键点批量任务时主要瓶颈在于底层 LLM 的推理速度。需要根据 Ollama 服务器的能力CPU/GPU合理设置并发数避免压垮服务。7. 资源占用与性能观察LangGraph 框架本身非常轻量其资源消耗主要来自两方面Python 进程内存用于维护工作流状态、消息传递等。通常很小几十到几百MB。大模型推理资源这是主要开销。取决于你使用的 Ollama 模型大小和硬件。性能观察方法Ollama 服务器监控运行ollama serve的终端会显示每个请求的处理时间。可以使用ollama ps查看正在运行的模型实例。工作流执行时间在代码中关键节点添加时间戳计算每个智能体的耗时。import time def analyst_node(state: State) - State: start time.time() # ... 原有逻辑 ... end time.time() print(f分析师节点耗时: {end - start:.2f}秒) return state系统资源监控Linux/macOS: 使用htop或nvidia-smi如有 GPU查看 CPU/GPU 和内存占用。Windows: 使用任务管理器。影响性能的关键因素模型大小llama3.2(3B参数) 比qwen2.5:14b快得多但能力可能较弱。提示词长度输入给模型的提示词越长推理耗时越长。工作流复杂度节点数量、条件分支数量、循环次数。每次循环都意味着重新调用模型。硬件GPU 推理远快于 CPU。优化建议本地部署使用量化后的模型如llama3.2:3b-instruct-q4_K_M平衡速度与质量。缓存对重复性高的子任务结果进行缓存。剪枝优化工作流减少不必要的节点调用或循环。异步调用如果工作流中某些节点不依赖 LLM如纯计算或数据库查询可将其设计为异步。8. 常见问题与排查方法问题现象可能原因排查方式解决方案导入langgraph失败Python 版本不兼容或虚拟环境未激活。检查 Python 版本 (python --version)确认已激活虚拟环境。使用 Python 3.10/3.11在正确环境下pip install。运行时报State相关错误TypedDict或Annotated导入错误或状态结构不匹配。检查state.py中的导入语句和类型定义。确保从typing导入TypedDict从langgraph.graph导入add_messages如果使用消息历史。Ollama 连接超时Ollama 服务未启动或端口被占用。运行ollama serve查看是否报错。用curl http://localhost:11434/api/tags测试。确保 Ollama 服务正常运行。检查防火墙是否阻止了 11434 端口。模型响应慢或无响应模型文件损坏或硬件资源不足。查看ollama serve日志。用ollama run llama3.2单独测试模型。尝试重新拉取模型 (ollama pull llama3.2)。关闭其他占用资源的程序。考虑使用更小的模型。工作流陷入无限循环条件边 (conditional_edges) 的逻辑有误导致无法满足结束条件。在decide_after_review等决策函数中添加详细日志打印判断依据。审查条件逻辑确保存在一条能到达END的路径。可以设置最大循环次数强制跳出。FastAPI 服务启动失败端口 8000 被占用或依赖未安装。检查端口占用 (netstat -ano | findstr :8000on Win)。确认安装了fastapi和uvicorn。更换端口 (uvicorn.run(..., port8001))。pip install fastapi uvicorn。智能体输出不符合预期提示词Prompt设计不佳或模型能力有限。单独测试每个智能体的提示词观察其输出。优化提示词使其指令更清晰。考虑升级模型或使用更强大的 API如 GPT-4但非本地。状态更新不正确节点函数没有正确返回更新后的state字典。在每个节点函数末尾打印state检查其内容。确保节点函数最后一行是return state且更新了目标字段。9. 最佳实践与使用建议从简单开始逐步复杂化先构建一个两三个节点的线性工作流并跑通再逐步添加条件分支、循环和并行节点。精心设计状态结构State是所有节点共享的“黑板”。设计时要想清楚哪些数据需要共享类型是什么。避免状态过于庞大。编写清晰的提示词多智能体的性能很大程度上取决于每个智能体的提示词。为每个角色分析师、程序员、审查员设计专业、明确的系统指令。实现健壮的错误处理在工作流的关键节点如调用外部 API、执行代码添加try...except并定义错误处理路径例如路由到一个“错误处理”节点。引入人工干预点对于关键决策或敏感操作使用Human-in-the-loop机制让工作流暂停并等待人工确认。LangGraph 对此有原生支持。使用 LangGraph Studio 进行调试这是官方可视化工具可以直观地看到工作流的执行路径、状态变化极大提升开发效率。版本控制与测试像管理普通代码一样管理你的工作流定义graph.py、state.py并为其编写单元测试模拟不同的输入状态。性能与成本监控在生产环境中记录每个工作流实例的执行时间、调用模型的 Token 消耗如果使用计费 API和最终结果用于分析和优化。安全与合规前置如果工作流最终生成的内容如代码、文本将对外发布或使用必须建立人工审核流程确保其安全性、合规性和准确性。10. 总结与下一步LangGraph 为我们提供了一套强大而直观的“乐高积木”用于搭建多智能体应用。它的核心优势在于将复杂流程可视化、模块化并通过有状态图的概念统一管理上下文和流程控制。通过本文的实战你应该已经掌握了核心概念状态State、节点Node、边Edge、条件路由Conditional Edge。搭建流程定义状态 → 实现节点函数 → 构建图并设置路由 → 编译运行。本地部署如何结合 Ollama 在本地运行完全开源的多智能体系统。服务化如何将工作流封装成 API并处理批量任务。最容易踩的坑忘记在节点函数中return state。条件路由的逻辑错误导致死循环。提示词不够精确导致智能体行为偏离预期。下一步可以探索的方向集成更强大的模型将 Ollama 节点替换为 OpenAI GPT、Claude 或国内大模型的 API 调用以提升智能体能力。添加更多工具让智能体能够调用搜索引擎、数据库、文件系统、代码解释器等使其能力从“思考”扩展到“行动”。实现并行执行使用langgraph.graph中的add_edge和add_conditional_edges组合实现多个节点并行运行提升效率。探索 LangGraph Studio使用图形化界面拖拽构建工作流这将是你开发复杂应用的神器。研究高级模式如子图Subgraph、消息持久化、检查点Checkpointing等用于构建更稳定、可恢复的长期运行应用。多智能体系统是构建下一代 AI 应用的关键架构。利用 LangGraph 上手实践是理解并掌握这一趋势的最佳途径。建议将本文的示例代码作为起点不断修改和扩展打造出适合你自己场景的自动化智能助手。