从LangChain到MCP与LangGraph：构建可运维AI Agent的工程实践

最近在折腾几个本地 AI 项目，发现一个挺有意思的现象：很多开发者，包括我自己在内，一开始都热衷于用 LangChain 去“拼装”一个看起来功能强大的 Agent。我们花大量时间研究各种工具链、记忆模块和复杂的执行流程，但往往在项目推进到一半时，就会遇到一个共同的瓶颈——上下文管理失控。

你可能会遇到这样的报错：“上下文过大，已进行多次自动总结但上下文大小仍超出限制。” 或者，Agent 的执行莫名其妙地终止，留下一句 “Agent execution terminated due to error.”。这时候，你面对的就不再是“如何添加新功能”，而是“如何让已有的功能稳定运行”。问题的核心，往往不在于 Agent 的逻辑有多复杂，而在于如何为它建立一个清晰、可控、可扩展的“工作台”。

这就是为什么 MCP（Model Context Protocol）和 LangGraph 的出现，不仅仅是 LangChain 生态里的两个新工具，它们实际上在重新定义我们构建 AI 应用的范式。过去，我们思考的是“如何让 LLM 调用工具”；现在，更值得思考的是“如何为 LLM 设计一个高效、安全、可管理的工作环境”。MCP 负责定义和接入这个环境里的“工具”与“资源”，而 LangGraph 则负责编排在这个环境里“人”与“Agent”的协作流程。

这篇文章，我们不打算平铺直叙地介绍概念或罗列 API。我想从一个更实际的角度出发：如何从“一次性脚本”思维，转向“可运维系统”思维，来构建你的 AI Agent。我们会以 LangChain 为基石，但重点会放在如何用 MCP 解决工具集成的混乱，以及如何用 LangGraph 解决执行流程的不可控。最终的目标，是让你能搭建一个不仅“能跑通”，而且“好调试、易扩展、可观察”的本地 AI 智能体。

1. 重新理解 LangChain：它不只是“胶水”，更是“脚手架”

很多人把 LangChain 简单地理解为连接大模型和外部工具的“胶水”。这个比喻没错，但容易让人低估它的价值。更准确的看法是，LangChain 提供了一套构建 AI 应用的标准脚手架和设计模式。在你开始写第一行业务逻辑之前，它已经帮你定义好了几个关键问题的处理方式：

• 工具（Tools）如何被标准化定义和调用？
• 记忆（Memory）如何以结构化的方式存储和读取？
• 链（Chains）如何将多个步骤组合成可复用的流程？

当你直接使用LLMChain或AgentExecutor时，你是在使用一个封装好的、但相对“黑盒”的解决方案。它能快速验证想法，就像用预制板搭建一个临时工棚。但当你需要在这个工棚里进行精细化作业（比如处理复杂对话、管理长上下文、集成异构工具）时，你就会发现墙壁不能随意开窗，管线难以调整。

LangChain 的真正价值，在于它公开了这些“预制板”的接口和组装逻辑。这让你可以在其基础上进行定制。例如，当默认的ConversationBufferMemory无法满足你对上下文窗口的精细控制时，你可以去实现自己的BaseMemory类。这就是从“使用框架”到“理解框架”的关键一步。

然而，即使理解了 LangChain 的组件，在集成外部工具时，我们依然面临挑战。传统方式是为每个工具（如数据库查询、API 调用、文件读取）编写一个适配函数，并注册到 Agent 中。这种方式在工具不多时可行，但随着工具数量增长，会出现以下问题：

1. 依赖耦合：Agent 代码严重依赖各个工具库的安装和版本。
2. 安全边界模糊：所有工具都在同一个 Python 进程中运行，权限控制困难。
3. 开发体验割裂：工具开发者需要深入理解 LangChain 的Tool类，才能提供适配。

MCP 协议的出现，正是为了从根本上解决工具集成的标准化和安全性问题。它让工具回归其本质——一个可以通过标准协议访问的服务，而不是必须被硬编码进应用逻辑的模块。

2. MCP 协议：为 AI Agent 建立标准化的“工具车间”

MCP 的全称是 Model Context Protocol。这个名字直击要害：它的核心是为模型（Model）管理上下文（Context）提供一种协议（Protocol）。这里“上下文”不仅指对话历史，更广义地指模型执行任务时所需的一切外部资源和信息，比如文件内容、数据库数据、API 返回结果等。

你可以把 MCP 想象成给 AI Agent 建立了一个标准化的“工具车间”。在这个车间里：

• 每个工具都是一个独立的“工作台”（MCP Server）：比如，一个专门读写文件的工作台，一个专门查询数据库的工作台，一个专门获取天气的工作台。这些工作台各自独立运行，可能用不同的语言（Python, Node.js, Go）编写，有独立的进程和环境。
• Agent 是“总调度员”（MCP Client）：它不需要知道每个工作台内部是如何运转的。它只需要掌握一套标准的“沟通语言”（MCP 协议），就可以向任何工作台发送指令（如“读取/data/report.md文件”），并接收标准化格式的结果。
• 协议是“通用工单”：MCP 协议规定了工单的格式。无论是要使用工具（tools/list，tools/call），还是要获取资源（resources/list，resources/read），都遵循同样的格式。这使得调度员可以无缝切换或增删工作台。

2.1 为什么需要 MCP？从三个痛点看价值

1. 解耦与安全：最直接的好处。一个文件操作的 MCP Server 崩溃了，不会导致你的主 Agent 进程崩溃。你可以为敏感的数据查询 Server 设置严格的网络访问控制（如仅允许本地 Socket 连接），而给相对安全的天气查询 Server 开放 HTTP 接口。这种隔离性是传统 LangChain Tool 方式难以实现的。
2. 开发与集成的标准化：工具开发者现在只需要关注一件事：按照 MCP 协议实现一个 Server，提供工具列表和调用接口。他们不需要关心这个工具最终是被 LangChain、LlamaIndex 还是其他任何支持 MCP 的框架使用。这极大地降低了生态工具的开发门槛，促进了工具市场的繁荣。对于集成方来说，集成一个新工具就是配置一个 MCP Server 连接地址那么简单。
3. 动态性与可观测性：MCP Server 可以动态地注册或注销工具。例如，一个连接了公司内部知识库的 Server，可以在知识库更新时，动态地添加新的搜索工具。同时，由于通信是跨进程的，你可以很方便地在协议层对所有的工具调用进行日志记录、性能监控和审计，这对于生产环境至关重要。

2.2 实战：快速搭建一个文件读取 MCP Server

理论之后，我们来点实际的。搭建一个 MCP Server 比想象中简单。下面是一个用 Python 编写的最简单的文件读取 Server：

# file_server.py import json from typing import Any, List from mcp import Server, Tool import mcp.server.stdio # 1. 定义工具 def read_file(arguments: dict) -> dict: """读取指定路径文件的内容""" filepath = arguments.get("filepath") if not filepath: return {"error": "Missing 'filepath' argument"} try: with open(filepath, 'r', encoding='utf-8') as f: content = f.read() return {"content": content} except Exception as e: return {"error": f"Failed to read file: {str(e)}"} # 2. 创建 Server 实例并注册工具 server = Server("file-operations-server") server.tool( name="read_file", description="Read the contents of a file from the local filesystem.", arguments={ "type": "object", "properties": { "filepath": { "type": "string", "description": "The path to the file to read." } }, "required": ["filepath"] } )(read_file) # 将函数绑定到工具定义 # 3. 通过标准输入输出运行 Server（这是最常见的与 Client 交互的方式） if __name__ == "__main__": with mcp.server.stdio.stdio_server() as (read_stream, write_stream): server.run(read_stream, write_stream)

这个 Server 通过标准输入输出（stdio）与外界通信。要运行它，你只需要执行python file_server.py。现在，任何一个支持 MCP 协议的 Client（比如配置好的 LangChain）都可以连接到这个 Server，并调用read_file工具。

关键理解：这个 Server 本身不包含任何 LLM 调用逻辑，它就是一个纯粹的、功能单一的服务。这才是 MCP 倡导的“单一职责”和“清晰边界”。

3. LangGraph：将线性“链”升级为可控“图”

解决了工具的标准化接入（MCP）后，我们面临下一个问题：如何组织复杂的、可能循环的、需要人工干预的 Agent 执行流程？传统的AgentExecutor是一个“黑盒”循环：思考 -> 行动 -> 观察 -> 再思考。这在简单任务中有效，但一旦流程复杂，就会出现以下问题：

• 状态管理困难：对话历史、中间结果、工具调用状态散落在各处。
• 流程难以定制：你想在 Agent 调用特定工具后插入一个确认步骤，或者根据工具结果分支到不同的处理流程，用标准的 Agent 很难优雅实现。
• 调试如同黑盒：你只知道最终输出或错误，很难知道在漫长的循环中，是哪一步的决策导致了问题。

LangGraph 应运而生。它将 Agent 的执行流程抽象为一个有向图（Graph）。图中的节点（Node）代表一个步骤（如“调用LLM”、“执行工具”、“等待用户输入”），边（Edge）代表步骤之间的流转条件。

3.1 LangGraph 的核心概念：状态（State）与流程（Flow）

1. 状态（State）：这是一个贯穿整个图执行过程的共享字典。你可以定义它的结构，例如{"messages": List, "intermediate_steps": List, "user_feedback": str}。每个节点都可以读取和修改这个状态。
2. 节点（Node）：一个普通的 Python 函数，它接收当前的State，执行一些操作（如调用 LLM、调用工具），然后返回更新后的State。
3. 边（Edge）：决定执行完一个节点后，下一步该去哪个节点。边可以是固定的（always_go_to），也可以根据State中的内容动态决定（conditional_edge）。

这种图结构带来了前所未有的清晰度和控制力。

3.2 实战：用 LangGraph 构建一个带人工审核的 Agent

假设我们要构建一个数据分析 Agent，它可以根据用户描述读取 CSV 文件并生成图表。但出于安全考虑，在写入最终报告文件前，需要人工审核图表内容。

用传统链式思维很难嵌入这个“中断点”，而用 LangGraph 则非常自然：

# graph_agent.py from typing import TypedDict, Annotated, List from langgraph.graph import StateGraph, END from langchain_core.messages import HumanMessage, AIMessage, ToolMessage from langchain_openai import ChatOpenAI from langchain_community.tools import DuckDuckGoSearchRun import operator # 1. 定义状态结构 class AgentState(TypedDict): messages: Annotated[List, operator.add] # 关键：这是一个消息累加器 next: str # 记录下一步该去哪个节点 # 2. 初始化组件 llm = ChatOpenAI(model="gpt-4o-mini") search_tool = DuckDuckGoSearchRun() # 3. 定义各个节点函数 def llm_node(state: AgentState): """节点：调用LLM，决定下一步行动""" # 从状态中获取对话历史 model_with_tools = llm.bind_tools([search_tool]) response = model_with_tools.invoke(state["messages"]) # 将LLM的回复添加到消息历史中 new_messages = [response] # 检查LLM是否想调用工具 if response.tool_calls: # 如果调用工具，下一步去 `tool_node` return {"messages": new_messages, "next": "tool_node"} else: # 否则，流程结束 return {"messages": new_messages, "next": "__end__"} def tool_node(state: AgentState): """节点：执行工具调用""" last_message = state["messages"][-1] tool_calls = last_message.tool_calls tool_messages = [] for tc in tool_calls: # 这里简化处理，实际应根据 tc['name'] 调用对应工具 if tc['name'] == 'duckduckgo_search': tool_result = search_tool.invoke(tc['args']['query']) else: tool_result = f"Tool {tc['name']} not found." # 将工具执行结果封装为 ToolMessage tool_messages.append(ToolMessage(content=str(tool_result), tool_call_id=tc['id'])) # 将工具结果添加到历史，下一步回到LLM进行总结 return {"messages": tool_messages, "next": "llm_node"} def human_review_node(state: AgentState): """节点：等待人工审核输入（模拟）""" # 在实际应用中，这里可能会暂停，等待来自Web界面或API的输入 # 此处我们模拟用户输入“同意” human_input = HumanMessage(content="审核通过，可以继续。") return {"messages": [human_input], "next": "generate_report_node"} def generate_report_node(state: AgentState): """节点：生成最终报告""" # 假设这里调用LLM生成报告 report = "这是最终的分析报告..." return {"messages": [AIMessage(content=report)], "next": "__end__"} # 4. 构建图 workflow = StateGraph(AgentState) # 添加节点 workflow.add_node("llm_node", llm_node) workflow.add_node("tool_node", tool_node) workflow.add_node("human_review_node", human_review_node) workflow.add_node("generate_report_node", generate_report_node) # 设置入口点 workflow.set_entry_point("llm_node") # 添加边 workflow.add_conditional_edges( "llm_node", # 根据 `state['next']` 决定下一个节点 lambda state: state["next"], { "tool_node": "tool_node", "__end__": END, "needs_review": "human_review_node" # 这是一个可能由LLM节点设置的路径 } ) workflow.add_edge("tool_node", "llm_node") # 工具执行完回到LLM workflow.add_edge("human_review_node", "generate_report_node") workflow.add_edge("generate_report_node", END) # 编译图 app = workflow.compile() # 5. 执行图 initial_state = AgentState(messages=[HumanMessage(content="帮我搜索一下LangGraph的最新动态。")], next="") final_state = app.invoke(initial_state) for msg in final_state["messages"]: print(f"{type(msg).__name__}: {msg.content}")

这个例子虽然简化，但清晰地展示了 LangGraph 如何将“LLM决策 -> 执行工具 -> 人工审核 -> 生成报告”这一系列步骤，组织成一个可视、可控的流程。你可以通过workflow.get_graph().draw_mermaid()生成流程图，直观地看到整个 Agent 的决策路径，这对于调试和协作至关重要。

LangGraph 的“Human-in-the-Loop”机制正是通过添加一个像human_review_node这样的节点来实现的。流程可以在此节点暂停，等待外部输入（通过回调、API或队列），然后再继续。这完美解决了需要人工确认、审核或提供额外信息的复杂场景。

4. 整合实战：构建一个基于本地模型的 AI 智能体

现在，我们将 MCP 和 LangGraph 结合起来，构建一个更贴近真实场景的本地 AI 智能体。这个智能体将：

1. 使用本地模型（通过 Ollama 运行）。
2. 通过 MCP 安全地访问文件系统。
3. 使用 LangGraph 管理一个包含“思考-行动-用户确认”的循环流程。

4.1 系统架构与准备工作

架构图（文字描述）：

用户输入 | v [LangGraph 编排引擎] <---(状态管理)---> [共享状态] | | | (通过LCEL调用) | v | [LangChain LLM (Ollama)] | | | | (决定调用工具) | v | [MCP Client] | | | | (标准MCP协议) | v | [MCP Server 1: 文件操作] | [MCP Server 2: 网络搜索] | ... | | | | (返回结果) | v | [更新状态，决定下一步] ------------+

环境准备：

1. 安装核心库：

   pip install langchain langgraph langchain-community mcp

2. 安装并运行 Ollama：从官网下载 Ollama，并拉取一个本地模型，例如llama3.2。

   ollama pull llama3.2

3. 启动 MCP Server：我们需要一个文件操作的 Server。可以使用官方示例或自己编写（如第 2.2 节的file_server.py）。假设我们使用一个简单的示例 Server，通过 stdio 运行。

4.2 代码实现：连接所有组件

# local_ai_agent.py import asyncio from typing import TypedDict, Annotated, List import operator from langgraph.graph import StateGraph, END from langchain_community.chat_models import ChatOllama from langchain_core.messages import HumanMessage, AIMessage, ToolMessage from langchain_core.tools import Tool # 假设我们有一个能连接MCP Server并生成LangChain Tools的工具类 # 这里使用一个模拟的MCP工具适配器 from mcp_client_adapter import get_mcp_tools # 这是一个假设的辅助函数 # 1. 定义状态 class AgentState(TypedDict): messages: Annotated[List, operator.add] next_step: str # 控制流程 # 2. 初始化LLM和工具 llm = ChatOllama(model="llama3.2", temperature=0) # 连接到MCP Server并获取工具列表 # 假设我们的文件Server运行在某个端口或stdio上 file_tools = get_mcp_tools(server_type="stdio", server_config={"command": "python", "args": ["file_server.py"]}) # file_tools 现在是一个LangChain Tool对象的列表 # 3. 定义图节点 async def supervisor_node(state: AgentState): """监督节点：分析状态，决定下一步是LLM思考还是需要人工介入""" last_msg = state["messages"][-1] if isinstance(last_msg, HumanMessage) and "审核" in last_msg.content: # 如果用户消息包含“审核”关键词，跳转到人工节点 return {"messages": [], "next_step": "human_intervention"} # 否则，交给LLM处理 return {"messages": [], "next_step": "llm_agent"} async def llm_agent_node(state: AgentState): """LLM代理节点：思考并决定行动""" # 将工具绑定给LLM model_with_tools = llm.bind_tools(file_tools) # 调用LLM response = await model_with_tools.ainvoke(state["messages"]) new_messages = [response] if response.tool_calls: # 有工具调用，下一步去执行工具 return {"messages": new_messages, "next_step": "execute_tools"} else: # 没有工具调用，流程结束 return {"messages": new_messages, "next_step": "__end__"} async def execute_tools_node(state: AgentState): """工具执行节点：调用MCP工具""" last_message = state["messages"][-1] tool_messages = [] for tc in last_message.tool_calls: # 根据 tool_call 的 name 找到对应的 Tool 对象 tool_to_use = next((t for t in file_tools if t.name == tc['name']), None) if tool_to_use: try: # 执行工具 tool_result = await tool_to_use.ainvoke(tc['args']) tool_messages.append(ToolMessage(content=str(tool_result), tool_call_id=tc['id'])) except Exception as e: tool_messages.append(ToolMessage(content=f"Error: {str(e)}", tool_call_id=tc['id'])) else: tool_messages.append(ToolMessage(content=f"Tool {tc['name']} not available.", tool_call_id=tc['id'])) # 执行完工具，回到监督节点决定下一步 return {"messages": tool_messages, "next_step": "supervisor"} async def human_intervention_node(state: AgentState): """人工干预节点（模拟）：这里可以连接Web界面或等待API调用""" # 在实际中，这里可能是一个等待外部输入的事件循环 # 此处我们模拟用户确认 user_confirmation = HumanMessage(content="我已审核，同意继续执行。") return {"messages": [user_confirmation], "next_step": "supervisor"} # 4. 构建图 workflow = StateGraph(AgentState) workflow.add_node("supervisor", supervisor_node) workflow.add_node("llm_agent", llm_agent_node) workflow.add_node("execute_tools", execute_tools_node) workflow.add_node("human_intervention", human_intervention_node) workflow.set_entry_point("supervisor") # 定义条件边 def route_after_supervisor(state): return state["next_step"] workflow.add_conditional_edges( "supervisor", route_after_supervisor, { "llm_agent": "llm_agent", "human_intervention": "human_intervention", "__end__": END } ) workflow.add_conditional_edges( "llm_agent", lambda s: s["next_step"], {"execute_tools": "execute_tools", "__end__": END} ) workflow.add_edge("execute_tools", "supervisor") workflow.add_edge("human_intervention", "supervisor") app = workflow.compile() # 5. 运行智能体 async def main(): initial_state = AgentState( messages=[HumanMessage(content="请读取当前目录下的 README.md 文件，并总结其内容。")], next_step="" ) async for event in app.astream(events=initial_state, stream_mode="values"): if "messages" in event: for msg in event["messages"]: print(f"[{msg.type}] {msg.content[:100]}...") # 打印消息摘要 if __name__ == "__main__": asyncio.run(main())

4.3 关键解析与避坑指南

1. MCP Client 适配：上面的get_mcp_tools是一个示意函数。在实际中，你需要使用mcp库的Client类来连接 Server，并将 Server 提供的工具描述转化为 LangChain 的Tool对象。这个过程可能涉及一些异步通信的封装。
2. 状态管理：我们使用Annotated[List, operator.add]来定义messages，这是 LangGraph 的一个妙处，它自动将每个节点返回的 messages 列表合并到总状态中，极大简化了对话历史管理。
3. 错误处理：在生产环境中，必须在execute_tools_node和llm_agent_node中加入更完善的错误处理、重试逻辑和超时控制。
4. 流程可控性：supervisor_node是一个简单的路由逻辑。在复杂应用中，它可以基于更复杂的规则（如工具调用次数、特定关键词、用户身份）来路由流程，甚至实现多 Agent 协作。
5. 持久化与可视化：LangGraph 支持将图的状态持久化到数据库，并提供了get_graph().draw_mermaid()方法来生成流程图，这对于调试和文档化至关重要。

5. 从项目到产品：工程化考量与进阶方向

当你成功运行起第一个整合了 MCP 和 LangGraph 的 Agent 后，恭喜你，你已经跨越了从“脚本”到“系统”的第一道门槛。但这距离一个可投入生产环境的“产品”还有距离。以下是你需要持续关注的工程化考量：

5.1 稳定性与可观测性

• 日志与追踪：为每个 MCP Server 调用、LLM 调用、图节点跳转添加结构化日志。考虑集成 OpenTelemetry 等标准，实现分布式追踪。
• 错误恢复：Agent 执行失败后，是重试、回滚还是转人工？在图设计中，可以添加专门的error_handler_node节点来处理异常分支。
• 资源限制：为 LLM 调用设置 Token 限制和超时；为 MCP 工具调用设置超时和并发控制。

5.2 性能与成本

• 上下文管理：这是开头提到的核心痛点。除了依赖模型的上下文窗口，要积极使用 LangChain 的ConversationSummaryMemory或ConversationBufferWindowMemory来压缩历史。对于 MCP 提供的资源（如长文档），优先使用resources/read的分页或摘要功能，避免一次性加载。
• 缓存策略：对昂贵的 LLM 调用或重复的 MCP 查询结果进行缓存，可以显著降低成本和提高响应速度。
• 异步优化：确保你的图节点和工具调用是异步的（async/await），以充分利用 I/O 等待时间。

5.3 架构扩展

• 多模态能力：MCP Server 不仅可以提供文本工具，也可以提供图像生成、语音处理、视频分析等能力。LangGraph 可以协调这些多模态工具的调用流程。
• 多 Agent 协作：LangGraph 可以轻松编排多个具有不同专长的 Agent 协同工作。例如，一个“研究员”Agent 负责搜索和信息整理，一个“写手”Agent 负责生成报告，一个“审核”Agent 负责质量控制。
• 与现有系统集成：将你的 AI 智能体作为微服务嵌入现有架构。通过 FastAPI 或 Flask 将 LangGraph 的app暴露为 HTTP 端点，前端或其他服务可以通过 API 与之交互。

5.4 学习路径建议

1. 第一步（入门）：吃透 LangChain 的核心概念（Models, Prompts, Chains, Agents, Memory）。先不用 MCP 和 LangGraph，用标准 Agent 完成几个小任务，感受其便利与局限。
2. 第二步（解耦）：引入 MCP。尝试将一两个外部依赖（如读取特定数据库、调用内部 API）改造成独立的 MCP Server。体会进程隔离和协议标准化带来的好处。
3. 第三步（可控）：引入 LangGraph。将一个简单的线性 Chain 重构成图。尝试添加一个条件分支或一个人工干预节点。感受状态管理的清晰和流程的可视化。
4. 第四步（整合）：将三者结合，构建一个完整的本地智能体。从 Ollama + 文件操作 MCP + 简单 LangGraph 开始。
5. 第五步（产品化）：围绕这个智能体，添加上文提到的日志、监控、缓存、API 层等工程化组件。

回过头看，LangChain、MCP、LangGraph 这个技术栈，解决的不仅仅是“让 AI 调用工具”的问题，它更是在帮助我们建立一种构建复杂、可靠、可维护 AI 应用的系统工程方法。它迫使我们将关注点从单一的 Prompt 技巧，转移到整体的架构设计、组件边界、数据流和异常处理上。这或许才是从 AI 爱好者迈向 AI 应用开发者的关键一步。