这次我们来看一个对后端开发者特别友好的 AI 应用开发框架LangGraph。如果你已经熟悉了 Spring Boot、状态机、微服务编排现在想切入 AI Agent 和多智能体系统开发这个项目就是为你准备的。它不是一个玩具而是一个能让你快速构建具备复杂状态流转、工具调用和人机交互能力的企业级 AI 应用的生产力工具。LangGraph 的核心价值在于它将 AI 应用开发中常见的“状态管理”和“流程编排”难题抽象成了后端开发者非常熟悉的“图”和“节点”概念。你不用再为 Agent 的记忆、工具调用顺序、条件分支而头疼而是像设计一个微服务调用链一样用代码清晰地定义整个智能体的工作流。这对于需要处理多轮对话、复杂决策或自动化流程的后端工程师来说学习曲线非常平缓。本文将带你从零开始系统掌握 LangGraph。我们会重点关注三个核心状态State的持久化与流转、工具Tools的定义与集成以及如何构建支持人机交互的智能体。整个过程会以实战代码演示为主告诉你每一步怎么配置、怎么跑通、遇到问题怎么排查。学完你就能着手开发自己的第一个企业级多智能体系统。1. 核心能力速览在深入代码之前我们先快速了解 LangGraph 能做什么以及它的技术门槛。能力项说明项目类型AI 应用编排框架基于 LangChain 生态核心抽象有向图Graph、节点Node、边Edge、状态State主要功能构建多步骤、带状态、可循环的 AI 工作流Agent硬件门槛无特殊要求。框架本身是 Python 库计算负载取决于集成的 AI 模型如 OpenAI API 或本地大模型。纯逻辑编排对 CPU/GPU 无要求。启动方式通过 Python 脚本或 Jupyter Notebook 启动工作流。可封装为 REST API 服务如使用 FastAPI。接口能力提供编程接口Python SDK来定义和运行图。可轻松集成到现有后端服务中。批量任务支持通过异步或并行处理对多个输入流执行相同的工作流图。适合场景客服对话机器人、复杂决策助手、自动化流程RPA、数据分析流水线、多智能体协作系统。前置知识基础 Python了解 LangChain 基础概念如 LLM、Prompt、Chain更佳。从上表可以看出LangGraph 的门槛主要在“软件设计”而非“硬件部署”。它帮你解决了 AI 应用中最复杂的流程控制问题让你能更专注于业务逻辑和提示工程。2. 适用场景与使用边界适合谁用后端/全栈开发者熟悉系统设计、API 开发和状态管理希望快速将 AI 能力集成到现有产品中。AI 应用工程师已经使用过 LangChain但发现简单的 Chain 无法处理复杂、多轮、带状态的交互场景。产品经理/技术负责人需要设计一个具有明确步骤和决策分支的 AI 流程原型。能解决什么问题复杂对话管理用户问题需要多轮追问、信息收集和确认的客服场景。自动化工作流根据输入内容自动调用不同工具搜索、数据库、计算并按顺序执行的场景如报告生成、数据审核。多智能体协作模拟多个角色分析师、审核员、执行者协作完成一项任务。游戏/模拟环境构建具有状态和规则的非玩家角色NPC或模拟环境。不适合什么场景简单的单次问答如果只是调用一次大模型 API 并返回结果使用 LangChain 的LLMChain或直接调用 SDK 更简单。对延迟极度敏感的实时系统图节点的执行、状态检查会引入额外开销。完全无状态的批处理任务如果任务间没有状态依赖使用普通函数或并发库可能更高效。安全与合规边界工具调用安全通过 LangGraph 集成的工具如网络搜索、数据库写入、命令执行必须经过严格审计和权限控制防止越权操作。数据隐私工作流中流转的状态State可能包含用户敏感信息需考虑加密存储和传输。模型输出审核对于生成内容尤其是面向公众的必须建立审核机制避免产生有害或违规信息。3. 环境准备与前置条件开始前请确保你的开发环境满足以下条件。我们将以一个相对干净的环境为例。操作系统Windows 10/11, macOS, 或 Linux (Ubuntu 20.04)。本文演示以 macOS/Linux 命令为主Windows 用户可使用 Git Bash 或 WSL。Python 版本Python 3.8 至 3.11。LangGraph 对 3.12 的兼容性请参考官方文档建议使用 3.10 或 3.11 以获得最佳兼容性。# 检查Python版本 python --version包管理工具使用pip或conda。推荐使用虚拟环境隔离项目依赖。# 创建并激活虚拟环境 (以 venv 为例) python -m venv langgraph-env source langgraph-env/bin/activate # Linux/macOS # langgraph-env\Scripts\activate # Windows基础依赖我们将安装 LangGraph 及其核心依赖。由于 LangGraph 常用于编排 LLM通常需要安装langchain和某个 LLM 提供商的库如openai。LLM 访问权限为了运行示例你需要一个可用的 LLM。本文示例将使用OpenAI API因此你需要一个有效的 OpenAI API Key。你也可以替换为其他兼容 LangChain 的模型如 Anthropic、本地部署的 Ollama 等。4. 安装部署与启动方式安装过程非常简单主要通过 pip 完成。我们安装 LangGraph 的核心库以及用于示例的 LangChain 和 OpenAI 集成。# 激活虚拟环境后安装核心包 pip install langgraph langchain-openai # 如果你需要更完整的 LangChain 生态如更多工具、文档加载器也可以安装 # pip install langchain langchain-community验证安装 创建一个简单的 Python 脚本test_import.pyimport langgraph print(fLangGraph version: {langgraph.__version__}) from langchain_openai import ChatOpenAI print(Imports successful!)运行它python test_import.py如果没有报错说明环境准备就绪。关于“启动”LangGraph 本身不是一个独立运行的服务而是一个库。你的“启动”就是执行定义了工作流的 Python 脚本。你可以将这个脚本封装在 Web 框架如 FastAPI、Flask中从而提供 HTTP API 服务。5. 功能测试与效果验证构建你的第一个智能体我们将通过三个循序渐进的例子验证 LangGraph 的核心功能状态管理、工具调用和人机交互。5.1 示例一基础状态流转State Management这个例子展示如何定义一个简单的、带状态的工作流。我们模拟一个“对话轮次计数器”。# basic_state_graph.py from typing import TypedDict, Annotated from langgraph.graph import StateGraph, END # 1. 定义状态结构 class AgentState(TypedDict): # Annotated 用于在图中声明该字段的缩减reduce方式add表示累加 messages: Annotated[list, add] # 消息列表以追加方式更新 turn_count: Annotated[int, add] # 对话轮次累加 # 2. 定义节点函数 def call_model(state: AgentState): 模拟AI模型响应并增加轮次计数 print(f[Model] 这是第 {state.get(turn_count, 0) 1} 轮对话。) # 模拟生成一个回复 new_message {role: assistant, content: f这是第 {state.get(turn_count, 0) 1} 轮回复。} return {messages: [new_message], turn_count: 1} # 返回要更新的状态部分 # 3. 构建图 builder StateGraph(AgentState) # 添加节点 builder.add_node(model, call_model) # 设置入口点 builder.set_entry_point(model) # 设置出口点执行完model节点后结束 builder.add_edge(model, END) # 编译图 graph builder.compile() # 4. 执行图 initial_state {messages: [], turn_count: 0} print(初始状态:, initial_state) # 执行一次 result graph.invoke(initial_state) print(第一次执行后状态:, result) # 再执行一次注意这里传入的是上一次的结果作为新状态 result2 graph.invoke(result) print(第二次执行后状态:, result2)运行与验证python basic_state_graph.py预期输出初始状态: {messages: [], turn_count: 0} [Model] 这是第 1 轮对话。 第一次执行后状态: {messages: [{role: assistant, content: 这是第 1 轮回复。}], turn_count: 1} [Model] 这是第 2 轮对话。 第二次执行后状态: {messages: [{role: assistant, content: 这是第 1 轮回复。}, {role: assistant, content: 这是第 2 轮回复。}], turn_count: 2}成功标准图成功编译并执行。turn_count状态在每次执行后正确累加1 - 2。messages列表正确追加了新的消息。这个例子演示了 LangGraph 状态管理的核心你定义了一个状态结构节点函数返回要更新的部分框架会自动根据注解如add合并到全局状态中。5.2 示例二集成工具调用Tool Calling这是 LangGraph 最强大的功能之一。我们创建一个能调用“计算器”和“网络搜索”模拟工具的智能体。# tool_calling_agent.py import os from typing import TypedDict, Annotated from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain.tools import tool from langgraph.prebuilt import ToolExecutor, ToolInvocation import json # 0. 设置 OpenAI API Key (请替换成你的) os.environ[OPENAI_API_KEY] your-api-key-here # 1. 定义工具 tool def calculator(expression: str) - str: 计算一个数学表达式的值。支持 , -, *, /, **。 # 警告在生产环境中直接eval是危险的此处仅为演示。 try: result eval(expression) return f计算结果: {result} except Exception as e: return f计算错误: {e} tool def search_web(query: str) - str: 模拟网络搜索。返回模拟结果。 # 模拟搜索延迟和结果 return f搜索 {query} 的模拟结果相关文章1相关文章2。 # 工具列表和执行器 tools [calculator, search_web] tool_executor ToolExecutor(tools) # 2. 定义状态 class AgentState(TypedDict): messages: Annotated[list, add] # 其他状态可以根据需要添加 # 3. 定义节点函数 def call_model(state: AgentState): 调用LLM决定是回复还是调用工具 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 将消息和历史传递给LLM并绑定工具 llm_with_tools llm.bind_tools(tools) message llm_with_tools.invoke(state[messages]) return {messages: [message]} def execute_tools(state: AgentState): 执行LLM要求调用的工具 last_message state[messages][-1] tool_calls last_message.tool_calls outputs [] for tool_call in tool_calls: # 构建工具调用请求 action ToolInvocation( tooltool_call[name], tool_inputtool_call[args], ) # 执行工具 output tool_executor.invoke(action) outputs.append(output) # 将工具执行结果作为新消息返回 return {messages: [{role: tool, content: str(output), tool_call_id: tool_call[id]} for output, tool_call in zip(outputs, tool_calls)]} # 4. 构建条件边决定下一个节点 def should_continue(state: AgentState) - str: 根据最后一条消息判断下一步调用工具还是结束 last_message state[messages][-1] if last_message.tool_calls: return tools # 有工具调用去执行工具 return END # 没有工具调用结束 # 5. 构建图 builder StateGraph(AgentState) builder.add_node(agent, call_model) # 思考节点 builder.add_node(tools, execute_tools) # 执行节点 builder.set_entry_point(agent) # 条件路由agent节点后根据判断决定去tools还是END builder.add_conditional_edges( agent, should_continue, ) # tools节点执行完后必须回到agent节点进行下一步思考 builder.add_edge(tools, agent) graph builder.compile() # 6. 执行测试 print( 测试1: 数学计算 ) initial_state {messages: [{role: user, content: 请计算 (15 7) * 3 的值是多少}]} result graph.invoke(initial_state) print(最终回复:, result[messages][-1].content) print(\n 测试2: 信息查询 ) initial_state2 {messages: [{role: user, content: LangGraph 是什么}]} result2 graph.invoke(initial_state2) print(最终回复:, result2[messages][-1].content)运行与验证将your-api-key-here替换为你的有效 OpenAI API Key。python tool_calling_agent.py预期输出内容可能因模型随机性略有不同 测试1: 数学计算 最终回复: (15 7) * 3 的值是 66。 测试2: 信息查询 最终回复: LangGraph 是一个用于构建复杂、有状态、多智能体应用程序的库它是 LangChain 生态系统的一部分。...成功标准智能体正确识别用户意图。对于计算问题成功调用了calculator工具并返回了正确结果66。对于查询问题成功调用了search_web工具模拟并整合了信息。工作流在“思考agent- 执行tools- 再思考”的循环中正确运行直到没有工具调用为止。这个例子演示了智能体的核心循环LLM 思考、决定调用工具、执行工具、将结果反馈给 LLM 进行下一步思考。LangGraph 优雅地管理了这个循环的状态和流程。5.3 示例三实现人机交互Human-in-the-loop在很多企业流程中需要人工审核或干预。LangGraph 可以很容易地将“人工确认”作为一个节点加入工作流。# human_in_loop.py from typing import TypedDict, Annotated, Literal from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI import os os.environ[OPENAI_API_KEY] your-api-key-here # 1. 定义状态 class ApprovalState(TypedDict): messages: Annotated[list, add] needs_approval: bool approved: bool draft_content: str final_content: str # 2. 定义节点函数 def generate_draft(state: ApprovalState): 生成内容草稿 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0.7) prompt f根据以下要求起草一份内容{state[messages][-1][content]} draft llm.invoke(prompt).content # 假设生成了草稿并标记为需要审核 return {draft_content: draft, needs_approval: True, messages: [{role: assistant, content: f草稿已生成{draft[:50]}...}]} def human_approval(state: ApprovalState): 模拟人工审核节点实际中这里会连接UI print(f\n 需要人工审核 ) print(f草稿内容{state[draft_content]}) # 模拟人工输入。实际应用中这里会等待来自UI或接口的输入。 # 我们这里用代码模拟人工批准。 human_input approve # 可以改为 reject 测试不同分支 print(f模拟人工输入: {human_input}) is_approved (human_input.lower() approve) return {approved: is_approved, messages: [{role: user, content: f人工审核结果: {human_input}}]} def revise_draft(state: ApprovalState): 如果被拒绝修改草稿 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0.7) revision_prompt f原草稿被拒绝。请修改以下内容{state[draft_content]}。使其更简洁。 revised_draft llm.invoke(revision_prompt).content return {draft_content: revised_draft, needs_approval: True, messages: [{role: assistant, content: f已修改草稿{revised_draft[:50]}...}]} # 修改后再次需要审核 def publish_content(state: ApprovalState): 审核通过后发布内容 final state[draft_content] [已发布] return {final_content: final, messages: [{role: assistant, content: f内容已发布{final}}]} # 3. 定义条件边 def after_draft(state: ApprovalState) - Literal[needs_approval, publish]: 生成草稿后判断是否需要审核 if state.get(needs_approval): return needs_approval return publish # 如果不需要审核直接发布 def after_approval(state: ApprovalState) - Literal[revise, publish, END]: 人工审核后判断下一步 if state.get(approved): return publish else: return revise # 被拒绝去修改 # 4. 构建图 builder StateGraph(ApprovalState) builder.add_node(draft, generate_draft) builder.add_node(human_check, human_approval) builder.add_node(revise, revise_draft) builder.add_node(publish, publish_content) builder.set_entry_point(draft) # 草稿后如果需要审核 - human_check否则 - publish builder.add_conditional_edges( draft, after_draft, {needs_approval: human_check, publish: publish} ) # 人工审核后如果批准 - publish如果拒绝 - revise builder.add_conditional_edges( human_check, after_approval, {publish: publish, revise: revise} ) # 修改后应回到人工审核节点 builder.add_edge(revise, human_check) # 发布后结束 builder.add_edge(publish, END) graph builder.compile() # 5. 执行测试 print(开始内容生成与审核流程...) initial_state {messages: [{role: user, content: 写一篇关于LangGraph的简短介绍。}], needs_approval: False, approved: False, draft_content: , final_content: } result graph.invoke(initial_state) print(\n 流程结束 ) print(最终状态:, {k: v for k, v in result.items() if k ! messages}) print(最终内容:, result.get(final_content, N/A))运行与验证替换 API Key。python human_in_loop.py预期输出开始内容生成与审核流程... 草稿已生成LangGraph 是一个用于构建复杂、有状态、多智能体应用程序的库... 需要人工审核 草稿内容LangGraph 是一个用于构建复杂、有状态、多智能体应用程序的库... 模拟人工输入: approve 内容已发布LangGraph 是一个用于构建复杂、有状态、多智能体应用程序的库... [已发布] 流程结束 最终状态: {needs_approval: True, approved: True, draft_content: ..., final_content: ... [已发布]} 最终内容: ... [已发布]你可以修改human_approval函数中的human_input reject来测试被拒绝后重新修改并再次审核的流程。成功标准工作流按照“起草 - 审核 - (批准/拒绝) - 发布/修改”的路径正确执行。条件边add_conditional_edges根据状态needs_approval,approved正确路由。模拟的人工干预节点成功集成到自动化流程中。这个例子展示了如何将不可预测的人工操作纳入到确定的自动化流程中这是实现复杂业务逻辑的关键。6. 接口 API 与批量任务6.1 封装为 API 服务将编译好的graph对象集成到 Web 框架中非常简单。以下是一个使用 FastAPI 的示例# api_server.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from .tool_calling_agent import graph # 导入之前编译好的图 import asyncio from contextlib import asynccontextmanager # 定义请求体 class GraphRequest(BaseModel): messages: list # 可以根据需要添加其他初始状态字段 asynccontextmanager async def lifespan(app: FastAPI): # 启动时可以加载模型等资源 yield # 关闭时清理资源 app FastAPI(lifespanlifespan) app.post(/invoke/) async def invoke_graph(request: GraphRequest): try: # 调用图传入初始状态 initial_state {messages: request.messages} # 注意graph.invoke 是同步的在异步环境中需在线程池中运行 loop asyncio.get_event_loop() result await loop.run_in_executor(None, graph.invoke, initial_state) return result except Exception as e: raise HTTPException(status_code500, detailstr(e)) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)启动服务后就可以通过POST /invoke/接口来调用智能体工作流。6.2 处理批量任务对于批量处理可以利用 Python 的并发库。但要注意 LangGraph 状态对象可能不是线程安全的通常建议为每个任务创建新的图实例或使用异步。# batch_processing.py import asyncio from .tool_calling_agent import builder # 导入builder为每个任务编译新图 async def process_one_task(user_input: str): 处理单个任务 graph builder.compile() # 为每个任务编译一个图实例轻量级操作 initial_state {messages: [{role: user, content: user_input}]} result graph.invoke(initial_state) return result[messages][-1].content async def process_batch(inputs: list): 并发处理一批任务 tasks [process_one_task(inp) for inp in inputs] results await asyncio.gather(*tasks, return_exceptionsTrue) # 处理结果和异常 for inp, res in zip(inputs, results): if isinstance(res, Exception): print(f输入 {inp} 处理失败: {res}) else: print(f输入 {inp} 的结果: {res[:100]}...) # 使用示例 if __name__ __main__: input_list [ 计算 2 的 10 次方, 搜索 Python 的最新版本, 今天的天气怎么样 ] asyncio.run(process_batch(input_list))7. 资源占用与性能观察LangGraph 框架本身的内存和 CPU 开销极低性能瓶颈主要在于LLM API 调用网络延迟和 Tokens 消耗成本是主要因素。使用流式响应streaming可以改善用户体验。工具执行时间如果你集成了运行缓慢的工具如复杂数据库查询、爬虫会成为瓶颈。状态大小如果State对象非常庞大例如存储了很长的对话历史在节点间序列化/反序列化传递时会增加开销。优化建议状态精简只在 State 中存储必要信息。对于冗长的历史考虑使用摘要或外部存储如数据库在 State 中只保留引用 ID。异步工具尽可能将工具函数定义为异步async并在图中使用异步节点以提高 I/O 密集型任务的并发能力。缓存对昂贵的 LLM 调用或工具查询结果实施缓存策略。超时与重试为 LLM 调用和工具调用设置合理的超时和重试机制增强鲁棒性。8. 常见问题与排查方法问题现象可能原因排查方式解决方案ImportError或ModuleNotFoundError依赖未安装或虚拟环境未激活。检查 pip listgrep langgraph确认包已安装。图编译失败状态类TypedDict定义错误节点函数签名与状态不匹配。仔细检查AgentState的字段类型和Annotated注解。检查节点函数是否接受state参数并返回字典。参考官方示例修正状态和节点函数定义。确保返回的字典键是状态的子集。执行时状态更新不符合预期状态字段的缩减reduce方式如add设置错误。打印每个节点执行前后的完整状态。理解add,replace等缩减操作的含义。对于列表add是追加对于其他类型可能需要replace。LLM 不调用工具1. 工具绑定不正确。2. LLM 模型不支持工具调用如gpt-3.5-turbo-instruct。3. Prompt 未引导模型使用工具。1. 检查llm.bind_tools(tools)。2. 确认使用gpt-3.5-turbo或gpt-4等 Chat 模型。3. 检查传递给模型的消息历史。1. 确保工具列表正确传入。2. 更换为支持工具调用的模型。3. 在系统提示词中明确要求模型使用工具。条件边add_conditional_edges路由错误条件函数返回的值与边映射的键不匹配。打印条件函数的输入state和输出值。确保条件函数返回的字符串如tools与add_conditional_edges中映射的键完全一致。图陷入无限循环节点间的边形成了环但没有终止条件。使用graph.get_graph().draw_mermaid()输出图结构可视化检查循环。在循环路径上添加条件判断确保在满足某个条件时能跳出循环走向END。API 服务调用超时工作流执行时间过长超过 HTTP 超时设置。在本地先测试工作流单次执行时间。1. 优化工作流性能。2. 在 Web 框架中增加超时时间。3. 改为异步接口并返回任务 ID通过轮询获取结果。9. 最佳实践与使用建议从简单开始先用一个节点、两个节点构建最小可行图确保状态流转正确再逐步增加复杂度。可视化你的图在开发调试时强烈建议使用graph.get_graph().draw_mermaid()生成 Mermaid 图表直观理解工作流。from IPython.display import Image, display # 在 Jupyter 中显示 display(Image(graph.get_graph().draw_mermaid_png()))状态设计要精简State 是工作流的“内存”只放必要数据。考虑将大块数据如长文档、图片存储在外部State 中只存 ID 或 URL。为节点函数编写单元测试每个节点函数应该是纯的、可测试的函数。单独测试它们确保输入特定的 State 能产生正确的输出 State。错误处理与持久化在生产环境中需要考虑工作流执行失败的情况。LangGraph 支持检查点Checkpoint可以实现状态的持久化和从失败点恢复。版本控制工作流图的定义代码应该进行版本控制。当业务逻辑变更时可以通过切换代码版本来切换 AI 行为。监控与日志在关键节点添加详细的日志记录 State 的变化、工具调用的输入输出这对于调试和审计至关重要。10. 总结与下一步LangGraph 为后端开发者提供了一个极其顺滑的路径来构建复杂的、生产级的 AI 应用。它把看似神秘的 Agent 设计转化为了熟悉的状态机和图论问题。最值得尝试的点用 LangGraph 将你手头的一个需要多步骤、有条件判断的脚本或业务流程重构一下。例如一个需要先查询数据库、再调用 API、最后根据结果发送通知的自动化任务。你会立刻感受到它带来的结构清晰度和可维护性优势。最先应该验证的功能状态管理和条件边。这是 LangGraph 区别于简单 Chain 的核心。确保你理解State的定义、节点函数如何更新它以及add_conditional_edges如何根据状态路由。最容易踩的坑状态更新混淆不理解Annotated中add和replace的区别导致列表不是追加而是被覆盖。循环缺失终止条件设计了一个循环图如 Agent - Tools - Agent但 LLM 永远返回需要调用工具导致死循环。务必设置最大迭代次数或超时。工具绑定遗漏忘记调用llm.bind_tools(tools)导致 LLM 不知道有哪些工具可用。后续扩展方向探索LangGraph Studio这是一个可视化编辑和调试 LangGraph 应用的 UI 工具能极大提升开发体验。集成向量数据库将检索增强生成RAG流程融入图中构建知识问答系统。实现多智能体定义多个不同的 Agent 节点让它们通过共享状态或消息进行协作。深入研究持久化学习使用Checkpointer将工作流状态保存到数据库实现长时记忆和断点续跑。建议将本文中的三个示例代码保存下来作为你开发 LangGraph 应用的脚手架。当你需要设计新的 AI 工作流时可以快速在此基础上修改和扩展。