ARTICLE DETAIL

资讯详情

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

LangChain+MCP+LangGraph+Agent:从零构建可控AI Agent应用

LangChain+MCP+LangGraph+Agent:从零构建可控AI Agent应用 这次我们来看一套 2026 年依然会继续演进的技术组合LangChain MCP LangGraph Agent。很多人学 AI 大模型应用开发时先被 LangChain 的 API 绕晕然后又遇到 LangGraph 的状态图再碰到 MCP 协议最后还要搞懂 Agent 到底怎么规划工具调用。这套教程不讲概念堆砌直接告诉你这几个工具分别是干什么的、它们之间怎么配合、怎么写一个能真正跑起来的 Agent以及上线前你需要注意什么。先说结论如果你只想快速调一个模型 API那不需要这些框架直接写requests就行。但一旦你的任务变成“让模型自主调用多个工具”“让流程可控可回滚”“让外部服务通过统一协议接入 Agent”LangChain、LangGraph 和 MCP 就是当前最主流的落地方案。这套组合的核心特点是模型负责理解任务LangChain 负责封装调用LangGraph 负责流程控制MCP 负责让 Agent 能接入各种外部工具。本文会带你完成四件事第一理解 Agent、LangChain、LangGraph、MCP 四者之间的关系第二搭建一套本地开发环境第三从零写一个能调用工具、带状态流转的 Agent 示例第四把它封装成 API 服务并验证批量调用。文章后面还会给出一份常见报错排查清单建议先收藏再看。1. 核心能力速览先给一张总表方便你快速判断这套技术栈能做什么、不能做什么。能力项说明技术类型AI 大模型应用开发框架与协议核心组件LangChain、LangGraph、MCP、Agent主要功能模型调用封装、工具调用、多步骤流程编排、外部服务接入、批量任务运行环境Python 3.9建议 3.10/3.11硬件要求调用云端 API 时无特殊 GPU 要求本地模型推理需按模型规格配置显卡显存占用取决于所接模型纯框架本身几乎不占显存支持平台Windows / macOS / Linux是否支持 API支持可封装为 FastAPI 服务是否支持批量任务支持可通过循环、队列或 LangGraph 批量执行适合场景智能客服、文档分析、代码辅助、数据分析、自动化流程编排不适合场景极简单次调用、对延迟极其敏感的超高频接口这套技术栈的定位是“编排层”不是“模型层”。你仍然需要一个大模型作为推理核心模型可以是 OpenAI、Anthropic、通义千问、DeepSeek 等云端 API也可以是本地部署的 Qwen、Llama 等开源模型。2. 四者关系LangChain、LangGraph、MCP、Agent 到底是什么很多新手最大的困惑不是某个框架学不会而是不知道这几个东西怎么区分。这里用一句话概括它们的定位。2.1 LangChain工具集与胶水层LangChain 是最早火起来的大模型应用开发框架它提供了一整套封装模型调用接口、Prompt 模板、输出解析器、文档加载器、向量存储封装、各种第三方工具集成。它的核心价值是把“模型调用”和“周边操作”统一成一套 API让你不必每次重新封装请求参数、处理返回格式。不过 LangChain 的问题也很明显它功能太多抽象层不少初学者经常为了一个简单功能引入大量依赖。我的建议是把它当“工具箱”用不需要所有模块都学用到哪个查哪个。2.2 LangGraph把 Agent 流程变成图LangGraph 是 LangChain 团队推出的流程编排框架定位是构建有状态、可控制的 Agent。它的核心概念是图节点Node负责执行具体操作边Edge决定流转方向状态State在节点之间传递。为什么有了 LangChain 还需要 LangGraph因为 LangChain 原生 Agent 的循环控制是封装好的你很难干预它的中间过程。LangGraph 把控制权交还给开发者你可以指定模型先做什么、后做什么、什么条件下重试、什么条件下结束。对于生产级应用来说这种可控性非常关键。LangChain 和 LangGraph 的区别可以这样理解LangChain 提供积木LangGraph 提供搭积木的图纸和流程。实际工程中二者经常配合使用。2.3 MCPAgent 接入外部服务的统一协议MCPModel Context Protocol是 Anthropic 提出的开放协议目的是让 AI 应用以统一方式接入外部工具和数据源。你可以把 MCP 理解为 AI 领域的 USB 接口模型侧和工具侧只要都支持 MCP 协议就能即插即用。在实际开发中MCP 解决了一个非常实际的问题之前每个 Agent 要接入一个外部系统都要单独写一套工具调用逻辑现在工具方把能力封装成 MCP ServerAgent 侧通过 MCP Client 直接发现并调用这些工具代码复用率和接入效率都会明显提升。2.4 Agent真正的决策者与执行者Agent 不是某个具体框架而是一种应用形态。它的工作方式是接收用户目标由大模型判断需要调用哪些工具执行工具调用后观察结果再决定下一步操作直到完成目标或达到终止条件。一个标准的 Agent 循环包括四步模型分析任务决定是否需要调用工具。如果调用则生成结构化工具调用参数。程序执行工具并把结果返回给模型。模型根据新信息继续推理或给出最终答案。这套循环可以先用 LangChain 快速实现流程控制升级后用 LangGraph 重构工具接入复杂后用 MCP 统一管理。3. 环境准备与前置条件开始写代码之前先确认你的机器满足基本条件。3.1 环境清单检查项要求说明Python3.9 以上建议 3.10 或 3.11兼容性更好包管理工具pip / uvuv 速度更快本文示例使用 uv模型 API Key按需需要准备一个可调用的大模型 API网络环境能访问模型 API国内可直接使用国内厂商服务Git可选用于拉取示例代码3.2 创建项目环境使用 uv 创建虚拟环境会更清爽。如果本机没有 uv先安装pip install uv然后创建项目并激活环境uv venv .venv source .venv/bin/activate # Linux / macOS .venv\Scripts\activate # Windows安装核心依赖uv pip install langchain langchain-openai langgraph mcp fastapi uvicorn python-dotenv需要注意LangChain 系列包的版本迭代很快安装时建议锁定版本。实际安装版本以 PyPI 当前最新稳定版为准不要照抄早期教程里的旧版本 API。将模型 API Key 写入.env文件MODEL_API_KEYyour_api_key_here MODEL_BASE_URLhttps://your-model-provider.com/v1 MODEL_NAMEyour-model-name也可以直接在代码中通过环境变量读取。4. Agent 基础实战模型 工具 Agent先实现一个最简单的 Agent让它具备“调用计算器工具”的能力。这一步不引入 MCP只用 LangChain 的基础能力目的是让你先看懂 Agent 循环的底层逻辑。4.1 定义一个自定义工具from langchain_core.tools import tool tool def add(a: float, b: float) - float: 计算两个数字的和。 return a b tool def multiply(a: float, b: float) - float: 计算两个数字的乘积。 return a * b tools [add, multiply]这里使用了tool装饰器LangChain 会自动把函数名、docstring、参数类型转换成模型可识别的工具描述。docstring 写清楚很重要因为模型会阅读这段文字来判断“这个工具是干什么用的”。4.2 创建模型并绑定工具import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI load_dotenv() llm ChatOpenAI( modelos.getenv(MODEL_NAME), api_keyos.getenv(MODEL_API_KEY), base_urlos.getenv(MODEL_BASE_URL), temperature0, ) llm_with_tools llm.bind_tools(tools)如果你的模型支持 OpenAI 兼容的 tools 接口这段代码就适用。国内多家云厂商都提供了 OpenAI 兼容端点配置好base_url即可不需要改代码。4.3 手动执行 Agent 循环先不用高层 Agent 封装手动模拟一次完整的 Agent 循环from langchain_core.messages import HumanMessage, AIMessage, ToolMessage messages [HumanMessage(content3 乘以 4然后加上 5结果是多少)] # 第一步模型判断需要调用工具 response llm_with_tools.invoke(messages) messages.append(response) print(模型响应:, response) # 第二步解析工具调用并执行 for tool_call in response.tool_calls: tool_name tool_call[name] tool_args tool_call[args] selected_tool {add: add, multiply: multiply}[tool_name] tool_result selected_tool.invoke(tool_args) messages.append(ToolMessage(contentstr(tool_result), tool_call_idtool_call[id])) # 第三步模型基于工具结果生成最终答案 final_response llm_with_tools.invoke(messages) print(最终答案:, final_response.content)这个例子展示了 Agent 最核心的三步模型决定调用工具、程序执行工具、模型解读工具结果。理解这三步后面用 LangGraph 和 MCP 都是在这个基础上扩展。4.4 使用 LangChain 内置 Agent手动循环便于理解实际开发可先用 LangChain 的预置 Agent 快速验证from langchain.agents import create_tool_calling_agent from langchain.agents import AgentExecutor from langchain_core.prompts import ChatPromptTemplate prompt ChatPromptTemplate.from_messages([ (system, 你是一个擅长使用工具的助手。), (placeholder, {chat_history}), (human, {input}), (placeholder, {agent_scratchpad}), ]) agent create_tool_calling_agent(llm, tools, prompt) agent_executor AgentExecutor(agentagent, toolstools, verboseTrue) result agent_executor.invoke({input: 365 乘以 24 等于多少}) print(result[output])这里用create_tool_calling_agent创建了一个工具调用型 Agent再用AgentExecutor执行。verboseTrue可以打印中间过程方便调试。5. MCP 实战让 Agent 即插即用外部工具第 4 节的工具是直接在代码里定义的属于“写死”的本地工具。真实业务中工具往往由独立团队维护比如查询订单系统、调用搜索服务、读取公司内部数据库。如果每个 Agent 都自己实现一遍接入逻辑维护成本会很高。MCP 的思路是统一协议工具提供方开发 MCP ServerAgent 侧通过 MCP Client 连接。下面用一个计算器服务为例演示 MCP Server 的编写与接入方式。5.1 使用 FastMCP 创建 MCP ServerFastMCP 是 MCP 官方 Python SDK 提供的高层封装写起来很简洁# mcp_calculator_server.py from mcp.server.fastmcp import FastMCP mcp FastMCP(calculator) mcp.tool() def add(a: float, b: float) - float: 计算两个数字的和。 return a b mcp.tool() def multiply(a: float, b: float) - float: 计算两个数字的乘积。 return a * b if __name__ __main__: mcp.run(transportstdio)这个 Server 通过 stdio 传输运行也就是标准输入输出。很多 AI 编程工具里的 MCP 插件用的就是这种模式。5.2 MCP Client 接入 LangChainLangChain 官方提供了 MCP 适配工具可以读取 MCP Server 暴露的工具列表并转换成 LangChain 的工具对象。这里以langchain-mcp-adapters为例from langchain_mcp_adapters.tools import load_mcp_tools from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client server_params StdioServerParameters( commandpython, args[mcp_calculator_server.py], ) 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) print(已加载工具:, [tool.name for tool in tools])这里涉及异步代码。如果是在 Jupyter 或普通脚本中运行需要借助asyncio.run()或把逻辑放进 async 函数。5.3 将 MCP 工具接入 Agent加载到的tools是标准 LangChain 工具对象列表可以直接传给AgentExecutorfrom langchain.agents import create_tool_calling_agent, AgentExecutor from langchain_core.prompts import ChatPromptTemplate prompt ChatPromptTemplate.from_messages([ (system, 你是一个擅长使用外部工具的助手。), (human, {input}), (placeholder, {agent_scratchpad}), ]) agent create_tool_calling_agent(llm, tools, prompt) agent_executor AgentExecutor(agentagent, toolstools, verboseTrue) result await agent_executor.ainvoke({input: 请计算 12 乘以 8 再乘以 3}) print(result[output])从 Agent 侧看它只知道自己有一批工具可用并不关心工具背后是本地函数还是 MCP 服务。这就是 MCP 协议带来的清晰边界接入一个 MCP Server就像插上一个统一接口的 USB 设备。5.4 MCP 在企业场景中的想象空间在企业内部MCP 的价值更大。搜索服务可以做成一个 MCP Server数据库查询可以做成一个 MCP Server企业内部 API 网关也可以暴露成 MCP Server。Agent 需要哪个能力就连接哪个 Server不需要为每个新系统单独开发 Agent 端代码。但也要注意安全边界MCP Server 一旦被 Agent 调用就相当于给了模型一个操作真实系统的入口。必须对工具做权限管控确认模型只能调用它应该调用的工具不能越权。6. LangGraph 实战把 Agent 流程变成可控图第 4 节的 Agent 虽然能跑但流程是黑盒。如果想在中间插入人工审核节点或者让模型在结果不满足条件时重试原生 AgentExecutor 扩展起来并不方便。LangGraph 的图结构天然适合这种需求。6.1 核心概念概念说明State在节点之间传递的状态对象类似全局变量Node具体执行单元输入输出都是 StateEdge节点之间的连线决定流程方向Conditional Edge条件分支根据状态决定下一步走向Graph把节点和边组装成可执行的图6.2 实现一个带人工审核的 Agent下面这个示例实现一个简单的生成 → 审阅 → 修正/通过的流程。模型生成一段文本后模拟人工审阅如果文本中检测到敏感词就进入修正节点否则直接通过。from typing import TypedDict, Literal from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI class AgentState(TypedDict): input: str output: str needs_revision: bool llm ChatOpenAI( modelos.getenv(MODEL_NAME), api_keyos.getenv(MODEL_API_KEY), base_urlos.getenv(MODEL_BASE_URL), ) def generate_node(state: AgentState) - AgentState: 根据输入生成文本。 response llm.invoke(f请写一段简短的产品介绍{state[input]}) return {output: response.content} def review_node(state: AgentState) - AgentState: 模拟人工审阅包含特定词则标记为需要修正。 output state[output] sensitive_words [绝对, 最好, 第一] state[needs_revision] any(word in output for word in sensitive_words) return state def revise_node(state: AgentState) - AgentState: 修改文本去掉绝对化表达。 revised llm.invoke(f请改写下面这段文本去掉绝对化表达{state[output]}) return {output: revised.content, needs_revision: False} def should_continue(state: AgentState) - Literal[revise, end]: 条件分支决定是否需要修正。 if state[needs_revision]: return revise return end # 构建图 graph StateGraph(AgentState) graph.add_node(generate, generate_node) graph.add_node(review, review_node) graph.add_node(revise, revise_node) graph.set_entry_point(generate) graph.add_edge(generate, review) graph.add_conditional_edges(review, should_continue, { revise: revise, end: END, }) graph.add_edge(revise, review) app graph.compile()执行这个图result app.invoke({input: 这是一款效率很高的编程IDE}) print(result[output])这个示例演示了 LangGraph 相比原生 Agent 的关键优势你可以在review_node中插入任何自定义逻辑比如人工审批接口、内容审核服务、业务规则校验。流程不是黑盒而是每个节点都清清楚楚便于日志记录、问题定位和流程变更。6.3 在 LangGraph 中调用工具LangGraph 的节点里也可以调用工具。通常做法是把工具调用封装在一个节点中模型在该节点内完成“决定调用 → 执行 → 返回结果”的循环。官方也提供了create_react_agent这类预置封装可以在 LangGraph 中使用 ReAct 模式from langgraph.prebuilt import create_react_agent agent create_react_agent( modelllm, tools[add, multiply], state_modifier你是一个计算助手请使用工具完成计算。, ) result agent.invoke({messages: [(human, 123 乘以 456 等于多少)]}) print(result[messages][-1].content)create_react_agent的优点是既保留了 LangGraph 的可控结构又不用自己手写工具调用循环。你可以在此基础上继续添加节点实现更复杂的流程。7. 接口 API 与批量任务Agent 在本地跑通只是第一步工程化的下一步是把它封装成 API 服务供业务系统调用。7.1 用 FastAPI 封装 Agent 服务下面提供一个最小可运行的 Agent API 服务示例# app.py import os from dotenv import load_dotenv from fastapi import FastAPI from pydantic import BaseModel from langchain_openai import ChatOpenAI from langchain_core.tools import tool from langgraph.prebuilt import create_react_agent load_dotenv() app FastAPI(titleAgent Service) tool def add(a: float, b: float) - float: 计算两个数字的和。 return a b tool def multiply(a: float, b: float) - float: 计算两个数字的乘积。 return a * b llm ChatOpenAI( modelos.getenv(MODEL_NAME), api_keyos.getenv(MODEL_API_KEY), base_urlos.getenv(MODEL_BASE_URL), ) agent create_react_agent(modelllm, tools[add, multiply]) class AgentRequest(BaseModel): message: str class AgentResponse(BaseModel): result: str app.post(/agent, response_modelAgentResponse) async def run_agent(request: AgentRequest): result await agent.ainvoke({ messages: [(human, request.message)] }) return AgentResponse(resultresult[messages][-1].content) if __name__ __main__: import uvicorn uvicorn.run(app, host127.0.0.1, port8000)启动服务uvicorn app:app --host 127.0.0.1 --port 80007.2 使用 curl 验证 APIcurl -X POST http://127.0.0.1:8000/agent \ -H Content-Type: application/json \ -d {message: 计算 156 乘以 89 的结果}预期返回 JSON 结构类似{ result: 156 乘以 89 的结果是 13884。 }7.3 Python 批量任务调用批量调用不需要改变服务端代码只需要在客户端循环请求。但要注意控制并发避免触发模型 API 限流。import asyncio import httpx async def call_agent(client: httpx.AsyncClient, message: str) - dict: resp await client.post( http://127.0.0.1:8000/agent, json{message: message}, timeout120, ) resp.raise_for_status() return {input: message, output: resp.json()[result]} async def batch_call(messages: list[str], concurrency: int 5): async with httpx.AsyncClient() as client: sem asyncio.Semaphore(concurrency) async def bounded_call(message: str): async with sem: return await call_agent(client, message) results await asyncio.gather(*[bounded_call(m) for m in messages]) return results if __name__ __main__: messages [ 计算 11, 计算 2*3, 计算 10*10, 计算 256/4, ] results asyncio.run(batch_call(messages, concurrency2)) for r in results: print(r)批量任务在工程上要注意三点一是并发数要低于模型 API 的速率限制二是每条任务要有唯一 ID方便追踪三是失败任务要支持重试推荐加入指数退避策略。7.4 批量任务失败重试设计一个简单的重试示例import time def call_with_retry(func, retries3, backoff1.0): for attempt in range(retries): try: return func() except Exception as e: if attempt retries - 1: raise sleep_time backoff * (2 ** attempt) print(f调用失败{sleep_time} 秒后重试{e}) time.sleep(sleep_time)8. 资源占用与性能观察很多读者会关心这套技术栈跑起来占多少资源。这里需要明确区分“框架资源”和“模型资源”。8.1 框架层资源占用LangChain、LangGraph、MCP 本身只是 Python 代码CPU 占用很低内存通常在几百 MB 级别取决于加载的工具和依赖数量。框架本身不占用 GPU 显存。8.2 模型层资源占用显存占用完全取决于你使用哪种模型调用云端 API本机几乎没有显存需求。本地部署 7B 量化模型通常需要 6G 以上显存。本地部署 14B 量化模型通常需要 12G 以上显存。本地部署更大参数模型需要更高配置。实际占用需以模型规格、量化方式和推理框架为准。建议首次测试时先用小模型或云端 API 跑通流程再根据需求升级模型。8.3 性能观察方法观察资源占用可以使用# 实时查看 CPU 和内存 top # 查看 GPU 使用情况 nvidia-smi -l 28.4 影响性能的关键因素因素影响模型推理速度Agent 整体延迟的主要来源工具调用次数每多一次工具调用就多一次模型推理上下文长度历史消息越长首字延迟越高并发请求数同时请求过多会触发 API 限流工具执行时间如果工具是外部 HTTP 服务网络延迟会叠加降低延迟的常用思路减少不必要的工具调用、缩短历史消息、使用更快的模型、对工具结果做缓存。9. 常见问题与排查方法下面这套排查清单是从实际开发经验中整理的遇到问题先对照表格定位方向。问题现象可能原因排查方式解决方案安装依赖失败Python 版本过低或包版本冲突检查 Python 版本查看报错信息升级到 Python 3.10锁定依赖版本模型返回 401 错误API Key 错误或环境变量未加载打印环境变量确认.env文件位置检查 Key 是否正确确认load_dotenv()已调用模型返回 404 错误base_url路径不对查看模型服务文档确认端点是否为/v1结尾Agent 不调用工具模型不支持 tools 接口或未绑定工具检查模型是否支持 function calling打印绑定后的消息更换支持工具的模型确认bind_tools已生效MCP 连接失败命令或参数配置错误路径不正确单独命令行启动 MCP Server检查报错确认 command 和 args 能正常启动 ServerLangGraph 图死循环条件分支没有通向 END打印每次节点状态检查分支逻辑增加最大步数限制修正条件分支API 请求超时模型推理慢或工具执行慢逐段计时查看日志增大 timeout优化提示词减少上下文批量任务卡住并发过高触发限流查看服务端日志确认限流状态降低并发数加入重试机制端口被占用其他进程占用了 8000 端口查看端口占用情况更换端口或停止占用进程端口占用排查命令# Windows netstat -ano | findstr :8000 # Linux / macOS lsof -i :8000如果 MCP Server 调试不便可以先单独运行 Server用官方客户端工具验证工具是否正常暴露再接入 LangChain。10. 最佳实践与合规边界这套技术栈虽然开发效率高但工程化落地时有一些经验值得提前知道。10.1 工程实践建议第一第一次接入新模型时先用最小脚本验证模型能正常响应再接入 Agent 框架。这样可以隔离问题是模型本身的问题还是框架封装的问题。第二LLM 调用建议统一走一层封装比如写一个get_llm()工厂函数环境变量统一管理。后续切换模型时只需要改配置不用改业务代码。第三LangGraph 的状态对象建议定义得精简只放必要字段。状态越长每次节点间传递的序列化开销越大调试时也不容易看清问题。第四Agent 日志要记录完整消息链用户输入、模型中间推理、工具调用参数、工具返回结果、最终输出。线上问题排查时这些信息缺一不可。第五批量任务要设计成可重入的。也就是说同一个任务重复执行多次不能产生重复副作用比如重复扣费、重复发送消息。10.2 合规与安全边界涉及 AI 大模型应用开发必须强调安全合规API Key 不要硬编码到代码里不要提交到公共仓库建议使用环境变量或密钥管理服务。Agent 接入外部工具时要做权限最小化设计。只给模型必要的工具防止越权操作。如果 Agent 接入公司内部系统必须设置身份认证和操作审计。如果工具会调用第三方服务或产生实际业务动作必须增加人工确认环节。涉及用户数据的场景要遵守数据隐私相关法规明确数据用途和存储边界。MCP 作为一个全新的工具接入协议安全模型还处于快速演进阶段。接入第三方 MCP Server 时要像评估第三方依赖一样谨慎确认 Server 的代码来源可信工具行为可审计。11. 总结与下一步这套 LangChain MCP LangGraph Agent 的组合解决了大模型应用开发中的三个核心问题模型接入繁琐、流程控制不透明、外部工具难复用。实际项目中的定位是LangChain 提供工具集LangGraph 负责流程控制MCP 统管外部工具接入Agent 作为最终的业务入口。建议你按这个顺序上手先用 LangChain 写一个带工具调用的简单 Agent理解模型决策与工具执行的循环然后把一个 MCP Server 接入 Agent理解协议的价值接着用 LangGraph 把流程改造成可控的图加上条件和人工审核节点最后封装成 API 服务用批量脚本验证稳定性。最容易踩的坑有三个一是模型不支持 tools 调用却硬套 Agent 框架二是 MCP Server 路径配置错误导致连接失败三是 LangGraph 条件分支没有终止条件导致死循环。只要把这三个问题提前想清楚整个学习过程会顺畅很多。后面可以继续扩展的方向包括引入记忆机制让 Agent 支持多轮对话接入向量数据库实现 RAG 检索增强使用可观测性平台追踪 Agent 运行轨迹以及用自动评测集验证 Agent 在不同场景下的效果。先把这篇文章里的最小闭环跑通再逐步加复杂度这条路比一开始就追求大而全要稳妥得多。
返回列表