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

pip install langchain langchain-community requests
# 假设我们使用OpenAI的模型
pip install openai

{
  "results": [
    {
      "id": "doc_123",
      "text": "产品的核心功能是...",
      "source": "user_guide",
      "score": 0.95
    }
  ]
}

from typing import List, Optional
from langchain.schema import BaseRetriever, Document
from langchain.callbacks.manager import CallbackManagerForRetrieverRun
import requests

class MCPRetriever(BaseRetriever):
    """一个从MCP服务器检索上下文的Retriever。"""
    
    base_url: str = "http://localhost:8080"
    """MCP服务器的基础URL。"""
    
    default_source: Optional[str] = None
    """默认检索的上下文源。"""
    
    def _get_relevant_documents(
        self, query: str, *, run_manager: CallbackManagerForRetrieverRun
    ) -> List[Document]:
        # 构造搜索请求URL
        url = f"{self.base_url}/search"
        params = {"query": query, "limit": 4}
        if self.default_source:
            params["source"] = self.default_source
            
        try:
            response = requests.get(url, params=params)
            response.raise_for_status() # 检查HTTP错误
            data = response.json()
        except requests.exceptions.RequestException as e:
            # 在实际应用中应有更完善的错误处理
            print(f"请求MCP服务器失败: {e}")
            return []
            
        # 将MCP返回的结果转换为LangChain的Document对象
        documents = []
        for item in data.get("results", []):
            # 将元数据存入Document的`metadata`字段
            doc = Document(
                page_content=item["text"],
                metadata={
                    "source": item.get("source"),
                    "id": item.get("id"),
                    "relevance_score": item.get("score")
                }
            )
            documents.append(doc)
            
        return documents

from langchain.chains import RetrievalQA
from langchain.chat_models import ChatOpenAI
from langchain.prompts import PromptTemplate

# 1. 初始化MCP Retriever
mcp_retriever = MCPRetriever(default_source="user_guide")

# 2. 初始化LLM
llm = ChatOpenAI(model_name="gpt-4", temperature=0)

# 3. （可选）定义自定义提示模板，指导模型如何使用检索到的上下文
prompt_template = """请根据以下提供的上下文信息来回答问题。如果你无法从上下文中找到答案，请如实说明你不知道，不要编造信息。

上下文：
{context}

问题：{question}
请给出有帮助的答案："""

PROMPT = PromptTemplate(
    template=prompt_template, input_variables=["context", "question"]
)

# 4. 创建检索式问答链
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff", # 简单地将所有检索到的文档“堆叠”到上下文中
    retriever=mcp_retriever,
    chain_type_kwargs={"prompt": PROMPT}, # 使用自定义提示
    return_source_documents=True # 返回用于生成答案的源文档，便于调试
)

# 5. 运行链
question = "产品如何实现数据备份？"
result = qa_chain({"query": question})
print(f"问题：{question}")
print(f"答案：{result['result']}")
print("\n--- 引用的源文档 ---")
for doc in result['source_documents']:
    print(f"来源：{doc.metadata.get('source')} | 片段：{doc.page_content[:100]}...")