ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

LangChain与LangGraph实战:构建带MCP工具和记忆的AI Agent

LangChain与LangGraph实战:构建带MCP工具和记忆的AI Agent 在实际的大模型应用开发中LangChain 和 LangGraph 已经不只是两个 Python 库的名称而是围绕 AI Agent、MCP 协议和智能体记忆 Memory 形成的一套完整开发链路。很多人刚开始接触时会被“LangChain 过时了”“LangGraph 替代 LangChain”“MCP 是什么”“记忆怎么存”这类说法绕晕。这篇文章从零开始用一个名为 DeepAgent 的示例项目完整跑通一条路径先搞清楚 LangChain 和 LangGraph 的分工再准备环境接着用 LangChain 写一个 Agent 雏形再用 LangGraph 重构成有状态的工作流然后接入 MCP 工具和 Memory 记忆最后给出运行验证、排错清单和生产落地建议。适合刚接触 Agent 开发、被概念和版本变动困住的读者。1. LangChain 和 LangGraph 到底解决什么问题1.1 从单次 Prompt 调用到 Agent 工作流早期的 LLM 应用通常是一个很直接的流程用户输入一段文字程序把这段文字拼进 Prompt调用模型拿到结果展示给用户。这种模式在简单问答、文本分类、内容生成场景下完全够用但当需求变成“让模型自己决定调用哪个工具”“多轮对话中记住上下文”“某个分支失败后自动重试”“多个 Agent 协作完成一个任务”时单次调用就不够了。Agent 的核心变化在于模型不再只是被动生成文本而是参与决策。它需要看到用户问题、分析可用工具、决定是否调用工具、根据工具结果继续推理最后输出答案。这个过程不是一次函数调用而是一个循环决策、执行、观察、再决策。要让这个循环可控、可追踪、可回放就需要把流程显式地建模出来。LangChain 解决的是组件标准化问题它把模型、Prompt、工具、检索器、记忆这些部件封装成统一接口。LangGraph 解决的则是流程编排问题它用“状态图”的方式描述 Agent 的节点和边让每一步都能被审计状态可以被持久化分支可以通过条件路由控制。两者不是替代关系而是上下游配合。1.2 LangChain 与 LangGraph 的分工和区别可以这样理解LangChain 提供零件LangGraph 提供流水线。零件包括ChatOpenAI、PromptTemplate、tool装饰器、Retriever、OutputParser等。流水线则把零件粘合成一个有状态、可循环、可持久化的图。在实际项目里常见的组合方式是使用 LangChain 的模型封装和工具标准使用 LangGraph 的StateGraph定义节点和路由使用 LangGraph 的 checkpointer 保存会话状态。两者的边界虽然没有完全固定但分工越来越清晰如果你只需要简单的多步骤链式调用用 LangChain 的RunnableSequence就够如果业务里有循环判断、分支跳转、多轮工具调用、长期状态保存就应该切到 LangGraph。对比维度LangChainLangGraph核心抽象Chain / RunnableStateGraph / Node / Edge是否显式建模流程偏线性适合链式调用有向图支持分支和循环状态管理依赖外部 Memory较松散State 为核心可持久化适用场景简单 RAG、Prompt 流水线、工具封装复杂 Agent、多轮工具调用、多 Agent 协作调试方式靠日志和中间结果可以查看每一步状态和路由结果1.3 认识 MCP 和智能体记忆MCP 全称 Model Context Protocol是一套把外部工具和数据源接入大模型应用的开放协议。它的思路和 USB 接口类似工具提供方实现一个 MCP ServerAgent 通过 MCP Client 统一发现和调用工具不用再为每个工具单独写一套接入代码。MCP 解决了“工具越来越多每家接口都不一样”的集成问题因此这几年在 Agent 生态里增长很快。智能体记忆 Memory 则解决的是“Agent 是否记得之前说过什么”的问题。记忆可以简单分成两层短期记忆保存当前会话的对话历史长期记忆保存跨会话的用户偏好、项目背景和历史结论。LangGraph 在短期记忆上做得比较规范通过 checkpointer 把 State 持久化长期记忆则通常要配合向量数据库、摘要存储和合适的写入时机。DeepAgent 这个示例项目正是围绕这三样东西展开用 LangChain 封装模型和工具用 LangGraph 编排流程用 MCP 接入外部工具用 Memory 保存长期上下文。2. 环境准备和依赖安装2.1 版本策略先定 Python再查依赖版本LangChain 和 LangGraph 的版本迭代速度非常快社区教程经常因为 API 变动而过期。开始之前先确定 Python 版本。建议使用 Python 3.10 到 3.12太老的版本可能装不上最新依赖太新的版本可能还没被完全兼容。推荐在虚拟环境中操作避免污染系统环境python -m venv .venv source .venv/bin/activateWindows 下激活命令是.venv\Scripts\activate激活后确认 pip 版本python -m pip install --upgrade pip2.2 安装依赖包DeepAgent 的最小依赖包括LangChain 核心、OpenAI 模型封装、LangGraph 工作流、MCP 适配层。可以一次性安装pip install langchain langchain-openai langgraph langchain-mcp-adapters python-dotenv如果要用 SQLite 保存 checkpointer可以额外安装pip install langgraph-checkpoint-sqlite如果需要使用社区工具、向量库或本地模型再按需补充pip install langchain-community langchain-ollama langchain-chroma这里要特别注意langchain是一个聚合包它会拉取langchain-core等核心依赖。langchain-openai是独立的模型适配包langchain-mcp-adapters负责把 MCP 工具转换成 LangChain 的 Tool 格式。不同版本之间可能存在 API 变动落地时以实际安装版的 README 或官方文档为准。2.3 配置模型连接DeepAgent 默认使用 OpenAI 兼容接口这样可以方便地切换不同模型服务。先在项目根目录创建.env文件OPENAI_API_KEYyour_api_key_here OPENAI_BASE_URLhttps://api.openai.com/v1 MODEL_NAMEgpt-4o-mini如果使用的是国内模型平台或本地 vLLM 服务把OPENAI_BASE_URL换成对应地址即可。然后写一个小的验证脚本确认模型能通import os from dotenv import load_dotenv load_dotenv() from langchain_openai import ChatOpenAI llm ChatOpenAI( modelos.getenv(MODEL_NAME), temperature0.2, ) resp llm.invoke(你好请回复连接成功。) print(resp.content)运行后如果输出“连接成功”或类似内容说明环境正常。这一步应该作为第一步检查点不要等到 Agent 跑不起来再排查模型配置。环境项推荐值说明Python3.10 - 3.12版本过老容易依赖冲突langchain最新稳定版不同小版本工具接口有差异langgraph最新稳定版节点和路由 API 变化较多操作系统Linux / macOS / WindowsWindows 注意路径和激活命令模型支持工具调用的模型传统补全模型无法使用bind_tools3. 用 LangChain 写第一个 Agent 雏形3.1 最小项目结构先搭建一个最简单但能运行的项目结构deepagent/ ├── .env ├── .venv/ ├── requirements.txt └── agent_chain.pyrequirements.txt可以直接写成依赖列表方便在别的机器上复现langchain langchain-openai langgraph langchain-mcp-adapters python-dotenv3.2 写一个带天气工具的 ReAct AgentDeepAgent 的第一步是用 LangChain 自带的create_react_agent写一个最小 Agent。这里的核心是模型通过 ReAct 风格的思考-行动-观察循环决定是否调用工具。import os from dotenv import load_dotenv load_dotenv() from langchain_openai import ChatOpenAI from langchain_core.tools import tool from langchain.agents import create_react_agent, AgentExecutor from langchain_core.prompts import PromptTemplate tool def get_weather(city: str) - str: 查询指定城市的天气city 使用中文城市名。 return f{city} 今天多云气温 18-26 摄氏度。 llm ChatOpenAI( modelos.getenv(MODEL_NAME, gpt-4o-mini), temperature0.2, ) tools [get_weather] prompt PromptTemplate.from_template( 你是 DeepAgent一个能查询天气的智能助手。\n 可用工具\n{tools}\n 工具名称\n{tool_names}\n 模型推理过程\n{agent_scratchpad}\n 用户问题{input} ) agent create_react_agent(llmllm, toolstools, promptprompt) executor AgentExecutor(agentagent, toolstools, verboseTrue) def main(): question 北京天气如何 result executor.invoke({input: question}) print(result[output]) if __name__ __main__: main()运行python agent_chain.py如果模型和工具调用正常日志里会看到类似Thought: 我需要查询北京的天气、Action: get_weather、Action Input: 北京这样的过程最终输出天气结果。3.3 这个雏形的问题在哪里AgentExecutor能跑通但有两个明显问题暴露在日志里一是流程逻辑藏在框架内部开发者只能通过 verbose 日志观察很难精确控制“下一步去哪”二是没有内置状态持久化用户多轮对话时需要另想办法维护历史否则第二轮问题“我刚才问的哪个城市”模型是不知道的。所以这个雏形的价值在于验证“模型 工具 AgentExecutor”这条链路能通并不适合直接作为生产代码。接下来用 LangGraph 重构正是为了解决这两个问题。4. 用 LangGraph 重构带状态的工作流4.1 StateGraph 的核心是 StateLangGraph 的StateGraph是一个带状态的图。图中的每个节点是一个函数函数输入当前 State返回一个 dictdict 中列出的字段会更新到下一轮 State。关键在于State 的字段可以定义 reducer最常见的 reducer 是add_messages它的作用不是覆盖消息列表而是把新消息追加进去。先定义一个 DeepAgent 的 Statefrom typing import Annotated, TypedDict from langgraph.graph.message import add_messages from langchain_core.messages import BaseMessage class AgentState(TypedDict): messages: Annotated[list[BaseMessage], add_messages]如果去掉Annotated节点每次返回{messages: [...]}都会把旧消息覆盖掉模型就看不到历史。这是新手最容易踩的坑。4.2 节点、边和条件路由LangGraph 的基础结构由节点、普通边和条件边组成。节点函数接收state返回更新内容。例如def call_agent(state: AgentState): response llm.invoke(state[messages]) return {messages: [response]}普通边用于固定方向条件边则根据状态决定下一步走哪个节点def route_after_agent(state: AgentState): last_message state[messages][-1] if getattr(last_message, tool_calls, None): return tools return end这里判断last_message.tool_calls。如果模型返回了这个字段说明它想调用工具就进入tools节点否则直接结束。4.3 用 LangGraph 写一个完整的工具调用 AgentDeepAgent 的 LangGraph 版本可以这样写from typing import Annotated, TypedDict from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langgraph.prebuilt import ToolNode from langchain_openai import ChatOpenAI from langchain_core.tools import tool from langchain_core.messages import BaseMessage, HumanMessage tool def get_weather(city: str) - str: 查询指定城市的天气city 使用中文城市名。 return f{city} 今天多云气温 18-26 摄氏度。 tools [get_weather] llm ChatOpenAI( modelgpt-4o-mini, temperature0.2, ).bind_tools(tools) class AgentState(TypedDict): messages: Annotated[list[BaseMessage], add_messages] def call_agent(state: AgentState): response llm.invoke(state[messages]) return {messages: [response]} def route_after_agent(state: AgentState): last_message state[messages][-1] if getattr(last_message, tool_calls, None): return tools return end graph StateGraph(AgentState) graph.add_node(agent, call_agent) graph.add_node(tools, ToolNode(tools)) graph.add_edge(START, agent) graph.add_conditional_edges(agent, route_after_agent, { tools: tools, end: END, }) graph.add_edge(tools, agent) app graph.compile() result app.invoke({ messages: [HumanMessage(content北京天气如何)] }) for msg in result[messages]: print(msg.type, msg.content)运行后消息列表会依次出现human北京天气如何ai带tool_calls的中间结果。tool天气工具返回的结果。ai最终的文案回答。这就是 LangGraph 的 Agent 循环用户消息进入 agent 节点模型决定调用工具进入 tools 节点工具结果作为新的消息再回到 agent 节点直到模型认为不需要工具为止。4.4 条件路由与循环的边界控制条件路由的强大之处在于可以控制循环次数和退出条件。上面的例子中route_after_agent返回end就会走到END返回tools就会进入工具执行。如果业务上需要限制最多调用 3 次工具可以在 State 里维护一个计数class AgentState(TypedDict): messages: Annotated[list[BaseMessage], add_messages] tool_call_count: int然后在call_agent节点里更新计数在条件路由里判断是否达到上限。这样就能避免模型陷入反复调用工具的循环。5. 给 Agent 接上 MCP 工具5.1 MCP 解决工具接入碎片化问题如果没有 MCP每接入一个外部服务基本都要写一套客户端代码调用 HTTP 接口、解析返回结构、转换错误格式、维护 token。工具少的时候还能接受工具一多Agent 代码会变得极其臃肿。MCP 的解决方案是统一协议。MCP Server 暴露能力MCP Client 发现并调用能力。Agent 程序不需要关心工具部署在哪、用的是什么语言只要通过 Client 拿到标准化 Tool 列表就行。DeepAgent 里接 MCP 的目的很简单把“天气查询”这个能力做成一个 MCP Server再让 LangGraph 里的 Agent 通过 MCP Client 使用它。5.2 用 FastMCP 写一个 MCP Server以下示例用 Python 快速实现一个 MCP Server文件名为mcp_server.pyfrom mcp.server.fastmcp import FastMCP mcp FastMCP(deepagent) mcp.tool() def get_weather(city: str) - str: 查询指定城市的天气。 return f{city}: 晴22 度 if __name__ __main__: mcp.run(transportstdio)这里transportstdio表示通过标准输入输出通信适合本地子进程启动。实际使用时FastMCP 的导入路径和run参数会随mcp包版本变化安装后可以先查看包内文档确认。5.3 在 LangGraph 中加载 MCP 工具langchain-mcp-adapters提供了 MCP Client 的 LangChain 适配层。常见用法是使用MultiServerMCPClient加载多个 MCP Server 的工具from langchain_mcp_adapters.client import MultiServerMCPClient async def load_mcp_tools(): async with MultiServerMCPClient( { weather: { command: python, args: [mcp_server.py], transport: stdio, } } ) as client: tools await client.get_tools() return tools加载回来的tools是标准的 LangChain Tool 列表可以直接传给ToolNodefrom langgraph.prebuilt import ToolNode mcp_tools await load_mcp_tools() tool_node ToolNode(mcp_tools)接入 MCP 之后原来的get_weather就不需要写在 Agent 代码里了工具的发现和调用逻辑统一由 MCP Client 管理。DeepAgent 后续增加企业通讯录、数据库查询、文档检索等能力时只需再启动对应 MCP Server然后注册到MultiServerMCPClient的配置里。5.4 MCP Server 的选择与安全边界MCP 支持两种常见传输方式。本地子进程服务使用stdio远程服务通常使用streamable_http或旧版sse。配置错误是最常见的接入失败原因。远程服务要特别注意网络、鉴权和超时问题不要给 Agent 暴露无权限限制的写操作工具。注意MCP 工具一旦被 Agent 加载模型就可能根据用户输入调用它。涉及删除、改密、支付、发送消息等敏感能力时必须在 MCP Server 或调用层增加权限校验和人工确认。6. 实现智能体记忆 Memory6.1 短期记忆用 Checkpointer 保存会话状态LangGraph 的短期记忆通过 checkpointer 实现。checkpointer 的作用是保存每一步 State这样同一个thread_id的多轮调用可以共享历史消息。先看内存版本from langgraph.checkpoint.memory import MemorySaver memory MemorySaver() app graph.compile(checkpointermemory) config {configurable: {thread_id: user-123}} # 第一轮 app.invoke( {messages: [HumanMessage(content我叫小明住在北京。)]}, config, ) # 第二轮 response app.invoke( {messages: [HumanMessage(content我叫什么住哪个城市)]}, config, ) print(response[messages][-1].content)由于两次调用使用同一个thread_idLangGraph 会把第一轮的 messages 保存下来并在第二轮的 State 中带给模型。如果使用不同thread_id两者之间互相隔离等于重新开始一段对话。MemorySaver只能用于学习和测试它把状态保存在进程内存中进程一退出就丢了。生产环境应该换成持久化的 checkpointer例如from langgraph.checkpoint.sqlite import SqliteSaver with SqliteSaver.from_conn_string(checkpoints.db) as checkpointer: app graph.compile(checkpointercheckpointer)SQLite 适合单机应用分布式场景通常使用 Postgres 或 Redis 等外部存储。实际包名和导入路径以安装版本为准。6.2 长期记忆不能只依赖对话历史短期记忆的问题是历史消息会越来越长很快超出一个模型的上下文窗口。而且它绑定thread_id用户换一个会话之前的偏好就不存在了。长期记忆需要把重要信息提取出来存到外部存储在后续会话中根据当前问题检索后注入 Prompt。DeepAgent 里可以设计这样一个简化流程retrieve_memory节点根据用户最新消息从向量库检索相关记忆。agent节点把检索到的记忆拼进 system prompt。store_memory节点在对话结束后对消息做摘要并写入向量库。向量库部分可以先用 FAISS 做本地示例from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import FAISS memory_store FAISS.from_texts( [用户住在北京, 用户希望回复使用中文], embeddingOpenAIEmbeddings(), )在agent节点里检索并注入记忆from langchain_core.messages import SystemMessage def call_agent_with_memory(state: AgentState): user_input state[messages][-1].content docs memory_store.similarity_search(user_input, k3) memory_context \n.join(doc.page_content for doc in docs) system SystemMessage( contentf以下是关于用户的历史记忆如果和当前问题相关请参考\n{memory_context} ) response llm.invoke([system] state[messages]) return {messages: [response]}这个写法展示了核心思路但并不是完整生产方案。生产环境还需要考虑记忆如何更新、旧记忆如何过期、多个用户之间的数据隔离、敏感信息如何处理。6.3 记忆写入的时机和一致性问题长期记忆写入不是“每轮都存原始对话”而是先做摘要再决定是否更新已有记忆。如果用户先说“我住在北京”隔天又说“我搬到上海了”系统不能同时保留两条冲突记忆需要更新或覆盖旧记录。一个可参考的时机是当一轮对话结束、Agent 完成最终回答后触发一个store_memory节点对最近的对话做一次摘要然后写入存储。如果对话没有完成或工具调用中途报错不建议写入记忆否则容易留下半截信息。注意不要把用户原始消息、工具返回结果里的大段文本直接塞进长期记忆。记忆粒度应该是“可复用的结论”例如用户偏好、常用术语、项目约束而不是完整聊天记录。7. 运行验证、日志和排错7.1 验证 Agent 是否真的走通了链路很多 Agent 项目“能启动”和“能正确工作”是两回事。DeepAgent 每次运行后建议先检查消息序列for msg in result[messages]: print(msg.type, msg.content)如果消息序列里没有tool类型说明模型根本没有调用工具如果消息序列里出现了多个ai和tool交替说明 Agent 在循环需要检查退出条件如果最终消息还是tool说明工具结果没有回到agent节点一般是边配置错误。还可以打印工具调用参数for msg in result[messages]: if getattr(msg, tool_calls, None): print(msg.tool_calls)这一步能确认模型生成的工具名、参数是否符合预期。7.2 常见问题现象和排查路径问题现象可能原因检查方式处理建议Agent 不调用工具模型不支持工具调用或工具描述不清楚打印 AIMessage 的tool_calls换支持 tool calling 的模型优化工具名称和描述messages 一直覆盖没有历史State 字段没有加 reducer检查TypedDict定义使用Annotated[list[BaseMessage], add_messages]第二轮不记得第一轮内容thread_id不一致或未编译 checkpointer检查 config 和 compile 参数使用同一个thread_id重新 compileMCP 工具加载不上传输方式配置错误单独运行 MCP Server 看日志确认stdio/streamable_http和 command、url 匹配上下文越来越长导致报错历史消息没有裁剪查看输入 token 数增加摘要或裁剪节点条件路由报 KeyError返回值没有对应分支名打印路由函数返回值确保映射 dict 里包含所有可能分支7.3 可复用的排错清单遇到 Agent 行为异常按这个顺序排查环境变量是否加载模型 API Key 是否有效。是否先跑通了最简单的llm.invoke。工具函数本身是否可执行是否抛出异常。状态字段是否按预期更新messages是否追加。条件路由函数返回的分支名是否存在于映射表。图中是否存在丢失的边例如tools是否回到agent。日志里是否有 HTTP 401、429、500 等模型服务错误。是否超过模型上下文窗口导致请求被拒。这个清单不只适用于 DeepAgent也适用于大多数 LangGraph 项目。先排除环境和输入问题再查逻辑和路由通常能快速定位问题。8. 学习环境和生产环境的差异8.1 学习环境优先跑通最小闭环学习阶段不要追求完整生产架构。Chekpointer 用MemorySaverMCP Server 用本地stdio向量库用本地文件日志直接 print。目标是让链路通、能观察中间状态、能看懂路由逻辑。学习环境的最小条件一个支持工具调用的模型。一个能跑通的最小工具。LangGraph 图中三个节点agent、tools、条件结束。一个thread_id固定的会话测试。8.2 生产环境必须补齐的能力生产环境和学习环境最大的区别不是代码能跑而是系统在异常情况下还能保持可控。生产环境至少要考虑以下几点能力学习环境生产环境CheckpointerMemorySaverSqliteSaver / Postgres / Redis记忆存储进程内列表或本地向量库Redis、数据库、企业级向量库模型配置.env手动维护配置中心、模型网关、多模型切换日志追踪print结构化日志、链路追踪、轨迹回放工具权限默认全部可调按用户、按角色、按操作类型控制异常处理直接抛错超时、重试、降级、人工兜底安全审计不关心记录调用者、工具参数、结果和费用模型调用和 MCP Server 调用都应该设置超时。Agent 节点如果一直不返回不能无限等待。条件路由里也要加上最大循环次数防止模型反复调用同一个工具。9. 最佳实践和下一步扩展9.1 可以直接落地的工程建议写 LangGraph Agent 时以下实践能明显降低维护成本State 字段要少而清晰。不要把任意对象都塞进 State能用messages管理对话历史就用单独字段管理业务状态。节点函数保持纯粹。不要在节点函数里依赖全局变量所有需要的数据尽量从 State 读取更新结果通过返回值写回。工具返回值尽量结构化。返回 JSON 字符串比返回散落的文本更容易被后续节点解析。工具描述写清楚。描述是模型决定是否调用工具的主要依据描述模糊会直接导致 Agent 不调用工具。使用last_message.tool_calls判断是否需要调用工具不要用字符串匹配content。长期记忆写入前先做摘要和查重避免保存冲突信息。每次修改图结构后先用固定输入做回归测试对比消息序列是否变化。9.2 从 DeepAgent 到产品化应用DeepAgent 只是一个起点。下一步可以按这个方向扩展用 FastAPI 封装 HTTP 接口把每次调用通过thread_id关联到具体用户。在 LangGraph 中引入子图把检索、问答、工具调用拆成独立模块。接入企业内的 MCP Server让 Agent 使用已有系统能力。增加 LangSmith 或其他 tracing 工具回放每一步 Agent 决策。对模型输出做校验敏感场景加人工审核环节。如果只记住一个关键点LangGraph 中真正决定 Agent 质量的是状态设计、路由控制和记忆策略不是模型参数调得多花哨。把最小图跑通再把节点、边、记忆一层层加上去Complexity 才不会失控。下一步建议把 DeepAgent 里的天气工具换成自己的业务工具先跑通一条主链路再增加分支和长期记忆逐步完善。
返回列表