自从大模型应用从“单轮问答”走向“多步自主执行”之后身边经常能看到这样三类问题有人把 LangChain 当成 Agent 的全部图省事把所有逻辑堆在一个 Chain 里结果稍微一复杂就开始失控有人知道 LangGraph 能编排流程却不知道它和 LangChain 的分工边界到底在哪还有人一听 MCP 就以为是又一个新框架其实它的定位非常清晰就是一套“Agent 与外部系统”之间的标准通信协议。本文不打算罗列概念而是围绕一条企业级项目落地链路来写先用 LangChain 处理模型调用、提示词和知识库检索再用 LangGraph 负责有状态、可循环、可分支的 Agent 主流程最后用 MCP 把订单、库存、数据库等企业内部系统安全地暴露给 Agent。标题里的 DeepAgent本质上也可以理解为企业对“Agent 运行时”的封装命名核心底座仍然是这套组合。本文适合有 Python 基础、想进入 AI Agent 开发或准备改造现有业务的开发者内容会从环境搭建一直写到可运行的项目代码和排错手册。1. 为什么企业级 AI Agent 开发绕不开这三件套1.1 从 Chatbot 到 Agent问题发生了什么变化传统的 LLM 应用典型形态是“用户提问题 → 模型生成答案”。优点是链路短、好调试缺点也很明显模型只能凭训练数据回答问题没法查实时订单没法查公司内部知识库更没法代表用户去执行一个完整的多步任务。Agent 的形态则完全不同。它把模型当成“调度大脑”让模型自己决定下一步该做什么是直接回答还是调用订单查询工具还是先去知识库检索资料。这个循环可以反复执行直到模型认为已经拿到足够信息才生成最终结果。所谓的 Agent本质上就是“模型 工具 流程控制 记忆”的组合。一旦进入这种模式开发复杂度会明显上升流程分支变多状态需要维护工具调用会失败模型可能会陷入死循环。如果还用简单的 Chain 去堆代码很快就会变成一锅粥。这时候LangGraph 这类偏流程编排的框架就变得非常必要。1.2 LangChain、LangGraph、MCP 分别解决什么问题LangChain 解决的是“模型应用组装”的问题。它把模型、提示词、输出解析、向量检索、工具调用等能力封装成统一组件让开发者可以快速拼出一条可用的 RAG 或工具调用链路。LangGraph 解决的是“流程编排”的问题。它把应用建模成一张状态图图中的每个节点是一个函数或模块边负责定义节点之间的流转关系。它的核心特性是支持条件分支、循环、子图、并行节点、状态持久化以及人工介入。对于 Agent 这种天然“循环执行”的场景状态图比链式结构更合适。MCP 解决的是“Agent 与外部系统如何通信”的问题。它的中文含义是模型上下文协议全称 Model Context Protocol。在 MCP 架构下自己的系统被封装成 MCP ServerAgent 作为 MCP Client 通过标准协议访问工具、资源和提示词。这样企业内部系统不需要为每个 Agent 框架单独开发适配层只要实现一次 MCP 协议就能被 LangChain、LangGraph 以及其他支持 MCP 的客户端复用。1.3 几个容易混淆的概念对比初学阶段最容易混淆 LangChain、LangGraph、MCP 和 Agent 的区别。这里给出一张对比表方便快速建立坐标系。技术定位典型作用LangChainLLM 应用开发框架统一模型、提示词、检索、工具等基础组件LangGraphAgent 流程编排框架用图结构组织状态流转、分支、循环、人工审核MCP外部能力接入协议标准化连接数据库、订单系统、SaaS 工具等Agent应用形态由模型驱动决策自主完成多步任务还有一个经常被提到的概念是 Agent Skill。如果把 Agent 比作一个员工Agent Skill 就是员工掌握的“技能包”比如“如何写 SQL”“如何做日报分析”。MCP 更像是“如何连接外部设备”的标准接口。两者不是互斥关系企业落地时可以同时建设Skill 负责沉淀可复用的能力MCP 负责把工具能力标准化暴露出来。2. 环境准备与项目初始化2.1 Python 环境与依赖安装本文示例基于 Python 3.10 及以上版本实际项目中建议使用虚拟环境隔离依赖。以下是安装步骤。如果你使用的是 Windows激活命令会稍有不同。mkdir deep-agent-demo cd deep-agent-demo python -m venv .venv # macOS / Linux source .venv/bin/activate # Windows PowerShell # .venv\Scripts\Activate.ps1在项目根目录创建requirements.txt写入以下依赖。langchain langchain-openai langchain-community langchain-text-splitters langgraph langchain-mcp-adapters mcp faiss-cpu python-dotenv然后安装依赖。pip install -r requirements.txt这里要特别说明一点LangChain、LangGraph 以及 MCP Python SDK 的迭代速度非常快版本号经常变化。文章不写死固定版本号建议直接安装最新稳定版。如果后续遇到 API 报错优先检查 Python 环境中安装的版本再对照官方文档调整写法。2.2 项目目录规划企业项目里目录结构直接影响维护成本。下面的结构适合中等规模的 Agent 项目既能让知识库、MCP 服务、Agent 核心逻辑彼此独立也方便后续替换组件。deep-agent-demo/ ├── .env ├── requirements.txt ├── app/ │ ├── __init__.py │ ├── config.py │ ├── build_index.py │ ├── rag.py │ ├── mcp_server.py │ └── agent.py └── data/ ├── docs/ │ └── company_policy.txt └── faiss_index/其中app/rag.py负责知识库加载与向量索引app/mcp_server.py对外提供订单查询工具app/agent.py是 LangGraph Agent 主流程app/build_index.py用于构建本地向量索引。2.3 模型配置与 .env在项目根目录创建.env文件写入模型相关的敏感配置。注意不要把.env提交到 Git 仓库。OPENAI_API_KEY你的_API_Key OPENAI_BASE_URLhttps://api.openai.com/v1 OPENAI_MODEL_NAMEgpt-4o-mini如果你的团队使用国内大模型服务只要服务商提供 OpenAI 兼容接口就可以把OPENAI_BASE_URL替换成对应地址并把OPENAI_MODEL_NAME改成实际模型名。整个示例代码不需要改动只需要保证环境变量正确即可。3. LangChain 与 LangGraph 核心机制拆解3.1 LangChain模型、提示词与链式调用LangChain 最常用的几个基础能力包括模型封装、提示词模板、输出解析、向量检索和工具封装。先看一个最简单的调用示例。from dotenv import load_dotenv load_dotenv() from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate llm ChatOpenAI(modelgpt-4o-mini, temperature0) prompt ChatPromptTemplate.from_messages([ (system, 你是一个企业知识库助手请根据给定的上下文回答问题。), (human, 上下文{context}\n问题{question}), ]) chain prompt | llm resp chain.invoke({ context: 公司名称叫智联未来科技主要业务是智能制造。, question: 公司主要做什么业务, }) print(resp.content)这里用|把提示词模板和模型组合成一条链这是 LangChain 新版本推荐的链式写法。实际项目中链式调用适合那些步骤固定的场景比如“检索上下文 → 生成回答”。一旦出现分支和循环建议进入 LangGraph 或者把流程拆成更小的 Agent 步骤。3.2 LangGraph用状态图编排 AgentLangGraph 的核心抽象是状态图。一个图包含多个节点节点之间通过边连接。每次执行时状态对象会在节点之间传递节点的返回值会更新状态。下面是最小示例。from typing import TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): messages: list def node_a(state: State) - State: print(执行节点 A) return {messages: state[messages] [A]} def node_b(state: State) - State: print(执行节点 B) return {messages: state[messages] [B]} graph StateGraph(State) graph.add_node(a, node_a) graph.add_node(b, node_b) graph.add_edge(START, a) graph.add_edge(a, b) graph.add_edge(b, END) app graph.compile() result app.invoke({messages: []}) print(result[messages])运行结果会依次输出“执行节点 A”“执行节点 B”最终messages是[A, B]。这个示例虽然简单但它揭示了 LangGraph 的核心思想状态是全局共享的每个节点只关心输入状态和输出状态节点之间不直接依赖而是通过图结构连接。3.3 条件路由与分支控制真实项目里Agent 经常需要判断“这个问题该走哪条路”。LangGraph 使用条件边来实现分支。条件边会根据当前状态返回一个字符串字符串决定下一步进入哪个节点。def route_by_intent(state: State) - str: if state.get(need_tool): return tool return end graph.add_conditional_edges( judge, route_by_intent, { tool: tool_node, end: END, }, )这段代码的含义是当judge节点执行完成后系统调用route_by_intent函数根据函数的返回值决定下一个节点。这种设计非常适合意图识别、异常兜底、人工审核等场景。比如判断用户是否需要查询订单如果需要就进入工具节点否则直接结束。3.4 长期记忆与状态持久化Agent 进入生产环境后记忆会成为一个关键问题。短期记忆比较容易处理直接把聊天记录放进messages列表即可。长期记忆则需要借助 LangGraph 的持久化能力比如MemorySaver或数据库 Checkpoint。from langgraph.checkpoint.memory import MemorySaver checkpointer MemorySaver() app graph.compile(checkpointercheckpointer)使用 Checkpoint 之后每次运行的状态都会被保存下来后续可以通过线程 ID 恢复历史状态。这在多轮对话场景中非常有用。不过要注意记忆不是越多越好过长的历史会浪费 Token还会降低模型对关键信息的注意力。建议结合摘要记忆和向量检索只保留对当前任务有意义的上下文。4. MCPAgent 接入企业系统的标准协议4.1 MCP 是什么MCP 由 Anthropic 提出但它已经发展成一种开放的协议标准。它的设计思路可以类比成“AI 世界的 USB-C 接口”鼠标、键盘、显示器各自负责不同功能但都通过同一种接口连接电脑。MCP 里外部系统是 MCP ServerAgent 应用是 MCP Client两者通过 JSON-RPC 协议通信。一个 MCP Server 可以暴露三类能力Tools可被模型调用的工具函数比如查询订单、创建工单。Resources可读取的数据资源比如文件、数据库记录。Prompts可复用的提示词模板。对企业来说MCP 的最大价值是“一次接入多处复用”。如果公司已经有订单、库存、CRM 等多个系统只需要为每个系统实现一个 MCP Server就能被多种 Agent 框架复用不需要为每个项目单独写一套集成代码。4.2 用 FastMCP 快速实现一个 MCP Server这里以“订单查询”为例创建一个标准的 MCP Server。使用 FastMCP 可以让代码非常简洁。# app/mcp_server.py from mcp.server.fastmcp import FastMCP mcp FastMCP(order-service) mcp.tool() def query_order(order_id: str) - str: 查询订单状态返回订单当前进度。 Args: order_id: 订单号例如 SO20260101001 # 真实项目中这里应替换为数据库查询或订单系统 HTTP 接口 order_db { SO20260101001: {status: 已发货, logistics: 顺丰 SF123456}, SO20260101002: {status: 待付款}, } order order_db.get(order_id) if order: return f订单 {order_id} 当前状态{order[status]}物流信息{order[logistics]} return f未查询到订单 {order_id} if __name__ __main__: mcp.run()重点看两处。第一mcp.tool()把这个函数暴露成一个工具。第二函数必须有完整的 docstring尤其是参数说明。因为模型在决定是否调用工具时主要依赖函数名、参数名和文档说明来判断用途。4.3 LangChain 客户端加载 MCP 工具MCP Server 写好后需要一个客户端来连接。langchain-mcp-adapters是 LangChain 官方提供的适配层可以把 MCP 工具转成 LangChain 工具。import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from langchain_mcp_adapters.tools import load_mcp_tools async def main(): server_params StdioServerParameters( commandpython, args[app/mcp_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) for t in tools: print(加载到工具, t.name) if __name__ __main__: asyncio.run(main())这里的stdio_client会启动一个新的 Python 子进程来运行 MCP Server然后通过标准输入输出进行通信。开发调试阶段这种方式很方便但生产环境更推荐使用 Streamable HTTP 传输模式避免每次连接都要启动子进程。4.4 MCP 调试建议MCP 调试最容易被忽视的是“先单独验证 Server再接入 Agent”。如果 Agent 里工具加载失败先直接运行一次 MCP Server确认它本身能正常启动。其次查看标准错误输出。使用stdio_client时Server 端的print输出会干扰协议通信因此调试日志应该写到日志文件而不是直接打印到标准输出。5. 企业级实战订单查询与知识库问答 Agent5.1 业务需求与整体设计下面进入核心实战。假设业务方要求做一个智能客服 Agent需要同时支持三类问题订单查询类用户报出订单号Agent 调用订单系统工具查询状态。知识库问答类用户咨询公司制度、售后政策Agent 先从知识库检索再回答。普通闲聊类不涉及外部工具Agent 直接回答。为了让示例既展示 RAG又展示 MCP 工具同时体现 LangGraph 的流程编排能力设计思路如下把所有能力都封装成工具由模型自主决定调用哪个工具。检索知识库也作为一个工具暴露给模型。这样LangGraph 就只需要维护一个“模型决策 → 工具执行 → 回到模型”的循环逻辑清晰且扩展容易。5.2 准备知识库数据在data/docs/company_policy.txt中放入一份简短的内部文档。智联未来科技公司售后政策 1. 自签收之日起 7 天内支持无理由退货。 2. 退货商品需保持完好不影响二次销售。 3. 因质量问题产生的退货运费由公司承担。 员工加班补贴制度 1. 工作日加班超过 2 小时可申请加班餐补。 2. 周末加班的调休时长按实际加班时间计算。这份数据仅用于演示。真实项目中文档可能分布在 Confluence、语雀、企业网盘或数据库里需要额外开发文档同步和清洗逻辑。5.3 构建知识库索引先编写app/rag.py实现文档加载、切分和向量索引构建。# app/rag.py import os from langchain_community.vectorstores import FAISS from langchain_openai import OpenAIEmbeddings from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_core.documents import Document def load_docs_from_dir(doc_dir: str) - list[Document]: docs [] for fname in os.listdir(doc_dir): if not (fname.endswith(.txt) or fname.endswith(.md)): continue file_path os.path.join(doc_dir, fname) with open(file_path, encodingutf-8) as f: content f.read() docs.append(Document(page_contentcontent, metadata{source: fname})) return docs def build_vectorstore(doc_dir: str, index_dir: str): os.makedirs(index_dir, exist_okTrue) docs load_docs_from_dir(doc_dir) splitter RecursiveCharacterTextSplitter(chunk_size500, chunk_overlap50) chunks splitter.split_documents(docs) embeddings OpenAIEmbeddings() vectorstore FAISS.from_documents(chunks, embeddings) vectorstore.save_local(index_dir) print(f索引构建完成共生成 {len(chunks)} 个文本片段保存至 {index_dir}) def load_vectorstore(index_dir: str data/faiss_index): embeddings OpenAIEmbeddings() vectorstore FAISS.load_local( index_dir, embeddings, allow_dangerous_deserializationTrue, ) return vectorstore if __name__ __main__: build_vectorstore(data/docs, data/faiss_index)其中allow_dangerous_deserializationTrue是 FAISS 本地索引加载的必要参数。这里必须提醒一句本地反序列化存在安全风险仅建议在可信环境下使用。生产环境建议使用 Postgres pgvector、Milvus 或云上向量数据库。接着编写app/build_index.py。# app/build_index.py from rag import build_vectorstore if __name__ __main__: build_vectorstore(data/docs, data/faiss_index)5.4 编写 MCP Server订单系统复用第 4 节的代码把app/mcp_server.py写成一个完整的订单查询工具。为了让示例更贴近真实可以再增加一个“查询物流轨迹”工具但为了保持篇幅可控这里只保留一个query_order工具。# app/mcp_server.py from mcp.server.fastmcp import FastMCP mcp FastMCP(order-service) mcp.tool() def query_order(order_id: str) - str: 查询订单状态返回订单当前进度和物流信息。 Args: order_id: 订单号例如 SO20260101001 order_db { SO20260101001: {status: 已发货, logistics: 顺丰 SF123456}, SO20260101002: {status: 待付款}, } order order_db.get(order_id) if order: return f订单 {order_id} 当前状态{order[status]}物流信息{order[logistics]} return f未查询到订单 {order_id} if __name__ __main__: mcp.run()5.5 编写 LangGraph Agent 主流程先写app/agent.py整体结构分为四部分。第一部分是状态定义。这里使用Annotated和add_messages它能在多轮工具调用后自动合并消息而不是覆盖旧消息。from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages]第二部分是知识库检索工具。它被封装成一个 LangChain 工具放在 Agent 的工具列表里由模型自主判断是否需要调用。from langchain_core.tools import tool from rag import load_vectorstore tool def search_knowledge_base(query: str) - str: 从企业内部知识库检索资料用于回答公司制度、产品说明、售后政策等问题。 vectorstore load_vectorstore(data/faiss_index) retriever vectorstore.as_retriever(search_kwargs{k: 3}) docs retriever.invoke(query) if not docs: return 知识库中没有检索到相关内容。 return \n\n.join(doc.page_content for doc in docs)第三部分是构建 LangGraph 图。这里使用ToolNode统一执行工具模型节点负责判断是否需要调用工具。from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, START, END from langgraph.prebuilt import ToolNode def build_graph(tools): llm ChatOpenAI(modelgpt-4o-mini, temperature0) llm_with_tools llm.bind_tools(tools) tool_node ToolNode(tools) def agent_node(state: AgentState): response llm_with_tools.invoke(state[messages]) return {messages: [response]} def should_continue(state: AgentState): last_message state[messages][-1] if getattr(last_message,