这次我们来看一个关于 LangGraph 智能体开发的实战教程。LangGraph 作为 LangChain 生态中用于构建复杂、有状态多智能体应用的核心框架正受到越来越多开发者的关注。它解决了传统链式调用在处理循环、分支和持久化状态时的局限性让构建能“思考”和“协作”的 AI 智能体变得更加直观。本文不是空谈概念而是聚焦于实战从核心架构拆解到代码逐行实现带你快速掌握如何用 LangGraph 搭建一个具备记忆、工具调用和决策能力的智能体并探讨其与 API、MCP 协议等生态的集成。如果你关心如何将大模型 API 转化为可执行、可协作的智能工作流或者正在寻找比简单提示工程更强大的智能体构建方案这篇文章值得你仔细阅读。我们将重点关注 LangGraph 的核心设计思想、关键组件的实际用法以及如何将其部署为一个可复用的服务。整个过程不依赖特定付费平台完全基于开源框架和标准 API你可以用自己的开发环境复现。1. 核心能力速览能力项说明项目类型智能体Agent开发框架用于构建有状态、可循环、多分支的 AI 应用工作流。核心价值将复杂的智能体逻辑如规划、执行、反思、协作图形化、模块化超越简单的线性链Chain。主要功能定义状态State、创建节点Node与边Edge、管理循环与条件分支、持久化对话历史与智能体状态。硬件门槛无特殊要求。本质是一个 Python 框架运行时消耗取决于所连接的大模型 API如 OpenAI GPT、DeepSeek、本地 Ollama 模型的调用开销。本地测试通常 CPU 和少量内存即可。启动方式通过 Python 脚本启动智能体应用也可封装为 FastAPI 等 Web 服务提供 API 接口。是否支持 API是。智能体逻辑本身可对外提供 API同时框架支持轻松集成外部工具和服务的 API。是否支持批量任务是。可以通过初始化多个图Graph实例或设计支持批量输入的状态结构来处理并发或批量任务。适合场景开发客服机器人、复杂任务规划与执行系统、多角色模拟与协作、自动化研究与分析流程、集成现有业务系统通过 MCP 等协议。2. 适用场景与使用边界LangGraph 适合需要超越简单问答和文本生成的场景。当你的应用逻辑涉及“先思考再行动根据结果决定下一步”的循环或者需要多个 AI “角色”分工协作时它就是理想选择。典型适用场景包括复杂任务分解与执行例如用户提出“帮我分析某公司的市场竞争力”智能体可以自动分解为“搜索最新财报”、“分析行业趋势”、“总结竞争优势与风险”等子任务并依次执行。持续对话与状态维护在客服或游戏场景中需要记住之前的对话历史和用户偏好并基于此做出连贯的后续响应。多智能体协作模拟一个团队例如一个“研究员”智能体负责搜集资料一个“分析师”智能体负责总结一个“评审员”智能体负责检查质量它们通过共享的状态进行交互。工具增强的AI应用智能体可以根据需要调用搜索引擎、数据库、代码执行环境、绘图工具等外部 API实现功能扩展。使用边界与注意事项并非万能对于简单的、无状态的文本生成任务使用基础的 Chain 或直接调用模型 API 可能更轻量、高效。复杂性管理图Graph的设计可能变得复杂需要良好的架构设计来保证可维护性。成本与延迟每个节点的执行都可能涉及一次大模型 API 调用在复杂图中可能导致较高的 token 消耗和累积延迟。合规与安全当智能体能够自动调用外部工具如网络搜索、数据库写入时必须实施严格的权限控制和输入验证防止越权操作或注入攻击。所有生成内容需符合法律法规。3. 环境准备与前置条件在开始编写 LangGraph 智能体之前你需要准备好基础的 Python 开发环境并确保能够访问所需的大模型服务。基础环境清单操作系统Windows 10/11, macOS, 或 Linux (推荐 Ubuntu 20.04)。Python 版本3.8 或更高版本。建议使用 3.10 以获得最佳兼容性。包管理工具pip或conda。网络能够访问你选择的大模型 API 服务如 OpenAI, Anthropic, DeepSeek 等。如果使用本地模型如通过 Ollama则需确保本地服务已启动。核心依赖包LangGraph 是 LangChain 生态系统的一部分通常需要一并安装。# 创建并激活虚拟环境推荐 python -m venv langgraph-env source langgraph-env/bin/activate # Linux/macOS # 或 langgraph-env\Scripts\activate # Windows # 安装核心包 pip install langgraph langchain langchain-core大模型 API 配置你需要一个可用的 API Key。本文以广泛使用的 OpenAI 兼容 API 为例也可以是 DeepSeek、通义千问等提供的兼容接口。# 安装 OpenAI SDK (如果使用 OpenAI 或兼容接口) pip install openai然后在代码中或通过环境变量设置 API Key# 在终端中设置环境变量临时 export OPENAI_API_KEYyour-api-key-here # Linux/macOS # set OPENAI_API_KEYyour-api-key-here # Windows CMD # $env:OPENAI_API_KEYyour-api-key-here # Windows PowerShell4. LangGraph 核心概念与架构拆解理解 LangGraph 的关键在于掌握其几个核心抽象它们共同构成了智能体的“骨架”。4.1 状态State状态是智能体的“记忆单元”是一个字典Dict在图的整个执行生命周期中传递和更新。你需要在定义图时声明状态的模式Schema。from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages import operator class State(TypedDict): # 消息历史add_messages 操作符能智能地追加消息 messages: Annotated[List[str], add_messages] # 其他自定义状态如当前任务步骤、工具调用结果等 current_step: str research_findings: List[str]Annotated和add_messages是 LangGraph 提供的语法糖用于简化消息列表的追加操作。4.2 节点Node节点是图中的一个执行单元通常是一个函数。它接收当前状态执行一些操作如调用 LLM、运行工具并返回一个更新后的状态字典。def llm_node(state: State) - dict: 一个简单的LLM响应节点 from langchain_openai import ChatOpenAI model ChatOpenAI(modelgpt-4o-mini, temperature0) # 从状态中获取对话历史 conversation_history state[messages] # 构造提示词 prompt f基于以下对话历史请给出专业回复\n{conversation_history[-1]} # 调用模型 response model.invoke(prompt) # 更新状态将模型回复添加到消息历史 new_messages state[messages] [response.content] return {messages: new_messages, current_step: responded}4.3 边Edge边决定了执行流程。分为两种条件边Conditional Edge根据某个条件函数的结果决定下一步走向哪个节点。普通边无条件地从一个节点指向下一个节点。4.4 图Graph图是节点和边的容器。你通过StateGraph类来构建图添加节点和边并最终编译成一个可执行的CompiledGraph。from langgraph.graph import StateGraph, END # 1. 创建图并指定状态结构 workflow StateGraph(State) # 2. 添加节点 workflow.add_node(llm_agent, llm_node) workflow.add_node(research_tool, research_node) # 假设有另一个研究工具节点 # 3. 设置入口点 workflow.set_entry_point(llm_agent) # 4. 添加边 workflow.add_edge(llm_agent, research_tool) # llm_agent 执行完后总是去 research_tool workflow.add_edge(research_tool, END) # research_tool 执行完后结束 # 5. 编译图 app workflow.compile()编译后的app就是一个可执行的智能体。你通过传入初始状态来运行它final_state app.invoke({messages: [用户问题]})。5. 实战构建一个具备研究与总结能力的智能体让我们构建一个相对完整的智能体它能够根据用户问题决定是否需要联网研究然后进行总结。5.1 定义状态与工具首先定义更丰富的状态并模拟一个“网络搜索”工具。from typing import TypedDict, Annotated, List, Literal from langgraph.graph.message import add_messages import operator class AgentState(TypedDict): messages: Annotated[List[str], add_messages] needs_research: bool # 是否需要研究 research_data: List[str] # 研究得到的数据 final_answer: str # 最终答案 # 模拟一个网络搜索工具函数 def web_search_tool(query: str) - str: 模拟工具根据查询返回模拟的网络搜索结果。 # 真实场景中这里会调用 SerperAPI、Google Search API 等 print(f[工具调用] 正在搜索: {query}) # 返回模拟数据 simulated_results [ f关于{query}的摘要信息点 A。, f关于{query}的最新报告指出观点 B。, f专家对{query}的看法 C。 ] return \n.join(simulated_results)5.2 创建决策节点路由这个节点负责分析用户问题决定是否需要调用研究工具。def decision_node(state: AgentState) - dict: 决策节点判断是否需要深入研究。 from langchain_openai import ChatOpenAI model ChatOpenAI(modelgpt-4o-mini, temperature0) last_message state[messages][-1] if state[messages] else prompt f 用户的问题是{last_message} 请判断回答这个问题是否需要实时或最新的网络信息进行研究。 如果你的判断是**需要**请回复单词 research。 如果你的判断是**不需要**仅凭你的知识足以回答请回复单词 answer。 只输出 research 或 answer不要有其他任何内容。 decision model.invoke(prompt).content.strip().lower() return {needs_research: decision research}5.3 创建研究节点如果决策节点判定需要研究则执行此节点。def research_node(state: AgentState) - dict: 研究节点调用搜索工具获取信息。 query state[messages][-1] search_results web_search_tool(query) return {research_data: [search_results]} # 将结果存入状态5.4 创建回答节点这个节点综合对话历史和研究数据如果有生成最终答案。def answer_node(state: AgentState) - dict: 回答节点生成最终答案。 from langchain_openai import ChatOpenAI model ChatOpenAI(modelgpt-4o-mini, temperature0) last_message state[messages][-1] research_info \n.join(state.get(research_data, [])) prompt f 请回答用户的问题。 用户问题{last_message} {f以下是相关的网络研究信息\n{research_info} if research_info else 请基于你的知识进行回答。} 请提供清晰、有条理、准确的回答。 response model.invoke(prompt) return {final_answer: response.content}5.5 组装智能体图现在我们将节点和条件边组合起来形成一个完整的智能体工作流。from langgraph.graph import StateGraph, END # 创建图 workflow StateGraph(AgentState) # 添加所有节点 workflow.add_node(decision, decision_node) workflow.add_node(research, research_node) workflow.add_node(answer, answer_node) # 设置入口点 workflow.set_entry_point(decision) # 添加条件边决策后路由 def decide_next_step(state: AgentState) - Literal[research, answer]: 根据决策节点的结果决定下一步是研究还是直接回答。 return research if state.get(needs_research) else answer workflow.add_conditional_edges( decision, decide_next_step, { research: research, answer: answer, } ) # 添加普通边 workflow.add_edge(research, answer) # 研究完成后进入回答节点 workflow.add_edge(answer, END) # 回答完成后结束 # 编译图 research_agent workflow.compile()5.6 运行与测试智能体编译完成后我们就可以像调用函数一样调用这个智能体。# 初始化输入 initial_state { messages: [请解释一下什么是 LangGraph以及它和 LangChain 的区别], needs_research: False, research_data: [], final_answer: } # 运行智能体 print(开始运行智能体...) try: final_state research_agent.invoke(initial_state) print(\n 智能体运行完成 ) print(f最终答案\n{final_state[final_answer]}) print(f是否触发了研究{final_state.get(needs_research)}) except Exception as e: print(f运行出错{e})运行上述代码你会看到智能体首先经过decision节点判断可能会输出[工具调用]信息然后根据判断结果要么走research - answer路径要么直接走answer路径最终输出一个结构化的答案。6. 接口 API 封装与批量任务处理将编译好的智能体图封装成 API 服务是投入生产环境的关键一步。这里我们使用 FastAPI 来创建一个简单的 Web 服务。6.1 创建 FastAPI 应用首先确保安装了 FastAPI 和 Uvicorn。pip install fastapi uvicorn然后创建一个app.py文件from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List, Optional import asyncio # 导入之前编译好的智能体 research_agent # 假设上面的代码已封装在一个函数 create_agent() 中 from your_agent_module import create_agent app FastAPI(titleLangGraph 智能体 API 服务) agent create_agent() # 获取编译好的图实例 # 定义请求体模型 class AgentRequest(BaseModel): message: str session_id: Optional[str] None # 用于区分不同会话实现状态隔离 class BatchAgentRequest(BaseModel): tasks: List[AgentRequest] # 单次查询接口 app.post(/v1/chat/completions) async def chat_completion(request: AgentRequest): try: # 这里可以基于 session_id 从数据库或缓存中恢复历史状态 # 为简化示例我们每次都是新的会话 initial_state { messages: [request.message], needs_research: False, research_data: [], final_answer: } # 注意invoke 是同步方法在异步环境中使用 asyncio.to_thread 避免阻塞 final_state await asyncio.to_thread(agent.invoke, initial_state) return { session_id: request.session_id, answer: final_state[final_answer], needs_research: final_state.get(needs_research, False) } except Exception as e: raise HTTPException(status_code500, detailf智能体执行失败: {str(e)}) # 批量任务接口简易版顺序执行 app.post(/v1/batch/completions) async def batch_chat_completion(batch_request: BatchAgentRequest): results [] for task in batch_request.tasks: try: initial_state { messages: [task.message], needs_research: False, research_data: [], final_answer: } final_state await asyncio.to_thread(agent.invoke, initial_state) results.append({ session_id: task.session_id, answer: final_state[final_answer], success: True }) except Exception as e: results.append({ session_id: task.session_id, answer: None, error: str(e), success: False }) return {results: results} if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)6.2 启动与调用 API 服务在终端中启动服务python app.py服务启动后你可以使用curl或 Pythonrequests库进行调用。# 单次调用 curl -X POST http://127.0.0.1:8000/v1/chat/completions \ -H Content-Type: application/json \ -d {message: 什么是MCP协议, session_id: test_123} # 批量调用 curl -X POST http://127.0.0.1:8000/v1/batch/completions \ -H Content-Type: application/json \ -d { tasks: [ {message: 解释一下AI幻觉, session_id: task1}, {message: Python的GIL是什么, session_id: task2} ] }# Python 调用示例 import requests import json url http://127.0.0.1:8000/v1/chat/completions payload {message: 什么是MCP协议, session_id: test_123} headers {Content-Type: application/json} response requests.post(url, datajson.dumps(payload), headersheaders) print(response.json())6.3 实现真正的状态持久化进阶上面的示例每次都是全新的状态。在生产环境中你需要根据session_id将会话状态即AgentState持久化到数据库如 Redis、PostgreSQL或文件系统中。在每次invoke时先加载历史状态执行后再保存新状态。这可以实现真正的多轮对话记忆。7. 与 MCPModel Context Protocol协议集成MCP 协议是 LangChain 推出的一种标准化协议用于让 AI 应用如智能体安全、可控地访问外部工具和数据源如数据库、文件系统、第三方 API。LangGraph 智能体可以很方便地集成 MCP Server 来扩展能力。核心思路部署或连接 MCP Server例如一个提供数据库查询工具的 MCP Server。在 LangGraph/LangChain 中配置 MCP Client通过langchain-mcp-adapters等包将 MCP Server 提供的工具注册到 LangChain 的工具列表中。在智能体节点中调用这些工具就像调用普通的tool装饰的函数一样。简化示例步骤# 假设已有一个运行在 localhost:8001 的 MCP Server (例如提供 SQL 查询) # 安装 MCP 适配器 pip install langchain-mcp-adaptersfrom langchain_mcp_adapters import ClientAdapter import httpx # 1. 创建 MCP 客户端连接到 MCP Server async with httpx.AsyncClient() as client: mcp_client ClientAdapter.create( clientclient, transportsse, # 或 stdio取决于 Server 类型 urlhttp://localhost:8001/sse # MCP Server 的 SSE 端点 ) # 2. 获取 MCP Server 提供的所有工具 tools await mcp_client.get_tools() # 3. 将工具绑定到 LLM from langchain_openai import ChatOpenAI model ChatOpenAI(modelgpt-4o-mini) model_with_tools model.bind_tools(tools) # 4. 在 LangGraph 的节点函数中就可以让模型决定是否调用这些 MCP 工具了 def agent_node_with_mcp(state: State): # ... 构造消息 ... response model_with_tools.invoke(messages) if response.tool_calls: # 执行工具调用这里会通过 MCP 协议与 Server 通信 results execute_tool_calls(response.tool_calls, tools) # ... 处理结果并更新状态 ... return new_state通过集成 MCP你的 LangGraph 智能体就能安全地访问企业内部数据库、CRM 系统等而无需将敏感逻辑直接写在智能体代码中。8. 资源占用、性能观察与优化LangGraph 框架本身是轻量级的性能瓶颈主要在于大模型 API 调用和工具执行。性能观察点Token 消耗每个节点中的 LLM 调用都会消耗 Token。在复杂图中累计成本可能很高。建议在关键节点打印或记录输入的 prompt 长度和返回的 response 长度。API 延迟网络延迟和模型推理时间占主导。对于顺序执行的图总耗时是各节点耗时的总和。考虑对无依赖的节点进行异步并行处理。状态大小如果状态中存储了大量历史消息或数据可能会影响序列化和传输效率。定期清理或摘要历史消息。工具调用开销如果工具是网络服务如搜索、数据库查询其延迟也需要计入。优化建议异步执行对于可以并行的节点例如同时查询多个不相关的数据源使用asyncio.gather在节点函数中并发执行。缓存对频繁查询且结果变化不大的工具调用如某些知识查询引入缓存机制如functools.lru_cache或 Redis 缓存。流式输出如果最终答案是文本生成考虑使用模型的流式响应提升用户体验。超时与重试为工具调用和模型调用设置合理的超时和重试机制增强鲁棒性。图结构优化审视你的图是否存在不必要的循环或分支能否合并一些简单的节点9. 常见问题与排查方法问题现象可能原因排查方式解决方案导入StateGraph等模块失败LangGraph 版本不兼容或未正确安装。检查 pip listgrep langgraph 确认版本。查看官方文档要求的版本。运行app.invoke()时报状态字段错误传入的初始状态字典结构与StateGraph(State)中定义的TypedDict不匹配。打印初始状态的键与State类中定义的字段逐一对比。确保初始状态字典包含State中定义的所有非可选字段且类型匹配。可使用state State(messages[], ...)方式构造。条件边Conditional Edge不生效总是走默认路径条件函数decide_next_step返回的值与add_conditional_edges中映射的键不匹配。在条件函数内打印其返回值确认是research、answer等确切的字符串。确保条件函数返回的文字与边映射字典的键完全一致大小写敏感。智能体陷入无限循环图中存在循环路径如 A-B, B-A但没有设置终止条件。检查图的结构特别是add_edge和add_conditional_edges的指向。确保每条执行路径最终都能到达END节点。可以在循环中设置最大迭代次数或在状态中增加step_count字段并在条件边中判断。调用 MCP 工具或外部 API 超时网络问题、服务端故障或未设置超时。使用try...except捕获超时异常并检查对应服务的日志和状态。在工具调用函数或 HTTP 请求中设置合理的timeout参数。实现重试逻辑如tenacity库。批量任务处理速度慢顺序执行每个任务没有利用并发。观察任务处理是顺序的。将 API 服务中的批量接口改为使用asyncio.gather并发调用智能体但要注意状态隔离和资源限制。显存/内存占用过高使用本地模型时同时运行多个智能体实例或加载了多个大模型。使用nvidia-smi或psutil监控资源。使用线程池或进程池限制并发数。考虑使用模型服务化如通过 Ollama 的 API让多个智能体共享同一个模型服务。10. 最佳实践与使用建议从简单开始逐步复杂化先构建一个只有2-3个节点的最小可行图MVP确保它能跑通。再逐步添加新的节点、分支和状态字段。状态设计要精简只把真正需要在节点间传递的数据放入状态。避免存储过大的中间结果如整张图片可以存储文件路径或引用 ID。为节点和边起好名字使用有意义的名称如classify_intent,call_search_api这能让图的可视化和调试更容易。实现完善的日志记录在每个节点的开始和结束记录日志包括输入状态摘要、工具调用详情、耗时等。这对于调试复杂工作流至关重要。进行彻底的单元测试为每个节点函数编写单元测试模拟输入状态验证输出状态。然后为整个图编写集成测试。版本化你的图当对智能体逻辑进行重大修改时考虑对图定义进行版本控制以便回滚和对比。安全第一特别是当智能体能够执行写操作如发送邮件、修改数据库或访问敏感信息时必须在工具调用前加入权限验证和用户确认机制可在状态中设计user_confirmed字段。成本监控在关键节点记录 LLM 调用的 Token 使用量并设置预算告警避免意外的高额费用。LangGraph 提供了一个强大而灵活的范式来构建新一代的 AI 智能体应用。它最大的优势在于将复杂的、有状态的交互流程清晰地建模成一张图使得开发、调试和维护都变得更加直观。要掌握它最好的方法就是动手实践从一个具体的任务出发画出它的执行流程图然后用 LangGraph 的节点和边将其实现出来。当你成功地将一个复杂的业务逻辑转化为一个自运行的智能体时你会感受到这种开发方式的巨大潜力。建议将本文的示例代码作为起点尝试改造和扩展构建属于你自己的智能体。