企业官网建设流程全解析

当前笔记围绕企业级 RAG（检索增强生成）知识库展开，从架构原理到检索、召回、重排、生成与工程化落地完整走一遍。适合正在做大模型应用、知识库问答、私有化部署的开发者参考，也适合想系统理解 RAG 全链路的新手。

1. RAG 是什么，为什么企业知识库需要它

1.1 先理解大模型的两大短板

大模型虽然能聊天、能写代码、能做摘要，但在企业知识问答场景里有两个明显问题：

第一，知识有时效性。大模型的训练数据有截止日期，之后发生的事情它不知道。公司内部新发布的制度、刚上线的产品规格、最新的法务条款，模型训练时根本没见过。

第二，生成内容不可控。模型回答问题时会把训练时学到的“常识”和“编造内容”混在一起，出现一本正经的胡说八道。这个问题在业内叫“幻觉”。对于企业知识库，幻觉是致命的，法务、财务、医疗、制造等场景尤其不能接受。

1.2 RAG 的核心思路：先检索，再生成

RAG（Retrieval-Augmented Generation，检索增强生成）的思路很直接：让大模型在回答前先“查资料”。

整个链路可以概括为：

1. 用户输入一个问题。
2. 系统从企业知识库中检索与问题最相关的内容片段。
3. 将这些内容片段拼接到 Prompt 中，一起交给大模型。
4. 大模型基于“检索到的内容 + 用户问题”生成答案。

这意味着模型不需要“记住”企业内部知识，它只需要“读懂”检索到的资料。知识的新增、更新、删除都发生在知识库侧，模型侧不需要重训练，维护成本大幅降低。

1.3 RAG 与模型微调的区别

很多人会把 RAG 和模型微调（Fine-tuning）放在一起比较。两者的定位完全不同：

实际项目中，两者可以结合。比如先用 RAG 解决知识来源问题，再用微调让模型学会特定的输出格式或行业表达习惯。但大多数知识库项目，RAG 是主链路。

2. RAG 全链路架构拆解

RAG 不是“单个模型”或者“单个 API”，而是一条完整的数据流水线。只有把整条链路拆清楚，才能定位问题、优化效果。

2.1 离线索引链路

离线链路负责把企业文档变成可检索的结构化数据。

文档接入 → 格式解析 → 文档清洗 → 文本切分 → 向量化 → 写入向量数据库 ↓ 建立关键词索引

核心环节包括：

• 文档接入：支持 PDF、Word、Markdown、TXT、HTML、Excel 等格式。
• 格式解析：PDF 需要抽取文本和表格，Word 需要保留标题层级，扫描件需要 OCR。
• 文档清洗：去掉页眉页脚、水印、无效换行、乱码字符。
• 文本切分（Chunking）：这是全链路里最容易影响效果的一步。
• 向量化（Embedding）：将文本转换成向量。
• 索引写入：向量入库，同时保留原文、元数据、文档来源。

2.2 在线推理链路

在线链路处理用户请求。

用户提问 → 查询改写 → 混合检索 → 重排 → 构造 Prompt → 大模型生成 ↓ 生成引用来源

每一步都有优化空间：

• 查询改写：把模糊问题改写成适合检索的表达，比如把“上次说的那个方案”改写成具体名词。
• 混合检索：同时使用关键词检索和向量检索，再合并结果。
• 重排（Rerank）：对检索结果做精细排序，把最相关的片段排到前面。
• Prompt 构造：在 Prompt 中限制模型只能使用检索内容作答。
• 生成：大模型输出答案，并附上引用来源。

2.3 完整流程图

用户问题 ↓ [查询预处理] → 纠错、改写、意图识别 ↓ [混合检索] → BM25 关键词检索 → 向量检索（KNN） ↓ [结果融合] → 去重、分数归一化、取 TopN ↓ [重排] → Cross-Encoder / Rerank 模型 ↓ [Prompt 构造] → 注入检索片段、设定回答边界 ↓ [大模型生成] → 输出答案 + 引用来源 ↓ [后处理] → 格式校验、敏感词过滤、记录日志

3. 环境准备与版本选型

RAG 项目涉及的组件比较多。为了避免“本地能跑，生产就崩”的问题，需要一开始就做好技术选型。

3.1 核心组件清单

3.2 版本注意事项

技术版本更新很快，所以这里不给写死的版本号，大家按自己项目的实际情况选择。需要特别注意几个原则：

• 向量数据库版本与 SDK 版本必须匹配。客户端 SDK 大版本升级后，连接方式、索引参数都可能有变化。
• Embedding 模型维度必须与索引配置一致。如果先建了 768 维的集合，换模型后向量变成 1024 维，就会写入失败。
• Python 版本建议使用 3.9+。新版 LangChain 和向量数据库 SDK 对旧版本支持越来越差。
• 生产环境避免使用 SQLite 类向量方案。Chroma 的原生持久化适合 Demo，高并发场景需要独立的向量数据库服务。

3.3 示例项目结构

rag-project/ ├── app/ │ ├── main.py # FastAPI 入口 │ ├── config.py # 配置文件 │ ├── ingestion/ │ │ ├── loader.py # 文档加载 │ │ ├── splitter.py # 文本切分 │ │ └── vectorize.py # 向量化与入库 │ ├── retrieval/ │ │ ├── keyword_search.py # 关键词检索 │ │ ├── vector_search.py # 向量检索 │ │ ├── hybrid_search.py # 混合检索与融合 │ │ └── rerank.py # 重排模块 │ ├── generation/ │ │ └── llm.py # Prompt 构造与生成 │ └── schemas.py # 请求响应模型 ├── data/ # 原始文档 ├── tests/ # 测试用例 └── requirements.txt

如果是 Java 技术栈，可以类比拆包：用 Spring Boot 作为 Web 层，用 langchain4j 做链路编排，用 Milvus Java SDK 做向量检索。

4. 完整实战：从零搭建一个 RAG 知识库

下面用一个可运行的 Python 示例，把 RAG 全链路走一遍。这里使用 FastAPI + Chroma + HuggingFace Embedding 和 Rerank 模型，重点演示架构思路。生产环境可以替换成 Milvus 或 Elasticsearch。

4.1 安装依赖

pip install fastapi uvicorn chromadb sentence-transformers pip install pypdf python-docx markdown pip install openai # 或者对接本地模型服务 pip install bm25s # 轻量 BM25 工具

一些库可能在不同版本中 API 有变化，安装时如果遇到兼容问题，可以把版本固定下来。

4.2 定义配置文件

# 文件路径：app/config.py import os EMBEDDING_MODEL = os.getenv("EMBEDDING_MODEL", "BAAI/bge-large-zh-v1.5") RERANK_MODEL = os.getenv("RERANK_MODEL", "BAAI/bge-reranker-v2-m3") LLM_BASE_URL = os.getenv("LLM_BASE_URL", "http://localhost:8000/v1") LLM_API_KEY = os.getenv("LLM_API_KEY", "EMPTY") LLM_MODEL = os.getenv("LLM_MODEL", "qwen2.5-7b-instruct") CHUNK_SIZE = 512 CHUNK_OVERLAP = 64 TOP_K = 20 # 检索阶段取回的数量 RERANK_TOP_N = 5 # 重排后保留的数量

4.3 文档加载与清洗

不同格式的文档需要不同的解析方式。这里先写一个统一的加载函数。

# 文件路径：app/ingestion/loader.py from pathlib import Path def load_document(file_path: str) -> str: path = Path(file_path) suffix = path.suffix.lower() if suffix == ".txt": return path.read_text(encoding="utf-8", errors="ignore") elif suffix == ".md": return path.read_text(encoding="utf-8", errors="ignore") elif suffix == ".pdf": from pypdf import PdfReader reader = PdfReader(str(path)) return "\n".join(page.extract_text() or "" for page in reader.pages) elif suffix == ".docx": from docx import Document doc = Document(str(path)) return "\n".join(p.text for p in doc.paragraphs) else: raise ValueError(f"暂不支持的文件格式：{suffix}")

文档清洗时，要注意：

• 去除页面页脚中的重复内容。
• 将全角字符统一转半角。
• 删除连续空行。
• 保留 Markdown 标题结构，方便后续切分。

4.4 文本切分

切分是 RAG 链路中最“细节”的环节。切太大，检索精度会下降；切太小，片段缺乏上下文，语义不完整。

推荐使用“递归字符切分 + 固定窗口重叠”的思路：

# 文件路径：app/ingestion/splitter.py from typing import List def recursive_split(text: str, chunk_size: int = 512, chunk_overlap: int = 64) -> List[str]: separators = ["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""] chunks = [] current = "" for para in text.split("\n"): if len(current) + len(para) < chunk_size: current += para + "\n" else: if current: chunks.append(current.strip()) current = para + "\n" if current.strip(): chunks.append(current.strip()) # 对过长的片段做二次切分并保留重叠 final_chunks = [] for chunk in chunks: if len(chunk) <= chunk_size: final_chunks.append(chunk) continue start = 0 while start < len(chunk): end = start + chunk_size final_chunks.append(chunk[start:end]) if end >= len(chunk): break start = end - chunk_overlap return final_chunks

这个实现保证：

• 以段落为基本单位优先合并。
• 过长段落按固定长度切分。
• 相邻 chunk 之间保留重叠，避免重要信息被截断。

实际项目里还可以结合标题层级切分，让同一章节的内容尽量在同一个 chunk 中。

4.5 向量化与入库

使用 sentence-transformers 加载 Embedding 模型，将文本转为向量后写入 Chroma。

# 文件路径：app/ingestion/vectorize.py import chromadb from sentence_transformers import SentenceTransformer from app.config import EMBEDDING_MODEL, CHUNK_SIZE, CHUNK_OVERLAP from app.ingestion.loader import load_document from app.ingestion.splitter import recursive_split client = chromadb.PersistentClient(path="./chroma_db") collection = client.get_or_create_collection(name="enterprise_rag") embedder = SentenceTransformer(EMBEDDING_MODEL) def index_document(file_path: str, doc_id: str, metadata: dict = None): text = load_document(file_path) chunks = recursive_split(text, CHUNK_SIZE, CHUNK_OVERLAP) embeddings = embedder.encode(chunks).tolist() ids = [f"{doc_id}_{i}" for i in range(len(chunks))] metadatas = [ { "doc_id": doc_id, "chunk_index": i, "source": file_path, **(metadata or {}), } for i in range(len(chunks)) ] collection.add( ids=ids, documents=chunks, embeddings=embeddings, metadatas=metadatas, ) return len(chunks)

注意：Chromat 的add方法在重复执行同一文档时会报 ID 冲突，所以更新文档时可以先collection.delete(where={"doc_id": doc_id})再重新写入。

4.6 混合检索：向量检索 + 关键词检索

只用向量检索的问题在于：一些精确的专有名词、编号、公式、报错代码，向量模型理解得并不好。比如搜索“BGP 邻居状态 Establised”，关键词匹配能精准命中，向量检索可能召回到语义相近但不够准确的内容。

所以企业级 RAG 通常使用混合检索。

向量检索代码：

# 文件路径：app/retrieval/vector_search.py from sentence_transformers import SentenceTransformer from app.config import EMBEDDING_MODEL embedder = SentenceTransformer(EMBEDDING_MODEL) def vector_search(query: str, top_k: int = 20): query_embedding = embedder.encode([query]).tolist()[0] results = collection.query( query_embeddings=[query_embedding], n_results=top_k, include=["documents", "metadatas", "distances"], ) return results

关键词检索使用 BM25 算法。这是一个经典的概率检索模型，对专有名词匹配非常有效。

# 文件路径：app/retrieval/keyword_search.py from bm25s import BM25 from app.config import TOP_K # 全局维护一个语料库，生产环境建议用 Elasticsearch bm25_index = None def build_bm25_index(corpus: list[str], doc_ids: list[str]): global bm25_index tokenized = [doc.split() for doc in corpus] bm25_index = BM25() bm25_index.index(tokenized, doc_ids=doc_ids) def keyword_search(query: str, top_k: int = TOP_K): if bm25_index is None: return [] tokenized_query = query.split() results = bm25_index.retrieve(tokenized_query, k=top_k) return results

混合检索的融合策略有很多种。最简单的是 “Score 加权求和”：

# 文件路径：app/retrieval/hybrid_search.py def hybrid_search(query: str, top_k: int = TOP_K, alpha: float = 0.5): vector_results = vector_search(query, top_k) keyword_results = keyword_search(query, top_k) # 统一分数区间后加权 scores = {} docs = {} for i, doc in enumerate(vector_results["documents"][0]): score = 1 / (1 + vector_results["distances"][0][i]) docs[doc] = vector_results["metadatas"][0][i] scores[doc] = scores.get(doc, 0) + alpha * score for doc_id, score in keyword_results.items(): if score <= 0: continue doc_text = doc_id_to_text.get(doc_id, "") if not doc_text: continue scores[doc_text] = scores.get(doc_text, 0) + (1 - alpha) * score docs.setdefault(doc_text, {"doc_id": doc_id}) sorted_docs = sorted(scores.items(), key=lambda x: x[1], reverse=True) return sorted_docs[:top_k]

注意，alpha表示向量检索的权重。不同业务需要调参：代码、编号、合同类内容，关键词权重应更高；语义问答、综述类内容，向量权重应更高。

4.7 重排：解决“召回多但排不准”的问题

向量检索和关键词检索都属于“粗召回”，它们的优势是效率高，但排序质量不够精细。重排（Rerank）的作用是：把召回的 TopK 片段，用更精确的模型重新打分排序。

核心思路是使用 Cross-Encoder 模型：把“问题”和“候选片段”拼在一起输入模型，模型输出一个相关性分数。这种方式比双塔向量模型更精确，但速度慢，所以只对前 20 或前 50 条候选结果做重排。

# 文件路径：app/retrieval/rerank.py from sentence_transformers import CrossEncoder from app.config import RERANK_MODEL reranker = CrossEncoder(RERANK_MODEL) def rerank(query: str, candidates: list[dict], top_n: int = 5): if not candidates: return [] pairs = [(query, candidate["text"]) for candidate in candidates] scores = reranker.predict(pairs) for candidate, score in zip(candidates, scores): candidate["rerank_score"] = float(score) ranked = sorted(candidates, key=lambda x: x["rerank_score"], reverse=True) return ranked[:top_n]

重排的效果提升非常明显。以相同召回集合为前提，使用 bge-reranker 类模型重排后，正确答案排进前三的概率通常比向量打分排序高 10 到 20 个百分点。不过具体数字与数据集有关，这里只强调这一环节的必要性。

4.8 构造 Prompt 与生成

重排后的片段要作为“参考资料”注入到 Prompt 中。Prompt 设计要满足三个要求：

1. 说明资料的内容边界。
2. 禁止模型使用资料之外的知识。
3. 要求模型在信息不足时明确回答“不知道”。

# 文件路径：app/generation/llm.py from openai import OpenAI from app.config import LLM_BASE_URL, LLM_API_KEY, LLM_MODEL client = OpenAI(base_url=LLM_BASE_URL, api_key=LLM_API_KEY) SYSTEM_PROMPT = """你是一个企业知识助手。你的回答必须严格基于提供的参考资料，不得使用参考资料之外的信息。 如果参考资料无法回答问题，请回答“根据现有资料无法回答该问题”。 回答时请用编号标注引用来源，例如[1][2]。""" def build_prompt(query: str, contexts: list[dict]) -> str: context_text = "\n\n".join( [ f"[{i + 1}] {item['text']}\n来源：{item.get('source', '未知')}" for i, item in enumerate(contexts) ] ) return f"用户问题：{query}\n\n参考资料：\n{context_text}" def generate_answer(query: str, contexts: list[dict]): prompt = build_prompt(query, contexts) response = client.chat.completions.create( model=LLM_MODEL, messages=[ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": prompt}, ], temperature=0.2, max_tokens=1024, ) return response.choices[0].message.content

temperature设置为 0.2 左右，可以显著降低模型随意发挥的概率。如果业务要求严格，还可以设为 0。

4.9 完整 API 服务

将上述模块串成 FastAPI 接口。

# 文件路径：app/main.py from fastapi import FastAPI from pydantic import BaseModel from app.retrieval.hybrid_search import hybrid_search from app.retrieval.rerank import rerank from app.generation.llm import generate_answer app = FastAPI(title="Enterprise RAG API") class QueryRequest(BaseModel): query: str top_k: int = 20 rerank_top_n: int = 5 class QueryResponse(BaseModel): answer: str sources: list[dict] @app.post("/api/query", response_model=QueryResponse) def query(request: QueryRequest): # 1. 混合检索 candidates = hybrid_search(request.query, request.top_k) # 2. 重排 reranked = rerank(request.query, candidates, request.rerank_top_n) # 3. 生成 answer = generate_answer(request.query, reranked) return QueryResponse( answer=answer, sources=[ { "text": item["text"][:200], "source": item.get("source", ""), "score": item.get("rerank_score", 0), } for item in reranked ], ) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)

启动服务：

uvicorn app.main:app --host 0.0.0.0 --port 8000

请求示例：

curl -X POST http://localhost:8000/api/query \ -H "Content-Type: application/json" \ -d '{"query": "公司请休假制度有哪些规定？"}'

4.10 运行结果说明

预期返回结果类似：

{ "answer": "根据制度文件[1]，员工累计工作满1年不满10年的，年休假5天；已满10年不满20年的，年休假10天。", "sources": [ { "text": "员工累计工作已满1年不满10年的，年休假5天……", "source": "data/员工考勤管理制度.pdf", "score": 8.23 } ] }

到这里，一个最小可运行的 RAG 知识库已经完成。接下来要考虑的是如何把这条链路做成企业级系统。

5. RAG 知识库质量的度量指标

很多项目上线后说不清“效果好不好”，这就是缺少指标定义。建议从以下维度建立评估体系。

5.1 检索质量指标

在没有标注数据的情况下，也可以用“回答引用率”来粗测：把线上日志中的 query 与最终引用来源做统计，看有多少请求使用了检索结果。

5.2 生成质量指标

线上建议建立“用户反馈”机制：答案下方放“有帮助 / 无帮助”按钮，无帮助的样本定期进入标注池，反哺检索与重排优化。

6. 工程化落地：从 Demo 到生产环境

Demo 能跑和项目能上线之间隔着很多工程问题。

6.1 文档更新与增量索引

企业文档每天都在变。文档更新要分成全量索引和增量索引：

• 全量索引：每天或每周固定时间，把所有文档重新处理一遍。
• 增量索引：监控文件系统、OA 系统或数据库表变化，新增和变更的文档实时处理。

增量索引要注意“删除”逻辑。文档变更后，旧向量仍然存在会导致检索到过期内容，所以先删除旧 ID，再写入新向量。

6.2 并发与性能

在线链路的瓶颈通常不在大模型，而在检索和重排。

• 向量检索使用近似最近邻索引，控制查询超时。
• 重排使用 GPU 或更强的 CPU 实例。
• 对高频重复问题增加缓存，缓存 key 可以用 query 的向量距离判断相似度。
• 大模型生成环节设置合理的max_tokens上限，避免慢请求占满连接池。

6.3 安全与权限

企业知识库的内容往往涉及权限分级。如果没有权限控制，用户可能会通过知识库问答间接读取到无权访问的文档。

工程上至少要做到：

• 文档级权限：向量库的 metadata 中记录文档权限组。
• 检索阶段过滤：查询时携带用户身份，检索时按权限组过滤。
• 生成后校验：答案中的引用来源必须是当前用户可见的文档。
• 访问审计：记录每次问答的 user、query、引用的文档 ID、时间戳。

6.4 可观测性

线上系统不能只靠“感觉”。需要监控：

• 检索耗时、重排耗时、生成耗时。
• 检索召回数量、重排后 Top1 分数。
• 大模型 API 的 token 消耗。
• 用户反馈数据。

建议将每个环节的指标写入日志系统，配合 Prometheus 和 Grafana 做大盘展示。

7. 常见问题与排查思路

7.1 检索结果不相关

排查顺序建议：先看召回结果，再看重排结果。如果召回结果里没有正确答案，问题在索引链路；如果召回有但重排后没了，问题在重排。

7.2 向量库写入失败

7.3 大模型回答幻觉

7.4 重排速度太慢

Cross-Encoder 重排的复杂度是 O(N) 次前向推理。当 N = 50 时，每个请求需要 50 次模型推理。优化方向：

• 先用向量和关键词粗召回，只保留前 20 条进重排。
• 重排服务单独部署，独立扩缩容。
• 如果重排模型过大，可以尝试量化版本或更小的 distill 版本。

8. 最佳实践与工程建议

8.1 从“检索正确”而不是“生成正确”入手

RAG 的上限由检索决定。大模型再强，喂进去的资料不对，答案也不可能对。所以效果优化应该把重心放在：

1. 先人工检查召回结果。
2. 再做重排优化。
3. 最后才调整 Prompt。

很多团队一上来就调 Prompt，结果发现调来调去回答还是不准确，原因就是检索环节根本没召回相关内容。

8.2 建立一个评测集

知识库项目上线前，必须准备一个评测集。评测集不需要很大，50 到 100 条真实问题即可，但要有标准答案和来源文档。

每次修改切分策略、Embedding 模型、重排模型、检索参数后，都跑一遍评测集，对比指标变化。这样可以避免“改一个参数，好了这个问题，坏了另一个问题”的情况。

8.3 保留原文溯源

向量库中除了存向量，一定要存原文 chunk、文档路径、页码、章节标题。这样生成答案时可以附带引用来源，方便用户核对，也方便事后审计。

8.4 用日志反哺优化

线上日志是免费的金矿。定期分析：

• 用户点了“有帮助”的 query 和引用的文档。
• 用户点了“无帮助”的 query。
• 多轮对话中重复提问的内容。

这些数据可以帮助识别知识库中的盲区，比如某类问题永远检索不到对应文档，那么很可能知识库里根本没有相关资料，或者有但切分和索引方式不合理。

9. 总结与学习路线

RAG 知识库并不是一个大模型 API 就能解决的问题，而是一条包含文档处理、文本切分、向量化、混合检索、重排、Prompt 工程和工程化部署的完整链路。

本文把全链路拆解成了离线索引和在线推理两条流程，并动手实现了一个可运行的 FastAPI 服务。你可以在这个基础上继续学习：

• Embedding 模型微调：用领域数据微调，让向量检索更懂业务语言。
• ** Query 改写**：引入大模型做查询改写，解决口语化表达和指代问题。
• Agent 化 RAG：让模型在检索不到答案时主动查询数据库、调用 API 或追问用户。
• GraphRAG / Ontology RAG：引入知识图谱结构，处理多跳问答和复杂关系推理。
• 分布式向量数据库：在 Milvus、Elasticsearch 上做大规模索引和部署。

如果你正在做企业级项目，建议依次关注评测集建设、权限控制、日志监控和文档增量更新这四件事。它们不一定能让 Demo 跑得更快，但决定了系统能不能真正投入生产。

动手搭一个最小版本，跑通全链路，然后把你不满意的环节逐一替换成更优方案——这是学习 RAG 最直接的方式。