企业知识库问答Agent实战:RAG+Agent+MCP架构与优化

📅 发布时间:2026/10/1 13:01:41
企业知识库问答Agent实战:RAG+Agent+MCP架构与优化
1. 企业知识库问答 Agent 的整体设计思路1.1 为什么企业需要自己的知识库问答 Agent企业知识库问答 Agent 这件事本质上解决的是一个非常朴素的问题公司里的文档太多了没人看得过来也没人知道哪份文档里写了什么。产品手册、运维手册、合同模板、历史项目复盘、会议纪要、FAQ、内部 Wiki这些东西散落在 Confluence、飞书文档、SharePoint、Git 仓库、甚至某些同事的本地硬盘里。新人入职第一周最痛苦的事情不是学技术而是找文档老员工最烦的事情不是干活而是被反复问“这个流程在哪份文档里写过”。传统的做法是建一个搜索系统比如 Elasticsearch 全文检索。但全文检索有个致命问题它匹配的是关键词不是语义。你搜“报销流程”它可能给你返回十份提到“报销”两个字的文档但真正讲流程的那份可能排在第八位。用户需要的是答案不是文档列表。这就是 RAGRetrieval-Augmented Generation检索增强生成要解决的核心痛点。RAG 的思路很直接先把企业文档切片、向量化、存进向量数据库用户提问时先把问题也向量化从库里召回最相关的若干片段再把这些片段作为上下文喂给大语言模型让模型基于这些片段生成答案。这样模型不需要记住所有知识它只需要会“阅读理解”就行。企业知识库问答 Agent 就是在 RAG 基础上再加一层 Agent 编排不只是问答还能根据问题类型决定要不要查库、查哪个库、要不要调用外部工具、要不要多轮追问。适合谁来参考这套方案我认为三类人最需要一是企业内部负责数字化或效率工具的同学想给公司搭一个能用的知识助手二是做 To B 产品的开发者客户经常提“能不能接入我们自己的文档”三是想学习 Agent 和 RAG 落地的中高级工程师因为企业知识库问答是目前 RAG 最成熟、最容易看到效果的应用场景之一。1.2 技术选型的核心考量为什么是 RAG Agent MCP先说 RAG。很多人会问现在大模型上下文窗口都到 128K 甚至 1M 了直接把所有文档塞进去不行吗理论上行实际上不行。第一成本扛不住每次提问都塞几十万字token 费用会爆炸第二效果反而会下降上下文太长时模型注意力会分散关键信息容易被淹没第三企业文档是动态更新的你不可能每次更新都重新训练或重新灌入全部内容。RAG 的本质是“按需检索”只把最相关的片段给模型既省钱又准。再说 Agent。纯 RAG 是一个线性流程检索、拼接、生成。但真实企业场景里用户的问题往往不是一句话能解决的。比如“上个月华东区的销售数据为什么下滑”这个问题需要先查销售报表再查市场活动记录可能还要对比去年同期数据。这就需要 Agent 来做任务分解和工具调用。Agent 的核心价值在于“决策”它决定下一步做什么而不是被动地走固定流程。最后说 MCPModel Context Protocol。MCP 是一个让模型和外部工具、数据源之间标准化通信的协议。你可以把它理解成“AI 世界的 USB 接口”。以前每接一个工具就要写一套适配代码MCP 把这些适配标准化了。企业知识库问答 Agent 接入 MCP 之后可以很方便地连接内部 Wiki、数据库、工单系统、CRM而不需要为每个系统单独开发插件。这也是为什么最近 MCP 在企业 Agent 落地里被频繁提及。注意MCP 是软件层面的协议标准不是硬件协议。硬件领域类似的概念叫总线或接口标准比如 USB、PCIe。两者不在一个层面不要混淆。1.3 整体架构分层我把这套系统分成四层从下往上依次是数据层企业文档的采集、清洗、切片、向量化、存储。这一层决定了知识库的上限。检索层向量检索、关键词检索、混合检索、重排序。这一层决定了召回质量。Agent 编排层意图识别、任务规划、工具调用、多轮对话管理。这一层决定了交互体验。接入层Web 界面、企业 IM 集成、API 接口、MCP Server。这一层决定了能不能被用起来。很多团队做知识库问答失败不是因为模型不够强而是因为数据层没做好。文档切片切得乱七八糟向量化用了一个不适合中文的模型检索层只做了单一向量检索没有重排序最后模型再强也救不回来。所以下面我会重点讲数据层和检索层的细节。2. 核心细节解析与实操要点2.1 文档采集与清洗脏数据是最大的敌人企业文档的来源极其复杂。PDF、Word、Excel、PPT、Markdown、HTML、扫描件、甚至聊天记录截图。不同格式的解析难度完全不同。PDF 里最麻烦的是多栏排版和表格很多解析库会把两栏文字混在一起读导致切片后语义断裂。扫描件需要 OCR但 OCR 的准确率直接影响后续检索。我的经验是采集阶段一定要做“来源分级”。把文档按质量分成三档来源类型解析难度建议处理方式Markdown / HTML低直接解析保留标题层级Word / PPT中用 python-docx / python-pptx 提取注意表格单独处理PDF文本型中高用 PyMuPDF 或 pdfplumber按版面分析切分PDF扫描型高OCR 后人工抽检错误率高的文档标记为低置信Excel中按 sheet 和表头转成自然语言描述不要直接塞表格清洗阶段要做几件事去掉页眉页脚、去掉重复的免责声明、统一全半角、修复断行。特别是 PDF 里的断行经常出现“企业知识\n库问答”这种情况如果不修复向量化后语义会偏。我一般用正则先把中文之间的换行去掉再保留段落之间的空行。实操心得不要试图一次性清洗所有文档。先拿 50 份核心文档跑通全流程验证效果后再批量处理。否则你会在清洗阶段耗掉大量时间却不知道最终效果如何。2.2 切片策略固定长度是下策语义切片是中策层级切片是上策切片是 RAG 里最容易被低估的环节。很多人直接用 LangChain 的 RecursiveCharacterTextSplitter设个 chunk_size500overlap50就完事了。这在简单场景能用但企业文档往往有复杂的层级结构比如“第 3 章 3.2 节 3.2.1 小节”固定长度切片会把层级信息切没。我的做法是“层级感知切片”先按标题层级切分成大块如果某一块超过阈值比如 800 字再按段落切如果段落还超再按句子切。每个切片都带上它的层级路径作为元数据比如{chapter: 第3章, section: 3.2, title: 报销流程}。这样检索时可以把层级路径一起展示给模型帮助它理解上下文。切片长度怎么定这取决于你的嵌入模型和生成模型。一般来说中文场景下 300 到 600 字是一个比较舒服的区间。太短了语义不完整太长了检索精度下降。overlap 建议设成 chunk_size 的 10% 到 15%保证跨切片的句子不会被切断语义。还有一个细节表格和代码块不要切。表格切了之后行列对应关系就丢了代码块切了之后语法就不完整了。遇到表格和代码块要么整体作为一个切片要么转成自然语言描述再切。2.3 向量化模型选择中文场景别直接用 OpenAI 的默认模型向量化模型决定了语义检索的质量。英文场景下 text-embedding-ada-002 或 text-embedding-3-small 都还行但中文场景下这些模型的表现会打折扣。国内可选的中文嵌入模型有不少比如 BGE 系列、M3E、GTE 等。我实测下来BGE-large-zh 在中文企业文档上的召回率明显优于通用英文模型。如果你要私有化部署BGE 系列可以本地跑用 sentence-transformers 加载就行。如果追求更高精度可以用 BGE-M3它支持多语言、多粒度还能同时做稠密检索和稀疏检索。代价是模型更大推理更慢需要 GPU。这里有个容易踩的坑向量维度。不同模型的输出维度不同BGE-large-zh 是 1024 维text-embedding-3-small 是 1536 维。如果你中途换模型必须重新向量化整个库否则新旧向量不在同一个空间里检索结果会完全乱掉。所以选模型要慎重最好一开始就定好。2.4 向量数据库选型别一上来就上分布式向量数据库的选择很多Milvus、Qdrant、Weaviate、Chroma、PGVector、Redis Vector 等等。我的建议是数据量在百万级以下直接用 PGVector 或 Chroma 就够了。PGVector 的好处是你不需要额外维护一个数据库直接用现有的 PostgreSQL 就能存向量运维成本低。Chroma 更轻量适合原型验证。数据量到千万级再考虑 Qdrant 或 Milvus。这两个都是专门做向量检索的性能更好支持分布式。但分布式带来的运维复杂度也是实打实的小团队没必要提前上。注意向量数据库的索引类型很关键。HNSW 索引查询快但占内存IVF 索引省内存但需要训练。数据量小的时候用 Flat 索引暴力检索反而最准因为数据少暴力检索的延迟可以接受。2.5 检索策略单一向量检索不够混合检索 重排序才是正解纯向量检索有个问题它对关键词不敏感。比如用户搜“ISO 27001 认证流程”向量检索可能召回一堆讲“信息安全认证”的文档但真正提到“ISO 27001”这个具体编号的文档反而排后面。这时候就需要关键词检索来补充。混合检索的做法是同时跑向量检索和 BM25 关键词检索各取 Top 20然后用 RRFReciprocal Rank Fusion算法融合排序。RRF 不需要调参直接把两个列表的排名做倒数加权求和效果稳定。融合之后还要做重排序。重排序模型Reranker是一个交叉编码器它把 query 和每个候选片段拼在一起打分精度比向量相似度高很多但速度慢。所以一般是先召回 50 个再用 Reranker 精排出 Top 5 给模型。BGE-reranker 系列在中文场景表现不错可以本地部署。这套组合拳下来召回率通常能从纯向量检索的 60% 左右提升到 85% 以上。别小看这 25 个百分点它直接决定了用户觉得这个系统“能用”还是“不能用”。3. 实操过程与核心环节实现3.1 环境准备与依赖安装假设我们用 Python 技术栈核心依赖如下pip install langchain langchain-community pip install sentence-transformers pip install pymupdf python-docx python-pptx pip install pgvector psycopg2-binary pip install rank-bm25 pip install fastapi uvicorn如果你要用本地大模型可以装 Ollama然后拉一个中文能力不错的模型比如 qwen2.5:7b 或 glm4:9b。如果要用 API就装对应厂商的 SDK。MCP 部分需要装 mcp 的 Python SDKpip install mcp数据库方面PostgreSQL 加 pgvector 扩展是最省事的方案。建表语句大概长这样CREATE EXTENSION IF NOT EXISTS vector; CREATE TABLE knowledge_chunks ( id SERIAL PRIMARY KEY, content TEXT NOT NULL, embedding vector(1024), metadata JSONB, source VARCHAR(255), created_at TIMESTAMP DEFAULT NOW() ); CREATE INDEX ON knowledge_chunks USING hnsw (embedding vector_cosine_ops);这里的 1024 对应 BGE-large-zh 的输出维度。如果你换模型这个数字要改。3.2 文档解析与切片代码实现先写一个通用的文档加载器根据文件后缀分派到不同的解析函数import os import fitz # PyMuPDF from docx import Document from pptx import Presentation def load_document(file_path): ext os.path.splitext(file_path)[1].lower() if ext .pdf: return load_pdf(file_path) elif ext .docx: return load_docx(file_path) elif ext .pptx: return load_pptx(file_path) elif ext in [.md, .txt]: return load_text(file_path) else: raise ValueError(fUnsupported format: {ext}) def load_pdf(file_path): doc fitz.open(file_path) pages [] for page in doc: text page.get_text(text) pages.append(text) return \n.join(pages) def load_docx(file_path): doc Document(file_path) paragraphs [p.text for p in doc.paragraphs if p.text.strip()] return \n.join(paragraphs)切片部分我写一个层级感知的切片器。核心思路是先用正则识别标题行比如以“第X章”“X.X”“一、”开头的行然后按标题切分import re HEADING_PATTERN re.compile(r^(第[一二三四五六七八九十][章节]|[0-9]\.[0-9]|[一二三四五六七八九十]、)) def split_by_heading(text): lines text.split(\n) chunks [] current_heading current_content [] for line in lines: if HEADING_PATTERN.match(line.strip()): if current_content: chunks.append({ heading: current_heading, content: \n.join(current_content) }) current_heading line.strip() current_content [] else: current_content.append(line) if current_content: chunks.append({ heading: current_heading, content: \n.join(current_content) }) return chunks然后再对每个 chunk 做长度控制超过 600 字的按段落再切def split_long_chunk(chunk, max_len600, overlap80): content chunk[content] if len(content) max_len: return [chunk] paragraphs content.split(\n\n) result [] buffer for para in paragraphs: if len(buffer) len(para) max_len and buffer: result.append({ heading: chunk[heading], content: buffer.strip() }) buffer buffer[-overlap:] \n\n para else: buffer \n\n para if buffer else para if buffer: result.append({ heading: chunk[heading], content: buffer.strip() }) return result3.3 向量化与入库用 sentence-transformers 加载 BGE 模型批量编码后写入 PGVectorfrom sentence_transformers import SentenceTransformer import psycopg2 from pgvector.psycopg2 import register_vector model SentenceTransformer(BAAI/bge-large-zh-v1.5) def embed_and_store(chunks, source): conn psycopg2.connect(dbnameknowledge userpostgres passwordxxx) register_vector(conn) cur conn.cursor() texts [c[content] for c in chunks] embeddings model.encode(texts, normalize_embeddingsTrue, batch_size32) for chunk, emb in zip(chunks, embeddings): cur.execute( INSERT INTO knowledge_chunks (content, embedding, metadata, source) VALUES (%s, %s, %s, %s), (chunk[content], emb, {heading: chunk[heading]}, source) ) conn.commit() cur.close() conn.close()注意normalize_embeddingsTrue这样向量就是单位向量余弦相似度可以直接用内积算检索更快。3.4 混合检索实现向量检索部分def vector_search(query, top_k20): query_emb model.encode(query, normalize_embeddingsTrue) cur.execute( SELECT id, content, metadata, source, 1 - (embedding %s) AS score FROM knowledge_chunks ORDER BY embedding %s LIMIT %s, (query_emb, query_emb, top_k) ) return cur.fetchall()BM25 部分我用 rank-bm25 在内存里建索引。如果数据量大建议用 Elasticsearch 或 PG 的全文检索from rank_bm25 import BM25Okapi import jieba def build_bm25_index(): cur.execute(SELECT id, content FROM knowledge_chunks) rows cur.fetchall() corpus [list(jieba.cut(row[1])) for row in rows] bm25 BM25Okapi(corpus) return bm25, rows def bm25_search(bm25, rows, query, top_k20): tokenized list(jieba.cut(query)) scores bm25.get_scores(tokenized) ranked sorted(zip(rows, scores), keylambda x: x[1], reverseTrue) return ranked[:top_k]RRF 融合def rrf_fusion(vector_results, bm25_results, k60): scores {} for rank, item in enumerate(vector_results): doc_id item[0] scores[doc_id] scores.get(doc_id, 0) 1 / (k rank 1) for rank, (item, _) in enumerate(bm25_results): doc_id item[0] scores[doc_id] scores.get(doc_id, 0) 1 / (k rank 1) return sorted(scores.items(), keylambda x: x[1], reverseTrue)3.5 Agent 编排与 MCP 接入Agent 的核心是一个循环理解用户意图、决定调用哪个工具、执行工具、根据结果决定下一步。我用一个简化的 ReAct 模式来实现class KnowledgeAgent: def __init__(self, llm, retriever, tools): self.llm llm self.retriever retriever self.tools tools def run(self, query, max_steps5): history [] for step in range(max_steps): prompt self.build_prompt(query, history) response self.llm.generate(prompt) if ACTION: in response: action self.parse_action(response) if action[name] search_knowledge: result self.retriever.search(action[query]) else: result self.tools[action[name]].run(action[params]) history.append({action: action, result: result}) else: return response return 抱歉我无法在限定步骤内找到答案。MCP 接入部分你需要实现一个 MCP Server把知识库检索暴露成标准工具from mcp.server import Server from mcp.types import Tool, TextContent server Server(knowledge-base) server.list_tools() async def list_tools(): return [ Tool( namesearch_knowledge, description在企业知识库中检索相关文档片段, inputSchema{ type: object, properties: { query: {type: string, description: 检索问题} }, required: [query] } ) ] server.call_tool() async def call_tool(name, arguments): if name search_knowledge: results retriever.search(arguments[query]) return [TextContent(typetext, textformat_results(results))]这样任何支持 MCP 的客户端都能直接调用你的知识库不需要为每个客户端单独写适配。4. 常见问题与排查技巧实录4.1 检索召回率低怎么办这是最常见的问题。排查顺序如下排查项检查方法常见问题切片质量随机抽 20 个切片人工看切片断裂、语义不完整嵌入模型用几个已知问题测试模型不适合中文或领域不匹配向量归一化检查是否 normalize未归一化导致相似度计算错误索引类型检查向量索引配置HNSW 参数设置不当查询改写看原始 query 是否太短用户问题太模糊需要改写查询改写是一个很实用的技巧。用户问“报销”你可以让 LLM 先把它改写成“企业员工费用报销的流程和所需材料是什么”再去检索召回率会明显提升。4.2 模型回答幻觉严重怎么办幻觉的根源通常是检索到的上下文里没有答案但模型硬要编。解决办法有三个一是在 prompt 里明确要求“如果上下文中没有答案直接说不知道”二是加一个相关性阈值检索分数低于阈值的片段不传给模型三是让模型在回答时引用来源比如“根据《报销管理制度》第 3.2 节”这样用户能自己判断。实操心得我在 prompt 里会加一句“你的回答必须能在给定文档片段中找到依据否则请回答‘根据现有知识库无法回答该问题’”。这句话能挡掉大部分幻觉。4.3 并发扛不住怎么办企业知识库问答的并发压力主要来自两方面LLM 推理和向量检索。LLM 推理如果是 API 调用瓶颈在厂商的限流如果是本地部署瓶颈在 GPU。向量检索的瓶颈在数据库连接和索引查询。优化手段向量检索加缓存相同 query 直接返回缓存结果LLM 调用加队列超过并发上限的请求排队等待切片和向量化离线做不要在线做。如果本地 GPU 不够可以考虑用 vLLM 做推理加速吞吐量能提升好几倍。4.4 文档更新后知识库不同步企业文档是动态的今天更新的制度明天就要能查到。我的做法是给每个文档算一个内容哈希定时扫描文档目录发现哈希变化就重新解析、切片、向量化并删除旧的切片。删除时按 source 字段批量删再插入新的。def sync_document(file_path): new_hash compute_hash(file_path) cur.execute(SELECT hash FROM documents WHERE path %s, (file_path,)) row cur.fetchone() if row and row[0] new_hash: return # 无变化 chunks process_document(file_path) cur.execute(DELETE FROM knowledge_chunks WHERE source %s, (file_path,)) embed_and_store(chunks, file_path) cur.execute( INSERT INTO documents (path, hash) VALUES (%s, %s) ON CONFLICT (path) DO UPDATE SET hash %s, (file_path, new_hash, new_hash) )4.5 多轮对话上下文丢失用户经常追问比如先问“报销流程是什么”再问“那需要哪些材料”。如果每次都用原始 query 去检索第二轮的“那需要哪些材料”没有主语检索效果会很差。解决办法是做 query 改写把多轮对话的历史一起给 LLM让它把当前问题改写成独立完整的 query 再去检索。def rewrite_query(history, current_query): prompt f 对话历史 {format_history(history)} 当前问题{current_query} 请把当前问题改写成不依赖历史、可以独立检索的完整问题。 只输出改写后的问题不要解释。 return llm.generate(prompt)这个改写步骤看起来简单但对多轮对话的体验提升非常明显。4.6 权限控制怎么做企业知识库不是所有人都能看所有文档。财务制度可能只有财务部能看人事制度只有 HR 能看。权限控制要在检索层做不能只在应用层做。具体做法是在切片元数据里存一个access_group字段检索时根据当前用户的组过滤SELECT id, content, metadata, source FROM knowledge_chunks WHERE metadata-access_group ANY(%s) ORDER BY embedding %s LIMIT 20这样即使用户绕过了应用层直接查数据库也拿不到没权限的内容。当然数据库本身的访问权限也要控制好。4.7 效果评估怎么做没有评估就没有优化。我一般建一个 50 到 100 条的问题集每条包含问题、标准答案、期望召回的文档 ID。然后跑自动化评估看两个指标召回率期望文档是否在 Top K 里和答案准确率生成的答案是否和标准答案一致。每次调整切片策略、换模型、改检索参数都跑一遍评估集用数据说话。评估集不用很大但一定要覆盖不同类型的问法事实型、流程型、对比型、多跳型。事实型比如“年假多少天”流程型比如“报销怎么走”对比型比如“A 方案和 B 方案的区别”多跳型比如“去年 Q3 的销售目标是多少实际完成了多少”。这四类问题的检索难度依次递增能全面反映系统能力。5. 部署与持续优化的一些经验5.1 私有化部署的硬件门槛如果全部本地部署包括嵌入模型、重排序模型、LLM硬件要求不低。嵌入模型 BGE-large 大概需要 2GB 显存重排序模型类似LLM 7B 量化后大概需要 6 到 8GB 显存。加起来至少需要一张 16GB 显存的卡比如 4080 或 A4000。如果 LLM 用 API那本地只需要跑嵌入和重排序8GB 显存的卡就够。CPU 推理也能跑但速度会慢很多。嵌入模型在 CPU 上编码 1000 个切片大概要几分钟可以接受因为这是离线做的。但重排序是在线做的CPU 上每次查询可能要几秒体验会差。所以重排序建议至少用 GPU。5.2 成本控制如果 LLM 用 API成本主要在 token 消耗。控制成本的手段一是控制检索片段数量Top 5 通常够用不要给 Top 20二是控制片段长度每个片段 500 字左右不要给整篇文档三是加缓存相同问题直接返回缓存答案四是把简单问题路由到小模型复杂问题才用大模型。我实测下来一个中等规模企业500 人左右每天 1000 次问答用 API 的成本大概在几十到一百多块人民币之间具体取决于模型选择和片段数量。这个成本对大多数企业来说是可以接受的。5.3 持续优化的方向系统上线只是开始。后续优化方向有几个一是扩充评估集把用户实际问的问题加进去二是分析 bad case看是检索问题还是生成问题三是优化切片根据 bad case 调整切片策略四是引入 GraphRAG对多跳问题做知识图谱增强五是做用户反馈闭环让用户对答案点赞点踩用反馈数据微调重排序模型。GraphRAG 是最近比较热的方向它把文档里的实体和关系抽出来建图检索时不仅看向量相似度还看实体之间的关联。对“A 和 B 有什么关系”这类问题效果很好。但 GraphRAG 的构建成本高实体抽取和关系抽取都需要 LLM而且图的质量很难保证。我的建议是先把基础 RAG 做扎实有明确的多跳需求再上 GraphRAG。5.4 一个容易被忽略的点用户教育技术做得再好用户不会用也白搭。我见过很多企业知识库上线后没人用原因是用户不知道该怎么问。所以上线时要配一份“提问指南”告诉用户什么样的问法效果好。比如“年假怎么算”比“年假”好“报销流程是什么”比“报销”好。还可以在界面上放几个示例问题引导用户。另外要让用户知道这个系统的边界。它不是万能的知识库里没有的内容它答不出来。与其让用户失望不如提前说明“当前知识库覆盖了哪些范围”。透明比假装聪明更重要。5.5 关于模型选择的个人体会国内企业做知识库问答模型选择上我倾向于用国产开源模型做私有化比如 Qwen 系列或 GLM 系列。原因有三一是中文能力确实好对中文企业文档的理解更到位二是可以私有化部署数据不出内网满足合规要求三是社区活跃遇到问题容易找到解决方案。如果非要用 API也要选国内厂商的 API延迟低、合规风险小。模型大小上7B 到 14B 的模型在知识库问答场景通常够用因为答案主要来自检索到的上下文模型只需要做阅读理解不需要它自己记住知识。32B 以上的模型当然更好但推理成本也高很多性价比不一定划算。最后分享一个我在实际项目中总结的小技巧在 prompt 里给模型一个“思考空间”让它先复述检索到的关键信息再给出答案。这个简单的步骤能显著降低幻觉率因为模型在复述时如果发现上下文里没有相关信息它就会倾向于说“不知道”而不是硬编。这个技巧不需要改模型只需要改 prompt成本极低效果立竿见影。