微信扫码
添加专属顾问
 
                        我要投稿
企业数字化转型中,如何高效管理和利用内部知识资产?MCP帮你打造高性能企业RAG系统,实现智能知识库应用。 核心内容: 1. RAG技术在企业知识管理中的挑战与局限 2. MCP解决方案的优势与企业知识管理需求 3. 基于MCP的企业RAG系统设计与实现目标
 
                                
在企业数字化转型浪潮中,如何有效管理和利用内部知识资产已成为关键挑战。随着大型语言模型(LLM)技术的成熟,检索增强生成(RAG)应用正逐渐成为连接企业知识与AI能力的重要桥梁。然而,传统RAG实现常面临检索质量不佳、实时更新困难等痛点问题。
本文将通过实战案例,详细介绍如何基于模型上下文协议(MCP)构建一套高性能企业RAG系统,帮助企业快速打造智能知识库应用。
传统RAG实现通常采用简单的"Embedding+检索+LLM生成"架构,存在以下限制:
基于MCP的RAG系统通过标准化协议,将知识检索服务解耦为独立模块,带来以下优势:
现代企业面临的知识管理挑战主要表现在以下几个方面:
针对这些问题,我们需要设计一个系统满足以下核心需求:
本项目旨在构建一个基于MCP的企业RAG系统,实现以下具体目标:
本项目系统设计参考自alibabacloud-tablestore-mcp-server[1],由于
alibabacloud-tablestore-mcp-server项目使用Tablestore存储和Java实现的MCP Server,不方便于后期扩展和迭代。本项目改造为Milvus存储和Python实现MCP Server和MCP Client,代码全部重写(cursor帮忙不少)。
以下设计和流程皆为
alibabacloud-tablestore-mcp-server内容,在此感谢@xjtushilei 开源的alibabacloud-tablestore-mcp-server。
我们构建的基于MCP的RAG系统主要包含三个核心部分:
主要分为两部分:知识库构建和检索。
相比传统的 Naive RAG,在知识库构建和检索分别做了一些常见的优化,包括 Chunk 切分优化、提取 FAQ、Query Rewrite、混合检索等。
本Agent整体架构分为三个部分:
项目结构分为两部分:
milvus-mcp-client: Python 实现的 Client 端,实现了与大模型进行交互,通过 MCP Client 获取 Tools,根据大模型的反馈调用 Tools 等基本能力。通过 Prompt 实现了知识库构建、检索和问答三个主要功能。milvus-mcp-server:** Python 实现的 Server 端,基于 MCP 框架实现的服务,提供了连接 Milvus 向量数据库的接口,支持知识库的存储和检索功能。接下来,我们将从环境搭建、服务部署到功能测试,全面介绍如何搭建一个基于MCP的RAG系统。
首先,确保满足以下系统要求:
git clone -b rag_0.1.1 https://github.com/FlyAIBox/mcp-in-action.gitMCP Server基于Milvus向量数据库,提供了知识库的存储和检索功能。
对于需要进行开发或调试的场景,可以选择本地部署:
# 进入项目目录
cd mcp-rag
# 先启动Milvus及依赖服务
docker compose up -d etcd minio standalone
# 创建Python虚拟环境
python -m venv env-mcp-rag
source env-mcp-rag/bin/activate  
# 安装依赖
pip install -r requirements.txt
# 启动服务
python -m app.mainMCP Server提供四个核心工具,支持知识库的读写操作:
让我们看看这些API的实际实现原理:
async defstore_knowledge(self, content: str, metadata: Dict[str, Any] = None) -> Dict[str, Any]:
    """存储知识内容到Milvus"""
    # 确保服务准备就绪
    awaitself.ready_for_connections()
    
    try:
        knowledge_content = KnowledgeContent(
            content=content,
            metadata=metadata or {}
        )
        self.milvus_service.store_knowledge(knowledge_content)
        return {"status": "success", "message": "Knowledge stored successfully"}
    except Exception as e:
        logger.error(f"Error storing knowledge: {e}")
        return {"status": "error", "message": str(e)}这段代码展示了storeKnowledge工具的实现:接收文本内容和元数据,创建知识内容对象,然后通过Milvus服务存储到向量数据库中。
接下来,我们需要实现一个RAG客户端,通过MCP协议与Server通信,实现知识库的构建和查询功能。
• 文本切分:对长文本进行智能切分,保证语义完整性 
• FAQ提取:从文档自动生成常见问题解答对 
• 向量化存储:将文本片段和FAQ转换为向量并存入Milvus 
文本切分代码示例:
def _chunk_text(self, text: str) -> List[str]:
    """将文本分割成chunk,保证语义完整性"""
    chunks = []
    
    # 处理文本小于chunk_size的简单情况
    iflen(text) <= self.chunk_size:
        chunks.append(text)
        return chunks
        
    # 使用重叠策略分割文本
    start = 0
    while start < len(text):
        # 获取chunk结束位置
        end = start + self.chunk_size
        
        # 调整结束位置,避免在句子中间切断
        if end < len(text):
            # 寻找句子边界(句号、问号、感叹号)
            sentence_end = max(
                text.rfind('. ', start, end),
                text.rfind('? ', start, end),
                text.rfind('! ', start, end)
            )
            
            # 如果找到句子结束,使用它作为chunk结束
            if sentence_end > start:
                end = sentence_end + 1# 包含句号
        
        # 添加chunk
        chunks.append(text[start:min(end, len(text))])
        
        # 移动开始位置到下一个chunk,考虑重叠
        start = end - self.chunk_overlap
        
        # 确保进度
        if start >= len(text) or start <= 0:
            break
            
    return chunksFAQ提取,通过LLM实现:
async def_extract_faqs(self, text: str) -> List[Dict[str, str]]:
    """从文本中提取FAQ"""
    # 对过长文本进行分块处理
    iflen(text) > 8000:
        chunks = self._chunk_text(text)
        faqs = []
        for chunk in chunks:
            chunk_faqs = awaitself._extract_faqs(chunk)
            faqs.extend(chunk_faqs)
        return faqs
        
     # FAQ提取的提示模板
     system_prompt = 
        """你是一位专业的知识提取专家。你的任务是从文本中提取可能的常见问题(FAQ)。
        这些问题应该是用户可能会问的关于文本内容的自然问题,答案应该能在文本中找到。
        提取的FAQ应该覆盖文本中最重要的概念和信息。
        请遵循以下规则:
        1. 每个FAQ由一个问题和一个答案组成
        2. 问题应该简短明了,直接针对主题
        3. 答案应该全面但简洁,提供文本中的相关信息
        4. 提取的FAQ数量应该基于文本长度和内容丰富度,通常不超过10个
        5. 确保提取的FAQ相互之间不重复
        6. 按照重要性排序,最重要的问题应该放在前面
        输出格式必须是一个JSON数组,每个FAQ是一个包含"question"和"answer"字段的对象,例如:
        [
          {
            "question": "问题1?",
            "answer": "答案1"
          },
          {
            "question": "问题2?",
            "answer": "答案2"
          }
        ]
        只输出JSON格式,不要有任何其他文本。"""
    user_prompt = f"""从以下文本中提取常见问题(FAQ):
    ```
    {text}
    ```
    请提取最相关、最有价值的FAQ,并按JSON格式返回。"""
   
   # 使用LLM提取FAQ
   response = self.llm_client.sync_generate(
       prompt=text,
       system_prompt=system_prompt,
       temperature=0.3
   )
   
   # 解析LLM响应获取FAQ
   # ...
与传统RAG不同,我们在检索环节引入了问题拆解、混合检索和结果筛选三个优化机制。
• 问题拆解:将复杂问题拆解为多个子问题 
• 混合检索:同时检索文本库和FAQ库,提高召回率 
• 结果筛选:对检索结果进行排序和筛选,优先保留高质量内容 
问题拆解示例:
async def_decompose_question(self, question: str) -> List[str]:
    """将复杂问题分解为更简单的子问题"""
    system_prompt = 
    """你是一位问题分析专家。你的任务是将复杂问题分解为更简单的子问题,以便更好地检索相关信息。
    请遵循以下规则:
    1. 分析用户的问题,识别其中包含的不同方面或概念
    2. 将复杂问题拆分成更简单、更具体的子问题
    3. 确保子问题覆盖原始问题的所有关键方面
    4. 提供2-4个子问题,具体数量取决于原始问题的复杂度
    5. 子问题应该是明确的、有针对性的
    6. 子问题之间应该尽量避免重复
    输出格式必须是一个JSON数组,包含所有子问题的字符串,例如:
    ["子问题1", "子问题2", "子问题3"]
    如果原始问题已经足够简单,不需要分解,则返回只包含原始问题的JSON数组:
    ["原始问题"]
    只输出JSON格式,不要有任何其他文本。"""
    user_prompt = f"""请将以下问题分解为更简单的子问题以便检索:{question}"""
    
    # 使用LLM生成子问题
    response = self.llm_client.sync_generate(
        prompt=user_prompt,
        system_prompt=system_prompt,
        temperature=0.3
    )
    
    # 解析响应获取子问题列表
    # ...结果筛选与生成回答的关键代码:
async def_filter_context(self, question: str, context_items: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """根据问题相关性筛选上下文"""
    # 简单筛选:去重和截断
    seen_contents = set()
    filtered_items = []
    
    # 优先处理FAQ类型
    faq_items = [item for item in context_items if item["type"] == "faq"]
    knowledge_items = [item for item in context_items if item["type"] == "knowledge"]
    
    # 先处理FAQ项
    for item in faq_items:
        # 去重处理
        # ...
    
    # 再处理知识项
    for item in knowledge_items:
        # 去重处理
        # ...
    
    # 限制上下文项总数
    max_context_items = 6
    iflen(filtered_items) > max_context_items:
        filtered_items = filtered_items[:max_context_items]
        
    return filtered_items部署完成后,让我们看看系统的实际运行效果:
python -m app.main build --file test.md --title "RAG基本介绍" --author "企业知识库" --tags "LLM,RAG,知识库"
执行结果:
2025-05-11 14:50:16 | INFO | app.knowledge_builder:build_from_text:52 - Split text into 2 chunks
2025-05-11 14:50:59 | INFO | app.knowledge_builder:build_from_text:72 - Extracted 8 FAQs from text
2025-05-11 14:51:00 | INFO | __main__:build_knowledge_base:48 - Stored 2/2 chunks to knowledge base
2025-05-11 14:51:00 | INFO | __main__:build_knowledge_base:50 - Extracted and stored 8 FAQspython -m app.main query --question "RAG相比企业传统的知识库有什么优势和缺点"
执行结果:
2025-05-11 15:01:46 | INFO | app.knowledge_retriever:query:39 - Decomposed question into 4 sub-questions
2025-05-11 15:01:47 | INFO | app.knowledge_retriever:query:67 - Filtered 28 context items to 6
================================================================================
问题: RAG相比企业传统的知识库有什么优势和缺点
--------------------------------------------------------------------------------
回答: 检索增强生成(RAG)是一种通过整合训练数据之外的权威知识库来优化大型语言模型(LLM)输出的技术。其核心在于允许LLM在生成响应前动态访问特定领域或组织的内部知识库,例如实时数据源、文档或专业数据库,而无需对模型本身进行重新训练。这种方式通过引入外部信息,显著提升了生成内容的相关性、准确性及实用性,同时保留了LLM的灵活性和泛化能力。
================================================================================根据实际项目经验,我们总结了以下最佳实践:
基于MCP实现的RAG系统代表了知识库建设的新方向。通过模型上下文协议,我们不仅解决了传统RAG系统中的诸多痛点,还为企业提供了一种低成本、高效率的知识管理方案。
未来,随着大模型技术的进步和MCP标准的完善,我们可以期待更多创新功能的出现:
对于企业而言,现在正是开始探索和应用这一技术的最佳时机。通过MCP-RAG,企业可以充分挖掘自身知识资产的价值,为员工和客户提供更智能、更精准的信息服务。
53AI,企业落地大模型首选服务商
产品:场景落地咨询+大模型应用平台+行业解决方案
承诺:免费POC验证,效果达标后再合作。零风险落地应用大模型,已交付160+中大型企业
 
            2025-09-15
2025-09-02
2025-08-05
2025-08-18
2025-08-25
2025-08-25
2025-08-25
2025-09-03
2025-08-20
2025-09-08
2025-10-04
2025-09-30
2025-09-10
2025-09-10
2025-09-03
2025-08-28
2025-08-25
2025-08-20