LangChain与LangGraph实战:打造可落地的AI-Agent与MCP记忆系统

发布时间:2026/8/31 3:16:38
LangChain与LangGraph实战:打造可落地的AI-Agent与MCP记忆系统 LangChain 的教程很多但大多数停留在“调用一次模型”的程度。这次我们不绕弯直接走一条能落地的路径从 LangChain 的模型调用开始进入 LangGraph 的状态图编排再做到 AI-Agent 实战并把 MCP 工具接入、智能体记忆 Memory 这两个容易卡人的模块一并过一遍。先说门槛。LangChain 和 LangGraph 不是“需要显卡才能跑”的模型而是纯 Python 的 LLM 应用开发框架CPU 机器就能开发和测试。真正吃资源的是底层大模型你可以接 OpenAI 等在线模型 API也可以用 Ollama 在本地跑开源模型。所以这篇文章没有显存门槛8GB 内存的电脑也能起步关键是理解状态、节点、工具调用这些工程概念。本文会带你完成以下内容环境安装与 API Key 配置第一个 LangChain 程序LangGraph 节点/边/条件路由带工具调用的 ReAct Agent短期与长期记忆MCP Server 工具接入FastAPI 接口封装以及批量任务。适合已经写过 Python、想从“调用模型”升级到“构建 Agent”的开发者。1. 核心能力速览能力项说明项目类型LLM 应用开发框架 Agent 编排框架开源情况LangChain、LangGraph 均为开源项目社区活跃度很高核心功能模型调用、提示词模板、检索增强 RAG、工具调用、Agent 编排、状态管理、多智能体协作、记忆持久化硬件门槛框架层纯 Python 实现CPU 即可显存由底层模型决定显存占用框架本身几乎不占用显存本地部署模型时显存占用由推理框架决定支持平台Windows / macOS / Linux启动方式Python 脚本 / Jupyter Notebook / FastAPI 服务API 支持支持封装成后端服务也支持调用 OpenAI / DeepSeek / Ollama 等模型服务批量任务支持 LangGraph 原生 batch 方法也可接任务队列适合场景AI-Agent 开发、RAG 知识库、流程编排、多智能体模拟、工具集成从这张表能看出LangChain LangGraph 的组合最核心的卖点不是“单个模型有多强”而是“把模型装进一个可控的、有状态的、可扩展的工程体系”。下面进入场景判断。2. 适用场景与使用边界这套技术栈适合以下人群和项目正在做 AI-Agent 产品原型的开发者需要让模型自主决定调用哪些工具、按什么顺序调用。在做 RAG 知识库问答的团队想在“文档检索 生成”之外加入工具调用和多轮记忆。需要搭建多智能体协作系统的工程师比如主控 Agent 调度多个子 Agent 完成复杂任务。想把 LLM 能力封装成内部 API 服务、批量处理任务的平台开发人员。不适合的情况也很清楚完全没写过 Python直接上手会卡在环境依赖和异步调用上建议先补 Python 基础。只想要一个开箱即用的可视化界面LangChain/LangGraph 本身不带 WebUI需要自己用 Gradio、Streamlit 或 FastAPI 搭建。对输出实时性要求极低、流程非常固定的场景可能直接写函数调用比引入框架更省事。使用边界必须单独说。Agent 一旦具备工具调用能力就相当于把系统的一部分执行权交给了模型。下面几点需要在实际项目中守住对用户私密信息、通讯录、文件系统等敏感资源的访问必须经过明确授权并在最小权限范围内运行。调用外部工具时要对工具返回内容做校验避免提示词注入导致 Agent 执行非预期操作。涉及人脸、声音、肖像、版权文本或图片的生成与处理必须确认授权来源不能用于未经同意的用途。自动化任务要记录日志方便回查 Agent 在每一步做了什么谁在什么时候触发过什么操作。3. 核心概念梳理LangChain、LangGraph、AI-Agent、MCP、Memory3.1 LangChain 和 LangGraph 有什么区别这个问题几乎每个入门者都会问。LangChain 是 LLM 应用开发框架提供模型封装、提示词模板、输出解析、检索器、记忆、工具等组件适合把“调用模型”这件事工程化。LangGraph 是 LangChain 团队推出的 Agent 编排框架核心思路是把应用建模成一张图StateGraph 维护状态Node 执行逻辑Edge 决定流转边可以是普通连接也可以是条件连接。通俗类比LangChain 是积木盒组件丰富适合搭固定流程LangGraph 是带状态机的流水线适合搭需要循环、分支、人工介入、多智能体协作的复杂流程。更准确地说LangGraph 不是用来替代 LangChain 的它依赖 LangChain 的模型封装和工具体系同时提供了更底层的执行编排能力。你可以在 LangGraph 的节点里直接使用 LangChain 的 ChatOpenAI、ChatPromptTemplate、Retriever 等组件。顺便回应一个热门问题LangChain、vLLM、PyTorch 是不是同一类东西不是。PyTorch 是深度学习底层框架vLLM 是模型推理服务框架LangChain/LangGraph 是应用开发框架。三者处于不同层次日常开发中它们可以组合使用但不互相替代。3.2 AI-Agent 的核心机制从工程实现看AI-Agent 就是让模型在一个循环里反复做三件事理解任务、决定调用哪个工具、观察工具返回结果再决定下一步。最常见的模式是 ReAct即 Reason Act 交替进行。LangGraph 里实现 Agent 有两种路径一种是直接用create_react_agent这种预置封装快速搭建另一种是在 StateGraph 里手动定义 agent 节点和 tools 节点通过条件边让两者循环直到模型认为不需要再调用工具。后者更灵活便于扩展记忆、人工审核和分支逻辑。3.3 MCP 解决什么问题MCP全称 Model Context Protocol是一个标准化模型与外部工具之间连接的协议。以前每接一个外部系统就要写一套私有集成逻辑MCP 的目标是把工具接口统一起来让模型应用可以像插 USB 设备一样接入工具。在 LangChain/LangGraph 项目中MCP 的价值在于工具提供方只需要实现一个 MCP Server应用方通过langchain-mcp-adapters就可以把 MCP Server 里的工具加载成 LangChain 的 Tool然后交给 Agent 调用。比如浏览器控制、设计稿读取、数据库查询、GitHub 操作等都有社区提供的 MCP Server 可以直接接入。也有朋友会问Agent Skill 和 MCP 有什么区别。简单说Skill 更像是给 Agent 准备的一份“技能说明书”或可复用流程模板指导模型怎么完成任务MCP 更像是一种“外部工具接口标准”把具体的功能调用能力暴露给模型。两者解决的是不同层面问题实际项目中可以同时使用。3.4 Memory 在 LangGraph 里的分层智能体记忆主要分两层。第一层是线程级短期记忆通过 Checkpointer 保存每一次对话后的完整状态只要传入相同的thread_id下一次调用就能读到之前的消息历史。这一层适合多轮对话场景。第二层是跨线程长期记忆通常存用户画像、偏好、关键事实等可以使用 LangGraph 的 Store 机制也可以自己用 Redis、PostgreSQL、向量数据库实现。长期记忆和短期记忆配合才能做出真正“记住老用户”的 Agent。4. 环境准备与前置条件4.1 Python 版本与依赖安装建议使用 Python 3.10 及以上版本避免部分异步语法和类型注解兼容问题。先确认本机 Python 版本python --version然后安装核心依赖pip install --upgrade langchain langchain-openai langgraph如果要做 MCP 接入再安装适配器pip install langchain-mcp-adapters mcp如果后续把 Agent 封装成 HTTP 服务需要安装 FastAPI 和 uvicornpip install fastapi uvicorn如果你倾向用 Jupyter Notebook 跟着跑也可以直接安装 jupyter 后逐段执行。4.2 模型服务准备在线 API 或本地 Ollama推荐准备两个通道之一。通道一在线模型 API。将 Key 写入环境变量避免硬编码在代码里。以 OpenAI 兼容接口为例export OPENAI_API_KEY你的API_KEY如果你用的是 DeepSeek、通义千问等 OpenAI 兼容服务可以在ChatOpenAI里传base_url指向对应服务地址。通道二本地 Ollama。先安装 Ollama再拉取模型ollama pull qwen2.5:7bOllama 启动后默认会在http://localhost:11434提供 OpenAI 兼容接口开发阶段不产生 API 费用适合反复测试。4.3 验证安装是否成功打开 Python 交互环境执行import langchain import langgraph print(langchain.__version__) print(langgraph.__version__)能正常输出版本号说明依赖安装完成。如果报ModuleNotFoundError回到上一步检查 pip 安装是否成功、当前 Python 环境是否与 pip 对应。5. 第一个 LangChain 程序模型调用与提示词模板先写一个最小的模型调用。这个程序的作用是验证模型接口通了没有不要一上来就搭复杂结构。from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage llm ChatOpenAI( modelgpt-4o-mini, temperature0.7, api_key你的API_KEY, # 生产环境建议从环境变量读取 ) resp llm.invoke([HumanMessage(content用一句话介绍 LangChain)]) print(resp.content)如果你走本地 Ollama只要修改模型名和接口地址llm ChatOpenAI( modelqwen2.5:7b, base_urlhttp://localhost:11434/v1, api_keyollama, # 本地服务不校验给任意值即可 )接下来验证 LangChain 的提示词模板和 LCEL 链式调用。prompt | llm是 LangChain 最常用的组合方式左侧传模板变量右侧接模型输出类型会沿着链自动传递。from langchain_core.prompts import ChatPromptTemplate prompt ChatPromptTemplate.from_messages([ (system, 你是一名{role}回答问题时必须简洁、直接、不啰嗦。), (human, {question}), ]) chain prompt | llm result chain.invoke({ role: 技术顾问, question: 什么是 MCP 协议 }) print(result.content)这段代码跑通后你已经掌握了 LangChain 最核心的组件模型封装、提示词模板、LCEL 组合。判断成功的标准是输出内容是紧扣“MCP 协议”且符合“简洁直接”要求的回答。如果这块还有问题先排查这三项API Key 是否有效、模型名是否真实存在、网络是否能正常访问对应模型服务域名。本地 Ollama 模式还要检查服务是否启动、端口是否为 11434。6. LangGraph 基础从链式调用到状态图LangChain 的链适合固定顺序但真实 Agent 流程经常需要循环、分支、并行。LangGraph 把这些能力抽象成一张状态图。6.1 状态、节点、边状态是 LangGraph 的核心。我们用一个TypedDict定义状态结构每个节点函数接收当前状态返回要更新的部分状态。返回的内容会和原状态合并。from typing import TypedDict from langgraph.graph import StateGraph, START, END class AgentState(TypedDict): messages: list next_step: str def collect_input(state: AgentState): return {next_step: process} def process(state: AgentState): return {messages: state[messages] [{role: assistant, content: 处理完成}]} graph StateGraph(AgentState) graph.add_node(collect, collect_input) graph.add_node(process, process) graph.add_edge(START, collect) graph.add_edge(collect, process) graph.add_edge(process, END) app graph.compile() result app.invoke({messages: [], next_step: }) print(result[messages])这段代码建立了最简单的一条流水线collect - process - END。注意节点函数必须返回 dict返回的 key 会更新到共享状态里这是 LangGraph 更新状态的基本方式。6.2 条件路由conditional_edges实际 Agent 经常要根据内容走不同分支。比如判断用户问题是否需要计算需要就进入计算节点不需要就进入普通对话节点。def route_by_content(state: AgentState): last_message state[messages][-1][content] if 计算 in last_message: return calculator return chat graph.add_conditional_edges( collect, route_by_content, { calculator: calculator, chat: chat, } )add_conditional_edges的第一个参数是起点节点第二个参数是路由函数第三个参数是路由结果到节点的映射。路由函数返回值必须能在映射表里找到否则会在运行时报错。6.3 循环与子图Agent 的核心循环是“模型判断要不要继续调用工具”。在 LangGraph 里这通过条件边从 tools 节点指回 agent 节点实现形成环。如果模型返回的工具调用列表为空就走END结束。当单个图太复杂时可以把一张已编译的子图加到另一张图的节点上app.add_node(sub_agent, sub_app) # sub_app 是另一个 compile() 后的图LangGraph 还支持并行扇出使用Send把不同输入动态分发到多个子节点。对于需要同时处理多个独立子任务的场景这是比写死 for 循环更可控的方案。具体 API 在不同版本略有差异使用时以官方文档为准。7. AI-Agent 实战带工具调用的智能体现在进入正题构建一个能自主调用工具的 AI-Agent。这里用 LangGraph 预置的create_react_agent它内部已经实现了 ReAct 循环我们只需要定义工具列表并传入模型。7.1 定义工具用tool装饰器定义一个函数函数名和 docstring 会作为模型调用工具的元信息因此描述要清晰。from langchain_core.tools import tool tool def get_weather(city: str) - str: 查询指定城市的天气入参为城市名称。 return f{city}晴气温 26°C tool def calculate(expression: str) - str: 计算数学表达式例如 12*78。 try: return str(eval(expression)) except Exception as e: return f计算失败{e}需要特别提醒eval直接执行字符串存在代码注入风险这里只是为了演示工具调用流程。生产环境要换成安全解析器比如用ast限制表达式范围或者调用专门的公式计算库。7.2 创建并调用 Agentfrom langgraph.prebuilt import create_react_agent from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini, api_key你的API_KEY) agent create_react_agent(llm, [get_weather, calculate]) response agent.invoke({ messages: [{role: user, content: 北京天气怎么样顺便算一下 12*78}] }) print(response[messages][-1].content)运行后模型应该会先调用get_weather拿到天气信息再调用calculate算出结果最后把两部分信息整理成一段完整回答。判断成功的关键不是看答案是否“漂亮”而是看在response[messages]里是否出现了工具调用记录和工具返回结果。7.3 验证工具调用过程想看清楚 Agent 内部发生了什么可以把响应里的中间消息打印出来for msg in response[messages]: print(msg.type, -, msg.content[:100])正常会看到ai - 包含工具调用请求、tool - 工具返回结果、ai - 最终回答这样的消息序列。如果只有一条ai消息且没有工具调用说明模型判定不需要工具或者工具描述不够清晰导致模型没有触发调用。7.4 关于标题里的 DeepAgent 方向标题里出现的 “DeepAgent”并不是 LangGraph 官方 API 名称更像是课程或博客对“深度 Agent 实战”的包装说法。它的本质是用 LangGraph 把多节点、多工具、带记忆、带条件路由、甚至多智能体协作的复杂系统拼装起来。学完本文的 StateGraph、工具调用和 Memory就具备了搭建这类 DeepAgent 的基础能力。真正深入的进阶方向是多 Agent 分工、人工审核节点、动态工具注册、状态压缩和长期记忆。8. 智能体记忆 Memory 实战从 Checkpoint 到长期记忆Agent 默认是无状态的每次调用都是一张白纸。要让它记住用户上一句说了什么需要启用记忆机制。8.1 Thread-level 短期记忆LangGraph 用 Checkpointer 保存每次调用后的完整状态。接入MemorySaver并在调用时传thread_id同一个会话内的多轮消息就会被保留。from langgraph.checkpoint.memory import MemorySaver memory MemorySaver() agent_with_memory agent.compile(checkpointermemory) config {configurable: {thread_id: session-001}} result1 agent_with_memory.invoke({ messages: [{role: user, content: 我叫小明喜欢打篮球}] }, config) result2 agent_with_memory.invoke({ messages: [{role: user, content: 我叫什么名字}] }, config) print(result2[messages][-1].content)第二次调用如果回答出“小明”说明线程级记忆已经生效。注意必须传入相同的thread_id并且compile时传入同一个 checkpointer 实例。换一个thread_id就是完全独立的会话。8.2 长期记忆跨会话用户画像线程级记忆只在一个会话内有效。跨会话记住用户信息需要长期记忆。LangGraph 官方方向是引入 Store 机制但不同版本 API 变化较快更稳妥的做法是结合项目实际情况选择存储方案。比如用一个简单的 Redis 或 PostgreSQL 表保存用户画像每次 Agent 启动时加载对话结束后回写# 伪代码长期记忆读写 user_profile load_profile(user_id) # 从数据库读取 system_prompt f当前用户信息{user_profile} result agent.invoke({messages: [...]}) save_profile(user_id, updated_profile) # 对话后回写把长期记忆和短期记忆分开治理短期记忆解决“当前对话上下文”长期记忆解决“用户是谁、偏好什么”。两者配合才能做出持续的个性化体验。9. MCP 接入实战让 Agent 通过标准协议调用外部工具当工具数量多、来源分散时MCP 的价值就体现出来了。一个 MCP Server 负责暴露一组工具LangGraph 侧的 Agent 通过langchain-mcp-adapters把这些工具加载成普通 Tool。先安装依赖pip install langchain-mcp-adapters mcp9.1 连接本地 MCP Server下面是一个基于 stdio 传输的加载示例。思路是创建一个MultiServerMCPClient配置 MCP Server 的启动命令然后获取工具列表。from langchain_mcp_adapters.client import MultiServerMCPClient async def load_mcp_tools(): client MultiServerMCPClient({ filesystem: { command: python, args: [path/to/mcp_filesystem_server.py], transport: stdio, } }) tools await client.get_tools() return tools这段代码是异步的调用时需要放在异步环境中或者用asyncio.run包一层。不同版本的langchain-mcp-adapters对异步支持方式有差异建议以当前版本的官方文档为准。9.2 常见的 MCP Server社区里已经有大量现成 MCP Server常见方向包括浏览器控制类Playwright MCP可以让 Agent 操作浏览器页面。设计稿类Figma MCP、蓝湖 MCP读取设计稿信息和标注。数据库类通过 MCP 暴露 SQL 查询工具让 Agent 安全执行只读查询。开发工具类GitHub MCP、GitLab MCP操作 Issue、PR。本地文件类读写指定目录文件使用时要严格限制路径范围。接入思路都一样有一个 MCP Server把它加进MultiServerMCPClient拿到工具后交给 Agent。真正麻烦的不是“连接”而是设计清楚哪些工具可以暴露给 Agent、调用权限怎么控制。9.3 MCP 与 Agent Skill 如何配合前面提到过MCP 负责“工具接口标准化”Agent Skill 负责“任务方法模板化”。举个例子你想让 Agent 用 Playwright 做网页自动化测试MCP 提供的是打开页面、点击元素、读取文本这些原子能力Skill 则是把“登录 - 填写表单 - 提交 - 截图”这套流程固化下来让模型知道遇到同类任务时按这个步骤执行。实际项目里两者可以组合Skill 定义流程MCP 提供执行能力。10. 接口 API 与批量任务Agent 写好之后通常要暴露成服务才能被业务系统调用。这里用 FastAPI 封装一个带记忆的 Agent 接口。10.1 FastAPI 封装 Agent 服务from fastapi import FastAPI from pydantic import BaseModel from langchain_openai import ChatOpenAI from langgraph.prebuilt import create_react_agent from langgraph.checkpoint.memory import MemorySaver app FastAPI() class ChatRequest(BaseModel): message: str thread_id: str default llm ChatOpenAI(modelgpt-4o-mini, api_key你的API_KEY) agent create_react_agent(llm, tools[]) memory MemorySaver() compiled_agent agent.compile(checkpointermemory) app.post(/chat) def chat(req: ChatRequest): config {configurable: {thread_id: req.thread_id}} result compiled_agent.invoke( {messages: [{role: user, content: req.message}]}, config ) return {reply: result[messages][-1].content}启动服务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: 你好介绍一下你自己}返回结果里的reply字段就是模型的回答。这个接口已经带上了线程级记忆同一个thread_id连续调用Agent 能记住上下文。10.2 批量任务LangGraph 内置了batch方法可以一次性传入多条输入适合离线批量处理场景inputs [] for i in range(10): inputs.append({ messages: [{role: user, content: f这是第{i}个问题} }) results compiled_agent.batch(inputs, config{max_concurrency: 5}) for r in results: print(r[messages][-1].content)max_concurrency控制并发数量能有效避免同时请求过多导致模型服务过载。生产环境中更建议把批量任务放进 Celery / Redis 队列由 Worker 异步消费失败任务自动重试这样不会因为单个请求异常阻塞整个队列。10.3 批量任务失败重试批量任务最容易遇到两类问题一是外部模型 API 限流二是某个样本输入格式错误导致整批失败。建议每个任务单独捕获异常不要因为一条失败就中断整个批次for input_item in inputs: try: result compiled_agent.invoke(input_item) save_result(result) except Exception as e: log_error(input_item, e) retry_inputs.append(input_item)11. 资源占用与性能观察很多读者关心显存先说结论LangChain 和 LangGraph 本身是纯 Python 框架运行时不加载大模型权重因此框架层几乎不占显存。资源占用主要由底层模型服务决定。使用在线 API本机只占用少量 CPU 和内存显存占用为 0。瓶颈在网络延迟和 API 限额。使用本地 Ollama显存占用取决于模型大小和量化精度比如 7B 量化模型通常需要 6GB 左右显存实际以本机nvidia-smi观察为准。使用本地 vLLM显存占用和推理吞吐都更高适合服务化部署不适合个人开发调试。开发时重点观察三个指标第一个是响应时间。LangGraph 的流程越长、工具调用次数越多响应越慢。排查耗时可以用graph.invoke的耗时日志也可以打印每条消息的时间戳定位是模型推理慢还是工具调用慢。第二个是内存占用。长对话场景下消息历史会不断累积Checkpointer 保存的状态也越来越大。可以通过定期压缩历史消息、只保留最近 N 轮摘要来控制内存。第三个是并发能力。FastAPI 接口服务默认是同步阻塞的高并发时需要把 Agent 调用放到异步线程池或者使用独立任务队列。批量任务要控制最大并发避免瞬时请求打满模型服务。12. 常见问题与排查方法问题现象可能原因排查方式解决方案安装 langgraph 失败Python 版本过低或依赖冲突确认python --version和 pip 环境使用 Python 3.10建议创建独立虚拟环境调用 OpenAI 接口报错API Key 无效、模型名错误、网络不可达检查环境变量、模型名、连接日志换有效 Key确认模型名检查网络连通性本地 Ollama 连接不上Ollama 未启动或端口不对curl http://localhost:11434验证启动 Ollama确认 base_url 正确LangGraph 节点状态不更新节点函数返回的不是 dict打印节点返回值和输入 state确保节点返回 dict键要与 State 定义一致conditional_edges 路由报错路由函数返回的路径不在映射表打印路由函数返回值给路由函数加默认分支覆盖所有可能返回值记忆没有生效未传 thread_id 或未加 checkpointer检查 compile 参数和 config编译时传入 memory调用时传configurable.thread_idMCP 工具加载不出来MCP Server 未启动或路径错误单独运行 MCP Server 验证确认启动命令和参数检查日志FastAPI 端口冲突8000 已被占用lsof -i:8000查看占用换端口启动本地模型显存不足模型过大或并发过高用nvidia-smi观察显存占用换更小模型、开启量化、降低并发工具调用不触发工具描述不清晰或模型版本不支持打印中间消息日志优化工具名和 docstring换更强的模型再回答一个