MCP 实战：从零构建可落地的问答机器人

一、实战目标：让 MCP 架构 “从理论到落地”

本集通过开发一个 “基于文档的问答机器人”，将前序讲解的 MCP “三层一储” 架构转化为可运行的实际应用。该机器人需实现三大核心功能：支持 PDF/Word 文档上传、理解用户基于文档的提问、生成结构化答案并支持导出 —— 整个开发过程严格遵循 MCP“组件独立、接口统一、数据驱动” 的原则，让初学者能清晰看到 “如何用组件拼出完整应用”。

二、问答机器人的 MCP 架构拆解：明确组件分工

首先需将机器人功能映射到 MCP 的 “三层一储”，确定每层需用到的组件及交互逻辑，避免开发中出现 “组件混乱、职责交叉” 的问题。具体拆解如下：

MCP 分层	核心任务	选用组件	组件核心作用
输入层	接收文档和用户提问	文档解析组件、文本输入组件	解析文档为结构化文本，清洗用户提问内容
处理层	生成基于文档的答案	LLM 调度组件、上下文管理组件	调用 LLM 结合文档生成答案，记录对话历史
输出层	展示答案并支持导出	文本格式化组件、文档导出组件	将答案转为清晰格式，支持导出为 Markdown
存储层	保存对话和文档信息	对话存储组件、文档元数据组件	存储用户 - 机器人对话，记录文档上传信息

三、分步开发：从组件初始化到应用运行

1. 第一步：环境准备与组件引入

需提前安装 MCP 组件库（课程提供的标准化库），确保所有组件遵循统一接口标准。环境依赖及安装命令如下：

bash

# 安装MCP核心组件库（含文档解析、LLM调度等基础组件）
pip install mcp-core==1.0.0
# 安装额外依赖（如PDF解析、Markdown导出所需库）
pip install pypdf python-docx python-markdown

引入所需组件的代码：

python

# 输入层组件
from mcp_core.input import DocumentParser, TextInputCleaner
# 处理层组件
from mcp_core.process import LLMCaller, ContextManager
# 输出层组件
from mcp_core.output import TextFormatter, DocExporter
# 存储层组件
from mcp_core.storage import ChatStorage, DocMetadataStorage

2. 第二步：组件初始化（按接口配置参数）

每个组件需通过 “初始化参数” 定义运行规则，参数需严格遵循组件的输入接口标准（避免因参数错误导致组件失效）。具体配置如下：

python

# ---------------------- 输入层组件初始化 ----------------------
# 文档解析组件：支持PDF/Word，允许解密，忽略扫描件
doc_parser = DocumentParser(
    supported_formats=["pdf", "docx"],  # 支持的文件格式
    allow_encrypted=True,              # 允许处理加密文件
    skip_scanned_pdf=True              # 跳过扫描件（无法提取文本）
)

# 文本输入组件：限制提问长度，过滤敏感词
text_cleaner = TextInputCleaner(
    max_length=500,                    # 提问最长500字
    filter_sensitive=True              # 过滤敏感词
)

# ---------------------- 处理层组件初始化 ----------------------
# LLM调度组件：使用GPT-4o，设置低随机性（确保答案贴合文档）
llm_caller = LLMCaller(
    model="gpt-4o",                    # 选用的LLM模型
    api_key="your-openai-api-key",     # 替换为自己的API密钥
    params={"temperature": 0.2}        # 低随机性，答案更精准
)

# 上下文管理组件：保留最近5轮对话（避免上下文过长）
context_manager = ContextManager(
    max_history=5                      # 最多保留5轮对话
)

# ---------------------- 输出层组件初始化 ----------------------
# 文本格式化组件：将答案转为“问题+答案+来源”的结构化格式
text_formatter = TextFormatter(
    format_template="""
    【用户问题】{user_question}
    【回答内容】{answer}
    【参考文档】{doc_name}（第{page_num}页）
    """
)

# 文档导出组件：支持导出为Markdown，文件名含时间戳
doc_exporter = DocExporter(
    export_format="markdown",          # 导出格式
    file_prefix="qa_result_",          # 文件名前缀
    add_timestamp=True                 # 加入时间戳（避免重名）
)

# ---------------------- 存储层组件初始化 ----------------------
# 对话存储组件：存储到本地JSON文件
chat_storage = ChatStorage(
    storage_type="local",              # 本地存储
    file_path="chat_history.json"      # 存储文件路径
)

# 文档元数据组件：记录文档名称、上传时间、页数
doc_meta_storage = DocMetadataStorage(
    storage_type="local",
    file_path="doc_metadata.json"
)

3. 第三步：核心逻辑开发（组件衔接与数据流转）

按 “输入→处理→输出→存储” 的顺序，编写组件间的衔接代码，核心逻辑分为 “文档上传处理” 和 “用户提问回答” 两大模块：

模块 1：文档上传与预处理

python

def process_uploaded_doc(doc_path):
    # 1. 输入层：解析文档
    parse_result = doc_parser.parse(file_path=doc_path)
    if parse_result["status"] == "failed":
        return f"文档解析失败：{parse_result['error']}"
    
    # 2. 存储层：记录文档元数据
    doc_name = doc_path.split("/")[-1]  # 提取文件名
    doc_meta = {
        "doc_name": doc_name,
        "upload_time": parse_result["data"]["parse_time"],
        "page_count": parse_result["data"]["page_count"],
        "file_path": doc_path
    }
    doc_meta_storage.save(metadata=doc_meta)
    
    # 3. 返回处理结果
    return f"文档处理成功！文档名：{doc_name}，页数：{parse_result['data']['page_count']}"

# 调用文档处理函数（示例：上传test.pdf）
print(process_uploaded_doc(doc_path="test.pdf"))

模块 2：用户提问与答案生成

python

def answer_user_question(user_question, doc_name):
    # 1. 输入层：清洗用户提问
    clean_result = text_cleaner.clean(input_text=user_question)
    if not clean_result["data"]["is_valid"]:
        return f"提问无效：{clean_result['error']}"
    cleaned_question = clean_result["data"]["cleaned_text"]
    
    # 2. 存储层：获取文档解析结果（从元数据找到文档路径）
    doc_meta = doc_meta_storage.get_by_name(doc_name=doc_name)
    if not doc_meta:
        return f"未找到文档：{doc_name}"
    doc_content = doc_parser.get_parsed_text(file_path=doc_meta["file_path"])  # 获取已解析的文本
    
    # 3. 处理层：获取对话上下文+生成答案
    # 3.1 加载历史对话
    history = context_manager.get_history(session_id="user_001")  # 假设用户ID为user_001
    # 3.2 构建LLM提示词（结合文档内容和历史对话）
    prompt = f"""
    基于以下文档内容，回答用户问题。若无法从文档中找到答案，直接说明“文档中未提及相关信息”。
    【文档内容】{doc_content}
    【历史对话】{history}
    【用户当前问题】{cleaned_question}
    """
    # 3.3 调用LLM生成答案
    llm_result = llm_caller.generate(prompt=prompt)
    if llm_result["status"] == "failed":
        return f"生成答案失败：{llm_result['error']}"
    raw_answer = llm_result["data"]["output_text"]
    
    # 4. 输出层：格式化答案
    formatted_answer = text_formatter.format(
        user_question=cleaned_question,
        answer=raw_answer,
        doc_name=doc_name,
        page_num=llm_result["data"]["reference_page"]  # LLM返回的参考页数
    )
    
    # 5. 存储层：保存本次对话
    chat_record = {
        "session_id": "user_001",
        "role": "user",
        "content": cleaned_question,
        "timestamp": llm_result["data"]["generate_time"]
    }
    chat_storage.save(record=chat_record)
    chat_storage.save(record={
        "session_id": "user_001",
        "role": "ai",
        "content": formatted_answer,
        "timestamp": llm_result["data"]["generate_time"]
    })
    
    # 6. 更新上下文
    context_manager.update_history(
        session_id="user_001",
        user_msg=cleaned_question,
        ai_msg=formatted_answer
    )
    
    return formatted_answer

# 调用提问函数（示例：基于test.pdf提问）
print(answer_user_question(user_question="文档中提到的MCP架构核心分层有哪些？", doc_name="test.pdf"))

模块 3：答案导出

python

def export_qa_result(session_id):
    # 1. 存储层：获取该用户的所有对话
    chat_history = chat_storage.get_by_session(session_id=session_id)
    if not chat_history:
        return "无对话记录可导出"
    
    # 2. 输出层：导出为Markdown
    export_result = doc_exporter.export(
        content=chat_history,
        save_dir="./exports"  # 导出目录
    )
    
    return f"答案已导出至：{export_result['data']['file_path']}"

# 调用导出函数
print(export_qa_result(session_id="user_001"))

四、实战调试与优化：解决常见问题

在实际运行中，需针对 MCP 组件的特性进行调试，以下是高频问题及解决方案：

1. 问题 1：文档解析后文本乱码

• 原因：文档编码格式不兼容（如 PDF 使用 GBK 编码，组件默认 UTF-8）。
• 解决方案：在文档解析组件初始化时，添加encoding参数：

  python

  doc_parser = DocumentParser(
      supported_formats=["pdf", "docx"],
      allow_encrypted=True,
      skip_scanned_pdf=True,
      encoding="gbk"  # 手动指定编码
  )
  

2. 问题 2：LLM 回答偏离文档内容

• 原因：prompt 中未明确 “仅基于文档回答” 的约束，LLM 引入外部知识。
• 解决方案：优化 LLM 提示词，强化 “文档依赖” 约束（如在 prompt 中加入 “严禁使用文档外的信息，不确定时直接说明”）。

3. 问题 3：对话上下文过长导致 LLM 调用超时

• 原因：上下文管理组件保留的历史对话过多，导致 prompt 长度超出 LLM 限制。
• 解决方案：减小max_history参数（如从 5 轮改为 3 轮），或在上下文管理组件中添加 “文本截断” 功能：

  python

  context_manager = ContextManager(
      max_history=3,
      truncate_long_text=True,  # 开启长文本截断
      max_text_length=1000      # 单条对话最长1000字
  )
  

五、实战总结与后续学习

1. 核心收获

• 掌握 MCP 实战流程：从 “架构拆解→组件选择→参数配置→逻辑衔接”，能独立将 MCP 理论应用到具体场景；
• 理解组件化优势：若需新增 “语音提问” 功能，仅需在输入层添加 “语音转文字组件”，无需修改其他模块；若需切换 LLM，仅需重新初始化llm_caller组件。

2. 后续衔接

下一节将聚焦 “MCP 组件的自定义开发”—— 当现有组件无法满足需求（如需支持特殊格式的文档解析、自定义 LLM 调用逻辑）时，如何按 MCP 接口标准开发专属组件，进一步提升应用的灵活性。