MCP Chatbot 与引用服务器连接：让 AI 拥有 “外部知识检索能力”

一、为什么需要连接引用服务器？—— 突破 Chatbot 的知识局限

前序开发的 MCP Chatbot 仅能基于用户上传的单份文档生成答案，面临 “知识范围窄、重复上传效率低、企业级知识库难复用” 的问题。引用服务器的核心价值，是为 Chatbot 提供集中化、可复用的外部知识源（如企业内部知识库、产品手册库、行业文档库），让 Chatbot 无需依赖用户手动上传文档，就能通过检索引用服务器的存量知识生成答案 —— 简单来说，就是给 Chatbot 配备 “外部知识检索引擎”，使其从 “只能用眼前文档” 升级为 “能查企业所有知识库”，同时解决 “多团队重复上传同一份文档” 的资源浪费问题。

引用服务器带来的核心改变可通过对比体现：

能力维度	未连接引用服务器的 Chatbot	连接引用服务器的 Chatbot
知识来源	仅用户单次上传的 1-2 份文档	引用服务器中的所有知识库（支持数千份文档）
知识复用	不同用户需重复上传同一份文档	文档一次上传到引用服务器，所有 Chatbot 用户可复用
回答准确性	依赖单份文档，易遗漏关键信息	可检索多份相关文档，交叉验证信息准确性
企业级适配	无法对接企业统一知识库	支持企业权限管控（如部门专属知识库仅部门成员可检索）

二、引用服务器的核心能力：为 Chatbot 提供 “知识检索服务”

引用服务器并非简单的 “文档存储库”，而是具备 “语义检索、知识结构化、权限管控” 的专业知识服务系统，其核心能力可拆解为三点：

1. 知识库管理：集中存储与结构化处理

• 文档批量导入：支持通过 API 或后台管理界面，批量上传 PDF/Word/Excel 等格式文档，自动完成 “文本提取、章节拆分、元数据标注”（如标注文档所属部门、更新时间、关键词）；
• 知识结构化：将非结构化文本（如产品手册）转化为 “问答对”“关键知识点” 等结构化数据（例：将 “年假申请流程” 文档拆分为 “年假天数标准”“申请步骤”“审批权限” 等知识点），提升后续检索效率；
• 版本管理：支持文档版本更新，旧版本自动归档，Chatbot 检索时可选择 “最新版本” 或 “历史版本”，避免因文档更新导致答案过时。

2. 语义检索：精准定位 “相关知识片段”

引用服务器的核心技术是语义检索（而非传统关键词匹配），通过向量数据库（如 FAISS、Milvus）将文本转化为向量，再基于 “语义相似度” 匹配用户问题对应的知识片段，解决 “关键词匹配漏检” 问题（例：用户问 “请假多少天需要总监审批”，即使文档中写 “年假超 5 天需总监签字”，语义检索也能精准匹配）。

检索流程可概括为：

1. 用户提问传入引用服务器；
2. 服务器将问题转化为向量；
3. 在向量数据库中匹配 “语义相似度 Top3” 的知识片段；
4. 返回片段内容、所属文档 ID、页码、相似度分数。

3. 权限与引用管控：确保知识安全与可追溯

• 细粒度权限：支持按 “用户角色 / 部门” 设置知识库访问权限（如 “研发部门文档仅研发人员可检索，全员文档所有用户可访问”），权限与 Chatbot 的session_id关联（通过用户身份绑定session_id实现权限校验）；
• 引用追溯：所有 Chatbot 调用的知识片段，均会生成唯一 “引用 ID”，记录 “哪个用户、何时、检索了哪份文档的哪个片段”，方便企业追溯知识使用情况，同时确保答案可溯源（用户可查看答案对应的原始文档片段）。

三、Chatbot 与引用服务器的交互架构：遵循 MCP “模块化衔接” 原则

连接过程严格遵循 MCP 架构的 “低耦合、接口统一” 原则，整体交互流程分为 “问题预处理→检索请求→结果整合→答案生成” 四步，核心是 Chatbot 的 “检索组件” 与引用服务器的 “API 接口” 对接，架构如下：

plaintext

[用户] → [MCP Chatbot（处理层：检索组件+LLM调度组件）] → [引用服务器API接口] → [引用服务器（知识库+语义检索引擎）]
                                 ↓                                  ↑
                                 ←———————  返回检索结果（知识片段） ————————→

1. 核心交互角色与职责

• MCP Chatbot（处理层新增 “检索组件”）：负责发起检索请求（将用户问题转化为检索参数）、接收并整合检索结果（筛选高相似度片段）、将片段融入 LLM prompt 生成答案；
• 引用服务器（API 层）：提供标准化检索 API（接收检索请求）、调用内部语义检索引擎（匹配知识片段）、返回结构化检索结果（含片段内容、引用 ID、文档信息）；
• 引用服务器（数据层）：存储知识库文档、向量数据、权限配置，为检索引擎提供数据支撑。

2. 关键数据流转格式

为确保 Chatbot 与引用服务器无缝衔接，需定义统一的 “请求 - 响应” 数据格式（JSON），核心字段需与 MCP 架构的 “session_id”“文档元数据” 保持一致：

（1）Chatbot→引用服务器的检索请求格式

json

{
  "session_id": "user_001",  // 关联用户身份，用于权限校验
  "user_question": "公司年假申请超过多少天需要总监审批？",  // 用户原始问题
  "retrieval_params": {
    "top_k": 3,  // 返回相似度Top3的知识片段
    "knowledge_base_ids": ["hr_base", "company_policy"],  // 指定检索的知识库ID（可选，默认检索所有有权限的库）
    "similarity_threshold": 0.7  // 相似度阈值（低于0.7的片段不返回，避免无关信息）
  }
}

（2）引用服务器→Chatbot 的检索响应格式

json

{
  "status": "success",
  "data": {
    "retrieval_id": "ret_8f2d7c",  // 本次检索的唯一ID（用于追溯）
    "references": [  // 匹配的知识片段列表
      {
        "reference_id": "ref_123",  // 片段唯一ID（用于答案溯源）
        "doc_id": "doc_456",  // 片段所属文档ID
        "doc_name": "公司考勤与休假政策2024.pdf",  // 文档名称
        "page_num": 12,  // 片段所在页码
        "content": "员工年假申请天数超过5天（含5天）时，需由部门总监审批；超过10天需由CEO审批。",  // 知识片段内容
        "similarity_score": 0.92  // 与用户问题的相似度（0-1）
      },
      {
        "reference_id": "ref_124",
        "doc_id": "doc_789",
        "doc_name": "研发部门休假补充说明.docx",
        "page_num": 3,
        "content": "研发部门员工因项目紧急需延后休假的，需提前3天在系统报备，审批流程与普通年假一致（超5天需总监审批）。",
        "similarity_score": 0.85
      }
    ]
  },
  "error": null
}

四、分步实现：MCP Chatbot 与引用服务器的连接

1. 前提准备：引用服务器 API 接口开发（基于 FastAPI）

首先需在引用服务器端开发标准化检索 API，确保 Chatbot 可通过 HTTP 请求调用。核心代码如下（延续 MCP 技术栈，使用 FastAPI 实现）：

python

运行

# 引用服务器检索API（server_reference/main.py）
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
import faiss  # 向量数据库，用于语义检索
import numpy as np
from typing import List, Dict

# 初始化FastAPI应用
app = FastAPI(title="MCP Reference Server")

# 模拟知识库（实际场景中从数据库/文件加载）
# 结构：{知识库ID: {文档ID: {doc_name: 文档名, content: 文档内容, vectors: 文本向量}}}
KNOWLEDGE_BASES = {
    "hr_base": {  # 人力资源知识库
        "doc_456": {
            "doc_name": "公司考勤与休假政策2024.pdf",
            "page_num": 12,
            "content": "员工年假申请天数超过5天（含5天）时，需由部门总监审批；超过10天需由CEO审批。",
            "vector": np.random.rand(768).astype("float32")  # 模拟768维向量（实际由LLM生成）
        }
    },
    "company_policy": {  # 公司通用政策知识库
        "doc_789": {
            "doc_name": "研发部门休假补充说明.docx",
            "page_num": 3,
            "content": "研发部门员工因项目紧急需延后休假的，需提前3天在系统报备，审批流程与普通年假一致（超5天需总监审批）。",
            "vector": np.random.rand(768).astype("float32")
        }
    }
}

# 初始化FAISS向量索引（模拟，实际需批量导入所有文档向量）
index = faiss.IndexFlatL2(768)  # 768维向量的L2距离索引
# 构建向量与文档的映射关系
vector_doc_map = {}  # {向量索引: {doc_id, knowledge_base_id, ...}}
vector_idx = 0
for kb_id, docs in KNOWLEDGE_BASES.items():
    for doc_id, doc_info in docs.items():
        index.add(np.array([doc_info["vector"]]))
        vector_doc_map[vector_idx] = {
            "kb_id": kb_id,
            "doc_id": doc_id,
            "doc_name": doc_info["doc_name"],
            "page_num": doc_info["page_num"],
            "content": doc_info["content"]
        }
        vector_idx += 1

# 定义检索请求模型（与Chatbot请求格式对齐）
class RetrievalRequest(BaseModel):
    session_id: str
    user_question: str
    retrieval_params: Dict = {
        "top_k": 3,
        "similarity_threshold": 0.7,
        "knowledge_base_ids": []  # 空表示检索所有有权限的知识库
    }

# 权限校验依赖（模拟：根据session_id判断用户是否有权限访问知识库）
def check_permission(session_id: str, target_kb_ids: List[str]) -> List[str]:
    # 实际场景中：从用户数据库查询session_id对应的用户角色，再匹配知识库权限
    # 此处简化：所有用户可访问"hr_base"和"company_policy"
    allowed_kbs = ["hr_base", "company_policy"]
    if not target_kb_ids:
        return allowed_kbs
    # 筛选用户有权限的知识库
    return [kb_id for kb_id in target_kb_ids if kb_id in allowed_kbs]

# 检索API（核心接口）
@app.post("/api/reference/retrieve")
def retrieve_knowledge(
    request: RetrievalRequest,
    allowed_kbs: List[str] = Depends(lambda req: check_permission(req.session_id, req.retrieval_params["knowledge_base_ids"]))
):
    # 1. 提取检索参数
    top_k = request.retrieval_params.get("top_k", 3)
    similarity_threshold = request.retrieval_params.get("similarity_threshold", 0.7)
    
    # 2. 模拟：将用户问题转化为向量（实际需调用LLM的Embedding接口，如OpenAI Embeddings）
    question_vector = np.random.rand(768).astype("float32").reshape(1, -1)
    
    # 3. 向量检索（FAISS计算相似度）
    distances, indices = index.search(question_vector, top_k)  # distances为距离（越小相似度越高）
    similarity_scores = 1 - (distances / np.max(distances))  # 将距离转化为0-1的相似度分数
    
    # 4. 筛选符合条件的引用片段（权限+相似度阈值）
    references = []
    for idx, vector_idx in enumerate(indices[0]):
        doc_info = vector_doc_map.get(vector_idx, None)
        if not doc_info or doc_info["kb_id"] not in allowed_kbs:
            continue
        score = similarity_scores[0][idx]
        if score >= similarity_threshold:
            references.append({
                "reference_id": f"ref_{vector_idx}_{idx}",
                "doc_id": doc_info["doc_id"],
                "doc_name": doc_info["doc_name"],
                "page_num": doc_info["page_num"],
                "content": doc_info["content"],
                "similarity_score": round(score, 2)
            })
    
    # 5. 返回检索结果
    if not references:
        return {
            "status": "success",
            "data": {
                "retrieval_id": f"ret_{request.session_id}_{hash(request.user_question)}",
                "references": [],
                "message": "未检索到相关知识片段，建议调整问题或上传补充文档"
            },
            "error": None
        }
    
    return {
        "status": "success",
        "data": {
            "retrieval_id": f"ret_{request.session_id}_{hash(request.user_question)}",
            "references": references
        },
        "error": None
    }

启动引用服务器：

bash

uvicorn server_reference.main:app --host 0.0.0.0 --port 8001  # 端口8001，与MCP主服务器区分

2. Chatbot 端新增 “引用检索组件”：对接引用服务器 API

在 MCP Chatbot 的处理层新增ReferenceRetriever组件，负责发起检索请求、处理检索结果，与原有LLMCaller组件协同工作。核心代码如下：

python

运行

# MCP Chatbot处理层：新增引用检索组件（mcp_core/process/reference_retriever.py）
import requests
from typing import Dict, List

class ReferenceRetriever:
    def __init__(self, reference_server_url: str):
        self.server_url = reference_server_url  # 引用服务器地址（如"http://127.0.0.1:8001"）
    
    def retrieve(self, session_id: str, user_question: str, retrieval_params: Dict = None) -> Dict:
        """
        调用引用服务器检索知识片段
        :param session_id: 用户会话ID（用于权限校验）
        :param user_question: 用户原始问题
        :param retrieval_params: 检索参数（top_k、知识库ID等）
        :return: 检索结果（含引用片段列表）
        """
        # 1. 构造请求数据（与引用服务器API格式对齐）
        request_data = {
            "session_id": session_id,
            "user_question": user_question,
            "retrieval_params": retrieval_params or {}
        }
        
        # 2. 发起HTTP请求（使用Requests库，保持与MCP其他组件一致）
        try:
            response = requests.post(
                url=f"{self.server_url}/api/reference/retrieve",
                json=request_data,
                timeout=10  # 超时时间10秒，避免阻塞Chatbot
            )
            response.raise_for_status()  # 检查请求是否成功（如404、500会抛出异常）
            return response.json()
        except Exception as e:
            # 异常处理：返回失败状态，避免影响Chatbot主流程
            return {
                "status": "failed",
                "data": {},
                "error": f"引用服务器检索失败：{str(e)}"
            }
    
    def format_references(self, references: List[Dict]) -> str:
        """
        将引用片段格式化为LLM可理解的文本（用于融入prompt）
        :param references: 引用服务器返回的references列表
        :return: 格式化后的引用文本
        """
        if not references:
            return "无相关外部知识引用。"
        
        formatted_text = "【外部知识引用】\n"
        for idx, ref in enumerate(references, 1):
            formatted_text += f"""
{idx}. 文档名称：{ref['doc_name']}（第{ref['page_num']}页）
相似度：{ref['similarity_score']}
内容：{ref['content']}
引用ID：{ref['reference_id']}
"""
        return formatted_text

3. 整合检索逻辑与 LLM 生成：实现 “检索 - 生成” 一体化

修改 Chatbot 的核心提问处理逻辑，新增 “先检索引用、再生成答案” 的流程 —— 用户提问后，Chatbot 先调用ReferenceRetriever获取相关知识片段，再将片段融入 LLM prompt，确保答案基于外部知识库生成。核心代码修改如下：

python

运行

# MCP Chatbot核心逻辑（整合引用检索）
from mcp_core.process import LLMCaller, ReferenceRetriever, ContextManager

# 初始化组件（新增引用检索组件）
llm_caller = LLMCaller(model="gpt-4o", api_key="your-api-key", params={"temperature": 0.2})
ref_retriever = ReferenceRetriever(reference_server_url="http://127.0.0.1:8001")  # 引用服务器地址
context_manager = ContextManager(max_history=5)

def handle_user_question(session_id: str, user_question: str) -> str:
    # 1. 步骤1：调用引用服务器检索外部知识
    retrieval_result = ref_retriever.retrieve(
        session_id=session_id,
        user_question=user_question,
        retrieval_params={
            "top_k": 2,  # 取Top2最相关的片段
            "knowledge_base_ids": ["hr_base"],  # 仅检索人力资源知识库
            "similarity_threshold": 0.65
        }
    )
    
    # 2. 步骤2：处理检索结果，格式化引用文本
    if retrieval_result["status"] == "success":
        references = retrieval_result["data"]["references"]
        formatted_refs = ref_retriever.format_references(references)
        retrieval_msg = f"已检索到{len(references)}条相关知识片段"
    else:
        formatted_refs = "无相关外部知识引用（检索失败：" + retrieval_result["error"] + "）"
        retrieval_msg = "外部知识检索失败，将基于历史对话生成答案"
    
    # 3. 步骤3：获取对话上下文
    history = context_manager.get_history(session_id=session_id)
    
    # 4. 步骤4：构建包含引用的LLM prompt
    prompt = f"""
基于以下信息回答用户问题：
1. 外部知识引用：
{formatted_refs}

2. 历史对话：
{history}

3. 用户当前问题：
{user_question}

回答要求：
- 优先使用外部知识引用的内容，若引用内容冲突，以相似度高的为准；
- 明确标注答案引用的片段（格式：【引用自：文档名（页码），引用ID】）；
- 若没有外部知识引用，需明确说明“未使用外部知识库”；
- 避免编造信息，无法回答时直接说明。
"""
    
    # 5. 步骤5：调用LLM生成答案
    llm_result = llm_caller.generate(prompt=prompt)
    if llm_result["status"] == "failed":
        return f"答案生成失败：{llm_result['error']}\n{retrieval_msg}"
    
    # 6. 步骤6：更新上下文与返回结果
    formatted_answer = llm_result["data"]["output_text"]
    context_manager.update_history(
        session_id=session_id,
        user_msg=user_question,
        ai_msg=f"{formatted_answer}\n\n{retrieval_msg}"
    )
    
    return formatted_answer

# 测试：处理用户提问
print(handle_user_question(
    session_id="user_001",
    user_question="公司年假申请超过多少天需要总监审批？"
))

生成的答案示例（含引用标注）：

plaintext

公司年假申请超过5天（含5天）时，需由部门总监审批；若超过10天，则需由CEO审批【引用自：公司考勤与休假政策2024.pdf（第12页），引用ID：ref_0_0】。
此外，研发部门员工因项目紧急需延后休假的，需提前3天在系统报备，审批流程与普通年假一致（超5天需总监审批）【引用自：研发部门休假补充说明.docx（第3页），引用ID：ref_1_1】。

已检索到2条相关知识片段

五、联调关键问题与解决方案

1. 问题 1：引用服务器检索结果为空（相似度低）

• 原因：1. 用户问题表述模糊（如 “请假审批流程” 未提 “年假”）；2. 文档向量生成质量低（如用低精度 LLM 生成 Embedding）；3. 相似度阈值设置过高（如 0.8 以上）。
• 解决方案：
  1. 在 Chatbot 端添加 “问题优化” 逻辑（如将 “请假审批流程” 补充为 “公司年假请假审批流程”）；
  2. 改用高精度 Embedding 模型（如 OpenAI text-embedding-3-large）生成文档向量；
  3. 动态调整相似度阈值（检索结果为空时自动降低至 0.6）。

2. 问题 2：Chatbot 无权限访问指定知识库

• 原因：引用服务器的权限校验未通过（如session_id对应的用户角色无 “研发知识库” 访问权限）。
• 解决方案：
  1. 在 Chatbot 端添加 “权限提示”（如检索结果返回 “无权限访问研发知识库，请联系管理员开通”）；
  2. 引用服务器 API 返回 “无权限” 时，自动检索用户有权限的知识库（而非直接返回空结果）。

3. 问题 3：引用服务器调用超时，阻塞 Chatbot

• 原因：引用服务器检索大量文档时耗时过长（如超过 10 秒），导致 Chatbot 无法响应。
• 解决方案：
  1. 在ReferenceRetriever中添加 “超时重试” 逻辑（超时后重试 1 次，仍失败则返回本地处理）；
  2. 引用服务器端优化检索性能（如对知识库按部门分区，减少单次检索的数据量）；
  3. 采用异步调用（如用asyncio+aiohttp替代requests），避免阻塞 Chatbot 主流程。

六、实战案例：企业 HR Chatbot 对接引用服务器

以 “企业 HR Chatbot” 为例，展示连接引用服务器后的完整流程：

1. 管理员操作：将 “公司考勤政策”“社保缴纳说明”“员工入职手册” 等 10 份文档批量上传到引用服务器的 “HR 知识库”，设置 “所有员工可访问” 权限；
2. 员工操作：员工通过 Chatbot 提问 “社保断缴后如何补缴？”；
3. Chatbot 流程：
   • 调用引用服务器检索 “HR 知识库”，匹配到 “社保缴纳说明.pdf” 的 2 条相关片段；
   • 将片段融入 LLM prompt，生成含引用标注的答案；
4. 员工反馈：员工点击答案中的 “引用 ID”，可跳转查看完整文档（引用服务器支持文档预览）。

七、总结与后续扩展

1. 本集核心收获

• 理解引用服务器的核心价值：为 Chatbot 提供集中化、可复用的外部知识库，突破单文档知识局限；
• 掌握连接实现流程：引用服务器 API 开发→Chatbot 检索组件开发→“检索 - 生成” 逻辑整合；
• 解决联调高频问题：相似度低、权限不足、调用超时等，确保连接稳定性。

2. 后续扩展方向

• 多引用服务器负载均衡：对接多台引用服务器（按知识库类型分区），通过负载均衡提升检索效率；
• 引用片段优先级排序：按 “文档更新时间 + 相似度” 综合排序（如最新文档的片段优先级高于旧文档）；
• 引用结果编辑功能：管理员可在引用服务器端编辑知识片段（如修正错误信息），Chatbot 实时获取更新后内容；
• 多模态引用支持：引用服务器新增 “图片 / 表格” 检索（如产品手册中的截图），Chatbot 可引用图片链接并在答案中展示。