新版LangChainLangGraph的教程看了不少但大多数还停留在链式调用和提示词拼接的阶段。真正值得投入时间的是LangGraph带来的有状态Agent编排能力——它把Agent从一条固定流水线变成一个可以循环决策、调用工具、分叉路由、持久化记忆的程序化系统。这篇文章的目标很直接从零基础开始带你跑通新版LangChainLangGraph的AI-Agent实战包括MCP工具接入、Memory记忆管理、条件路由、批量任务和API化部署。先说结论LangGraph不是要替代LangChain而是在LangChain的基础上补上“状态管理”和“图编排”这两块短板。新版LangChain中LangGraph已经作为官方推荐的Agent编排方案深度集成。你会需要一点Python基础不需要GPUCPU也能跑通框架本身因为推理部分由LLM API或本地模型服务承担。文章会引导你完成环境准备、LangChain核心语法、LangGraph状态图构建、DeepAgent实战、MCP接入、记忆持久化、接口发布和问题排查每一步都给可复制代码。如果你正在做AI-Agent落地、工具调用类应用或者准备把Agent封装成服务给业务复用这篇文章建议收藏后按章节实操。1. 核心能力速览能力项说明项目定位LangChain LangGraph 的大模型应用与Agent编排开发框架核心能力链式调用、图状态编排、条件路由、工具调用、MCP协议接入、记忆持久化、批量任务、API化部署适合人群有Python基础、想从LangChain过渡到LangGraph、或准备做Agent落地的开发者硬件要求框架本身依赖CPU和内存即可运行推理由LLM端点或本地模型服务提供显存占用取决于推理后端使用本地大模型时显存由模型服务占用与LangGraph框架本身无直接关系支持平台Windows / Linux / macOS建议Linux或macOS做生产部署推荐Python版本Python 3.9 及以上启动方式Python脚本 / Jupyter Notebook / API服务是否支持API支持可基于FastAPI封装工具调用接口是否支持批量任务支持可使用LangGraph内置batch能力或自建批量处理队列开源情况LangChain、LangGraph均为开源项目MCP为开放协议典型场景客户问答、文档分析、代码辅助、数据处理、多工具协同Agent从能力表可以看出来这套技术栈的价值不在“调用模型”而在“把一次模型对话变成一套可控制、可跟踪、可恢复的业务流程”。2. LangChain与LangGraph的关系与选型很多人一上来就问LangGraph出来了LangChain是不是过时了LangChain和LangGraph到底有什么区别结论是LangChain仍然是LangGraph的基础设施。LangGraph是LangChain团队推出的Agent编排框架它不是在重写LangChain而是用“状态图”的方式重新组织LangChain已有的模型调用、工具、检索和记忆能力。2.1 LangChain擅长什么LangChain做了三件基础工作统一模型接入接口无论是OpenAI、Anthropic、DeepSeek、Qwen还是本地vLLM都可以通过ChatOpenAI等封装统一调用。提供提示词模板、输出解析器、文档加载器、向量存储抽象让RAG和文档问答的开发成本大幅降低。在0.3版本之后LangChain已经把老旧的AgentExecutor作为历史方案冻结推荐迁移到LangGraph。如果你只是做“文档问答、数据提取、简单对话”这类偏线性任务LangChain基础组件完全够用。2.2 LangGraph解决什么问题LangGraph的核心是状态图StateGraph。它把Agent的一次运行过程拆成多个节点Node节点之间用边Edge连接运行时通过共享的State传递信息。这样做带来了三个关键能力循环Agent可以反复执行“模型推理 - 工具调用 - 模型再推理”的闭环而不是一次调用就结束。条件分支根据模型输出或业务规则动态决定下一步走哪个节点实现真正的“Agent自主决策”。持久化状态可以存到SQLite、PostgreSQL等存储器里任务中断后可以恢复也可以实现跨会话记忆。2.3 选型建议任务类型推荐方案单轮问答 / 简单RAGLangChain基础组件即可多轮对话 / 需要记录上下文LangGraph MemorySaverAgent工具调用 / 多步骤任务LangGraph ToolNode复杂业务流 / 多人协作 / 需要人工审核节点LangGraph 子图 条件路由接入外部工具生态LangGraph MCP Adapters所以如果你准备做Agent不要再用老式的AgentExecutor直接从LangGraph开始。3. 环境准备与依赖安装3.1 环境检查清单检查项推荐配置操作系统Windows 10/11、Ubuntu 20.04、macOS 12PythonPython 3.9 - 3.12 均可建议3.11包管理工具pip 或 poetry、uvGit可选用于拉取示例代码模型API需要可用的LLM API或本地部署模型服务3.2 创建虚拟环境建议每个项目使用独立虚拟环境避免依赖冲突。下面以venv为例python -m venv .venv source .venv/bin/activate # Windows 下使用 .venv\Scripts\activate3.3 安装核心依赖pip install -U langchain langgraph langchain-openai pip install langgraph-checkpoint-sqlite langchain-mcp-adapters pip install jupyter # 可选用于Notebook调试安装完成后检查版本python -c import langchain; import langgraph; print(langchain.__version__); print(langgraph.__version__)LangGraph版本更新较快如果你的版本与教程示例有出入以官方文档为准。遇到安装失败时优先排查Python版本和依赖冲突不建议强制覆盖安装langchain-core。4. LangChain基础与LCEL实践在进入LangGraph之前先快速过一遍新版LangChain的核心写法LCEL表达式。4.1 基本调用示例from langchain_openai import ChatOpenAI from langchain_core.output_parsers import StrOutputParser from langchain_core.prompts import ChatPromptTemplate # 这里配置你的模型服务地址和API Key llm ChatOpenAI( modelgpt-4o-mini, api_keyyour-api-key, base_urlyour-model-endpoint, # 本地或第三方兼容地址 temperature0.7 ) prompt ChatPromptTemplate.from_messages([ (system, 你是一个严谨的技术助手。), (human, {question}) ]) chain prompt | llm | StrOutputParser() result chain.invoke({question: 用一句话解释LangGraph是什么}) print(result)这段代码使用了管道符|把“提示词模板、模型、输出解析器”串成一条链这就是LCEL的核心思想。每个组件都实现了Runnable接口可以单独invoke也可以组合调用。4.2 绑定工具的模型LangGraph的Agent通常需要模型支持工具调用在LangChain中通过bind_tools实现from langchain_core.tools import tool tool def get_weather(city: str) - str: 查询指定城市的实时天气 return f{city} 当前天气晴朗气温28度 llm_with_tools llm.bind_tools([get_weather])注意bind_tools依赖模型本身支持function calling或tool calling。如果使用本地部署的开源模型需要确认后端服务提供了工具调用协议否则模型不会输出tool_calls。5. LangGraph核心概念与第一个Agent5.1 核心概念LangGraph有五个核心概念概念作用State全局共享状态节点之间通过它传递数据Node一个函数或可执行对象接收State并返回State变更Edge节点之间的连接决定流程方向Conditional Edge根据条件动态选择下一步节点Checkpointer状态持久化机制实现记忆和中断恢复5.2 构建一个带工具调用的Agent下面构建一个最小可运行的Agent用户提问 - 模型决定是否调用工具 - 调用工具 - 回到模型生成最终回答。from typing import TypedDict, Annotated, Literal from langgraph.graph import StateGraph, START, END from langgraph.prebuilt import ToolNode, tools_condition from langchain_core.messages import BaseMessage, HumanMessage from langchain_openai import ChatOpenAI # 定义状态 class AgentState(TypedDict): messages: Annotated[list, add_messages] # 工具定义 tool def get_weather(city: str) - str: 查询指定城市的天气 return f{city} 今天多云转晴最高气温30度。 tool def get_stock_price(symbol: str) - str: 查询股票代码的当前价格 return f{symbol} 当前价格 128.5 元。 # 初始化模型并绑定工具 llm ChatOpenAI(modelgpt-4o-mini, api_keyyour-api-key, base_urlyour-model-endpoint) llm_with_tools llm.bind_tools([get_weather, get_stock_price]) # Agent节点模型根据当前消息列表决定是回答问题还是调用工具 def agent_node(state: AgentState): response llm_with_tools.invoke(state[messages]) return {messages: [response]} # 构建图 graph StateGraph(AgentState) # 添加节点 graph.add_node(agent, agent_node) graph.add_node(tools, ToolNode([get_weather, get_stock_price])) # 添加边 graph.add_edge(START, agent) graph.add_conditional_edges(agent, tools_condition) graph.add_edge(tools, agent) # 编译 app graph.compile()执行一次询问result app.invoke({ messages: [HumanMessage(content北京天气怎么样顺便查一下600519的股价)] }) for msg in result[messages]: print(f{msg.type}: {msg.content})tools_condition是LangGraph预置的条件路由函数它会检查模型输出中是否有tool_calls如果有进入tools节点如果没有直接跳到结束。5.3 判断Agent是否成功判断标准很明确模型输出正确识别出“需要调用两个工具”。每个工具调用都返回了对应结果。最终回答综合了天气和股价信息。整个过程中messages列表完整记录了每一步。如果最终回答没有工具结果优先查看模型是否真的输出了tool_calls。很多问题出在模型端点不支持工具调用或者绑定工具格式与模型不兼容。6. 条件路由、分支控制与子图6.1 自定义条件路由内置的tools_condition只判断是否有工具调用但真实业务往往需要更灵活的规则。比如用户输入包含敏感词时先走审核节点问题太复杂时先走检索节点。这时可以自己写路由函数def route_after_agent(state: AgentState) - Literal[tools, audit, END]: last_message state[messages][-1] # 如果模型请求调用工具 if getattr(last_message, tool_calls, None): return tools # 如果用户输入包含需要审核的关键词 user_text state[messages][0].content.lower() if 激进 in user_text or 风险 in user_text: return audit return END graph.add_conditional_edges(agent, route_after_agent, { tools: tools, audit: audit, __end__: END })条件路由的关键是路由函数接收当前State返回一个字符串然后在映射表里指定返回字符串对应的目标节点。6.2 子图子图用于把复杂流程拆成独立模块。比如一个客服Agent可以拆成“订单查询子图”和“售后处理子图”。主图根据用户的意图进入不同子图。# 先定义一个子图 sub_graph StateGraph(AgentState) sub_graph.add_node(agent, agent_node) sub_graph.add_node(tools, ToolNode([get_weather])) sub_graph.add_edge(START, agent) sub_graph.add_conditional_edges(agent, tools_condition) sub_graph.add_edge(tools, agent) sub_compiled sub_graph.compile() # 主图把子图作为节点 main_graph StateGraph(AgentState) main_graph.add_node(intent_router, intent_router_node) main_graph.add_node(weather_agent, sub_compiled) main_graph.add_edge(START, intent_router) main_graph.add_conditional_edges(intent_router, route_by_intent)子图的好处是隔离复杂度每个子图可以单独测试、单独调试、单独复用。如果某个子图崩了不影响主图其他分支。6.3 并行分支LangGraph也支持并行执行多个节点。比如一个分析任务同时进行“关键词提取”和“情感分析”两个节点处理完汇总到合并节点。并行能减少整体耗时但要注意线程安全和状态更新的原子性。并行分支的写法是在路由映射表中让多个节点同时返回再通过add_edge合并到汇总节点。具体语法建议结合当前版本文档查看因为并行分支的API在各版本之间有微调。7. AI-Agent实战接入MCP工具生态7.1 什么是MCPMCP是模型上下文协议解决的是“Agent如何统一接入外部工具”的问题。之前每个工具都要自己写一套调用代码接入方还需要了解每个工具的参数格式。MCP相当于给工具行业提供了一个统一接口规范Agent通过MCP协议就能发现、调用、组合外部能力。可以这样理解MCP是工具层的“统一接口”Agent是业务层的“调度大脑”。7.2 在LangGraph中接入MCPlangchain-mcp-adapters提供了从MCP服务器加载工具的适配器。下方是通用流程from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from langchain_mcp_adapters.tools import load_mcp_tools # 配置MCP服务器启动参数 server_params StdioServerParameters( commandpython, args[./mcp_server.py], # 你的MCP服务器脚本 envNone ) async def load_tools(): async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools await load_mcp_tools(session) return tools加载到的tools可以直接传给ToolNode。这样你的Agent就能通过MCP协议调用外部系统比如文件管理、GitHub仓库操作、数据库查询、蓝湖设计稿信息读取等。7.3 自建MCP服务器的思路如果你内部系统需要暴露成MCP服务可以用官方提供的Python SDK编写。一个最小MCP服务器结构如下from mcp.server.fastmcp import FastMCP mcp FastMCP(my-service) mcp.tool() def query_order(order_id: str) - str: 查询订单状态 # 这里连接内部业务系统 return f订单 {order_id} 状态已发货 if __name__ __main__: mcp.run()启动后你的Agent、Claude Desktop或者其他MCP客户端都能按协议接入这个工具。自建MCP服务时要注意内部系统的鉴权、数据权限、接口限流都需要在MCP服务层做控制不能把内部敏感数据无差别暴露给模型。7.4 Agent Skill与MCP的区别一个经常被问到的问题是Agent Skill和MCP有什么区别。简单来说Skill是“给Agent的一段技能指令或调用手册”解决的是“Agent要不要用、怎么用”的问题偏提示词和流程逻辑MCP解决的是“工具怎么连、参数怎么传”的问题偏通信协议和工具标准化。实际项目中两者可以结合用Skill定义Agent的能力边界用MCP完成具体工具接入。8. 智能体记忆Memory短期与长期记忆设计Agent没有记忆能力相当于每次对话都是陌生人。LangGraph在记忆上提供了比较完整的方案这里分开说。8.1 短期记忆会话内上下文LangGraph把对话消息放在State的messages字段中Agent运行时通过Checkpointer保存检查点。只要传入相同thread_id下一次调用可以继续读取之前的消息记录。from langgraph.checkpoint.memory import MemorySaver memory MemorySaver() app_with_memory graph.compile(checkpointermemory) config {configurable: {thread_id: user-abc-001}} # 第一轮对话 app_with_memory.invoke( {messages: [HumanMessage(content我的订单号是A1001)]}, configconfig ) # 第二轮对话模型仍然记得订单号 app_with_memory.invoke( {messages: [HumanMessage(content这个订单发货了吗)]}, configconfig )MemorySaver把记忆存在内存里适合开发调试和单机短生命周期场景。生产环境建议使用持久化实现比如SqliteSaver。from langgraph.checkpoint.sqlite import SqliteSaver with SqliteSaver.from_conn_string(./checkpoints.sqlite) as saver: app_with_sqlite_memory graph.compile(checkpointersaver)用SQLite保存检查点之后重启进程聊天记忆也不会丢这是多轮对话产品化最基础的一步。8.2 长期记忆跨会话知识库短期记忆解决“这个会话内记得住”长期记忆解决“用户下次来你还认识他、记得他的偏好”。LangGraph中可以通过Store实现长期记忆把用户偏好、业务标签、历史摘要写入独立存储在新会话启动时主动加载。# 伪代码展示长期记忆的读写思路 store_key fuser:{user_id} # 写入记忆 store.put((users, user_id), preferences, {city: 北京, care_about: [发货时间]}) # 读取记忆 memory_data store.get((users, user_id), preferences)需要说明的是长期记忆在LangGraph里的具体API仍随版本调整不建议把代码死记硬背。更重要的是理解分层记忆的设计思路记忆层次存储位置生命周期典型内容会话内消息State messages单次会话对话历史、中间结果短期记忆Checkpointer以thread_id为粒度的会话周期多轮对话上下文长期记忆Store/数据库跨会话长期保留用户偏好、业务事实、摘要记忆设计的核心不是“全部塞进上下文”而是只把关键信息从长期存储中抽取到当前会话避免上下文膨胀、推理变慢和token成本上升。8.3 记忆与上下文管理如果对话很长消息列表会越来越大最终触发上下文窗口上限。常用方案有两种消息压缩用LLM把历史对话总结成摘要替换原始历史。滑动窗口只保留最近N轮消息再加一段早期摘要。LangGraph中可以通过自定义节点拦截消息列表在做模型调用前完成压缩或裁剪。这一步在真实业务中基本必须做否则长会话一定会卡在上下文长度上。9. 接口API与批量任务9.1 封装成API服务使用FastAPI把Agent封装成HTTP接口是接入业务系统最快的方式。from fastapi import FastAPI from pydantic import BaseModel from langchain_core.messages import HumanMessage app FastAPI(titleLangGraph Agent API) class ChatRequest(BaseModel): message: str thread_id: str default class ChatResponse(BaseModel): reply: str app.post(/chat, response_modelChatResponse) def chat(req: ChatRequest): config {configurable: {thread_id: req.thread_id}} result app_with_memory.invoke( {messages: [HumanMessage(contentreq.message)]}, configconfig ) reply result[messages][-1].content return ChatResponse(replyreply) # 启动uvicorn main:app --host 0.0.0.0 --port 8000启动服务后可以通过curl验证curl -X POST http://127.0.0.1:8000/chat \ -H Content-Type: application/json \ -d {message: 帮我查一下北京天气, thread_id: test-001}接口服务要注意访问控制。如果暴露在公网需要加API Key鉴权或网关认证防止被未授权调用消耗模型额度。9.2 批量任务LangGraph支持通过batch方法批量处理多个输入inputs [ {messages: [HumanMessage(contentf总结第{i}篇文档)]} for i in range(10) ] results app.batch(inputs, max_concurrency5)批量任务推荐设计思路环节建议输入使用Excel、JSONL或数据库表批量导入并发根据API限流和模型服务承载能力设置max_concurrency日志每个任务记录thread_id、状态、耗时、失败原因失败重试网络超时或限流时指数退避重试1-3次输出结果写回数据库或导出JSON文件如果批量任务量很大建议用消息队列Redis/RabbitMQ加Worker进程消费而不是直接在主进程里batch否则长任务会阻塞API服务。10. 资源占用与性能观察LangGraph本身是一个轻量编排框架它占用的CPU和内存并不多。真正的资源开销来自两个地方一是LLM推理服务二是状态中保存的消息内容。10.1 内存占用观察在开发机上可以通过memory_profiler观察Python进程内存pip install memory_profiler mprof run python your_agent.py mprof plot如果进程出现OutOfMemory通常不是因为LangGraph框架本身而是因为State中messages列表无限增长历史消息全量保留。工具返回的大文本被塞进State比如把整个网页内容放进去。批量任务并发过高多个进程同时持有大量消息对象。解决方案消息裁剪、压缩历史、控制单次工具返回大小、降低并发数、使用持久化存储把历史消息卸载到磁盘或数据库。10.2 显存占用如果你使用本地模型服务显存占用由模型推理引擎决定与LangGraph无关。比如在本地部署Qwen、DeepSeek等模型时显存占用取决于模型参数规模和推理框架的量化配置。日常开发建议先用API方式完成Agent逻辑再进行本地模型替换。10.3 推理延迟优化影响Agent响应速度的主要因素模型第一次推理的TTFT与模型服务配置相关。工具调用轮数每轮工具调用都增加一次模型推理尽量避免无意义的多次工具调用。上下文长度历史消息越多推理越慢。减少不必要的历史消息能明显改善延迟。并发控制过高并发会导致模型服务排队单请求延迟反而升高。11. 常见问题与排查方法问题现象可能原因排查方式解决方案安装依赖时报冲突Python版本不匹配或依赖包版本冲突查看报错堆栈检查Python版本升级Python到3.11使用独立虚拟环境安装模型不返回工具调用结果模型端点不支持function calling打印模型原始输出确认有无tool_calls字段替换支持工具调用的模型或使用本地代理服务转换协议LangGraph节点不执行边连接配置错误起始节点未设置检查add_edge(START, ...)和条件路由映射表重新检查节点映射关系给每个条件分支添加日志多轮对话不记忆上下文编译时未传checkpointer或thread_id不一致检查是否调用了compile(checkpointer...)添加MemorySaver或SqliteSaver确保thread_id一致上下文越来越长API报超限messages列表无限增长统计每轮消息长度增加消息压缩或滑动窗口节点批量任务卡住某个工具调用超时或模型服务限流给每个节点加日志查看卡在哪个节点设置工具调用超时加入重试机制API服务启动后拒绝访问防火墙、端口绑定或鉴权问题curl本机接口查看uvicorn日志确认--host绑定地址配置鉴权中间件数据库持久化失败SQLite文件路径无权限或并发写入冲突查看SQLite依赖日志使用文件路径权限或切换到PostgreSQL存储子图调试困难子图内部错误没有向上传递先单独调用子图再挂载到主图给每个子图配置独立日志最小化复现问题一句话总结排查原则先从最小链路开始验证再逐步增加节点和工具。最小链路跑通后再加条件分支、记忆、MCP等复杂特性崩溃时定位范围就会小很多。12. 最佳实践与使用建议12.1 设计层面的建议第一次写Agent先用一个工具跑通全流程再并行接入多个工具。状态设计不要太“胖”只放必要的业务数据。大块文本尽量用引用或外部存储ID而不是直接塞进State。条件路由函数尽量保持纯函数不要在里面调用外部API否则排错时很难判断是路由问题还是外部服务问题。子图命名清晰主图保持精简否则节点多了以后维护成本极高。12.2 工程化层面的建议每个节点的输入输出都要有日志建议记录节点名、耗时、关键字段摘要。批量任务必须做失败重试和任务状态表不要只打印日志。接口服务必须做鉴权、限流和请求体大小限制。模型API Key不要硬编码在代码里使用环境变量或密钥管理服务。每个会话的thread_id要有业务含义比如用户ID会话ID方便回溯。12.3 合规与安全边界使用MCP接入外部系统时务必遵守目标系统的服务条款和授权范围。涉及用户隐私数据、企业业务数据时要做脱敏和权限收敛确保Agent只访问被授权的数据。涉及版权素材、人脸信息、声音信息等场景必须取得合法授权。发布商用Agent前要对模型的输出做人工抽检避免误导性内容或不当回复流向业务场景。13. 总结与下一步新版LangChainLangGraph最值得投入的部分是它把Agent从“写死流程”变成了“可决策、可循环、可记忆、可恢复”的程序化系统。你先要用StateGraph跑通一个最小工具调用Agent再逐步加入条件路由、子图、MCP、持久化记忆和批量任务。最容易踩的坑有三个模型端点不支持工具调用、消息历史无限膨胀导致OutOfMemory、没有设置checkpointer导致对话记忆丢失。下一步建议按顺序做三件事第一把最小Agent跑通并加日志第二接入一个真实业务工具用MCP或直接工具绑定都可以第三把Agent封装成FastAPI服务让业务方可以通过HTTP调用。跑通这三步你的LangChainLangGraph就算真正入门了。建议收藏备用后续做Agent项目时可以直接参考文章里的模板和排查清单。