这次我们来看一个关于 LangGraph 智能体开发的实战教程。LangGraph 作为 LangChain 生态中用于构建复杂、有状态多步骤应用智能体的框架正成为连接大模型 API 与具体业务逻辑的关键桥梁。它的核心价值在于让你能用清晰的图Graph结构来定义智能体的决策流和工作流告别传统 prompt 工程中状态管理混乱的痛点。对于开发者而言最关心的问题通常是它能不能快速集成现有的大模型 API如 OpenAI、DeepSeek、智谱等代码结构是否清晰易上手能否处理需要记忆和工具调用的复杂任务以及它和 LangChain 到底是什么关系该用哪个这篇文章将直接切入这些核心问题通过架构剖析和代码实战带你快速掌握 LangGraph 构建智能体的核心方法。无论你是想开发一个自动化的数据分析助手还是一个能调用外部 API 的客服机器人这里的内容都能提供直接的参考。本文将围绕以下几个核心部分展开首先快速梳理 LangGraph 的核心概念与适用场景然后深入其架构中的关键组件State、Node、Edge接着通过一个完整的代码示例演示如何构建一个具备工具调用和记忆能力的智能体最后会讨论如何将其部署为 API 服务并分享开发中的常见问题与调试技巧。目标是让你看完后能立即动手搭建自己的第一个 LangGraph 智能体。1. 核心能力速览在深入细节之前我们先通过一个表格快速了解 LangGraph 的核心特性和能力边界这有助于判断它是否适合解决你手头的问题。能力项说明项目类型用于构建有状态、多步骤 AI 应用智能体的框架/库。核心抽象图Graph将应用逻辑定义为节点Node和边Edge组成的工作流。关键特性1.有状态State在整个工作流中持久化和传递数据。2.循环与条件分支支持基于状态的动态路由if-else, loops。3.并行与异步节点可以并行执行。4.与 LangChain 深度集成可直接使用 LangChain 的模型 I/O、工具Tools、记忆Memory等组件。主要功能构建复杂决策逻辑的智能体如客服机器人、数据分析流水线、自动化研究助手、游戏 NPC 等。硬件/环境门槛无特殊要求。作为 Python 库运行依赖主要在于所选的大模型 API如 OpenAI和 LangChain。可在普通开发机、服务器甚至容器中运行。启动/部署方式1.作为库集成在 Python 脚本中导入使用。2.作为服务可自行封装为 FastAPI/Flask API 服务或通过 LangServe 部署。是否支持 API框架本身提供编程接口Python API。构建的应用可以轻松暴露为 REST API。是否支持批量/异步任务支持。可以设计工作流处理批量输入并利用异步节点提升效率。适合场景1. 任务步骤超过3步且步骤间有状态依赖。2. 需要根据中间结果动态决定下一步行动工具调用、分支。3. 需要构建具备长期记忆或知识检索的智能体。4. 作为更灵活、可维护的替代方案替代冗长且难以维护的单一 Prompt 工程。2. 适用场景与使用边界LangGraph 并非所有 AI 应用的首选。理解其适用边界能帮你更高效地选择技术栈。它非常适合以下场景复杂多轮对话系统客服机器人需要根据历史对话、用户意图和知识库检索结果决定是回答问题、转人工还是询问更多信息。自动化工作流一个数据分析智能体需要依次执行“接收查询 - 规划分析步骤 - 查询数据库 - 执行计算 - 生成图表 - 撰写报告”。游戏与模拟NPC 需要根据环境状态、玩家交互和自身目标规划一系列动作移动、对话、使用物品。研究/写作助手智能体接收一个主题然后执行“搜索资料 - 总结要点 - 起草大纲 - 撰写章节 - 润色修改”的流程。它可能不是最佳选择或需要搭配其他技术的场景简单的单次问答QA如果只是输入问题模型直接输出答案使用 LangChain 的LLMChain或直接调用模型 API 更简单。单纯的文本生成/翻译没有复杂状态流转和工具调用用基础模型接口即可。对延迟极其敏感的实时应用图结构的调度会引入额外开销纯推理场景应追求最小化延迟。完全无状态的批处理任务如果任务只是对一批独立数据做相同处理用常规脚本或并行计算框架更直接。合规与安全边界提醒当使用 LangGraph 构建智能体时尤其是涉及以下能力必须特别注意工具调用智能体可以调用外部 API、操作数据库、执行代码。必须严格限制其权限避免未授权访问或破坏性操作。在生产环境中应对工具进行沙箱化处理或严格的输入校验。长期记忆如果智能体存储用户对话或数据需明确隐私政策遵守数据保护法规如 GDPR并提供数据清除机制。内容生成确保智能体生成的内容符合法律法规和平台政策避免产生侵权、虚假或有害信息。建议增加内容过滤层。模型选择所选大模型 API 的服务条款和内容政策同样适用于你的智能体。3. 环境准备与前置条件开始 LangGraph 实战之前需要准备好开发环境。以下是一个通用的环境清单你可以根据实际项目进行调整。基础环境要求操作系统Windows 10/11, macOS, 或 Linux (推荐 Ubuntu 20.04)。Python版本 3.8 或更高。建议使用 3.9 以获得最佳兼容性。包管理工具pip或conda。核心依赖安装LangGraph 通常与 LangChain 一起使用。我们将安装核心的 LangChain 包、LangGraph 以及一个用于与大模型交互的社区包。这里以使用 OpenAI API 为例。# 创建并激活一个虚拟环境强烈推荐 python -m venv langgraph-env # Windows: langgraph-env\Scripts\activate # macOS/Linux: source langgraph-env/bin/activate # 升级 pip pip install --upgrade pip # 安装核心依赖 pip install langgraph langchain langchain-openai # 可选安装用于构建API服务的框架 pip install fastapi uvicornAPI Key 配置你需要准备大模型服务的 API Key。这里以 OpenAI 为例你需要一个有效的 OpenAI API Key。将 Key 设置为环境变量是最安全、通用的做法。# Windows (PowerShell) $env:OPENAI_API_KEYyour-api-key-here # macOS/Linux export OPENAI_API_KEYyour-api-key-here也可以在代码中直接设置但不推荐将密钥硬编码在源码中。验证安装创建一个简单的 Python 脚本test_env.py来测试环境是否就绪。import os from langchain_openai import ChatOpenAI # 确保已设置环境变量 OPENAI_API_KEY llm ChatOpenAI(modelgpt-3.5-turbo) try: response llm.invoke(Hello, world!) print(环境验证成功) print(f模型回复: {response.content}) except Exception as e: print(f环境验证失败错误信息: {e}) print(请检查1. API Key 2. 网络连接 3. 依赖包是否安装正确)运行此脚本如果看到成功的回复说明基础环境已配置完成。4. LangGraph 核心架构与概念解析理解 LangGraph 的架构是高效使用它的关键。其核心思想是将智能体的执行流程建模为一个有向图。下面我们拆解其中的核心组件。4.1 State状态State 是贯穿整个图执行过程的“共享内存”。它定义了智能体工作流中需要传递和更新的所有数据。State 通常是一个 TypedDict 或 Pydantic 模型。from typing import TypedDict, Annotated from typing_extensions import TypedDict import operator class AgentState(TypedDict): # 用户输入的问题 input: str # 智能体生成的中间思考或最终答案 output: str # 对话历史记录 chat_history: list # 工具调用的结果列表 tool_results: Annotated[list, operator.add] # 这是一个“归约器”用于追加列表Annotated用于声明某个字段的“归约”方式。operator.add表示当多个节点修改tool_results时其结果会被追加append到列表中而不是覆盖。这是实现“记忆”或“收集结果”的关键机制。4.2 Node节点Node 是图中的一个执行单元是一个普通的 Python 函数或可调用对象。它接收当前的 State执行一些操作如调用 LLM、使用工具然后返回一个包含 State 更新内容的字典。def call_model(state: AgentState): 节点调用大模型生成思考或回答 from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-3.5-turbo) # 构建提示词可以包含历史、问题等来自 State 的信息 messages [ (system, 你是一个有帮助的助手。), (human, state[input]) ] # 调用模型 response llm.invoke(messages) # 返回要更新到 State 的内容 return {output: response.content} def use_tool(state: AgentState): 节点调用某个工具例如查询天气 # 假设我们有一个工具函数 get_weather tool_result get_weather(state[input]) # 返回结果tool_results 字段会因为归约器而追加这个结果 return {tool_results: [tool_result]}4.3 Edge边Edge 定义了节点之间的流转逻辑。它决定了一个节点执行完毕后下一个该执行哪个节点。边可以是固定的也可以是基于 State 内容的条件边。固定边graph.add_edge(“node_a”, “node_b”)表示node_a执行完后总是执行node_b。条件边通过graph.add_conditional_edges添加它需要一个路由函数根据 State 返回下一个节点的名称。def should_use_tool(state: AgentState) - str: 路由函数判断是否需要调用工具 output state.get(output, ) # 简单逻辑如果模型输出中包含“查询天气”关键词则路由到工具节点 if 天气 in output: return use_tool_node else: return end_node # 否则结束4.4 Graph图与 Compilation编译将节点和边组装起来就形成了一个StateGraph。最后需要调用graph.compile()来生成一个可执行的、优化过的CompiledGraph对象这才是我们实际运行的东西。from langgraph.graph import StateGraph, END # 1. 创建图并指定 State 的类型 workflow StateGraph(AgentState) # 2. 添加节点 workflow.add_node(“call_model”, call_model) workflow.add_node(“use_tool”, use_tool) # 3. 设置入口点 workflow.set_entry_point(“call_model”) # 4. 添加边和条件边 workflow.add_conditional_edges( “call_model”, should_use_tool, # 路由函数 { “use_tool_node”: “use_tool”, # 如果返回 “use_tool_node”则跳转到 use_tool 节点 “end_node”: END # 如果返回 “end_node”则结束图执行 } ) workflow.add_edge(“use_tool”, END) # 工具节点执行完后结束 # 5. 编译图 app workflow.compile()编译后的app就是一个可以调用的智能体。你可以通过app.invoke(initial_state)来运行它。5. 实战构建一个具备工具调用能力的智能体现在我们将构建一个完整的智能体。它的功能是回答用户问题当问题涉及“计算”时调用一个计算器工具否则直接由模型回答。5.1 定义 State 和工具from typing import TypedDict, Annotated, List import operator from langchain_core.tools import tool # 1. 定义 State class CalculatorState(TypedDict): question: str # 用户问题 model_thought: str # 模型的思考过程 final_answer: str # 给用户的最终答案 used_tools: Annotated[List[str], operator.add] # 记录使用了哪些工具 # 2. 定义一个简单的计算器工具 tool def calculator(expression: str) - str: 计算一个数学表达式的值。支持加减乘除和括号。例如(3 5) * 2 try: # 警告实际生产中应对表达式进行严格安全检查避免代码注入 result eval(expression) return f“计算器结果{expression} {result}” except Exception as e: return f“计算错误{e}” # 工具列表 tools [calculator]5.2 创建节点函数我们需要两个关键节点一个负责让模型决定行动思考另一个负责执行工具。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_core.prompts import ChatPromptTemplate, MessagesPlaceholder # 初始化模型 llm ChatOpenAI(model“gpt-3.5-turbo”, temperature0) # 构建 ReAct 风格的提示词模板 prompt ChatPromptTemplate.from_messages([ (“system”, “你是一个聪明的助手可以回答问题或使用计算器工具。请严格按照以下格式回复\nThought: 你的思考过程\nAction: 要使用的工具名如果是最终答案则填 ‘Final Answer’\nAction Input: 工具的输入参数如果是最终答案则直接给出答案”), (“user”, “{input}”), MessagesPlaceholder(variable_name“agent_scratchpad”), ]) # 节点1代理Agent节点 - 决定思考、行动或直接回答 def agent_node(state: CalculatorState): # 将工具绑定到LLM llm_with_tools llm.bind_tools(tools) # 创建ReAct智能体 agent create_react_agent(llm_with_tools, tools, prompt) agent_executor agent | ReActSingleInputOutputParser() # 调用智能体传入当前问题 result agent_executor.invoke({“input”: state[“question”]}) # 解析结果 if hasattr(result, ‘tool’) and result.tool: # 如果需要调用工具 return { “model_thought”: result.log, “action”: result.tool, “action_input”: result.tool_input } else: # 如果是最终答案 return { “model_thought”: result.log, “final_answer”: result.return_values[‘output’] } # 节点2工具执行节点 def tool_node(state: CalculatorState): # 从state中获取要执行的动作 tool_name state[“action”] tool_input state[“action_input”] # 查找对应的工具 tool_map {tool.name: tool for tool in tools} tool_to_use tool_map[tool_name] # 执行工具 observation tool_to_use.invoke(tool_input) # 更新状态记录工具使用和观察结果 return { “used_tools”: [tool_name], “tool_observation”: observation }5.3 构建并运行图from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from langgraph.checkpoint import MemorySaver # 创建图 workflow StateGraph(CalculatorState) # 添加节点 workflow.add_node(“agent”, agent_node) workflow.add_node(“execute_tool”, tool_node) # 设置入口点 workflow.set_entry_point(“agent”) # 定义边逻辑根据 agent 节点的输出决定下一步 def route_after_agent(state: CalculatorState): # 如果 agent 节点产生了 final_answer则结束 if “final_answer” in state and state[“final_answer”]: return “end” # 否则说明需要执行工具 else: return “continue_to_tool” workflow.add_conditional_edges( “agent”, route_after_agent, { “continue_to_tool”: “execute_tool”, “end”: END } ) # 工具执行完后应回到 agent 节点进行下一步思考ReAct循环 workflow.add_edge(“execute_tool”, “agent”) # 可选添加记忆检查点使智能体能在多轮对话中记住历史 memory MemorySaver() app workflow.compile(checkpointermemory) # 运行智能体 initial_state {“question”: “请问 (12 8) * 3 等于多少”} final_state app.invoke(initial_state, config{“configurable”: {“thread_id”: “test-1”}}) print(“用户问题”, final_state[“question”]) print(“模型思考”, final_state.get(“model_thought”)) print(“使用的工具”, final_state.get(“used_tools”, [])) print(“最终答案”, final_state.get(“final_answer”))运行上述代码你会看到智能体经历了 “Thought - Action - Tool - Observation - Thought - Final Answer” 的完整 ReAct 循环最终给出了正确答案。这个图结构清晰地定义了智能体的决策流。6. 部署为 API 服务与批量任务处理将 LangGraph 智能体封装成 API 服务是将其集成到 Web 应用或其他系统的标准做法。我们使用 FastAPI 来实现。6.1 创建 FastAPI 应用# main.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Optional import asyncio from your_langgraph_module import app # 导入之前编译好的 LangGraph app # 定义请求和响应模型 class AgentRequest(BaseModel): question: str thread_id: Optional[str] None # 用于支持多轮对话的会话ID class AgentResponse(BaseModel): thread_id: str final_answer: str used_tools: list[str] model_thought: Optional[str] None # 创建 FastAPI 实例 fastapi_app FastAPI(title“LangGraph 智能体 API”) fastapi_app.post(“/chat”, response_modelAgentResponse) async def chat_with_agent(request: AgentRequest): try: # 准备初始状态 initial_state {“question”: request.question} config {“configurable”: {“thread_id”: request.thread_id or “default-thread”}} # 调用编译好的图。注意invoke 是同步的在异步环境中使用 run_in_executor 避免阻塞 loop asyncio.get_event_loop() final_state await loop.run_in_executor( None, app.invoke, initial_state, config ) # 构建响应 response AgentResponse( thread_idconfig[“configurable”][“thread_id”], final_answerfinal_state.get(“final_answer”, “未生成答案”), used_toolsfinal_state.get(“used_tools”, []), model_thoughtfinal_state.get(“model_thought”) ) return response except Exception as e: raise HTTPException(status_code500, detailf“智能体执行失败: {str(e)}”) fastapi_app.get(“/health”) async def health_check(): return {“status”: “healthy”}6.2 启动服务使用 Uvicorn 启动这个 FastAPI 应用。uvicorn main:fastapi_app --host 0.0.0.0 --port 8000 --reload启动后你可以通过http://localhost:8000/docs访问自动生成的 API 文档并测试/chat接口。6.3 批量任务处理对于批量处理大量问题简单的循环调用 API 可能效率低下且难以管理。我们可以设计一个简单的异步批量处理器。# batch_processor.py import aiohttp import asyncio from typing import List import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) async def process_single_question(session: aiohttp.ClientSession, api_url: str, question: str, thread_id: str): 处理单个问题 payload {“question”: question, “thread_id”: thread_id} try: async with session.post(api_url, jsonpayload) as response: if response.status 200: result await response.json() logger.info(f“成功处理: ‘{question}’ - {result[‘final_answer’][:50]}...”) return result else: error_text await response.text() logger.error(f“处理失败 ‘{question}’: HTTP {response.status} - {error_text}”) return None except Exception as e: logger.error(f“请求异常 ‘{question}’: {e}”) return None async def batch_process_questions(questions: List[str], api_url: str “http://localhost:8000/chat”, max_concurrent: int 5): 批量处理问题列表控制并发数 connector aiohttp.TCPConnector(limitmax_concurrent) timeout aiohttp.ClientTimeout(total60) async with aiohttp.ClientSession(connectorconnector, timeouttimeout) as session: tasks [] for idx, q in enumerate(questions): # 为每个问题生成一个唯一的 thread_id或根据业务逻辑决定 thread_id f“batch-{idx}” task asyncio.create_task(process_single_question(session, api_url, q, thread_id)) tasks.append(task) # 等待所有任务完成 results await asyncio.gather(*tasks, return_exceptionsTrue) # 整理结果 successful_results [] for q, r in zip(questions, results): if isinstance(r, Exception): logger.error(f“任务异常 for ‘{q}’: {r}”) elif r is not None: successful_results.append(r) logger.info(f“批量处理完成。总计: {len(questions)}, 成功: {len(successful_results)}”) return successful_results # 使用示例 if __name__ “__main__”: question_list [ “计算 25 * 4 的值” “北京今天的天气怎么样” # 这个可能需要其他工具此处仅为示例 “(18 - 7) / 2 等于多少” ] # 运行批量处理 final_results asyncio.run(batch_process_questions(question_list, max_concurrent3)) for res in final_results: print(res)这个批量处理器使用了aiohttp进行异步 HTTP 调用并通过max_concurrent参数控制并发数避免对 API 服务造成过大压力。7. 资源占用、性能观察与优化LangGraph 本身是一个轻量级的编排框架其资源占用主要取决于大模型 API 调用这是主要的耗时和成本来源。响应时间取决于模型提供商和网络。工具执行如果你集成了计算密集型或网络 I/O 密集型的工具它们会成为瓶颈。Python 运行时图本身和状态管理开销通常很小。性能观察点延迟使用time模块记录app.invoke()的总耗时拆分为模型调用耗时和工具执行耗时。API 调用次数监控大模型 API 的调用次数和 Token 消耗这是成本核心。状态大小如果 State 中存储了大量历史消息或中间结果可能会影响内存和序列化/反序列化性能。优化建议精简 State只保留必要的数据在 State 中避免存储过大的对象如图片二进制数据。异步工具如果工具涉及网络请求如查询数据库、调用外部 API将其实现为异步函数并在图中使用异步节点可以显著提升吞吐量。缓存对于重复性的工具调用或模型查询结果可以考虑引入缓存如functools.lru_cache或 Redis。流式输出如果最终答案是文本且模型支持可以使用流式响应Streaming来提升用户体验。检查点Checkpoint优化如果使用了MemorySaver等检查点注意其存储后端。对于高并发可能需要使用数据库如 PostgreSQL替代内存存储。8. 常见问题与排查方法在开发 LangGraph 智能体时你可能会遇到以下典型问题。问题现象可能原因排查方式解决方案app.invoke()报错KeyErrorState 中缺少某个节点函数期望的键。检查节点函数的输入参数和返回值。确保所有节点更新 State 时使用的键名一致。统一 State 的 TypedDict 定义并在节点返回的字典中使用正确的键名。使用.get()方法提供默认值避免 KeyError。智能体陷入无限循环图中的边逻辑有误导致两个节点间循环调用。1. 打印每个节点执行后的 State。2. 检查条件边add_conditional_edges的路由函数逻辑。确保存在明确的终止条件路由到END。在路由函数中加入最大循环次数限制。工具调用失败1. 工具函数签名或参数不匹配。2. 模型生成的Action Input格式错误。1. 打印模型生成的Action和Action Input。2. 单独测试工具函数。1. 确保工具使用tool装饰器正确定义。2. 在提示词中更明确地指导模型输出正确的工具输入格式。多轮对话记忆混乱检查点Checkpointer未正确配置或thread_id未传递。检查app.invoke()时是否传入了包含thread_id的 config。确保每次属于同一会话的调用都使用相同的thread_id。检查 MemorySaver 的配置。图编译很慢或内存占用高1. 图结构非常复杂节点/边极多。2. 在编译前加载了大型模型。1. 分析图结构复杂度。2. 观察编译阶段的资源使用。1. 尝试简化图或将大图拆分为子图。2. 将模型初始化移到节点函数内部或使用懒加载。API 服务并发性能差1. 智能体本身是同步的阻塞了事件循环。2. 模型 API 调用慢。使用压力测试工具如locust测试 API。1. 如 6.1 节所示使用run_in_executor将同步调用放到线程池中执行。2. 考虑使用异步模型客户端如openai.AsyncOpenAI并实现异步节点。langchain或langgraph导入错误版本不兼容或依赖缺失。检查pip list确认已安装的版本。查看错误堆栈信息。创建新的虚拟环境严格按照官方文档或本文 3. 节的推荐版本安装。使用pip install langgraph[all]安装常用扩展。9. 最佳实践与进阶建议掌握了基础之后遵循一些最佳实践能让你的 LangGraph 项目更加健壮和可维护。State 设计要精简而明确State 是你的智能体的“内存”。只定义工作流真正需要共享和传递的数据。使用Annotated和归约器如operator.add来优雅地处理列表追加等操作。节点函数保持纯净与可测试每个节点函数应尽可能只做一件事。避免在节点内部处理复杂的副作用。这样便于单元测试和调试。充分利用可视化调试LangGraph 提供了app.get_graph().draw_mermaid()功能可以生成 Mermaid 图表直观展示你的工作流。这在调试复杂图时非常有用。为生产环境配置检查点开发时可以用MemorySaver但生产环境建议使用持久化存储如SqliteSaver或PostgresSaver以确保会话状态不会丢失。实现严格的错误处理与回退在节点中对模型调用和工具调用进行try...catch。可以设计一个专门的“错误处理”节点当任何节点失败时路由到这里进行清理或提供友好回复。安全第一尤其是工具调用。永远不要盲目执行模型生成的代码或系统命令。对工具输入进行严格的验证、清洗和权限控制。考虑在沙箱环境中运行不可信的工具。版本化你的图当智能体逻辑更新时图的定义可能改变。考虑如何管理不同版本的图以及如何迁移已保存的检查点状态。监控与日志在关键节点添加详细的日志记录记录 State 的变化、工具调用结果和模型响应。这有助于追踪问题和分析智能体的决策过程。LangGraph 将智能体的开发从“写一大段脆弱的 Prompt”变成了“组装可维护、可测试的组件”。它可能不是最简单上手的但对于需要清晰逻辑和状态管理的复杂 AI 应用来说它提供的结构和灵活性是无可替代的。从本文的简单计算器智能体出发你可以尝试集成更复杂的工具网络搜索、数据库、代码解释器、设计多智能体协作的图、或者构建具备长期记忆和个性化能力的对话系统。