LangGraph 是 LangChain 生态中用于构建复杂、有状态、多步骤 AI 应用尤其是智能体的框架。它把智能体的执行过程建模成一张图Graph节点是执行步骤如调用 LLM、使用工具边是步骤之间的流转逻辑。这让你能清晰地设计出包含循环、分支、并行、人工干预的智能体工作流远超传统单次问答的简单模式。如果你正在寻找一个能构建企业级、可维护、可观测 AI 智能体的方案LangGraph 是目前最值得投入的技术栈之一。它解决了传统链式调用在复杂逻辑和长期记忆上的短板特别适合客服对话、数据分析流水线、自动化决策等需要多轮交互和状态管理的场景。本文不会停留在概念而是直接带你从环境搭建、核心概念到多智能体系统实战用代码和架构图把 LangGraph 讲透。我们将重点关注 LangGraph 如何定义工作流、管理状态、集成工具并最终构建一个模拟“技术团队评审”的多智能体协作系统。整个过程无需昂贵硬件在普通开发机上即可运行重点在于理解其架构思想和工程化实践。1. 核心能力速览能力项说明项目类型AI 智能体Agent与工作流Workflow编排框架核心价值将复杂的、有状态的智能体逻辑用“图”来清晰定义和管理关键特性支持循环Loop、条件分支Condition、并行Parallel、人工干预Human-in-the-loop硬件门槛无特殊要求依赖后端 LLM 服务如 OpenAI API、本地 Ollama。本地运行 Ollama 时需注意内存。启动方式Python 库安装通过代码定义和运行图。提供 LangGraph Studio 进行可视化开发。接口能力核心是 Python API。构建的图可暴露为可调用的函数易于集成到 FastAPI 等 Web 服务中。状态管理内置状态State管理支持自定义状态结构在节点间持久化和传递数据。适合场景多轮对话系统、自动化业务流程、决策支持系统、多角色协作智能体2. 适用场景与使用边界LangGraph 并非所有 AI 项目的首选它针对的是特定复杂度的场景。最适合的场景多步骤任务任务需要拆解为多个顺序或并行的步骤例如信息收集 - 分析 - 报告生成。有状态交互后续步骤依赖前序步骤的结果或整个对话历史例如客服对话、面试模拟。需要循环或分支根据 LLM 的输出或工具调用的结果决定重复执行还是走不同路径例如持续优化代码直到测试通过。多智能体协作需要多个具备不同角色和能力的智能体共同完成一项任务例如产品经理、工程师、测试员协同开发。需要人工审核在流程的关键节点插入人工确认或输入。不适合或需谨慎使用的场景简单的一次性问答如果只是调用一次 LLM 接口直接使用 LangChain 的 LCEL 或原始 SDK 更简单。对延迟极其敏感图的调度、状态管理会引入额外开销在毫秒级响应的场景下可能不适用。逻辑极其简单用图来管理一个线性三步流程可能显得“杀鸡用牛刀”。无状态的批量处理单纯对一批数据各自进行独立处理用传统脚本或并行计算框架更高效。合规与安全边界LLM 责任LangGraph 是编排框架内容生成的责任在于其集成的 LLM如 GPT、Claude。需确保 LLM 的使用符合其服务条款。工具安全智能体可以调用外部工具如执行代码、访问数据库。必须严格审查和沙箱化工具权限防止越权操作。数据隐私工作流状态中可能包含用户输入的敏感信息。需做好状态数据的加密、脱敏和访问控制。人工干预涉及人工审核的流程需明确审核人员的权责和操作日志。3. 环境准备与前置条件在开始编码前需要准备好以下环境。本文以 OpenAI 的 GPT 模型作为 LLM 后端为例。操作系统Windows 10/11, macOS, 或 Linux (如 Ubuntu 20.04)。LangGraph 是 Python 库跨平台兼容性好。Python 版本推荐 Python 3.10 或 3.11。确保已安装pip。LLM 服务接入方案一推荐简单准备一个有效的 OpenAI API Key 。后续代码将使用chatopenai模型。方案二本地安装并运行 Ollama 拉取一个模型如llama3.2或qwen2.5。这需要本地有足够内存通常 8GB。虚拟环境推荐使用venv或conda创建独立环境避免包冲突。网络能访问 OpenAI API 或下载 Ollama 模型。4. 安装部署与启动方式LangGraph 的“启动”即是安装 Python 包并运行你的脚本。我们首先安装核心依赖。步骤 1创建并激活虚拟环境# 使用 venv python -m venv langgraph-env # Windows langgraph-env\Scripts\activate # macOS/Linux source langgraph-env/bin/activate步骤 2安装 LangGraph 及 LLM 集成包pip install langgraph langchain-openailanggraph: 核心框架。langchain-openai: 用于集成 OpenAI 模型。如果你计划使用 Ollama则安装pip install langgraph langchain-community步骤 3验证安装创建一个简单的 Python 脚本test_import.pyfrom langgraph.graph import StateGraph, END print(LangGraph 导入成功)运行python test_import.py无报错即说明安装成功。关于 LangGraph Studio LangChain 提供了可视化的开发工具 LangGraph Studio可以通过 Docker 快速启动用于可视化编辑和调试图。对于初学者建议先通过代码理解概念再使用 Studio。# 通过 Docker 启动 Studio (可选) docker run -d -p 3000:3000 langchain/langgraph-studio访问http://localhost:3000即可使用。5. 核心概念与第一个智能体要掌握 LangGraph必须先理解三个核心概念State状态、Node节点、Edge边。State一个字典或 Pydantic 模型定义了在整个工作流中传递和更新的数据。它是图的“记忆”。Node一个函数接收当前的 State执行操作如调用 LLM、工具并返回更新后的 State。Edge连接节点的线决定流程的走向。可以是固定的也可以根据条件Condition动态决定。让我们构建一个最简单的“聊天机器人”智能体它只是重复用户的话。步骤 1定义状态状态决定了智能体能记住什么。我们定义一个非常简单的状态只包含对话轮次和最新消息。from typing import TypedDict, List class AgentState(TypedDict): 定义智能体的状态结构 messages: List[str] # 记录所有消息 turn_count: int # 记录对话轮次步骤 2创建节点函数节点是执行单元。这里创建两个节点一个处理用户输入一个生成回复。def user_node(state: AgentState) - AgentState: 模拟用户输入节点。在实际应用中这里可能连接Web接口。 user_input input(f[Turn {state[turn_count]}] 用户: ) new_messages state[messages] [f用户: {user_input}] return {messages: new_messages, turn_count: state[turn_count]} def assistant_node(state: AgentState) - AgentState: 助理节点。这里我们简单地将最后一条用户消息重复一遍。 last_message state[messages][-1] # 模拟 LLM 处理这里只是简单重复 response f助理: 我听到你说的是 {last_message} new_messages state[messages] [response] return {messages: new_messages, turn_count: state[turn_count] 1}步骤 3构建图并设置流转逻辑这是 LangGraph 的核心将节点组装成图并定义它们如何连接。from langgraph.graph import StateGraph, END # 1. 创建一个图构建器并指定状态结构 workflow StateGraph(AgentState) # 2. 添加节点 workflow.add_node(user, user_node) workflow.add_node(assistant, assistant_node) # 3. 设置入口点 workflow.set_entry_point(user) # 4. 添加边定义流程 workflow.add_edge(user, assistant) # 用户节点后总是执行助理节点 workflow.add_edge(assistant, END) # 助理节点后结束 # 5. 编译图得到一个可执行的对象 app workflow.compile()步骤 4运行智能体现在我们可以初始化状态并运行这个图。# 初始化状态 initial_state: AgentState {messages: [], turn_count: 0} # 运行图 print(--- 开始简单聊天机器人 ---) for _ in range(3): # 模拟3轮对话 # app.invoke 会从入口点开始执行直到遇到 END result app.invoke(initial_state) # 更新状态以便下一轮使用 initial_state result print(f当前消息记录: {result[messages][-2:]}\n) print(--- 对话结束 ---)运行这段代码你将看到一个简单的三轮对话流程。这虽然简单但完整展示了 LangGraph 构建智能体的模式定义状态 - 创建节点 - 组装成图 - 运行。6. 集成真实 LLM 与工具调用一个真正的智能体需要连接大语言模型LLM和外部工具。我们将升级上面的例子让助理节点调用 OpenAI GPT 来生成回复并集成一个计算器工具。步骤 1设置 OpenAI 客户端首先确保你已设置环境变量OPENAI_API_KEY或在代码中指定。import os from langchain_openai import ChatOpenAI # 初始化 LLM。推荐使用 gpt-3.5-turbo 进行测试成本低。 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 如果你用 Ollama可以这样初始化 # from langchain_community.chat_models import ChatOllama # llm ChatOllama(modelllama3.2)步骤 2定义工具LangChain 提供了便捷的工具装饰器。我们定义一个简单的计算器。from langchain.tools import tool tool def calculator(expression: str) - str: 计算一个数学表达式的值。例如calculator(3 5 * 2)。 try: # 警告实际生产中应对 eval 进行严格的安全限制或使用更安全的库如 ast.literal_eval。 result eval(expression, {__builtins__: None}, {}) return f计算结果: {result} except Exception as e: return f计算错误: {e}步骤 3创建具有工具调用能力的智能体节点我们将使用 LangChain 的create_react_agent模式来构建一个能自主决定是否使用工具的智能体节点。from langgraph.prebuilt import create_react_agent from langchain.agents import AgentExecutor # 创建智能体执行器 tools [calculator] agent_executor create_react_agent(llm, tools) # 定义一个新的助理节点函数它使用这个执行器 def smart_assistant_node(state: AgentState): 集成了LLM和工具的智能助理节点 # 获取最新的用户消息 last_user_message state[messages][-1] if state[messages] else 你好 # 构造给智能体的输入 agent_input {input: last_user_message} # 调用智能体执行器 response agent_executor.invoke(agent_input) # 提取输出 assistant_response response.get(output, 我没有得到回复。) # 更新状态 new_messages state[messages] [f助理: {assistant_response}] return {messages: new_messages, turn_count: state[turn_count] 1}步骤 4重构并运行增强版智能体现在我们用新的smart_assistant_node替换之前的简单节点。# 重新构建图 workflow_v2 StateGraph(AgentState) workflow_v2.add_node(user, user_node) # 沿用之前的用户节点 workflow_v2.add_node(assistant, smart_assistant_node) # 使用新的智能节点 workflow_v2.set_entry_point(user) workflow_v2.add_edge(user, assistant) workflow_v2.add_edge(assistant, END) app_v2 workflow_v2.compile() # 运行测试 print(--- 启动增强版智能体集成GPT与工具---) test_state: AgentState {messages: [用户: 3加5乘以2等于多少], turn_count: 1} result app_v2.invoke(test_state) print(f最终回复: {result[messages][-1]})运行后智能体会分析问题“3加5乘以2等于多少”识别出需要计算调用calculator(35*2)工具得到结果 13并组织语言回复给你。这演示了智能体的核心能力理解、规划、执行工具、响应。7. 构建多智能体协作系统实战多智能体系统MAS是 LangGraph 的杀手级应用。我们模拟一个“技术方案评审会”包含三个角色产品经理PM、后端工程师BE、前端工程师FE。流程是PM 提出需求 - BE 和 FE 并行给出技术方案 - PM 汇总并给出结论。步骤 1定义多智能体协作状态状态需要能容纳多个角色的输出和最终结论。from typing import TypedDict, List, Optional from datetime import datetime class TeamReviewState(TypedDict): 技术评审团队状态 requirement: str # 产品需求 pm_analysis: Optional[str] # 产品经理分析 backend_plan: Optional[str] # 后端方案 frontend_plan: Optional[str] # 前端方案 final_decision: Optional[str] # 最终结论 discussion_log: List[str] # 讨论日志 current_step: str # 当前步骤步骤 2创建各角色的智能体节点每个角色都是一个独立的“智能体”拥有自己的 LLM 和系统提示词角色设定。def create_agent_node(role: str, system_prompt: str): 工厂函数创建指定角色的智能体节点 # 为不同角色定制 LLM这里使用同一个也可用不同模型 role_llm ChatOpenAI(modelgpt-3.5-turbo, temperature0.7) # 创建带有角色设定的 Chain from langchain.prompts import ChatPromptTemplate from langchain.schema.output_parser import StrOutputParser prompt ChatPromptTemplate.from_messages([ (system, system_prompt), (human, {input}) ]) chain prompt | role_llm | StrOutputParser() def node_function(state: TeamReviewState): 节点函数根据当前步骤和状态执行角色任务 log_entry f[{datetime.now().strftime(%H:%M:%S)}] {role} 开始工作。 state[discussion_log].append(log_entry) # 根据角色和当前步骤决定输入 if role PM: input_text f产品需求是{state[requirement]}。请分析需求的合理性和核心目标。 result chain.invoke({input: input_text}) state[pm_analysis] result state[current_step] analysis_done elif role Backend: input_text f需求{state[requirement]}。产品经理分析{state.get(pm_analysis)}。请给出后端技术方案包括API设计、数据库选型等。 result chain.invoke({input: input_text}) state[backend_plan] result state[current_step] backend_done elif role Frontend: input_text f需求{state[requirement]}。产品经理分析{state.get(pm_analysis)}。请给出前端技术方案包括技术栈、页面结构、状态管理等。 result chain.invoke({input: input_text}) state[frontend_plan] result state[current_step] frontend_done return state return node_function # 定义各角色的系统提示词 pm_system_prompt 你是一名资深产品经理。你的任务是理解用户需求分析其商业价值和技术可行性并清晰地定义核心功能点。输出应结构化。 be_system_prompt 你是一名资深后端工程师精通微服务和云原生架构。你的任务是根据需求和产品分析设计稳健、可扩展的后端技术方案。 fe_system_prompt 你是一名资深前端工程师精通现代前端框架。你的任务是根据需求设计用户体验良好、性能优异的前端技术方案。 # 创建角色节点 pm_node create_agent_node(PM, pm_system_prompt) backend_node create_agent_node(Backend, be_system_prompt) frontend_node create_agent_node(Frontend, fe_system_prompt)步骤 3创建决策与汇总节点我们需要一个“协调员”节点来检查并行任务是否完成并触发最终汇总。def decision_node(state: TeamReviewState): 决策节点检查后端和前端方案是否都已完成并决定下一步 log_entry f[{datetime.now().strftime(%H:%M:%S)}] 协调员检查任务进度。 state[discussion_log].append(log_entry) # 检查两个并行任务的结果 if state.get(backend_plan) and state.get(frontend_plan): state[current_step] ready_for_summary print(协调员后端和前端方案已就绪进入汇总阶段。) else: state[current_step] waiting print(协调员正在等待后端或前端方案...) return state def summary_node(state: TeamReviewState): 汇总节点由PM或另一个LLM生成最终评审结论 log_entry f[{datetime.now().strftime(%H:%M:%S)}] 产品经理进行最终汇总。 state[discussion_log].append(log_entry) summary_llm ChatOpenAI(modelgpt-3.5-turbo, temperature0.5) input_text f 基于以下信息生成一份综合的技术评审结论 1. 原始需求{state[requirement]} 2. 产品分析{state[pm_analysis]} 3. 后端方案{state[backend_plan]} 4. 前端方案{state[frontend_plan]} 结论应包括技术可行性评估、主要风险点、建议排期、所需资源。 final_decision summary_llm.invoke(input_text).content state[final_decision] final_decision state[current_step] completed return state步骤 4构建包含并行和条件边的复杂图这是 LangGraph 最强大的部分我们可以定义backend_node和frontend_node在pm_node之后并行执行。from langgraph.graph import StateGraph, START, END from langgraph.graph import add_messages workflow_mas StateGraph(TeamReviewState) # 添加所有节点 workflow_mas.add_node(pm, pm_node) workflow_mas.add_node(backend, backend_node) workflow_mas.add_node(frontend, frontend_node) workflow_mas.add_node(decision, decision_node) workflow_mas.add_node(summary, summary_node) # 设置流程 workflow_mas.set_entry_point(pm) # PM 分析完后后端和前端并行开始 workflow_mas.add_edge(pm, backend) workflow_mas.add_edge(pm, frontend) # 后端和前端的下一步都是到决策节点 workflow_mas.add_edge(backend, decision) workflow_mas.add_edge(frontend, decision) # 决策节点根据状态决定是等待还是汇总 def decide_after_decision(state): if state[current_step] ready_for_summary: return summary # 去汇总节点 else: return decision # 继续等待循环 workflow_mas.add_conditional_edges( decision, decide_after_decision, { summary: summary, decision: decision, # 指向自己形成等待循环 } ) # 汇总节点后结束 workflow_mas.add_edge(summary, END) # 编译图 mas_app workflow_mas.compile()步骤 5运行多智能体系统# 初始化状态 initial_mas_state: TeamReviewState { requirement: 开发一个企业内部的知识库问答系统支持文档上传、智能检索和对话式问答。, pm_analysis: None, backend_plan: None, frontend_plan: None, final_decision: None, discussion_log: [], current_step: start } print(*50) print(启动多智能体技术评审系统...) print(f评审需求{initial_mas_state[requirement]}) print(*50) # 运行图 final_state mas_app.invoke(initial_mas_state) print(\n *50) print(评审完成) print(*50) print(\n--- 产品经理分析 ---) print(final_state.get(pm_analysis, N/A)) print(\n--- 后端方案 ---) print(final_state.get(backend_plan, N/A)) print(\n--- 前端方案 ---) print(final_state.get(frontend_plan, N/A)) print(\n--- 最终结论 ---) print(final_state.get(final_decision, N/A)) print(\n--- 讨论日志 ---) for log in final_state[discussion_log]: print(log)运行这个系统你会看到三个角色智能体依次被激活后端和前端并行工作协调员检查进度最终由产品经理汇总出结论。这完整演示了一个基于 LangGraph 的、有状态、有分支、有并行的多智能体协作系统。8. 接口 API 与生产部署开发完成后你需要将 LangGraph 智能体暴露为 API 服务供其他系统调用。使用 FastAPI 是常见选择。步骤 1创建 FastAPI 应用# 文件: app.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Dict, Any import asyncio # 假设你已经定义并编译好了你的智能体图 app (来自前面的例子) # from your_workflow import app # 初始化 FastAPI api_app FastAPI(titleLangGraph 智能体 API) # 定义请求体模型 class AgentRequest(BaseModel): 调用智能体的请求 input_state: Dict[str, Any] # 初始状态 config: Dict[str, Any] {configurable: {}} # 可配置参数 class AgentResponse(BaseModel): 智能体响应 output_state: Dict[str, Any] execution_log: str # 定义 API 端点 api_app.post(/run, response_modelAgentResponse) async def run_agent(request: AgentRequest): 运行智能体工作流 try: # 调用编译好的 LangGraph app # 注意app.invoke 是同步的在异步上下文中使用 asyncio.to_thread result_state await asyncio.to_thread(app.invoke, request.input_state, request.config) # 这里可以提取或格式化执行日志如果图有记录的话 log 执行成功 return AgentResponse(output_stateresult_state, execution_loglog) except Exception as e: raise HTTPException(status_code500, detailf智能体执行失败: {str(e)}) api_app.get(/health) async def health_check(): 健康检查端点 return {status: healthy, framework: LangGraph}步骤 2使用 Uvicorn 运行服务# 安装依赖 pip install fastapi uvicorn # 运行服务 (假设文件名为 app.py) uvicorn app:api_app --host 0.0.0.0 --port 8000 --reload服务启动后访问http://localhost:8000/docs可以看到自动生成的 API 文档。步骤 3使用 curl 或 Python 客户端调用# 使用 curl 调用 curl -X POST http://localhost:8000/run \ -H Content-Type: application/json \ -d { input_state: { requirement: 测试需求, messages: [], turn_count: 0 } }# 使用 Python requests 调用 import requests import json url http://localhost:8000/run payload { input_state: { requirement: 开发一个用户反馈分析仪表盘, messages: [], turn_count: 0 } } headers {Content-Type: application/json} response requests.post(url, datajson.dumps(payload), headersheaders) print(response.json())生产部署建议使用 Gunicorn/Uvicorn Worker对于生产环境使用gunicorn配合多个uvicorn worker提高并发。gunicorn -w 4 -k uvicorn.workers.UvicornWorker app:api_app设置超时与重试在客户端和服务器端设置合理的超时并对 transient 错误实现重试机制。状态持久化对于长时间运行或重要的流程应将状态State保存到数据库如 Redis、PostgreSQL而非仅内存中。可以自定义 State 的序列化与存储。监控与日志集成 Prometheus、OpenTelemetry 等工具对图的执行步骤、耗时、错误进行监控和记录。安全API 端点应添加认证如 API Key、JWT和速率限制。9. 常见问题与排查方法在开发和运行 LangGraph 应用时你可能会遇到以下典型问题。问题现象可能原因排查方式解决方案导入StateGraph失败LangGraph 版本不匹配或未安装。检查 pip listgrep langgraph。运行app.invoke()时报状态字段错误状态State的类型定义与实际返回的字典键不匹配。打印state的结构与TypedDict定义对比。确保所有节点函数返回的字典都包含状态定义的所有键或使用Annotated设置默认值。智能体陷入无限循环条件边Conditional Edge的逻辑有误导致无法跳转到END。在节点函数中打印state[current_step]观察流转。仔细检查add_conditional_edges的条件函数确保存在出口条件。使用langgraph的检查点或可视化工具调试。调用 OpenAI API 超时或报错网络问题、API Key 无效、额度不足、请求频率超限。查看完整的错误信息检查环境变量OPENAI_API_KEY。配置网络代理如需、更换 Key、升级账户、降低请求频率或增加超时时间。工具调用失败或结果不符合预期工具函数的参数解析错误或工具内部异常。在工具函数内部添加详细打印。检查 LLM 生成的工具调用字符串。确保工具的描述清晰使用tool装饰器正确。考虑使用更严格的参数解析库如 Pydantic。多智能体并行执行顺序混乱对add_edge的理解有误并行需要连接到同一个后续节点来“同步”。画出示意图确认期望的并行和汇聚点。并行分支的起点应相同或通过条件触发终点应汇聚到同一个决策节点。使用add_edge明确连接。LangGraph Studio 无法连接或显示错误Docker 容器未正确启动端口冲突或浏览器缓存。检查 Docker 容器状态docker ps查看日志docker logs container_id。确保端口 3000 未被占用重启容器清除浏览器缓存。状态数据过大导致内存溢出在messages等列表中不断追加内容未做清理。监控进程内存使用情况。实现状态清理策略例如只保留最近 N 轮对话或将历史存储到外部数据库。部署为 API 后性能低下同步调用app.invoke阻塞了事件循环LLM 调用慢。使用异步性能分析工具。1. 在 FastAPI 中使用asyncio.to_thread包装同步调用。2. 为 LLM 调用配置缓存。3. 考虑使用更快的模型或优化提示词。10. 最佳实践与使用建议从简单开始逐步复杂化不要一开始就设计庞大的图。先构建一个最小可行的工作流确保状态流转正确再逐步添加节点、分支和并行逻辑。精心设计状态结构状态是图的“脊柱”。使用TypedDict或Pydantic BaseModel明确定义所有字段和类型。避免使用过于复杂嵌套的结构。为节点函数命名在add_node时使用有意义的名称如validate_input,call_llm这将在调试和可视化时极大提升可读性。利用可视化在开发过程中频繁使用langgraph的get_graph().draw_mermaid_png()输出流程图或直接使用 LangGraph Studio 进行可视化编辑和调试。实现健壮的错误处理在节点函数内部使用 try-catch并考虑如何将错误信息反映到状态中以便图能根据错误类型走不同的恢复路径。管理 LLM 成本与延迟对于复杂流程LLM 调用可能是主要开销。考虑缓存 LLM 响应、使用更小/更快的模型处理简单步骤、设置合理的超时和重试。测试与模拟编写单元测试来测试单个节点函数。对于整个图可以模拟 LLM 和工具的响应进行集成测试。版本控制你的图图的定义节点、边、条件也是代码。将其纳入 Git 版本控制并考虑当业务逻辑变更时如何平滑迁移正在运行中的长周期工作流状态。LangGraph 将一个复杂的智能体系统抽象为一张清晰可控的图这是它最大的优势。它可能不是最简单的入门方式但绝对是构建可持续维护、逻辑复杂的企业级 AI 应用时最有力的框架之一。从定义一个状态和两个节点开始亲手运行起来你就能切身感受到它将抽象思维转化为可执行代码的强大能力。 LangGraph实战：构建有状态多智能体协作系统的核心指南
LangGraph 是 LangChain 生态中用于构建复杂、有状态、多步骤 AI 应用尤其是智能体的框架。它把智能体的执行过程建模成一张图Graph节点是执行步骤如调用 LLM、使用工具边是步骤之间的流转逻辑。这让你能清晰地设计出包含循环、分支、并行、人工干预的智能体工作流远超传统单次问答的简单模式。如果你正在寻找一个能构建企业级、可维护、可观测 AI 智能体的方案LangGraph 是目前最值得投入的技术栈之一。它解决了传统链式调用在复杂逻辑和长期记忆上的短板特别适合客服对话、数据分析流水线、自动化决策等需要多轮交互和状态管理的场景。本文不会停留在概念而是直接带你从环境搭建、核心概念到多智能体系统实战用代码和架构图把 LangGraph 讲透。我们将重点关注 LangGraph 如何定义工作流、管理状态、集成工具并最终构建一个模拟“技术团队评审”的多智能体协作系统。整个过程无需昂贵硬件在普通开发机上即可运行重点在于理解其架构思想和工程化实践。1. 核心能力速览能力项说明项目类型AI 智能体Agent与工作流Workflow编排框架核心价值将复杂的、有状态的智能体逻辑用“图”来清晰定义和管理关键特性支持循环Loop、条件分支Condition、并行Parallel、人工干预Human-in-the-loop硬件门槛无特殊要求依赖后端 LLM 服务如 OpenAI API、本地 Ollama。本地运行 Ollama 时需注意内存。启动方式Python 库安装通过代码定义和运行图。提供 LangGraph Studio 进行可视化开发。接口能力核心是 Python API。构建的图可暴露为可调用的函数易于集成到 FastAPI 等 Web 服务中。状态管理内置状态State管理支持自定义状态结构在节点间持久化和传递数据。适合场景多轮对话系统、自动化业务流程、决策支持系统、多角色协作智能体2. 适用场景与使用边界LangGraph 并非所有 AI 项目的首选它针对的是特定复杂度的场景。最适合的场景多步骤任务任务需要拆解为多个顺序或并行的步骤例如信息收集 - 分析 - 报告生成。有状态交互后续步骤依赖前序步骤的结果或整个对话历史例如客服对话、面试模拟。需要循环或分支根据 LLM 的输出或工具调用的结果决定重复执行还是走不同路径例如持续优化代码直到测试通过。多智能体协作需要多个具备不同角色和能力的智能体共同完成一项任务例如产品经理、工程师、测试员协同开发。需要人工审核在流程的关键节点插入人工确认或输入。不适合或需谨慎使用的场景简单的一次性问答如果只是调用一次 LLM 接口直接使用 LangChain 的 LCEL 或原始 SDK 更简单。对延迟极其敏感图的调度、状态管理会引入额外开销在毫秒级响应的场景下可能不适用。逻辑极其简单用图来管理一个线性三步流程可能显得“杀鸡用牛刀”。无状态的批量处理单纯对一批数据各自进行独立处理用传统脚本或并行计算框架更高效。合规与安全边界LLM 责任LangGraph 是编排框架内容生成的责任在于其集成的 LLM如 GPT、Claude。需确保 LLM 的使用符合其服务条款。工具安全智能体可以调用外部工具如执行代码、访问数据库。必须严格审查和沙箱化工具权限防止越权操作。数据隐私工作流状态中可能包含用户输入的敏感信息。需做好状态数据的加密、脱敏和访问控制。人工干预涉及人工审核的流程需明确审核人员的权责和操作日志。3. 环境准备与前置条件在开始编码前需要准备好以下环境。本文以 OpenAI 的 GPT 模型作为 LLM 后端为例。操作系统Windows 10/11, macOS, 或 Linux (如 Ubuntu 20.04)。LangGraph 是 Python 库跨平台兼容性好。Python 版本推荐 Python 3.10 或 3.11。确保已安装pip。LLM 服务接入方案一推荐简单准备一个有效的 OpenAI API Key 。后续代码将使用chatopenai模型。方案二本地安装并运行 Ollama 拉取一个模型如llama3.2或qwen2.5。这需要本地有足够内存通常 8GB。虚拟环境推荐使用venv或conda创建独立环境避免包冲突。网络能访问 OpenAI API 或下载 Ollama 模型。4. 安装部署与启动方式LangGraph 的“启动”即是安装 Python 包并运行你的脚本。我们首先安装核心依赖。步骤 1创建并激活虚拟环境# 使用 venv python -m venv langgraph-env # Windows langgraph-env\Scripts\activate # macOS/Linux source langgraph-env/bin/activate步骤 2安装 LangGraph 及 LLM 集成包pip install langgraph langchain-openailanggraph: 核心框架。langchain-openai: 用于集成 OpenAI 模型。如果你计划使用 Ollama则安装pip install langgraph langchain-community步骤 3验证安装创建一个简单的 Python 脚本test_import.pyfrom langgraph.graph import StateGraph, END print(LangGraph 导入成功)运行python test_import.py无报错即说明安装成功。关于 LangGraph Studio LangChain 提供了可视化的开发工具 LangGraph Studio可以通过 Docker 快速启动用于可视化编辑和调试图。对于初学者建议先通过代码理解概念再使用 Studio。# 通过 Docker 启动 Studio (可选) docker run -d -p 3000:3000 langchain/langgraph-studio访问http://localhost:3000即可使用。5. 核心概念与第一个智能体要掌握 LangGraph必须先理解三个核心概念State状态、Node节点、Edge边。State一个字典或 Pydantic 模型定义了在整个工作流中传递和更新的数据。它是图的“记忆”。Node一个函数接收当前的 State执行操作如调用 LLM、工具并返回更新后的 State。Edge连接节点的线决定流程的走向。可以是固定的也可以根据条件Condition动态决定。让我们构建一个最简单的“聊天机器人”智能体它只是重复用户的话。步骤 1定义状态状态决定了智能体能记住什么。我们定义一个非常简单的状态只包含对话轮次和最新消息。from typing import TypedDict, List class AgentState(TypedDict): 定义智能体的状态结构 messages: List[str] # 记录所有消息 turn_count: int # 记录对话轮次步骤 2创建节点函数节点是执行单元。这里创建两个节点一个处理用户输入一个生成回复。def user_node(state: AgentState) - AgentState: 模拟用户输入节点。在实际应用中这里可能连接Web接口。 user_input input(f[Turn {state[turn_count]}] 用户: ) new_messages state[messages] [f用户: {user_input}] return {messages: new_messages, turn_count: state[turn_count]} def assistant_node(state: AgentState) - AgentState: 助理节点。这里我们简单地将最后一条用户消息重复一遍。 last_message state[messages][-1] # 模拟 LLM 处理这里只是简单重复 response f助理: 我听到你说的是 {last_message} new_messages state[messages] [response] return {messages: new_messages, turn_count: state[turn_count] 1}步骤 3构建图并设置流转逻辑这是 LangGraph 的核心将节点组装成图并定义它们如何连接。from langgraph.graph import StateGraph, END # 1. 创建一个图构建器并指定状态结构 workflow StateGraph(AgentState) # 2. 添加节点 workflow.add_node(user, user_node) workflow.add_node(assistant, assistant_node) # 3. 设置入口点 workflow.set_entry_point(user) # 4. 添加边定义流程 workflow.add_edge(user, assistant) # 用户节点后总是执行助理节点 workflow.add_edge(assistant, END) # 助理节点后结束 # 5. 编译图得到一个可执行的对象 app workflow.compile()步骤 4运行智能体现在我们可以初始化状态并运行这个图。# 初始化状态 initial_state: AgentState {messages: [], turn_count: 0} # 运行图 print(--- 开始简单聊天机器人 ---) for _ in range(3): # 模拟3轮对话 # app.invoke 会从入口点开始执行直到遇到 END result app.invoke(initial_state) # 更新状态以便下一轮使用 initial_state result print(f当前消息记录: {result[messages][-2:]}\n) print(--- 对话结束 ---)运行这段代码你将看到一个简单的三轮对话流程。这虽然简单但完整展示了 LangGraph 构建智能体的模式定义状态 - 创建节点 - 组装成图 - 运行。6. 集成真实 LLM 与工具调用一个真正的智能体需要连接大语言模型LLM和外部工具。我们将升级上面的例子让助理节点调用 OpenAI GPT 来生成回复并集成一个计算器工具。步骤 1设置 OpenAI 客户端首先确保你已设置环境变量OPENAI_API_KEY或在代码中指定。import os from langchain_openai import ChatOpenAI # 初始化 LLM。推荐使用 gpt-3.5-turbo 进行测试成本低。 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 如果你用 Ollama可以这样初始化 # from langchain_community.chat_models import ChatOllama # llm ChatOllama(modelllama3.2)步骤 2定义工具LangChain 提供了便捷的工具装饰器。我们定义一个简单的计算器。from langchain.tools import tool tool def calculator(expression: str) - str: 计算一个数学表达式的值。例如calculator(3 5 * 2)。 try: # 警告实际生产中应对 eval 进行严格的安全限制或使用更安全的库如 ast.literal_eval。 result eval(expression, {__builtins__: None}, {}) return f计算结果: {result} except Exception as e: return f计算错误: {e}步骤 3创建具有工具调用能力的智能体节点我们将使用 LangChain 的create_react_agent模式来构建一个能自主决定是否使用工具的智能体节点。from langgraph.prebuilt import create_react_agent from langchain.agents import AgentExecutor # 创建智能体执行器 tools [calculator] agent_executor create_react_agent(llm, tools) # 定义一个新的助理节点函数它使用这个执行器 def smart_assistant_node(state: AgentState): 集成了LLM和工具的智能助理节点 # 获取最新的用户消息 last_user_message state[messages][-1] if state[messages] else 你好 # 构造给智能体的输入 agent_input {input: last_user_message} # 调用智能体执行器 response agent_executor.invoke(agent_input) # 提取输出 assistant_response response.get(output, 我没有得到回复。) # 更新状态 new_messages state[messages] [f助理: {assistant_response}] return {messages: new_messages, turn_count: state[turn_count] 1}步骤 4重构并运行增强版智能体现在我们用新的smart_assistant_node替换之前的简单节点。# 重新构建图 workflow_v2 StateGraph(AgentState) workflow_v2.add_node(user, user_node) # 沿用之前的用户节点 workflow_v2.add_node(assistant, smart_assistant_node) # 使用新的智能节点 workflow_v2.set_entry_point(user) workflow_v2.add_edge(user, assistant) workflow_v2.add_edge(assistant, END) app_v2 workflow_v2.compile() # 运行测试 print(--- 启动增强版智能体集成GPT与工具---) test_state: AgentState {messages: [用户: 3加5乘以2等于多少], turn_count: 1} result app_v2.invoke(test_state) print(f最终回复: {result[messages][-1]})运行后智能体会分析问题“3加5乘以2等于多少”识别出需要计算调用calculator(35*2)工具得到结果 13并组织语言回复给你。这演示了智能体的核心能力理解、规划、执行工具、响应。7. 构建多智能体协作系统实战多智能体系统MAS是 LangGraph 的杀手级应用。我们模拟一个“技术方案评审会”包含三个角色产品经理PM、后端工程师BE、前端工程师FE。流程是PM 提出需求 - BE 和 FE 并行给出技术方案 - PM 汇总并给出结论。步骤 1定义多智能体协作状态状态需要能容纳多个角色的输出和最终结论。from typing import TypedDict, List, Optional from datetime import datetime class TeamReviewState(TypedDict): 技术评审团队状态 requirement: str # 产品需求 pm_analysis: Optional[str] # 产品经理分析 backend_plan: Optional[str] # 后端方案 frontend_plan: Optional[str] # 前端方案 final_decision: Optional[str] # 最终结论 discussion_log: List[str] # 讨论日志 current_step: str # 当前步骤步骤 2创建各角色的智能体节点每个角色都是一个独立的“智能体”拥有自己的 LLM 和系统提示词角色设定。def create_agent_node(role: str, system_prompt: str): 工厂函数创建指定角色的智能体节点 # 为不同角色定制 LLM这里使用同一个也可用不同模型 role_llm ChatOpenAI(modelgpt-3.5-turbo, temperature0.7) # 创建带有角色设定的 Chain from langchain.prompts import ChatPromptTemplate from langchain.schema.output_parser import StrOutputParser prompt ChatPromptTemplate.from_messages([ (system, system_prompt), (human, {input}) ]) chain prompt | role_llm | StrOutputParser() def node_function(state: TeamReviewState): 节点函数根据当前步骤和状态执行角色任务 log_entry f[{datetime.now().strftime(%H:%M:%S)}] {role} 开始工作。 state[discussion_log].append(log_entry) # 根据角色和当前步骤决定输入 if role PM: input_text f产品需求是{state[requirement]}。请分析需求的合理性和核心目标。 result chain.invoke({input: input_text}) state[pm_analysis] result state[current_step] analysis_done elif role Backend: input_text f需求{state[requirement]}。产品经理分析{state.get(pm_analysis)}。请给出后端技术方案包括API设计、数据库选型等。 result chain.invoke({input: input_text}) state[backend_plan] result state[current_step] backend_done elif role Frontend: input_text f需求{state[requirement]}。产品经理分析{state.get(pm_analysis)}。请给出前端技术方案包括技术栈、页面结构、状态管理等。 result chain.invoke({input: input_text}) state[frontend_plan] result state[current_step] frontend_done return state return node_function # 定义各角色的系统提示词 pm_system_prompt 你是一名资深产品经理。你的任务是理解用户需求分析其商业价值和技术可行性并清晰地定义核心功能点。输出应结构化。 be_system_prompt 你是一名资深后端工程师精通微服务和云原生架构。你的任务是根据需求和产品分析设计稳健、可扩展的后端技术方案。 fe_system_prompt 你是一名资深前端工程师精通现代前端框架。你的任务是根据需求设计用户体验良好、性能优异的前端技术方案。 # 创建角色节点 pm_node create_agent_node(PM, pm_system_prompt) backend_node create_agent_node(Backend, be_system_prompt) frontend_node create_agent_node(Frontend, fe_system_prompt)步骤 3创建决策与汇总节点我们需要一个“协调员”节点来检查并行任务是否完成并触发最终汇总。def decision_node(state: TeamReviewState): 决策节点检查后端和前端方案是否都已完成并决定下一步 log_entry f[{datetime.now().strftime(%H:%M:%S)}] 协调员检查任务进度。 state[discussion_log].append(log_entry) # 检查两个并行任务的结果 if state.get(backend_plan) and state.get(frontend_plan): state[current_step] ready_for_summary print(协调员后端和前端方案已就绪进入汇总阶段。) else: state[current_step] waiting print(协调员正在等待后端或前端方案...) return state def summary_node(state: TeamReviewState): 汇总节点由PM或另一个LLM生成最终评审结论 log_entry f[{datetime.now().strftime(%H:%M:%S)}] 产品经理进行最终汇总。 state[discussion_log].append(log_entry) summary_llm ChatOpenAI(modelgpt-3.5-turbo, temperature0.5) input_text f 基于以下信息生成一份综合的技术评审结论 1. 原始需求{state[requirement]} 2. 产品分析{state[pm_analysis]} 3. 后端方案{state[backend_plan]} 4. 前端方案{state[frontend_plan]} 结论应包括技术可行性评估、主要风险点、建议排期、所需资源。 final_decision summary_llm.invoke(input_text).content state[final_decision] final_decision state[current_step] completed return state步骤 4构建包含并行和条件边的复杂图这是 LangGraph 最强大的部分我们可以定义backend_node和frontend_node在pm_node之后并行执行。from langgraph.graph import StateGraph, START, END from langgraph.graph import add_messages workflow_mas StateGraph(TeamReviewState) # 添加所有节点 workflow_mas.add_node(pm, pm_node) workflow_mas.add_node(backend, backend_node) workflow_mas.add_node(frontend, frontend_node) workflow_mas.add_node(decision, decision_node) workflow_mas.add_node(summary, summary_node) # 设置流程 workflow_mas.set_entry_point(pm) # PM 分析完后后端和前端并行开始 workflow_mas.add_edge(pm, backend) workflow_mas.add_edge(pm, frontend) # 后端和前端的下一步都是到决策节点 workflow_mas.add_edge(backend, decision) workflow_mas.add_edge(frontend, decision) # 决策节点根据状态决定是等待还是汇总 def decide_after_decision(state): if state[current_step] ready_for_summary: return summary # 去汇总节点 else: return decision # 继续等待循环 workflow_mas.add_conditional_edges( decision, decide_after_decision, { summary: summary, decision: decision, # 指向自己形成等待循环 } ) # 汇总节点后结束 workflow_mas.add_edge(summary, END) # 编译图 mas_app workflow_mas.compile()步骤 5运行多智能体系统# 初始化状态 initial_mas_state: TeamReviewState { requirement: 开发一个企业内部的知识库问答系统支持文档上传、智能检索和对话式问答。, pm_analysis: None, backend_plan: None, frontend_plan: None, final_decision: None, discussion_log: [], current_step: start } print(*50) print(启动多智能体技术评审系统...) print(f评审需求{initial_mas_state[requirement]}) print(*50) # 运行图 final_state mas_app.invoke(initial_mas_state) print(\n *50) print(评审完成) print(*50) print(\n--- 产品经理分析 ---) print(final_state.get(pm_analysis, N/A)) print(\n--- 后端方案 ---) print(final_state.get(backend_plan, N/A)) print(\n--- 前端方案 ---) print(final_state.get(frontend_plan, N/A)) print(\n--- 最终结论 ---) print(final_state.get(final_decision, N/A)) print(\n--- 讨论日志 ---) for log in final_state[discussion_log]: print(log)运行这个系统你会看到三个角色智能体依次被激活后端和前端并行工作协调员检查进度最终由产品经理汇总出结论。这完整演示了一个基于 LangGraph 的、有状态、有分支、有并行的多智能体协作系统。8. 接口 API 与生产部署开发完成后你需要将 LangGraph 智能体暴露为 API 服务供其他系统调用。使用 FastAPI 是常见选择。步骤 1创建 FastAPI 应用# 文件: app.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Dict, Any import asyncio # 假设你已经定义并编译好了你的智能体图 app (来自前面的例子) # from your_workflow import app # 初始化 FastAPI api_app FastAPI(titleLangGraph 智能体 API) # 定义请求体模型 class AgentRequest(BaseModel): 调用智能体的请求 input_state: Dict[str, Any] # 初始状态 config: Dict[str, Any] {configurable: {}} # 可配置参数 class AgentResponse(BaseModel): 智能体响应 output_state: Dict[str, Any] execution_log: str # 定义 API 端点 api_app.post(/run, response_modelAgentResponse) async def run_agent(request: AgentRequest): 运行智能体工作流 try: # 调用编译好的 LangGraph app # 注意app.invoke 是同步的在异步上下文中使用 asyncio.to_thread result_state await asyncio.to_thread(app.invoke, request.input_state, request.config) # 这里可以提取或格式化执行日志如果图有记录的话 log 执行成功 return AgentResponse(output_stateresult_state, execution_loglog) except Exception as e: raise HTTPException(status_code500, detailf智能体执行失败: {str(e)}) api_app.get(/health) async def health_check(): 健康检查端点 return {status: healthy, framework: LangGraph}步骤 2使用 Uvicorn 运行服务# 安装依赖 pip install fastapi uvicorn # 运行服务 (假设文件名为 app.py) uvicorn app:api_app --host 0.0.0.0 --port 8000 --reload服务启动后访问http://localhost:8000/docs可以看到自动生成的 API 文档。步骤 3使用 curl 或 Python 客户端调用# 使用 curl 调用 curl -X POST http://localhost:8000/run \ -H Content-Type: application/json \ -d { input_state: { requirement: 测试需求, messages: [], turn_count: 0 } }# 使用 Python requests 调用 import requests import json url http://localhost:8000/run payload { input_state: { requirement: 开发一个用户反馈分析仪表盘, messages: [], turn_count: 0 } } headers {Content-Type: application/json} response requests.post(url, datajson.dumps(payload), headersheaders) print(response.json())生产部署建议使用 Gunicorn/Uvicorn Worker对于生产环境使用gunicorn配合多个uvicorn worker提高并发。gunicorn -w 4 -k uvicorn.workers.UvicornWorker app:api_app设置超时与重试在客户端和服务器端设置合理的超时并对 transient 错误实现重试机制。状态持久化对于长时间运行或重要的流程应将状态State保存到数据库如 Redis、PostgreSQL而非仅内存中。可以自定义 State 的序列化与存储。监控与日志集成 Prometheus、OpenTelemetry 等工具对图的执行步骤、耗时、错误进行监控和记录。安全API 端点应添加认证如 API Key、JWT和速率限制。9. 常见问题与排查方法在开发和运行 LangGraph 应用时你可能会遇到以下典型问题。问题现象可能原因排查方式解决方案导入StateGraph失败LangGraph 版本不匹配或未安装。检查 pip listgrep langgraph。运行app.invoke()时报状态字段错误状态State的类型定义与实际返回的字典键不匹配。打印state的结构与TypedDict定义对比。确保所有节点函数返回的字典都包含状态定义的所有键或使用Annotated设置默认值。智能体陷入无限循环条件边Conditional Edge的逻辑有误导致无法跳转到END。在节点函数中打印state[current_step]观察流转。仔细检查add_conditional_edges的条件函数确保存在出口条件。使用langgraph的检查点或可视化工具调试。调用 OpenAI API 超时或报错网络问题、API Key 无效、额度不足、请求频率超限。查看完整的错误信息检查环境变量OPENAI_API_KEY。配置网络代理如需、更换 Key、升级账户、降低请求频率或增加超时时间。工具调用失败或结果不符合预期工具函数的参数解析错误或工具内部异常。在工具函数内部添加详细打印。检查 LLM 生成的工具调用字符串。确保工具的描述清晰使用tool装饰器正确。考虑使用更严格的参数解析库如 Pydantic。多智能体并行执行顺序混乱对add_edge的理解有误并行需要连接到同一个后续节点来“同步”。画出示意图确认期望的并行和汇聚点。并行分支的起点应相同或通过条件触发终点应汇聚到同一个决策节点。使用add_edge明确连接。LangGraph Studio 无法连接或显示错误Docker 容器未正确启动端口冲突或浏览器缓存。检查 Docker 容器状态docker ps查看日志docker logs container_id。确保端口 3000 未被占用重启容器清除浏览器缓存。状态数据过大导致内存溢出在messages等列表中不断追加内容未做清理。监控进程内存使用情况。实现状态清理策略例如只保留最近 N 轮对话或将历史存储到外部数据库。部署为 API 后性能低下同步调用app.invoke阻塞了事件循环LLM 调用慢。使用异步性能分析工具。1. 在 FastAPI 中使用asyncio.to_thread包装同步调用。2. 为 LLM 调用配置缓存。3. 考虑使用更快的模型或优化提示词。10. 最佳实践与使用建议从简单开始逐步复杂化不要一开始就设计庞大的图。先构建一个最小可行的工作流确保状态流转正确再逐步添加节点、分支和并行逻辑。精心设计状态结构状态是图的“脊柱”。使用TypedDict或Pydantic BaseModel明确定义所有字段和类型。避免使用过于复杂嵌套的结构。为节点函数命名在add_node时使用有意义的名称如validate_input,call_llm这将在调试和可视化时极大提升可读性。利用可视化在开发过程中频繁使用langgraph的get_graph().draw_mermaid_png()输出流程图或直接使用 LangGraph Studio 进行可视化编辑和调试。实现健壮的错误处理在节点函数内部使用 try-catch并考虑如何将错误信息反映到状态中以便图能根据错误类型走不同的恢复路径。管理 LLM 成本与延迟对于复杂流程LLM 调用可能是主要开销。考虑缓存 LLM 响应、使用更小/更快的模型处理简单步骤、设置合理的超时和重试。测试与模拟编写单元测试来测试单个节点函数。对于整个图可以模拟 LLM 和工具的响应进行集成测试。版本控制你的图图的定义节点、边、条件也是代码。将其纳入 Git 版本控制并考虑当业务逻辑变更时如何平滑迁移正在运行中的长周期工作流状态。LangGraph 将一个复杂的智能体系统抽象为一张清晰可控的图这是它最大的优势。它可能不是最简单的入门方式但绝对是构建可持续维护、逻辑复杂的企业级 AI 应用时最有力的框架之一。从定义一个状态和两个节点开始亲手运行起来你就能切身感受到它将抽象思维转化为可执行代码的强大能力。