AI Agent 和 RAG 放在一起已经成为企业级知识库项目最常见的落地组合。类飞书文档知识库的本质是把散落在内部文档里的资料统一接入让用户通过自然语言提问系统先检索到相关知识再由大模型生成带引用的回答。与普通搜索相比RAG 强调“检索增强”与普通聊天相比AI Agent 强调“根据任务调用工具”。本文从前端开发者视角拆解这条链路文档如何进库、向量如何生成、混合检索如何召回、Agent 如何把检索结果交给模型以及前端如何把流式回答与引用证据呈现给用户。全文会从概念讲到实现再讲到排错。你不需要先精通大模型推理也不需要先看懂 LangChain 全部源码。只要愿意跟读代码就能把“文档上传 - 解析切块 - 向量化 - 向量检索 BM25 混合召回 - Agent 编排 - 前端流式渲染”这条链路完整跑起来。文中代码以 Python 3.11 FastAPI React Qdrant 为例适合作为企业内部知识库、智能客服、内部 Wiki 等场景的起步模板。1. 先理解 RAG 与 AI Agent 为什么需要联动1.1 RAG 解决的是“模型不知道的知识”问题大模型的训练数据有截止时间企业内部私有文档也不会出现在训练集里。直接让模型回答“我们公司的休假制度是什么”如果没有相关资料模型只会凭记忆编造。RAG 的思路是把用户问题先从知识库里检索出候补片段再把这些片段拼进提示词让模型基于片段作答。这套机制有两个明显收益私有知识可以被查询不需要重新训练模型。模型回答可以附带证据来源便于用户核对。RAG 不是单一算法而是一条流水线。流水线中每个环节都会影响最终效果。常见的问题是只优化了模型提示词却忽略了切块、向量检索和召回排序导致回答看起来流畅但没有命中知识。1.2 AI Agent 在 RAG 链路里的角色普通 RAG 可以写成固定流程拿到问题直接检索拼接上下文生成回答。AI Agent 的介入让这个过程变成动态决策。Agent 会判断当前问题是否需要检索选择哪个知识库工具组合多轮对话记忆甚至根据检索结果决定是否继续追问。这类“由 Agent 决定何时检索、检索几次、用哪个工具”的架构通常被称作 Agentic RAG。在类飞书文档知识库中Agent 的价值体现在两点用户可以问“上个月的产品评审结论是什么”Agent 需要先判断问题牵涉的资料范围。用户可以追问“那对应的负责人是谁”Agent 需要结合前一轮检索结果重新召回。如果只是把知识库接口直接暴露给大模型而不经过 Agent 编排就会出现答非所问或引用错位。正确做法是把“知识库检索”注册成 Agent 的一个工具由 Agent 根据用户意图决定是否调用。1.3 前端在“类飞书文档知识库”里的职责很多 RAG 教程只写后端容易让前端开发者产生距离感。实际上类飞书产品的前端要承担五件事文档上传与管理界面展示解析、切块、向量化进度。知识库配置页面管理文档集、检索参数和测试区。会话页面展示提问、流式回答和引用来源。打开页面时初始化连接处理断线重连。在页面内展示检索中间态帮助调试“问题到底有没有命中”。前端不是简单调接口还要理解后端返回的数据结构。例如流式接口返回的是 SSE 事件中间可能穿插引用信息。把这些信息用 UI 表达清楚比只渲染一段 Markdown 更接近企业级产品。2. 企业级知识库架构与数据模型2.1 全链路组成文档流水线、存储、检索、问答类飞书文档知识库可以拆成四个模块文档流水线负责上传、解析、清洗、切块、向量化。存储层业务数据、切块文本、向量数据、会话记录。检索层向量检索、BM25 召回、融合重排。问答层Agent 编排、大模型调用、流式输出。整体流程是前端上传文档 - 后端解析并切块 - 对每个切块生成向量 - 写入向量数据库 - 用户提问 - 检索层召回候选片段 - Agent 组装上下文 - 大模型生成回答 - SSE 推送给前端。这里要特别注意“异步”与“同步”的取舍。文档数量少时可以同步等待入库完成文档量大时必须使用任务队列。任务队列的引入会增加复杂度但可以避免用户上传后长时间等待 HTTP 连接。2.2 核心表与集合设计企业级实现至少要管理五类数据文档、切块、向量、会话、消息。向量数据库里存向量业务数据库里存文本和状态。数据对象存储位置关键字段作用DocumentMySQL/PostgreSQLid, name, doc_type, parse_status, created_at管理原始文档信息与状态ChunkPostgreSQL/ESid, document_id, seq, content, token_count保存切块后的文本VectorQdrant/Milvuspoint_id, chunk_id, vector, metadata保存嵌入向量与检索元数据ConversationPostgreSQLid, user_id, title, created_at保存会话MessagePostgreSQLid, conversation_id, role, content, citations保存问答消息与引用切块表与向量集合需要保持一致性。写入数据时先保存 Chunk再写入 Qdrant PointPoint 的 metadata 里记录 chunk_id。删除文档时先查该文档的 chunk_id 列表再批量删除向量否则会出现“业务库里已经删除向量库里却还能检索到”的脏数据。2.3 混合检索向量检索 BM25 多路召回向量检索擅长语义相似比如用户问“年假怎么计算”能召回包含“年假折算”的文档。BM25 擅长关键词精确匹配比如用户输入“Q3 财报”如果向量模型没学好这个组合词BM25 可以直接命中。真实的企业文档里很多问题既包含语义表达又包含强关键词。只做向量检索会在产品名称、合同编号、人名等专有名词上失分。只做 BM25又会漏掉同义词和语义改写。因此生产系统通常采用“多路召回 融合排序”。多路召回不是把两路结果直接拼起来而是使用 RRFReciprocal Rank Fusion等算法合并排序。RRF 的基本思想是文档在一路结果里排名越靠前融合分越高最终分数是各路排名倒数之和。这样即使某一路没有命中另一路的强结果也不会被网络问题拖累。2.4 已有 ES 与知识库数据要不要同步很多企业已经用 Elasticsearch 存了文档内容。这里要区分“业务搜索库”和“RAG 向量库”。业务搜索库保存的是经过清洗后的结构化业务数据面向客服查询、后台列表、统计分析。RAG 向量库保存的是适合大模型阅读的切块文本和向量面向问答检索。两者字段和用途不一样不建议直接保持强同步。推荐做法是文档解析后同时发布到 ES 和向量库。ES 只管关键词检索和业务过滤向量库只管语义检索和知识问答。同步方式通过事件队列实现避免一个模块失败影响另一个模块。同步时记录每个 Chunk 在 ES 和向量库中的 ID失败时重试。3. 环境准备与项目结构3.1 依赖清单与版本说明下面示例基于常见开源版本编写。落地前要结合团队实际依赖确认版本尤其是 sentence-transformers 和 Qdrant 客户端的兼容性。软件版本建议用途Python3.11后端开发FastAPI0.111API 服务Qdrant1.10向量数据库sentence-transformers3.x生成文本向量rank-bm250.2.2BM25 关键词召回LangChain0.2.xAgent 工具封装与提示词组装Node.js18前端构建React18前端界面后端依赖可以集中写在 requirements.txtfastapi uvicorn[standard] python-multipart pypdf python-docx markdown sentence-transformers qdrant-client rank-bm25 langchain langchain-openai sse-starlette3.2 docker-compose 启动 Qdrant为了快速跑通先启动一个单机版 Qdrant。生产环境建议使用高可用部署并开启持久化存储。version: 3.8 services: qdrant: image: qdrant/qdrant:v1.10.0 container_name: knowledge-qdrant ports: - 6333:6333 - 6334:6334 volumes: - ./qdrant_storage:/qdrant/storage启动命令docker compose up -d检查是否正常运行curl http://localhost:6333/collections返回结果里可以看到当前集合列表。刚启动时集合还没有创建需要由后端代码在启动时初始化。3.3 后端项目结构目录结构要方便扩展。示例结构如下backend/ ├── app/ │ ├── main.py │ ├── config.py │ ├── models/ │ │ ├── document.py │ │ ├── chunk.py │ │ └── chat.py │ ├── services/ │ │ ├── parser.py │ │ ├── splitter.py │ │ ├── embedder.py │ │ ├── vector_store.py │ │ ├── bm25_store.py │ │ └── agent.py │ └── api/ │ ├── document.py │ └── chat.py ├── data/ ├── requirements.txt └── .env从这个结构可以看出文档解析、切块、嵌入、向量库、BM25、Agent 各自独立。后续要替换嵌入模型或者把 BM25 换成 Elasticsearch都不需要改动全部代码。4. 文档解析、切块与向量化入库4.1 文档解析PDF、Word、Markdown前端上传文件后后端先保存到临时目录再根据扩展名选择解析方式。这里演示三种常见格式。from pathlib import Path import markdown from pypdf import PdfReader from docx import Document def parse_document(file_path: Path) - str: suffix file_path.suffix.lower() if suffix .pdf: reader PdfReader(str(file_path)) return \n.join(page.extract_text() or for page in reader.pages) if suffix .docx: doc Document(str(file_path)) return \n.join(paragraph.text for paragraph in doc.paragraphs) if suffix .md: return markdown.markdown(Path(file_path).read_text(encodingutf-8), output_formattext) if suffix .txt: return Path(file_path).read_text(encodingutf-8) raise ValueError(f暂不支持的文件类型: {suffix})这里要注意一个常见坑PDF 的extract_text对扫描件返回空字符串。如果读取结果为空需要接入 OCR 服务或者提示用户上传带文本层的 PDF。不要把空字符串直接存入知识库否则后续检索会得到无意义片段。4.2 切块策略与参数切块影响召回质量。最简单的策略是固定字符切块但容易截断句子。推荐使用 LangChain 的递归字符切分器它会优先按段落、句子、字符层级切分。from langchain.text_splitter import RecursiveCharacterTextSplitter splitter RecursiveCharacterTextSplitter( chunk_size400, chunk_overlap80, separators[\n\n, \n, 。, , , ., !, ?, , ], ) chunks splitter.split_text(text)参数含义如下参数含义常见值调大影响调小影响chunk_size切块目标字符数300-500上下文信息多但可能引入无关内容上下文集中但可能截断语义chunk_overlap相邻块重叠字符数50-100减少截断损失但增加存储开销节省存储但可能丢失跨块信息切块后要给每个块编号并记录原始文档 ID 和页码信息。这些元数据在检索结果里做证据展示时非常关键。常见坑是只按固定字数切块导致一个文档块里包含多个无关主题。比如产品说明书的“安装步骤”和“故障排查”如果被切在同一块用户问安装问题时会召回一半无关内容。建议先按标题段落结构切分再对超长段落做二次切分。4.3 向量化并写入 Qdrant使用 sentence-transformers 生成向量。中文场景建议选择支持中文的模型比如BAAI/bge-m3。模型首次运行会下载权重生产环境最好提前缓存到本地。from sentence_transformers import SentenceTransformer model SentenceTransformer(BAAI/bge-m3) def embed_texts(texts): return model.encode(texts, normalize_embeddingsTrue).tolist()写入 Qdrant 前先创建集合。集合的向量维度必须和模型输出维度一致。bge-m3的输出维度是 1024如果使用all-MiniLM-L6-v2维度是 384。from qdrant_client import QdrantClient from qdrant_client.models import Distance, VectorParams, PointStruct client QdrantClient(urlhttp://localhost:6333) def create_collection(collection_name: str, vector_size: int 1024): client.recreate_collection( collection_namecollection_name, vectors_configVectorParams(sizevector_size, distanceDistance.COSINE), )写入向量时把 chunk_id、document_id、page、content 全部放入 payload。points [ PointStruct( idpoint_id, vectorvector, payload{ chunk_id: chunk.id, document_id: chunk.document_id, page: chunk.page, content: chunk.content, }, ) for chunk, vector in zip(chunks, vectors) ] client.upsert(collection_nameknowledge, pointspoints)这里最容易出错的是 point_id 重复。如果重复上传同一份文档建议使用可预测的 IDf{document_id}-{chunk_seq}。同时需要先删除该文档的旧向量再写入新向量。5. 混合检索实现从查询到召回5.1 向量检索用户提交问题后先对问题做嵌入再从 Qdrant 中查询相近向量。def vector_search(query: str, top_k: int 5): query_vector embed_texts([query])[0] results client.query_points( collection_nameknowledge, queryquery_vector, limittop_k, with_payloadTrue, ).points return [ { chunk_id: hit.payload[chunk_id], content: hit.payload[content], score: hit.score, } for hit in results ]很多问题不是“有没有查到”而是“查到的顺序不对”。向量检索的分数和具体模型相关不要直接拿分数作为绝对阈值。先观察日志里命中的内容与问题之间是否语义相关。5.2 BM25 关键词召回在项目早期可以用 rank-bm25 在内存里建立索引。它适合百万级以下的中小知识库。from rank_bm25 import BM25Okapi class BM25Store: def __init__(self): self.corpus [] self.bm25 None def build_index(self, chunks): self.corpus [chunk[content] for chunk in chunks] tokenized_corpus [content.split() for content in self.corpus] self.bm25 BM25Okapi(tokenized_corpus) def search(self, query: str, top_k: int 5): tokenized_query query.split() scores self.bm25.get_scores(tokenized_query) top_indices sorted(range(len(scores)), keylambda i: scores[i], reverseTrue)[:top_k] return [ {chunk_id: self.corpus[i][chunk_id], content: self.corpus[i][content], score: scores[i]} for i in top_indices ]中文场景直接split()会按空格切词对连续中文效果不好。这里为了演示先使用split()生产环境建议引入 jieba 分词或使用 ES 的标准分词器。索引构建后如果新增了文档需要重建 BM25 索引。5.3 RRF 融合排序RRF 不需要把两路分数归一化只根据排名计算融合分。代码如下def rrf_fusion(vector_results, bm25_results, k60, top_k5): score_map {} for rank, item in enumerate(vector_results): chunk_id item[chunk_id] score_map[chunk_id] score_map.get(chunk_id, 0) 1 / (k rank 1) score_map[chunk_id] 0 # 占位下面记录来源 for rank, item in enumerate(bm25_results): chunk_id item[chunk_id] score_map[chunk_id] score_map.get(chunk_id, 0) 1 / (k rank 1) merged sorted(score_map.items(), keylambda x: x[1], reverseTrue)[:top_k] return [chunk_id for chunk_id, _score in merged]如果同一 chunk 在两个结果中都出现RRF 会提高它的排序。如果一路结果没有命中不影响另一路。融合后需要回查 chunk 表把完整文本和元数据取出来。5.4 检索结果验证检索接口可以用一个简单的 API 暴露出来方便前端和调试工具单独测试。app.post(/api/retrieval) async def retrieval(payload: RetrievalRequest): vector_hits vector_search(payload.query) bm25_hits bm25_store.search(payload.query) merged_ids rrf_fusion(vector_hits, bm25_hits) return { vector_hits: vector_hits, bm25_hits: bm25_hits, merged_ids: merged_ids, }返回 JSON 示例{ vector_hits: [ { chunk_id: doc123-7, content: 年假折算规则员工入职满一年后可享有 5 天年假。, score: 0.82 } ], bm25_hits: [ { chunk_id: doc123-8, content: 年假与调休不能重复申请申请入口在 OA 系统。, score: 2.34 } ], merged_ids: [doc123-7, doc123-8] }验证时先看两路结果是否各自有意义。如果向量检索召回相关但 BM25 召回垃圾说明分词需要优化。如果向量召回垃圾说明切块或嵌入模型不适合当前文本。6. Agent 编排与智能问答接口6.1 把知识库注册成 Agent 的 tool不要在 system prompt 里堆检索逻辑正确做法是把检索封装成工具。LangChain 的tool装饰器可以快速实现。from langchain_core.tools import tool tool def knowledge_base_search(query: str) - str: 在企业知识库中检索相关内容适用于制度、产品文档、技术方案类问题。 vector_hits vector_search(query) bm25_hits bm25_store.search(query) merged_ids rrf_fusion(vector_hits, bm25_hits) return \n\n.join(get_chunk_content(cid) for cid in merged_ids)Agent 的目标是根据用户问题决定是否调用这个工具。如果用户说“你好”Agent 不需要检索。如果用户问公司请假制度Agent 应该调用工具并获取上下文。这里要区分“skill”和“tool”两个概念。如果团队在做 Agent 技能开发知识库问答更适合做成 tool它输入输出明确可以被多个 Agent 复用。而 skill 更像是封装一段复杂工作流适合多步骤任务。简单场景不要过度设计。6.2 FastAPI 问答接口与 SSE 流式响应问答接口需要接收用户问题和历史消息返回流式回答。这里使用StreamingResponse通过 SSE 格式输出。from fastapi import FastAPI from fastapi.responses import StreamingResponse from pydantic import BaseModel app FastAPI() class ChatRequest(BaseModel): conversation_id: str query: str history: list[dict] [] def chat_stream(payload: ChatRequest): # 1. Agent 决定是否检索得到上下文 context knowledge_base_search.invoke(payload.query) # 2. 组装 system prompt messages [ {role: system, content: f你是一个企业知识库助手请基于以下资料回答\n{context}}, ] payload.history [{role: user, content: payload.query}] # 3. 调用大模型并把输出流返回 for token in llm_stream(messages): yield fdata: {token}\n\n yield data: [DONE]\n\n app.post(/api/chat) async def chat(payload: ChatRequest): return StreamingResponse(chat_stream(payload), media_typetext/event-stream)示例里的llm_stream需要替换成具体模型 SDK。OpenAI 兼容接口通常都支持streamTrue逐字返回 token。使用 SSE 的关键点每条消息用data:开头以两个换行结尾。前端解析时不能只监听message事件还要处理数据被 TCP 分包的情况。6.3 引用溯源格式企业级知识库最重要的体验是“回答有依据”。回答内容要和引用片段关联起来而不是只给出纯文本。后端在生成回答前可以把检索结果中的 chunk 列表作为引用信息一并返回给前端。消息结构可以设计成{ conversation_id: conv_123, content: 根据知识库中的年假折算规则员工入职满一年后享有 5 天年假。, citations: [ { chunk_id: doc123-7, document_name: 2025年假期管理制度.docx, page: 2, snippet: 员工入职满一年后可享有 5 天年假。 } ] }前端收到引用信息后在回答右下角渲染成“引用了 3 份文档”的卡片。点击卡片可以看到对应的原文片段。这个能力看起来简单却是判断 RAG 产品是否可用的关键指标。7. React 前端实现上传、会话与流式渲染7.1 文件上传组件前端使用 React 18 Vite。上传组件需要展示进度和解析状态。这里用FormData提交。import { useState } from react; async function uploadDocument(file: File) { const formData new FormData(); formData.append(file, file); const response await fetch(/api/document/upload, { method: POST, body: formData, }); if (!response.ok) { throw new Error(上传失败); } return response.json(); }上传只是第一步。后端解析、切块、向量化可能需要几秒到几十秒。接口设计成上传后立即返回任务 ID前端轮询任务状态避免让用户一直停留在上传页。如果上传大文件原生 fetch 没有内置的分片能力。常见方案是用 Worker 读取文件再分片计算哈希和上传分片。分片上传能提升稳定性但会显著增加后端接口复杂度。前期优先支持 50MB 以内的文档后续再考虑分片。7.2 问答页与 SSE 读取问答页面使用 fetch 读取流比 EventSource 更灵活因为可以携带 POST 请求体。async function sendQuestion(query: string, history: Message[]) { const response await fetch(/api/chat, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ conversation_id, query, history }), }); const reader response.body!.getReader(); const decoder new TextDecoder(); let done false; while (!done) { const { value, done: readerDone } await reader.read(); done readerDone; if (value) { const text decoder.decode(value, { stream: true }); const lines text.split(\n\n); lines.forEach((line) { if (line.startsWith(data: )) { const data line.replace(data: , ); if (data [DONE]) return; setAnswer((prev) prev data); } }); } } }这段代码展示了前端手动解析 SSE 数据的过程。注意TextDecoder的stream: true参数它可以正确处理多字节字符被拆到两个网络包里的情况。7.3 引用卡片渲染回答末尾的引用信息可以独立于流式文本返回。前端在回答结束后再拉取一次消息详情或者从接口响应头里读取引用信息。简化设计是在流结束后发送一个包含citations的事件。前端解析到该事件时更新引用状态。if (data [CITATIONS]) { const citations JSON.parse(nextData); setCitations(citations); }引用卡片渲染时建议显示文档名称、页码和相似片段。用户点击卡片后展开原文这样能建立“系统不是瞎编”的信任感。7.4 前端联调常见坑前端与后端联调时有三个高频问题问题现象常见原因处理建议请求被 CORS 拦截后端未配置允许跨域开发和测试环境由 FastAPI 添加 CORSMiddleware流式回答中途停止用户关闭页面或网络断开前端监听read()异常并提示重新连接中文乱码解码时未使用 UTF-8使用new TextDecoder(utf-8)不要直接toString()本地开发时可以给 Vite 配置代理把/api转发到后端端口避免频繁处理 CORS。export default defineConfig({ server: { proxy: { /api: { target: http://localhost:8000, changeOrigin: true, }, }, }, });8. 常见问题排查与企业级落地建议8.1 从“检索无结果”倒推整条链路知识库问答最典型的问题是用户问了一个问题系统回答“没有找到相关信息”。排查顺序要从数据源头开始而不是先改模型提示词。确认文档已经解析成功。查看后端日志里的字符数如果字符数为 0说明 PDF 缺少文本层。确认切块后的 chunk 数。文档非常短时可能只有 1 个 chunk检索时可能因为相似度阈值被过滤。确认向量已经写入 Qdrant。用 Qdrant 的 Web UI 或 API 查询集合总点数和 chunk 数量比对。确认查询向量和写入向量维度一致。模型更换后旧集合仍然存在会报维度错误。用/api/retrieval单独测试检索观察向量结果和 BM25 结果。如果单独一路有结果说明融合逻辑有问题。可以按这个表格快速定位现象可能原因检查方式向量结果为空chunk 未写入或集合为空查询 Qdrant 集合点数向量结果有BM25 为空BM25 索引未重建或分词弱打印分词结果两路结果都有最终回答仍找不到Agent 没有调用工具或 prompt 拼接错误打印 Agent 工具调用日志前端一直转圈SSE 未关闭检查网络面板里是否收到[DONE]8.2 切块与嵌入导致回答质量差的排错回答质量差往往不是大模型问题而是检索上下文不对。建议做三层验证第一层验证切块随机抽取文档段落确认每块是否语义完整。如果有句子被硬切调整chunk_overlap。第二层验证检索用真实用户问题测试/api/retrieval看前三条结果是否都与问题相关。如果前三条里混入无关内容调大top_k并在融合后展示更多候选或者引入重排模型。第三层验证提示词把检索结果直接发给模型不经过 Agent看模型是否能基于资料回答。如果不能说明提示词里没有强调“只使用资料内容回答”。生产环境建议加入一个简单重排环节用交叉编码器对召回的候选片段重新评分。重排可以显著提升答案准确率代价是多一次模型推理。8.3 生产环境必备的能力清单从演示项目到生产系统必须补齐以下能力权限控制文档级别和知识库级别都要做访问控制避免越权查看。日志与追踪记录每次查询的文档 ID、chunk ID、命中的引用、模型输出。版本管理模型权重、切块参数、提示词版本都要可回滚。敏感信息过滤对文档内容做脱敏防止个人隐私被检索出来。监控告警向量库连接失败、嵌入服务超时、切块任务堆积都要有告警。数据备份Qdrant 的 snapshot 和业务数据库备份要定期执行。生产环境不建议把所有文档全量加载到内存。BM25 使用内存方案只适合小规模场景文档量大时切换到 Elasticsearch。Agent 的依赖也要升级到可观测组件方便追踪每次工具调用。8.4 可复用的上线前检查清单上线前可以按以下清单逐项确认文档上传接口是否在 10s 内返回任务 ID。解析失败的文档是否会进入失败队列而不是静默丢失。切块后的重复内容是否会被去重。向量集合的维度是否与当前嵌入模型一致。混合检索结果是否经过人工抽查。Agent 的工具调用是否在日志中完整记录。SSE 流式接口是否会设置合理的超时。前端断线后是否能恢复会话。引用展示是否包含文档名和页码。生产环境是否配置了 CORS 白名单、鉴权和敏感信息过滤。最终要守住一条原则RAG 项目的核心不是“答得流畅”而是“答得有据可查”。Agent 只是让这条链路更灵活混合检索只是让召回更稳定。前端作为用户第一接触层要把“检索到了什么、引用了什么、回答依据是什么”完整展示出来。下一步可以继续扩展的方向是引入重排模型、支持多租户隔离、接入 WPS 或飞书开放平台文档以及在检索结果中加入时间过滤和权限过滤。