LangChain 接入 MCP 完整指南：解锁模型上下文管理新范式

AILangChainMCPPythonTutorialLLM
•

在构建基于大语言模型（LLM）的复杂应用时，开发者常常面临一个核心挑战：如何高效、可靠地管理不断增长的对话或文档上下文？传统的上下文拼接方法不仅容易触及模型的令牌限制，还会导致信息丢失和成本激增。Model Context Protocol (MCP) 的出现，为这一问题提供了优雅的解决方案。本文将带你深入理解 MCP，并手把手教你如何将其无缝集成到流行的 LangChain 框架中，构建更智能、更健壮的 AI 应用。

什么是 Model Context Protocol (MCP)？

Model Context Protocol (MCP) 是一个开放的协议和工具集，旨在标准化和优化大语言模型上下文的管理方式。它核心解决了“上下文窗口”的瓶颈问题。

MCP 的核心优势包括：

  1. 智能上下文压缩与摘要：自动识别和压缩冗余或次要信息，保留关键内容，而非简单截断。
  2. 外部知识库集成：将超长文档、数据库记录等存储在外部向量库中，通过动态检索相关片段来扩展有效上下文，而非全部塞入提示词。
  3. 结构化上下文管理：将上下文视为可查询、可更新的结构化数据，而非纯文本字符串。
  4. 降低成本与延迟：通过减少每次请求传递的令牌数，直接降低 API 调用成本并提升响应速度。

简单来说，MCP 让 LLM 应用能够“聪明地”记住和利用海量信息，而不是笨拙地背负所有历史包袱。

LangChain 框架简介

LangChain 是一个用于开发由语言模型驱动的应用程序的框架。它通过“链”（Chains）、“代理”（Agents）、“记忆”（Memory）和“检索”（Retrieval）等抽象概念，极大地简化了构建复杂 LLM 工作流的难度。LangChain 的模块化设计使其成为集成 MCP 等先进协议的理想平台。

在 LangChain 中接入 MCP 的详细步骤

我们将通过一个示例项目来演示集成过程：构建一个能够处理长文档问答的聊天机器人。

步骤 1：环境准备与安装

首先，确保你的 Python 环境（建议 3.8+）并安装必要的包。

pip install langchain langchain-community openai tiktoken
# 假设我们使用一个实现了 MCP 的库，例如 `mcp-client` (此处为示例，请根据实际 MCP 服务器选择)
# pip install mcp-client

为了演示，我们将使用一个模拟的 MCP 客户端。在实际应用中，你需要连接到一个 MCP 兼容的服务器（如专用的上下文管理服务）。

步骤 2：创建自定义 MCP 记忆（Memory）类

LangChain 的 BaseMemory 类是其记忆系统的基石。我们将继承它来创建一个使用 MCP 协议管理上下文的记忆类。

from typing import Any, Dict, List
from langchain.schema import BaseMemory
from langchain.schema.messages import BaseMessage, get_buffer_string

class MCPMemory(BaseMemory):
    """一个使用 MCP 协议管理对话历史的记忆类。"""

    def __init__(self, mcp_client, max_compressed_tokens: int = 2000):
        super().__init__()
        self.mcp_client = mcp_client  # MCP 客户端实例
        self.max_compressed_tokens = max_compressed_tokens
        # 内部存储原始消息，用于初始化和更新 MCP 状态
        self.buffer: List[BaseMessage] = []

    @property
    def memory_variables(self) -> List[str]:
        """定义此记忆组件输出的变量名。"""
        return ["mcp_context"]

    def load_memory_variables(self, inputs: Dict[str, Any]) -> Dict[str, str]:
        """加载记忆变量。当链需要上下文时被调用。"""
        # 将当前 buffer 中的消息同步到 MCP 服务器
        if self.buffer:
            raw_context = get_buffer_string(self.buffer)
            # 调用 MCP 客户端，获取压缩/优化后的上下文
            # 这里模拟 MCP 的“压缩”和“检索”功能
            optimized_context = self.mcp_client.optimize_context(
                raw_context,
                max_tokens=self.max_compressed_tokens,
                query=inputs.get("input", "")  # 可以根据当前查询进行针对性检索
            )
        else:
            optimized_context = ""

        return {"mcp_context": optimized_context}

    def save_context(self, inputs: Dict[str, Any], outputs: Dict[str, Any]) -> None:
        """保存一轮对话的输入和输出到记忆。"""
        # 这里需要根据你的应用逻辑，将输入和输出转换为 BaseMessage 并存入 buffer
        # 例如，使用 HumanMessage 和 AIMessage
        from langchain.schema.messages import HumanMessage, AIMessage

        human_input = inputs.get("input", inputs.get("question"))
        ai_output = outputs.get("output", outputs.get("answer"))

        if human_input:
            self.buffer.append(HumanMessage(content=human_input))
        if ai_output:
            self.buffer.append(AIMessage(content=ai_output))

        # 可选：在保存一定轮数后，主动触发 MCP 的归档或摘要功能
        if len(self.buffer) > 10:  # 示例阈值
            self._archive_old_conversation()

    def clear(self) -> None:
        """清空记忆。"""
        self.buffer.clear()
        # 通知 MCP 服务器重置上下文
        self.mcp_client.clear_context()

    def _archive_old_conversation(self):
        """将早期对话归档到 MCP 的长期存储中。"""
        # 模拟：将前5条消息摘要后存入长期记忆，并从 buffer 中移除
        if len(self.buffer) > 5:
            to_archive = self.buffer[:5]
            archive_summary = self.mcp_client.create_summary(get_buffer_string(to_archive))
            self.mcp_client.save_to_long_term_memory(archive_summary)
            self.buffer = self.buffer[5:]

步骤 3：模拟 MCP 客户端

由于真实的 MCP 服务器依赖具体实现，这里我们创建一个模拟客户端来演示接口。

class MockMCPClient:
    """一个模拟的 MCP 客户端，用于演示。"""

    def optimize_context(self, raw_context: str, max_tokens: int, query: str = "") -> str:
        """模拟 MCP 的上下文优化功能。"""
        # 在实际中，这里会调用 MCP 服务器进行：
        # 1. 基于查询的向量检索（如果连接了知识库）
        # 2. 文本压缩和摘要
        # 3. 令牌计数与裁剪
        print(f"[MCP] 优化上下文。原始长度: {len(raw_context)} 字符， 查询: '{query}'")
        # 简单模拟：如果太长，就取最后一部分并加个说明
        if len(raw_context) > max_tokens * 4:  # 粗略字符数估计
            optimized = f"[之前的对话已由 MCP 摘要] ... {raw_context[-500:]}"
        else:
            optimized = raw_context
        return optimized[:max_tokens * 4]  # 再次确保长度

    def clear_context(self):
        print("[MCP] 上下文已清空。")

    def create_summary(self, text: str) -> str:
        return f"摘要: {text[:100]}..."  # 模拟摘要

    def save_to_long_term_memory(self, summary: str):
        print(f"[MCP] 已归档到长期记忆: {summary}")

步骤 4：在 LangChain 链中使用 MCP 记忆

现在，我们将 MCP 记忆集成到一个简单的对话链中。

from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate
from langchain.chains import LLMChain

# 1. 初始化组件
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
mcp_client = MockMCPClient()
memory = MCPMemory(mcp_client=mcp_client, max_compressed_tokens=1500)

# 2. 创建提示模板，其中包含 `{mcp_context}` 变量
prompt_template = PromptTemplate(
    input_variables=["mcp_context", "input"],
    template="""你是一个有帮助的助手。请根据以下历史对话上下文（可能经过压缩）来回答问题。

相关上下文：
{mcp_context}

当前问题：{input}

回答："""
)

# 3. 创建链
conversation_chain = LLMChain(
    llm=llm,
    prompt=prompt_template,
    memory=memory,  # 关键：注入我们的 MCP 记忆
    verbose=True  # 查看内部过程
)

# 4. 运行对话
print("=== 第一轮对话 ===")
response1 = conversation_chain.run(input="我叫小明。我喜欢编程和爬山。")
print(f"AI: {response1}\n")

print("=== 第二轮对话 (依赖记忆) ===")
# 这次提问依赖于之前的上下文
response2 = conversation_chain.run(input="我刚刚提到我喜欢什么运动？")
print(f"AI: {response2}\n")

# 模拟一个很长的上下文积累后...
print("=== 经过多轮对话后... ===")
# memory.buffer 模拟增长，触发 MCP 优化
for i in range(8):
    memory.save_context({"input": f"测试消息{i}"}, {"output": f"测试回复{i}"})

response3 = conversation_chain.run(input="总结一下我的爱好。")
print(f"AI: {response3}")

常见问题与解决方案

  1. Q: MCP 服务器如何选择？ A: 目前有多种实现方式：a) 使用云服务商提供的托管 MCP 服务；b) 自建开源 MCP 服务器（如某些向量数据库提供的插件）；c) 使用 LangChain 生态中其他工具（如 ConversationSummaryBufferMemory）模拟部分功能。评估时需考虑性能、成本和控制粒度。

  2. Q: 集成后响应变慢？ A: MCP 的优化和检索操作需要时间。解决方案：a) 对 MCP 操作进行异步处理；b) 实现缓存层，对相似查询返回缓存的优化上下文；c) 调整 max_compressed_tokens，在质量和速度间取得平衡。

  3. Q: 如何保证上下文压缩后的信息保真度？ A: 这是 MCP 的核心挑战。建议：a) 在 MCP 服务器中实现可调的压缩策略（如提取实体、关键句）；b) 保留一个“原始记录”的指针，当 AI 回答置信度低时，可以回查；c) 定期让 AI 自己生成对话摘要来替代自动压缩。

  4. Q: 与 LangChain 其他记忆组件（如 ConversationBufferWindowMemory）冲突吗？ A: 不冲突，但通常不需要同时使用。MCPMemory 旨在提供更高级的功能来替代基础记忆组件。你可以根据场景选择：简单窗口记忆用内置的，需要智能长上下文管理则用 MCP。

最佳实践建议

  1. 分层上下文策略：结合使用 MCP 的短期压缩、中期摘要和长期向量检索，为不同“年龄”的信息匹配不同的处理方式。
  2. 查询感知优化：如示例所示，将用户的当前查询 query 传递给 optimize_context 方法，使 MCP 能进行基于相似度的精准检索，而非无差别压缩。
  3. 与 LangChain Agent 结合：将 MCPMemory 用于 Agent 的 memory 参数，可以显著提升多步骤任务中 Agent 的连贯性和知识追溯能力。
  4. 监控与评估：记录 MCP 压缩前后的令牌数、关键信息保留情况以及最终回答质量，持续优化 MCP 服务器的参数和策略。
  5. 备选方案：在 MCP 服务不可用时，应有降级方案（如自动切换回传统的窗口记忆），保证应用的鲁棒性。

总结

通过将 Model Context Protocol (MCP) 集成到 LangChain 框架，我们为 AI 应用装上了“智能记忆管理器”。它不再是简单遗忘或机械堆砌，而是能够主动提炼、归档和检索关键信息。本文提供的 MCPMemory 类是一个起点，你可以在此基础上连接真实的 MCP 服务器，实现更复杂的上下文管理逻辑，从而构建出能够处理超长对话、深入分析大型文档的真正强大的 LLM 应用。拥抱 MCP 这类协议，是突破当前 LLM 上下文限制、迈向更可靠 AI 应用的关键一步。

← 返回博客列表