这次我们来看一个对后端开发者特别友好的 AI 开发框架LangGraph。如果你已经熟悉了 Spring Boot、微服务、状态管理和 API 设计现在想高效地切入 AI 应用开发特别是构建能自主决策、调用工具、与人交互的智能体Agent那么 LangGraph 可能是你当前的最优路线。它不是一个玩具而是一个用于构建有状态、多步骤工作流的框架核心解决了 AI 应用开发中的“状态管理”和“工具编排”难题。你可以把它理解为 AI 领域的“工作流引擎”或“状态机”让大模型的能力能够被结构化和工程化地调用。本文将带你从后端开发的视角快速掌握 LangGraph 的核心概念并通过实战演示如何用它开发一个具备工具调用和人机交互能力的企业级多智能体系统。1. 核心能力速览对于后端开发者最关心的是技术栈的成熟度、学习成本和工程化能力。下表概括了 LangGraph 的核心特性能力项说明与后端映射技术栈基于 Python深度集成 LangChain。对于 Java 后端可通过 Spring AI 或 HTTP 服务进行桥接。核心模型状态机StateGraph。开发者定义状态State和节点Node框架负责调度执行非常类似 BPMN 或 Activiti 的工作流思想。关键特性1. 显式状态管理所有中间数据、历史记录都保存在一个可序列化的 State 对象中告别 Prompt 中混乱的上下文拼接。2. 工具调用编排将大模型的函数调用Function Calling能力封装成可复用的工具Tool并在图中按条件或循环调用。3. 循环与条件分支支持while循环和if/else分支能构建复杂的、多轮交互的 Agent。4. 多智能体协作可定义多个具备不同能力的 Agent 节点让它们通过共享状态协同完成任务。硬件门槛无特殊要求。LangGraph 是框架层计算负载取决于你接入的大模型如 OpenAI GPT、本地部署的 Llama 等。本地测试时CPU 即可运行框架部分。启动与部署作为 Python 库安装在代码中实例化和运行。可以轻松封装成 RESTful API 服务如使用 FastAPI集成到现有后端系统中。是否支持 API是。框架本身提供 Python API 来构建和运行图。最佳实践是将整个图封装为一个服务端点。是否支持批量/异步任务是。可以并发处理多个独立的图执行实例非常适合任务队列场景。状态对象便于持久化实现断点续跑。适合场景1. 复杂对话机器人客服、导购2. 自动化流程数据分析、报告生成3. 多智能体系统模拟会议、游戏NPC4. 需要严格状态追踪和审计的 AI 应用。2. 适用场景与使用边界LangGraph 适合谁后端开发者熟悉系统设计、状态机和 API 开发希望快速将 AI 能力工程化。全栈工程师需要构建前后端一体化的 AI 应用原型或产品。AI 应用架构师设计需要复杂逻辑编排、多模型协作的解决方案。它能解决什么问题上下文管理混乱传统链式调用中对话历史、中间结果需要手动拼接并塞入 Prompt容易丢失或混乱。LangGraph 的状态对象是唯一的“真相来源”。流程控制薄弱简单的if-else和循环在纯链式结构中难以优雅实现。LangGraph 提供了图结构来描述复杂逻辑。工具调用不直观将多个工具调用、模型决策组合成一个连贯的工作流代码难以维护。LangGraph 用“节点”和“边”清晰地定义了执行路径。多角色协作困难实现多个 AI 智能体分工协作如一个分析、一个写作、一个审核需要复杂的协调逻辑。LangGraph 的多节点图天然支持此模式。它不适合什么场景极其简单的单次问答如果只是调用一次大模型 API 并返回结果使用 LangChain 的简单链或直接调用 SDK 更轻量。对延迟极其敏感的实时场景图的调度本身有开销对于毫秒级响应的场景需要精心优化节点逻辑和模型选择。完全无状态的批处理如果任务间完全独立无需共享状态或顺序执行传统的批处理脚本可能更直接。合规与边界提醒模型责任LangGraph 是编排框架生成内容的质量、安全性和合规性取决于你所接入的大模型如 OpenAI、Claude 或开源模型。你必须确保模型的使用符合其服务条款。工具调用安全当 Agent 能够调用外部工具如执行代码、访问数据库、发送邮件时必须实施严格的权限控制和输入验证防止越权操作。用户数据状态对象中可能包含用户输入和会话历史需遵循数据隐私法规做好加密存储和访问控制。3. 环境准备与前置条件作为后端开发你的环境很可能已经就绪。以下是快速检查清单Python 环境LangGraph 需要 Python 3.8 或更高版本。建议使用conda或venv创建独立的虚拟环境。# 检查Python版本 python --version # 创建虚拟环境 python -m venv langgraph-env # 激活环境 (Windows) langgraph-env\Scripts\activate # 激活环境 (Mac/Linux) source langgraph-env/bin/activate包管理工具使用pip进行安装。大模型访问权限你需要一个能够调用的大模型服务。本文示例将使用OpenAI API因为它最通用。你需要一个 OpenAI 账号。有效的 API Key。确保你的网络环境能够访问 OpenAI 服务注意必须通过合法合规的渠道使用国际互联网服务。代码编辑器任何你熟悉的 IDEVS Code, PyCharm 等即可。可选LangChain 基础了解 LangChain 的基本概念如 LLM、Prompt、Chain会有所帮助但不是必须的因为 LangGraph 的概念更上层。4. 安装部署与启动方式安装过程非常简单本质上就是安装两个 Python 包。# 在激活的虚拟环境中执行 pip install langgraph langchain-openailanggraph: 核心框架。langchain-openai: LangChain 提供的 OpenAI 集成包方便我们调用 GPT 模型。验证安装import langgraph print(langgraph.__version__) # 应能正常输出版本号LangGraph 没有“服务启动”的概念。它的启动方式是在你的 Python 代码中定义图State, Nodes, Edges。编译图。传入初始状态运行图。运行结果就是一个包含最终状态和输出的对象。接下来我们通过构建一个智能体来实际感受这个过程。5. 功能测试与效果验证构建第一个智能体我们将构建一个具备“思考-行动-观察”循环的 ReAct 智能体。它能根据用户问题决定是直接回答还是需要调用一个工具比如计算器来获取信息。5.1 定义状态State状态是所有节点共享的数据结构。我们定义一个字典类来承载状态。from typing import TypedDict, Annotated, List import operator # 定义状态结构 class AgentState(TypedDict): # 用户输入的问题 input: str # 所有节点的输出将汇总到这里 context: Annotated[List[str], operator.add] # 大模型生成的下一步动作如“FinalAnswer”或“Calculator” next: strAnnotated[List[str], operator.add]这是一个 LangGraph 的语法糖表示context字段是一个列表每个节点的输出会通过operator.add即列表的extend操作自动追加到这个列表中。这实现了状态的自动聚合。5.2 定义工具Tools工具是智能体可以调用的函数。我们先定义一个简单的计算器工具。from langchain.tools import tool tool def calculator(expression: str) - str: 计算一个数学表达式的值。支持 , -, *, / 和括号。 # 警告实际生产中eval 有安全风险此处仅用于演示。 # 应使用更安全的表达式求值库如 ast.literal_eval 或自定义解析器。 try: result eval(expression) return f计算结果: {expression} {result} except Exception as e: return f计算错误: {e}5.3 创建大模型和工具绑定我们需要一个大模型来驱动决策。这里使用 OpenAI 的 GPT-3.5-turbo。from langchain_openai import ChatOpenAI from langchain.agents import create_react_agent from langchain.agents.format_scratchpad import format_log_to_str from langchain.agents.output_parsers import ReActSingleInputOutputParser from langchain.tools.render import render_text_description from langchain.prompts import PromptTemplate import os # 设置你的 OpenAI API Key (请从环境变量读取不要硬编码) os.environ[OPENAI_API_KEY] 你的-api-key-here # 初始化大模型 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 将工具列表准备好 tools [calculator] tool_names [tool.name for tool in tools] tool_map {tool.name: tool for tool in tools} # 构建 ReAct 智能体的 Prompt prompt_template 你是一个有帮助的助手可以调用工具。 你可以调用的工具有 {tools} 请严格按照以下格式回答 问题用户输入的问题 思考你需要分析是否需要使用工具以及使用哪个工具 行动要调用的工具名称必须是[{tool_names}]中的一个如果没有工具可用就写“FinalAnswer” 行动输入调用工具的输入 观察工具返回的结果或你的最终答案 ... (这个 思考/行动/行动输入/观察 的循环可以重复多次) 当你有足够的信息给出最终答案时你的行动必须是“FinalAnswer”。 开始 问题{input} {agent_scratchpad} prompt PromptTemplate.from_template(prompt_template).partial( toolsrender_text_description(tools), tool_names, .join(tool_names) )5.4 定义图节点Nodes节点是图的基本执行单元。我们需要定义几个关键节点。from langgraph.graph import StateGraph, END # 初始化图构建器 graph_builder StateGraph(AgentState) # 1. 调用模型节点决定下一步做什么 def call_model(state: AgentState): 调用大模型决定下一步是调用工具还是直接回答。 # 构建当前上下文 intermediate_steps state.get(context, []) scratchpad format_log_to_str(intermediate_steps) if intermediate_steps else # 调用大模型 model_input { input: state[input], agent_scratchpad: scratchpad } response llm.invoke(prompt.format(**model_input)) # 解析响应获取下一步动作和输入 parsed_response ReActSingleInputOutputParser().parse(response.content) action parsed_response.tool action_input parsed_response.tool_input # 更新状态 return { context: [f思考: {parsed_response.log}], next: action, action_input: action_input } # 2. 执行工具节点根据模型决策调用工具 def execute_tool(state: AgentState): 执行模型选择的工具。 action state[next] action_input state.get(action_input, ) if action FinalAnswer: # 如果是最终答案直接将其作为观察 result action_input else: # 否则调用对应的工具 tool_to_use tool_map[action] result tool_to_use.invoke(action_input) # 更新上下文 return {context: [f行动: {action}\n行动输入: {action_input}\n观察: {result}]} # 将节点添加到图中 graph_builder.add_node(model, call_model) graph_builder.add_node(action, execute_tool)5.5 定义边Edges与条件路由边决定了节点间的执行流向。我们需要设置一个循环模型决策 - 执行动作 - 返回模型进行下一轮决策直到模型决定结束。# 设置入口点 graph_builder.set_entry_point(model) # 定义从“model”节点出发的路由逻辑 def route_after_model(state: AgentState): 根据模型输出的‘next’字段决定下一步去‘action’节点还是结束。 next_node state[next] if next_node FinalAnswer: # 如果是最终答案结束图执行 return END else: # 否则去执行工具 return action # 添加从“model”出发的条件边 graph_builder.add_conditional_edges( model, route_after_model ) # 添加从“action”到“model”的固定边执行完工具后继续让模型思考 graph_builder.add_edge(action, model)5.6 编译并运行图现在图已经定义完成我们可以编译它并运行了。# 编译图 graph graph_builder.compile() # 运行图处理一个需要计算的问题 initial_state {input: 请问 (12 34) * 2 等于多少, context: []} result graph.invoke(initial_state) print(最终状态中的上下文记录) for step in result[context]: print(- * 40) print(step) # 从上下文中提取最终答案通常是最后一个“观察” final_observation [s for s in result[context] if s.startswith(观察:)] if final_observation: print(\n智能体最终答案, final_observation[-1].replace(观察: , ))预期输出与效果验证 运行上述代码你应该能看到类似以下的输出展示了智能体的完整“思考-行动-观察”过程---------------------------------------- 思考: 用户问了一个数学计算问题我需要计算 (12 34) * 2 的值。我有一个计算器工具可以使用。我应该调用计算器工具。 行动: calculator 行动输入: (12 34) * 2 观察: 计算结果: (12 34) * 2 92 ---------------------------------------- 思考: 我已经通过计算器得到了结果 92。现在我可以给出最终答案了。 行动: FinalAnswer 行动输入: (12 34) * 2 等于 92。 智能体最终答案 (12 34) * 2 等于 92。判断成功的标准图成功编译并运行没有抛出异常。智能体正确识别出需要计算并调用了calculator工具。工具返回了正确结果92。智能体在获得结果后决定以FinalAnswer结束并输出了包含答案的文本。常见失败原因API Key 错误OPENAI_API_KEY未设置或无效导致llm.invoke调用失败。网络问题无法连接到 OpenAI 服务。工具解析错误模型输出的格式不符合ReActSingleInputOutputParser的预期。可以打印response.content检查原始输出。状态字段不匹配AgentState中定义的字段与节点中读取或写入的字段名不一致。6. 接口 API 与批量任务封装对于后端系统我们通常需要将 LangGraph 智能体暴露为 API 服务并支持批量异步处理。6.1 使用 FastAPI 封装为 RESTful 服务以下是一个简单的 FastAPI 应用示例将上述智能体图封装成一个 HTTP 端点。# app.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Any, Dict import asyncio from your_graph_module import graph # 假设你的图定义在另一个模块中 app FastAPI(titleLangGraph Agent API) class AgentRequest(BaseModel): input: str # 可以添加更多控制参数如 session_id, config 等 class AgentResponse(BaseModel): success: bool output: str steps: list error: str None app.post(/v1/agent/run, response_modelAgentResponse) async def run_agent(request: AgentRequest): 运行智能体图处理单个请求。 try: # 准备初始状态 initial_state {input: request.input, context: []} # 同步调用图如果在异步环境可考虑使用线程池 result graph.invoke(initial_state) # 提取最终答案 final_output for step in reversed(result[context]): if step.startswith(观察:): # 取最后一个“观察”作为最终输出 final_output step.replace(观察: , ) # 简单处理如果是工具调用的观察可能不是最终答案。 # 更健壮的做法是解析“FinalAnswer”对应的行动输入。 if FinalAnswer in result[context][-2]: # 检查前一步是否是FinalAnswer break response AgentResponse( successTrue, outputfinal_output, stepsresult[context] ) return response except Exception as e: raise HTTPException(status_code500, detailstr(e)) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)启动服务python app.py访问http://localhost:8000/docs可以看到自动生成的 API 文档并测试/v1/agent/run接口。6.2 批量任务处理对于批量任务核心是管理多个图实例的并发执行和状态持久化。# batch_processor.py import asyncio from concurrent.futures import ThreadPoolExecutor from your_graph_module import graph import json import os class BatchProcessor: def __init__(self, max_workers5): self.executor ThreadPoolExecutor(max_workersmax_workers) self.output_dir ./batch_outputs os.makedirs(self.output_dir, exist_okTrue) def process_single(self, task_id: str, user_input: str): 处理单个任务并保存结果到文件。 try: initial_state {input: user_input, context: []} result graph.invoke(initial_state) # 提取最终答案简化逻辑实际需根据业务调整 final_answer 未找到明确答案 for step in reversed(result[context]): if step.startswith(观察:) and FinalAnswer in step: final_answer step.replace(观察: , ) break output_data { task_id: task_id, input: user_input, output: final_answer, steps: result[context], status: success } # 保存结果 output_path os.path.join(self.output_dir, f{task_id}.json) with open(output_path, w, encodingutf-8) as f: json.dump(output_data, f, ensure_asciiFalse, indent2) return True, output_data except Exception as e: error_data { task_id: task_id, input: user_input, error: str(e), status: failed } output_path os.path.join(self.output_dir, f{task_id}_error.json) with open(output_path, w, encodingutf-8) as f: json.dump(error_data, f, ensure_asciiFalse, indent2) return False, error_data def process_batch(self, tasks: list): 并发处理一批任务。tasks格式: [{id:001, input:问题1}, ...] futures [] for task in tasks: future self.executor.submit(self.process_single, task[id], task[input]) futures.append((task[id], future)) results [] for task_id, future in futures: success, data future.result() # 这里会阻塞等待所有任务完成 results.append(data) print(f任务 {task_id} 处理完成状态: {data[status]}) return results # 使用示例 if __name__ __main__: processor BatchProcessor(max_workers3) batch_tasks [ {id: task_001, input: 计算 100 的平方根}, {id: task_002, input: 今天天气怎么样}, {id: task_003, input: 请写一首关于春天的诗}, ] all_results processor.process_batch(batch_tasks) print(f批量处理完成共 {len(all_results)} 个任务。)关键点并发控制使用ThreadPoolExecutor控制并发度避免同时发起过多 API 请求导致速率限制。状态隔离每个任务使用独立的initial_state保证任务间不干扰。结果持久化将每个任务的结果包括完整步骤保存为 JSON 文件便于审计和调试。错误处理单个任务失败不应影响整体批次错误信息单独保存。7. 资源占用与性能观察LangGraph 框架本身非常轻量资源消耗主要来自两方面大模型调用这是性能瓶颈和成本的主要来源。OpenAI API 调用有网络延迟和 Token 成本。本地部署的模型则消耗 GPU/CPU 和内存。Python 运行时内存图的状态对象和中间数据会驻留在内存中。对于超长对话或复杂状态需注意内存增长。性能观察与调优建议监控模型调用延迟在调用llm.invoke前后记录时间统计平均响应时间。控制上下文长度state[“context”]列表会不断增长。对于超长对话需要实现“摘要”或“滑动窗口”机制将过长的历史压缩后再放入 Prompt而不是无限制追加。优化工具调用工具的执行可能是 I/O 密集型如网络请求或计算密集型。考虑对工具调用进行超时设置、重试和缓存。图的编译开销graph_builder.compile()有一定开销但只需在服务启动时执行一次。编译后的图对象可重复使用。异步支持LangGraph 支持异步节点。如果工具或模型调用是异步的如async/await可以显著提升并发吞吐量。使用langgraph.graph.StateGraph的add_node时传入异步函数即可。8. 常见问题与排查方法问题现象可能原因排查方式解决方案导入langgraph失败未安装或虚拟环境未激活Python 版本过低。检查pip list | grep langgraph检查 Python 版本。在正确的虚拟环境中执行pip install langgraph升级 Python 至 3.8。调用llm.invoke时报错API Key 未设置或错误网络不通模型名称错误。检查os.environ[“OPENAI_API_KEY”]用curl测试 API 连通性。设置正确的环境变量检查网络代理设置确认模型名可用。图运行后state[‘next’]始终为空或错误Prompt 设计问题导致模型输出格式不符合解析器预期。打印response.content查看大模型的原始输出。调整 Prompt使模型输出严格遵循思考/行动/行动输入/观察格式或自定义输出解析器。工具未被调用直接跳转到 END条件路由函数route_after_model逻辑有误或模型输出的next字段值不是工具名。在route_after_model函数中打印state[‘next’]的值。确保模型在需要时输出正确的工具名检查工具名列表tool_names是否与模型输出匹配。状态字段更新不符合预期Annotated注解使用错误或节点返回的字典键与状态定义不匹配。在每个节点函数中打印输入和返回的状态。仔细核对AgentState的字段定义和每个节点返回值中的键名。Annotated用于列表自动追加普通字段直接赋值。批量处理时 API 被限速并发请求数过高触发 OpenAI 的速率限制。观察错误信息中是否包含rate limit。降低ThreadPoolExecutor的max_workers在请求间添加随机延迟使用指数退避重试。服务化后性能低下同步阻塞调用导致。使用异步框架如 FastAPI并配合 LangGraph 的异步节点。将节点函数定义为async def并在其中使用await调用异步的 LLM 或工具客户端。9. 最佳实践与使用建议从简单开始逐步复杂先构建一个只有 2-3 个节点的简单图并跑通再逐步添加分支、循环和多智能体。状态设计要精简状态对象应只包含必要的数据。避免将整个对话历史原文都塞进去考虑使用摘要或向量化存储。工具调用要安全尤其是执行代码、操作数据库、发送邮件的工具必须进行严格的输入验证和权限控制。切勿在生产环境中使用eval。实现持久化存储对于需要长时间运行或断点续跑的场景将state对象序列化如 JSON后存入数据库如 Redis、PostgreSQL。加入监控和日志记录每个节点的输入输出、执行时间便于调试和性能分析。LangGraph 提供了Checkpointer和Trace等内置机制。编写单元测试为每个节点函数编写单元测试模拟不同的输入状态确保其行为符合预期。图的整体流程也可以进行集成测试。版本化管理图配置图的结构节点、边也是一种配置。考虑将其定义为 JSON 或 YAML 文件便于版本控制和不同环境部署。明确人机交互边界在涉及多轮交互的 Agent 中清晰定义哪些步骤需要用户确认避免 Agent 在未获授权的情况下执行敏感操作。10. 总结与下一步对于后端开发者LangGraph 最大的价值在于它将 AI 应用的“逻辑”和“状态”从杂乱的 Prompt 工程中解放出来用熟悉的“图”和“状态机”的概念进行建模。你不再需要费力地管理越来越长的对话历史字符串而是通过定义清晰的状态结构和节点逻辑来控制流程。最值得尝试的点显式的状态管理所有数据流动一目了然。可视化的调试LangGraph 提供了可视化工具如graph.get_graph().draw_mermaid()可以生成流程图直观看到执行路径。强大的编排能力轻松实现 if-else、循环、多智能体协作这是构建复杂 AI 应用的基石。最先应该验证的功能 按照本文的步骤成功运行那个具备“思考-行动”循环的 ReAct 智能体。这是理解 LangGraph 工作流的基石。最容易踩的坑状态字段不匹配定义、读取、写入的字段名务必一致。模型输出格式Prompt 必须让模型输出能被OutputParser正确解析的格式。工具调用安全永远不要信任未经净化的模型输出直接作为工具参数。后续扩展方向集成更多工具为你的智能体接入搜索引擎、数据库、内部 API 等扩展其能力边界。实现多智能体创建多个具有不同专业能力的 Agent 节点让它们通过共享状态协作完成复杂任务如一个分析数据一个撰写报告一个审核质量。探索 LangGraph Studio使用 LangGraph 官方提供的 Web UI 来交互式地构建、调试和监控你的图。与现有后端集成将编译好的图对象封装成微服务通过 gRPC 或消息队列与你的 Java/Go 后端系统通信。替换底层模型尝试接入 Claude、Gemini 或本地部署的 Llama、Qwen 等开源模型比较效果和成本。掌握 LangGraph意味着你掌握了将大语言模型转化为可靠、可维护、可扩展的生产级应用的关键工具。建议将本文的示例代码作为起点结合你的业务场景进行改造和深化。
LangGraph：后端开发者构建AI智能体的状态管理与工作流编排框架
这次我们来看一个对后端开发者特别友好的 AI 开发框架LangGraph。如果你已经熟悉了 Spring Boot、微服务、状态管理和 API 设计现在想高效地切入 AI 应用开发特别是构建能自主决策、调用工具、与人交互的智能体Agent那么 LangGraph 可能是你当前的最优路线。它不是一个玩具而是一个用于构建有状态、多步骤工作流的框架核心解决了 AI 应用开发中的“状态管理”和“工具编排”难题。你可以把它理解为 AI 领域的“工作流引擎”或“状态机”让大模型的能力能够被结构化和工程化地调用。本文将带你从后端开发的视角快速掌握 LangGraph 的核心概念并通过实战演示如何用它开发一个具备工具调用和人机交互能力的企业级多智能体系统。1. 核心能力速览对于后端开发者最关心的是技术栈的成熟度、学习成本和工程化能力。下表概括了 LangGraph 的核心特性能力项说明与后端映射技术栈基于 Python深度集成 LangChain。对于 Java 后端可通过 Spring AI 或 HTTP 服务进行桥接。核心模型状态机StateGraph。开发者定义状态State和节点Node框架负责调度执行非常类似 BPMN 或 Activiti 的工作流思想。关键特性1. 显式状态管理所有中间数据、历史记录都保存在一个可序列化的 State 对象中告别 Prompt 中混乱的上下文拼接。2. 工具调用编排将大模型的函数调用Function Calling能力封装成可复用的工具Tool并在图中按条件或循环调用。3. 循环与条件分支支持while循环和if/else分支能构建复杂的、多轮交互的 Agent。4. 多智能体协作可定义多个具备不同能力的 Agent 节点让它们通过共享状态协同完成任务。硬件门槛无特殊要求。LangGraph 是框架层计算负载取决于你接入的大模型如 OpenAI GPT、本地部署的 Llama 等。本地测试时CPU 即可运行框架部分。启动与部署作为 Python 库安装在代码中实例化和运行。可以轻松封装成 RESTful API 服务如使用 FastAPI集成到现有后端系统中。是否支持 API是。框架本身提供 Python API 来构建和运行图。最佳实践是将整个图封装为一个服务端点。是否支持批量/异步任务是。可以并发处理多个独立的图执行实例非常适合任务队列场景。状态对象便于持久化实现断点续跑。适合场景1. 复杂对话机器人客服、导购2. 自动化流程数据分析、报告生成3. 多智能体系统模拟会议、游戏NPC4. 需要严格状态追踪和审计的 AI 应用。2. 适用场景与使用边界LangGraph 适合谁后端开发者熟悉系统设计、状态机和 API 开发希望快速将 AI 能力工程化。全栈工程师需要构建前后端一体化的 AI 应用原型或产品。AI 应用架构师设计需要复杂逻辑编排、多模型协作的解决方案。它能解决什么问题上下文管理混乱传统链式调用中对话历史、中间结果需要手动拼接并塞入 Prompt容易丢失或混乱。LangGraph 的状态对象是唯一的“真相来源”。流程控制薄弱简单的if-else和循环在纯链式结构中难以优雅实现。LangGraph 提供了图结构来描述复杂逻辑。工具调用不直观将多个工具调用、模型决策组合成一个连贯的工作流代码难以维护。LangGraph 用“节点”和“边”清晰地定义了执行路径。多角色协作困难实现多个 AI 智能体分工协作如一个分析、一个写作、一个审核需要复杂的协调逻辑。LangGraph 的多节点图天然支持此模式。它不适合什么场景极其简单的单次问答如果只是调用一次大模型 API 并返回结果使用 LangChain 的简单链或直接调用 SDK 更轻量。对延迟极其敏感的实时场景图的调度本身有开销对于毫秒级响应的场景需要精心优化节点逻辑和模型选择。完全无状态的批处理如果任务间完全独立无需共享状态或顺序执行传统的批处理脚本可能更直接。合规与边界提醒模型责任LangGraph 是编排框架生成内容的质量、安全性和合规性取决于你所接入的大模型如 OpenAI、Claude 或开源模型。你必须确保模型的使用符合其服务条款。工具调用安全当 Agent 能够调用外部工具如执行代码、访问数据库、发送邮件时必须实施严格的权限控制和输入验证防止越权操作。用户数据状态对象中可能包含用户输入和会话历史需遵循数据隐私法规做好加密存储和访问控制。3. 环境准备与前置条件作为后端开发你的环境很可能已经就绪。以下是快速检查清单Python 环境LangGraph 需要 Python 3.8 或更高版本。建议使用conda或venv创建独立的虚拟环境。# 检查Python版本 python --version # 创建虚拟环境 python -m venv langgraph-env # 激活环境 (Windows) langgraph-env\Scripts\activate # 激活环境 (Mac/Linux) source langgraph-env/bin/activate包管理工具使用pip进行安装。大模型访问权限你需要一个能够调用的大模型服务。本文示例将使用OpenAI API因为它最通用。你需要一个 OpenAI 账号。有效的 API Key。确保你的网络环境能够访问 OpenAI 服务注意必须通过合法合规的渠道使用国际互联网服务。代码编辑器任何你熟悉的 IDEVS Code, PyCharm 等即可。可选LangChain 基础了解 LangChain 的基本概念如 LLM、Prompt、Chain会有所帮助但不是必须的因为 LangGraph 的概念更上层。4. 安装部署与启动方式安装过程非常简单本质上就是安装两个 Python 包。# 在激活的虚拟环境中执行 pip install langgraph langchain-openailanggraph: 核心框架。langchain-openai: LangChain 提供的 OpenAI 集成包方便我们调用 GPT 模型。验证安装import langgraph print(langgraph.__version__) # 应能正常输出版本号LangGraph 没有“服务启动”的概念。它的启动方式是在你的 Python 代码中定义图State, Nodes, Edges。编译图。传入初始状态运行图。运行结果就是一个包含最终状态和输出的对象。接下来我们通过构建一个智能体来实际感受这个过程。5. 功能测试与效果验证构建第一个智能体我们将构建一个具备“思考-行动-观察”循环的 ReAct 智能体。它能根据用户问题决定是直接回答还是需要调用一个工具比如计算器来获取信息。5.1 定义状态State状态是所有节点共享的数据结构。我们定义一个字典类来承载状态。from typing import TypedDict, Annotated, List import operator # 定义状态结构 class AgentState(TypedDict): # 用户输入的问题 input: str # 所有节点的输出将汇总到这里 context: Annotated[List[str], operator.add] # 大模型生成的下一步动作如“FinalAnswer”或“Calculator” next: strAnnotated[List[str], operator.add]这是一个 LangGraph 的语法糖表示context字段是一个列表每个节点的输出会通过operator.add即列表的extend操作自动追加到这个列表中。这实现了状态的自动聚合。5.2 定义工具Tools工具是智能体可以调用的函数。我们先定义一个简单的计算器工具。from langchain.tools import tool tool def calculator(expression: str) - str: 计算一个数学表达式的值。支持 , -, *, / 和括号。 # 警告实际生产中eval 有安全风险此处仅用于演示。 # 应使用更安全的表达式求值库如 ast.literal_eval 或自定义解析器。 try: result eval(expression) return f计算结果: {expression} {result} except Exception as e: return f计算错误: {e}5.3 创建大模型和工具绑定我们需要一个大模型来驱动决策。这里使用 OpenAI 的 GPT-3.5-turbo。from langchain_openai import ChatOpenAI from langchain.agents import create_react_agent from langchain.agents.format_scratchpad import format_log_to_str from langchain.agents.output_parsers import ReActSingleInputOutputParser from langchain.tools.render import render_text_description from langchain.prompts import PromptTemplate import os # 设置你的 OpenAI API Key (请从环境变量读取不要硬编码) os.environ[OPENAI_API_KEY] 你的-api-key-here # 初始化大模型 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 将工具列表准备好 tools [calculator] tool_names [tool.name for tool in tools] tool_map {tool.name: tool for tool in tools} # 构建 ReAct 智能体的 Prompt prompt_template 你是一个有帮助的助手可以调用工具。 你可以调用的工具有 {tools} 请严格按照以下格式回答 问题用户输入的问题 思考你需要分析是否需要使用工具以及使用哪个工具 行动要调用的工具名称必须是[{tool_names}]中的一个如果没有工具可用就写“FinalAnswer” 行动输入调用工具的输入 观察工具返回的结果或你的最终答案 ... (这个 思考/行动/行动输入/观察 的循环可以重复多次) 当你有足够的信息给出最终答案时你的行动必须是“FinalAnswer”。 开始 问题{input} {agent_scratchpad} prompt PromptTemplate.from_template(prompt_template).partial( toolsrender_text_description(tools), tool_names, .join(tool_names) )5.4 定义图节点Nodes节点是图的基本执行单元。我们需要定义几个关键节点。from langgraph.graph import StateGraph, END # 初始化图构建器 graph_builder StateGraph(AgentState) # 1. 调用模型节点决定下一步做什么 def call_model(state: AgentState): 调用大模型决定下一步是调用工具还是直接回答。 # 构建当前上下文 intermediate_steps state.get(context, []) scratchpad format_log_to_str(intermediate_steps) if intermediate_steps else # 调用大模型 model_input { input: state[input], agent_scratchpad: scratchpad } response llm.invoke(prompt.format(**model_input)) # 解析响应获取下一步动作和输入 parsed_response ReActSingleInputOutputParser().parse(response.content) action parsed_response.tool action_input parsed_response.tool_input # 更新状态 return { context: [f思考: {parsed_response.log}], next: action, action_input: action_input } # 2. 执行工具节点根据模型决策调用工具 def execute_tool(state: AgentState): 执行模型选择的工具。 action state[next] action_input state.get(action_input, ) if action FinalAnswer: # 如果是最终答案直接将其作为观察 result action_input else: # 否则调用对应的工具 tool_to_use tool_map[action] result tool_to_use.invoke(action_input) # 更新上下文 return {context: [f行动: {action}\n行动输入: {action_input}\n观察: {result}]} # 将节点添加到图中 graph_builder.add_node(model, call_model) graph_builder.add_node(action, execute_tool)5.5 定义边Edges与条件路由边决定了节点间的执行流向。我们需要设置一个循环模型决策 - 执行动作 - 返回模型进行下一轮决策直到模型决定结束。# 设置入口点 graph_builder.set_entry_point(model) # 定义从“model”节点出发的路由逻辑 def route_after_model(state: AgentState): 根据模型输出的‘next’字段决定下一步去‘action’节点还是结束。 next_node state[next] if next_node FinalAnswer: # 如果是最终答案结束图执行 return END else: # 否则去执行工具 return action # 添加从“model”出发的条件边 graph_builder.add_conditional_edges( model, route_after_model ) # 添加从“action”到“model”的固定边执行完工具后继续让模型思考 graph_builder.add_edge(action, model)5.6 编译并运行图现在图已经定义完成我们可以编译它并运行了。# 编译图 graph graph_builder.compile() # 运行图处理一个需要计算的问题 initial_state {input: 请问 (12 34) * 2 等于多少, context: []} result graph.invoke(initial_state) print(最终状态中的上下文记录) for step in result[context]: print(- * 40) print(step) # 从上下文中提取最终答案通常是最后一个“观察” final_observation [s for s in result[context] if s.startswith(观察:)] if final_observation: print(\n智能体最终答案, final_observation[-1].replace(观察: , ))预期输出与效果验证 运行上述代码你应该能看到类似以下的输出展示了智能体的完整“思考-行动-观察”过程---------------------------------------- 思考: 用户问了一个数学计算问题我需要计算 (12 34) * 2 的值。我有一个计算器工具可以使用。我应该调用计算器工具。 行动: calculator 行动输入: (12 34) * 2 观察: 计算结果: (12 34) * 2 92 ---------------------------------------- 思考: 我已经通过计算器得到了结果 92。现在我可以给出最终答案了。 行动: FinalAnswer 行动输入: (12 34) * 2 等于 92。 智能体最终答案 (12 34) * 2 等于 92。判断成功的标准图成功编译并运行没有抛出异常。智能体正确识别出需要计算并调用了calculator工具。工具返回了正确结果92。智能体在获得结果后决定以FinalAnswer结束并输出了包含答案的文本。常见失败原因API Key 错误OPENAI_API_KEY未设置或无效导致llm.invoke调用失败。网络问题无法连接到 OpenAI 服务。工具解析错误模型输出的格式不符合ReActSingleInputOutputParser的预期。可以打印response.content检查原始输出。状态字段不匹配AgentState中定义的字段与节点中读取或写入的字段名不一致。6. 接口 API 与批量任务封装对于后端系统我们通常需要将 LangGraph 智能体暴露为 API 服务并支持批量异步处理。6.1 使用 FastAPI 封装为 RESTful 服务以下是一个简单的 FastAPI 应用示例将上述智能体图封装成一个 HTTP 端点。# app.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Any, Dict import asyncio from your_graph_module import graph # 假设你的图定义在另一个模块中 app FastAPI(titleLangGraph Agent API) class AgentRequest(BaseModel): input: str # 可以添加更多控制参数如 session_id, config 等 class AgentResponse(BaseModel): success: bool output: str steps: list error: str None app.post(/v1/agent/run, response_modelAgentResponse) async def run_agent(request: AgentRequest): 运行智能体图处理单个请求。 try: # 准备初始状态 initial_state {input: request.input, context: []} # 同步调用图如果在异步环境可考虑使用线程池 result graph.invoke(initial_state) # 提取最终答案 final_output for step in reversed(result[context]): if step.startswith(观察:): # 取最后一个“观察”作为最终输出 final_output step.replace(观察: , ) # 简单处理如果是工具调用的观察可能不是最终答案。 # 更健壮的做法是解析“FinalAnswer”对应的行动输入。 if FinalAnswer in result[context][-2]: # 检查前一步是否是FinalAnswer break response AgentResponse( successTrue, outputfinal_output, stepsresult[context] ) return response except Exception as e: raise HTTPException(status_code500, detailstr(e)) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)启动服务python app.py访问http://localhost:8000/docs可以看到自动生成的 API 文档并测试/v1/agent/run接口。6.2 批量任务处理对于批量任务核心是管理多个图实例的并发执行和状态持久化。# batch_processor.py import asyncio from concurrent.futures import ThreadPoolExecutor from your_graph_module import graph import json import os class BatchProcessor: def __init__(self, max_workers5): self.executor ThreadPoolExecutor(max_workersmax_workers) self.output_dir ./batch_outputs os.makedirs(self.output_dir, exist_okTrue) def process_single(self, task_id: str, user_input: str): 处理单个任务并保存结果到文件。 try: initial_state {input: user_input, context: []} result graph.invoke(initial_state) # 提取最终答案简化逻辑实际需根据业务调整 final_answer 未找到明确答案 for step in reversed(result[context]): if step.startswith(观察:) and FinalAnswer in step: final_answer step.replace(观察: , ) break output_data { task_id: task_id, input: user_input, output: final_answer, steps: result[context], status: success } # 保存结果 output_path os.path.join(self.output_dir, f{task_id}.json) with open(output_path, w, encodingutf-8) as f: json.dump(output_data, f, ensure_asciiFalse, indent2) return True, output_data except Exception as e: error_data { task_id: task_id, input: user_input, error: str(e), status: failed } output_path os.path.join(self.output_dir, f{task_id}_error.json) with open(output_path, w, encodingutf-8) as f: json.dump(error_data, f, ensure_asciiFalse, indent2) return False, error_data def process_batch(self, tasks: list): 并发处理一批任务。tasks格式: [{id:001, input:问题1}, ...] futures [] for task in tasks: future self.executor.submit(self.process_single, task[id], task[input]) futures.append((task[id], future)) results [] for task_id, future in futures: success, data future.result() # 这里会阻塞等待所有任务完成 results.append(data) print(f任务 {task_id} 处理完成状态: {data[status]}) return results # 使用示例 if __name__ __main__: processor BatchProcessor(max_workers3) batch_tasks [ {id: task_001, input: 计算 100 的平方根}, {id: task_002, input: 今天天气怎么样}, {id: task_003, input: 请写一首关于春天的诗}, ] all_results processor.process_batch(batch_tasks) print(f批量处理完成共 {len(all_results)} 个任务。)关键点并发控制使用ThreadPoolExecutor控制并发度避免同时发起过多 API 请求导致速率限制。状态隔离每个任务使用独立的initial_state保证任务间不干扰。结果持久化将每个任务的结果包括完整步骤保存为 JSON 文件便于审计和调试。错误处理单个任务失败不应影响整体批次错误信息单独保存。7. 资源占用与性能观察LangGraph 框架本身非常轻量资源消耗主要来自两方面大模型调用这是性能瓶颈和成本的主要来源。OpenAI API 调用有网络延迟和 Token 成本。本地部署的模型则消耗 GPU/CPU 和内存。Python 运行时内存图的状态对象和中间数据会驻留在内存中。对于超长对话或复杂状态需注意内存增长。性能观察与调优建议监控模型调用延迟在调用llm.invoke前后记录时间统计平均响应时间。控制上下文长度state[“context”]列表会不断增长。对于超长对话需要实现“摘要”或“滑动窗口”机制将过长的历史压缩后再放入 Prompt而不是无限制追加。优化工具调用工具的执行可能是 I/O 密集型如网络请求或计算密集型。考虑对工具调用进行超时设置、重试和缓存。图的编译开销graph_builder.compile()有一定开销但只需在服务启动时执行一次。编译后的图对象可重复使用。异步支持LangGraph 支持异步节点。如果工具或模型调用是异步的如async/await可以显著提升并发吞吐量。使用langgraph.graph.StateGraph的add_node时传入异步函数即可。8. 常见问题与排查方法问题现象可能原因排查方式解决方案导入langgraph失败未安装或虚拟环境未激活Python 版本过低。检查pip list | grep langgraph检查 Python 版本。在正确的虚拟环境中执行pip install langgraph升级 Python 至 3.8。调用llm.invoke时报错API Key 未设置或错误网络不通模型名称错误。检查os.environ[“OPENAI_API_KEY”]用curl测试 API 连通性。设置正确的环境变量检查网络代理设置确认模型名可用。图运行后state[‘next’]始终为空或错误Prompt 设计问题导致模型输出格式不符合解析器预期。打印response.content查看大模型的原始输出。调整 Prompt使模型输出严格遵循思考/行动/行动输入/观察格式或自定义输出解析器。工具未被调用直接跳转到 END条件路由函数route_after_model逻辑有误或模型输出的next字段值不是工具名。在route_after_model函数中打印state[‘next’]的值。确保模型在需要时输出正确的工具名检查工具名列表tool_names是否与模型输出匹配。状态字段更新不符合预期Annotated注解使用错误或节点返回的字典键与状态定义不匹配。在每个节点函数中打印输入和返回的状态。仔细核对AgentState的字段定义和每个节点返回值中的键名。Annotated用于列表自动追加普通字段直接赋值。批量处理时 API 被限速并发请求数过高触发 OpenAI 的速率限制。观察错误信息中是否包含rate limit。降低ThreadPoolExecutor的max_workers在请求间添加随机延迟使用指数退避重试。服务化后性能低下同步阻塞调用导致。使用异步框架如 FastAPI并配合 LangGraph 的异步节点。将节点函数定义为async def并在其中使用await调用异步的 LLM 或工具客户端。9. 最佳实践与使用建议从简单开始逐步复杂先构建一个只有 2-3 个节点的简单图并跑通再逐步添加分支、循环和多智能体。状态设计要精简状态对象应只包含必要的数据。避免将整个对话历史原文都塞进去考虑使用摘要或向量化存储。工具调用要安全尤其是执行代码、操作数据库、发送邮件的工具必须进行严格的输入验证和权限控制。切勿在生产环境中使用eval。实现持久化存储对于需要长时间运行或断点续跑的场景将state对象序列化如 JSON后存入数据库如 Redis、PostgreSQL。加入监控和日志记录每个节点的输入输出、执行时间便于调试和性能分析。LangGraph 提供了Checkpointer和Trace等内置机制。编写单元测试为每个节点函数编写单元测试模拟不同的输入状态确保其行为符合预期。图的整体流程也可以进行集成测试。版本化管理图配置图的结构节点、边也是一种配置。考虑将其定义为 JSON 或 YAML 文件便于版本控制和不同环境部署。明确人机交互边界在涉及多轮交互的 Agent 中清晰定义哪些步骤需要用户确认避免 Agent 在未获授权的情况下执行敏感操作。10. 总结与下一步对于后端开发者LangGraph 最大的价值在于它将 AI 应用的“逻辑”和“状态”从杂乱的 Prompt 工程中解放出来用熟悉的“图”和“状态机”的概念进行建模。你不再需要费力地管理越来越长的对话历史字符串而是通过定义清晰的状态结构和节点逻辑来控制流程。最值得尝试的点显式的状态管理所有数据流动一目了然。可视化的调试LangGraph 提供了可视化工具如graph.get_graph().draw_mermaid()可以生成流程图直观看到执行路径。强大的编排能力轻松实现 if-else、循环、多智能体协作这是构建复杂 AI 应用的基石。最先应该验证的功能 按照本文的步骤成功运行那个具备“思考-行动”循环的 ReAct 智能体。这是理解 LangGraph 工作流的基石。最容易踩的坑状态字段不匹配定义、读取、写入的字段名务必一致。模型输出格式Prompt 必须让模型输出能被OutputParser正确解析的格式。工具调用安全永远不要信任未经净化的模型输出直接作为工具参数。后续扩展方向集成更多工具为你的智能体接入搜索引擎、数据库、内部 API 等扩展其能力边界。实现多智能体创建多个具有不同专业能力的 Agent 节点让它们通过共享状态协作完成复杂任务如一个分析数据一个撰写报告一个审核质量。探索 LangGraph Studio使用 LangGraph 官方提供的 Web UI 来交互式地构建、调试和监控你的图。与现有后端集成将编译好的图对象封装成微服务通过 gRPC 或消息队列与你的 Java/Go 后端系统通信。替换底层模型尝试接入 Claude、Gemini 或本地部署的 Llama、Qwen 等开源模型比较效果和成本。掌握 LangGraph意味着你掌握了将大语言模型转化为可靠、可维护、可扩展的生产级应用的关键工具。建议将本文的示例代码作为起点结合你的业务场景进行改造和深化。