前几天接了个需求：把散落在官网、产品手册、内部FAQ里的内容，做成一个能用自然语言直接提问的企业私有知识库。需求一句话能说完，真正落地却牵扯出爬取、清洗、切分、向量化、检索、重排、生成一长串环节。我用的技术栈也简单粗暴——Python跑全流程，RAG（检索增强生成）做问答核心。这篇就把我在项目里的完整链路、踩过的坑和可以直接复用的代码一起整理出来，给正在做同类需求的同学参考。如果你想给自己的文档库加一个“AI问答”，这套东西完全可以拿来改。

1. 先拆需求：私有知识库问答到底在做什么

在做任何开发之前，我习惯先把“看起来高大上”的需求翻译成具体问题。所谓企业私有知识库问答，本质上就是三个子问题：

• 内容从哪里来？可能是官网页面、产品手册、售后文档、工单记录，甚至是一堆PDF。
• 内容怎么存？要支持快速检索，还不能泄露到外部。
• 问题怎么答？用户提问之后，系统能不能给出有依据、可溯源的回答。

这三个问题对应三条技术线：采集清洗、索引构建、问答生成。串起来就是一套标准的RAG流程。

1.1 为什么是RAG而不是微调

很多人第一反应是把文档拿去微调大模型，我直接否掉这个选项。

微调适合的场景是“教会模型某种能力”，比如让它模仿某种写作风格、学会特定格式的输出。但知识库问答需要的是“准确回忆某条知识”，而且知识还在频繁更新——今天上线的新产品参数，明天就要能查到。微调一次成本高、周期长，知识更新还得重新训练，完全不划算。

RAG的思路是把“记忆”外置：文档先被切块、向量化、存入向量数据库，用户提问时先检索出相关片段，再把片段和问题一起交给大模型组织答案。模型不需要“记住”你的文档，它只需要“读懂”检索回来的内容。这套机制决定了它天然适合私有知识库这类动态、可追溯的需求。

另外，RAG还有两个很现实的优点：一是所有答案都能标注来源，出了问题能倒查；二是如果发现答得不对，直接改文档、改检索结果就行，不用动模型。

1.2 全链路模块划分

整个系统我分成了四层，每一层的职责和产出都很明确：

这里要特别强调：RAG的效果天花板不在大模型，而在前两层。文档本身乱七八糟，或者检索回来的片段压根不相关，后面换再强的模型也白搭。

2. 采集层：只做合规且能落地的爬取

采集是整个流程的地基。这个环节做不好，后面所有步骤都是在“垃圾上盖楼”。

2.1 爬什么与怎么爬

企业私有知识库的数据来源，我把它分为三类：官网/帮助中心的静态页面、内部文档系统（比如Wiki、飞书知识库）、离线文件（Word、PDF、Markdown）。

三类来源里，静态页面最适合用爬虫处理。但动手之前必须先确认两个事：这个网站是你自己的，或者你至少获得了授权；网站上的robots协议允许抓取。别碰版权有争议的内容，也别去爬需要登录才能访问的内部系统——那应该走接口对接，而不是爬虫。

合规之后才是技术问题。我的采集脚手架一般长这样：

import requests import time from urllib.parse import urljoin, urlparse from bs4 import BeautifulSoup UA = "Mozilla/5.0 (compatible; YourBot/1.0; +https://your-website.com/bot)" def fetch_page(url): resp = requests.get(url, headers={"User-Agent": UA}, timeout=15) resp.raise_for_status() return resp.text def get_links(page_url, html): """提取站内链接，限定同域名，避免爬到外部去""" soup = BeautifulSoup(html, "html.parser") domain = urlparse(page_url).netloc links = set() for a in soup.find_all("a", href=True): href = urljoin(page_url, a["href"]) if urlparse(href).netloc == domain: links.add(href) return links # 核心：BFS遍历 + 限速 start_url = "https://docs.example.com" visited = set() queue = [start_url] while queue: url = queue.pop(0) if url in visited: continue visited.add(url) try: html = fetch_page(url) # 保存html到本地，后续解析 save_raw(url, html) for link in get_links(url, html): if link not in visited: queue.append(link) time.sleep(0.5) # 控制请求频率，别把服务器打挂 except Exception as e: log_error(url, e)

这段代码有几个细节值得说。

• 限定同域名很重要。如果不加过滤，爬虫会顺着外链跑偏，抓回来一堆无关内容。
• time.sleep(0.5)是给服务器留的喘息时间，也是给自己留的退路。别为了省几分钟把对方站点搞崩。
• save_raw和log_error我建议一定要实现。原始HTML保留一份，方便后续重新解析；错误日志保留一份，方便排查哪些页面没抓到。

2.2 页面解析与正文抽取

HTML页面拿到手，不能直接拿去切分向量化，里面有导航、页脚、广告、脚本标签这些噪音。我见过不少第一次做这个的同学，直接把整个HTML喂给模型，结果回答里全是乱码和链接，所以正文抽取这步省不了。

常规方案是用BeautifulSoup做结构解析：

def extract_main_content(html): soup = BeautifulSoup(html, "html.parser") # 先去掉明显是噪音的标签 for tag in soup.find_all(["script", "style", "nav", "footer", "aside"]): tag.decompose() # 优先按常见正文容器匹配 main = soup.find("article") or soup.find("main") or soup.find("div", class_=["content", "doc-content"]) if main: return main.get_text("\n", strip=True) # 兜底：取body下所有文本 body = soup.find("body") return body.get_text("\n", strip=True) if body else ""

如果页面结构不统一，这套办法会有点吃力。这时候我一般用trafilatura这个库，它是专门做网页正文抽取的，准确率比我手写规则高不少：

import trafilatura def extract_with_trafilatura(html): downloaded = trafilatura.extract(html, include_comments=False, include_tables=True) return downloaded or ""

两者怎么选？页面结构规整、量也不大的场景，BeautifulSoup完全够用；页面来源五花八门，比如既有帮助中心又有博客又有论坛，直接上trafilatura，省心。

2.3 数据清洗与“脏数据”处理

正文提取完，还需要做一轮清洗。这一步的目标很单纯：把文本变成干净的、语义完整的自然语言段落。我常用的清洗操作包括：

• 去除不可见字符和乱码，统一换行符；
• 去掉多余空行、表格的线框符（|---|---|---这类）；
• 处理编码问题，latin-1读出来变乱码的，强制转utf-8时要用errors="replace"避免崩溃；
• 对FAQ页面，把“问：”“答：”统一成对话格式，方便后续切分时保留语义；
• 去重：同一份内容可能在不同URL下重复存在，用正文的哈希值做去重。

再贴一个我刚入行时没写、后来补上的细节：标题要一起存。很多爬虫只保存了正文文本，把页面标题丢掉了，结果后续检索时，模型看到一段文字却不知道它属于哪个文档、哪个章节。标题是元数据的一部分，必须保留，下面这段代码请直接抄进项目里：

import hashlib import re def clean_text(text): # 去掉控制字符和奇怪空白 text = re.sub(r"[\x00-\x1f\x7f]", "", text) text = re.sub(r"[ \t]+", " ", text) text = re.sub(r"\n{3,}", "\n\n", text) return text.strip() def dedup_by_hash(text, seen_hashes): digest = hashlib.md5(text.encode("utf-8")).hexdigest() if digest in seen_hashes: return True seen_hashes.add(digest) return False

把这个函数用在每次保存之前。清洗完的文档要统一格式，我的习惯是一个文档一个JSON文件，包含id、url、title、content、last_updated五个字段，后面所有环节都从这个结构取数据。

3. 索引层：切分、向量化与存储

数据清洗干净，接下来就是RAG最核心的环节：把文档转化成向量索引。这里每一步都影响最终问答质量，我分别说。

3.1 文档切分：最容易被低估的一步

很多做RAG的人把精力全放在模型选择上，却忽略了切分策略。实际上，切分好坏直接决定检索命中率。切块太大，一块里混了多个主题，检索时容易捞回一堆不相关内容；切块太小，单块语义不完整，模型读不懂上下文。

我用的默认策略是以“语义段落”为单位切分，每块控制在500到800字之间，块与块之间保留150字左右的重叠。重叠的作用是防止一个完整知识点刚好被截断成两半，检索时哪一半都答不完整。

代码示例：

import re def split_document(text, chunk_size=700, overlap=150): # 先按段落拆，段落足够大的再按句子切 paragraphs = [p.strip() for p in text.split("\n\n") if p.strip()] chunks = [] current = "" for para in paragraphs: if len(current) + len(para) < chunk_size: current += para + "\n" else: if current: chunks.append(current.strip()) # 保留尾部 overlap 作为下一块的开始 current = current[-overlap:] + "\n" + para + "\n" if overlap else para + "\n" if current.strip(): chunks.append(current.strip()) return chunks

这里要注意，current[-overlap:]的切法只是“取字符”，可能不完整。工业级做法是用句子为单位来做重叠，保证不会在句子中间截断。更好的策略是直接使用现成的递归字符切分器，它会把长段落继续切分，同时尽量保留语义完整。

还有一个进阶技巧：结构化文档可以按标题切分。Markdown的#、##天然是语义边界，先按标题分成大节，每节内部再按长度切，这样检索回来的片段自带上下文标题，模型能看懂这段在讲什么。

3.2 Embedding模型选型

把文本变成向量，这一步叫向量化，用的是Embedding模型。选Embedding模型有几个关键指标：

• 中文效果如何；
• 向量维度多少（维度越高，存储和计算成本越大）；
• 是否开源、能否私有化部署；
• 对长文档的支持（有些模型上限512 token，长文本会被截断）。

我个人建议，企业场景优先选可以本地部署的开源Embedding模型，原因只有一个：私有不外泄。数据在本地处理，不用把企业内部文档发到外部API，心里踏实。如果团队没有GPU资源，再用商用API，但要注意数据安全协议。

下面是接入Embedding模型的通用姿势：

# 这里以调用本地Embedding服务为例 import requests def get_embedding(texts): resp = requests.post( "http://localhost:8000/embedding", json={"texts": texts} ) resp.raise_for_status() return resp.json()["vectors"] # shape: [len(texts), dim]

无论你最后选哪个模型，建议做一个小测试：把几条相似但措辞不同的句子丢进去，算一算向量余弦相似度，人工看一眼结果是否符合预期。这一步能帮你提前发现模型对中文长句理解力不足的问题。

3.3 向量存储与元数据设计

向量存哪里？我这里列几个主流方案：

我这次用的是FAISS，数据量不大，部署最轻。

存储时除了向量本身，一定要把元数据一起存进去。最关键的是这几个字段：doc_id、url、title、chunk_index。为什么？因为问答阶段需要用url和title来生成引用来源，需要用doc_id做去重，用chunk_index保证同一文档内多个片段能按顺序取回。

import faiss import numpy as np class VectorStore: def __init__(self, dim): self.index = faiss.IndexFlatIP(dim) # 内积，配合归一化向量就是余弦相似度 self.metadata = [] # 与索引位置一一对应 def add_document(self, chunks, meta): vectors = get_embedding(chunks) # 归一化 vectors = np.array(vectors).astype("float32") faiss.normalize_L2(vectors) self.index.add(vectors) for i in range(len(chunks)): self.metadata.append({**meta, "chunk_index": i, "chunk_text": chunks[i]}) def search(self, query, top_k=10): q_vec = get_embedding([query])[0] q_vec = np.array([q_vec]).astype("float32") faiss.normalize_L2(q_vec) scores, indices = self.index.search(q_vec, top_k) results = [] for score, idx in zip(scores[0], indices[0]): if idx != -1: results.append((float(score), self.metadata[idx])) return results

注意一个问题：FAISS的IndexFlatIP是暴力检索，数据量大之后速度会变慢。当文档块数量到几十万级别，就该换成IVF索引或者用向量数据库了。做索引的时候把这件事放在心里，避免上线后才发现性能瓶颈。

4. 问答层：检索增强生成的具体实现

索引建完，终于到效果直接可见的问答环节。这个阶段的核心工作只有三件：检索、拼接Prompt、调用大模型。但对效果的追求可以让它再复杂一点。

4.1 先向量检索，再重排序

直接拿用户的query去做向量检索，然后把这个结果一股脑丢给大模型，这是新手最常见的做法。问题是：向量检索的结果在语义上相似，不代表在答案层面真的有用。有些片段可能只是提到了和query一样的词，实际上并不能回答问题。

正确做法是两步走：第一步，向量检索召回尽量多的候选，比如top_k=50；第二步，用一个重排序模型（Reranker）对这50个候选按相关性精确打分，只保留最相关的5到10个片段，再送给大模型。

def rerank(query, candidates, top_n=5): # candidates: list of {"text": ...} pairs = [(query, c["text"]) for c in candidates] scores = reranker.score(pairs) # 返回相关性分数 sorted_pairs = sorted(zip(scores, candidates), key=lambda x: -x[0]) return [c for s, c in sorted_pairs[:top_n]]

重排序模型通常比Embedding模型更重，但对“相关不相关”的判断能力强一个档次。加了这一步之后，回答准确率提升特别明显。

4.2 不写LangChain也能搭RAG链路

框架确实能提速，但我觉得入门阶段最好先自己用原生Python把链路写一遍，你能清楚每一步在干什么。这里给一个最小可用的问答函数，没有用任何框架：

import json import requests def answer_question(question, store): # 1. 召回候选 candidates = store.search(question, top_k=50) texts = [meta["chunk_text"] for score, meta in candidates] # 2. 重排序 final_chunks = rerank(question, texts, top_n=5) # 3. 组装上下文 context = "\n\n".join( f"[来源:{meta['title']}]({meta['url']})\n{chunk}" for chunk, score, meta in final_chunks ) # 4. 调用大模型API prompt = build_prompt(question, context) resp = requests.post( "你的大模型API端点", headers={"Authorization": "Bearer your-api-key"}, json={ "model": "your-model-name", "messages": [{"role": "user", "content": prompt}], "temperature": 0.2 } ) return resp.json()["choices"][0]["message"]["content"]

这段代码里，最关键的是第3步。把来源URL和标题放进上下文里，模型就能在答案里自然地引用出处，用户在页面上也能看到“这段回答来自哪个文档”。

4.3 Prompt构造与流式输出

Prompt写得好不好，对回答质量的影响不亚于检索。我几经迭代之后，最终稳定在一个模板上，这里原样分享：

你是一个企业知识库问答助手。请根据下面提供的文档片段回答用户问题。 要求： 1. 只能使用文档片段中的信息，不要编造。 2. 如果文档片段不足以回答问题，请直接说“根据现有资料无法回答”。 3. 回答要简洁、准确，分点列出关键信息。 4. 每个关键信息后面，用 [来源编号] 标注出自哪个片段。 文档片段： [1] 来自《产品发布说明》：... [2] 来自《FAQ》：... [3] 来自《API文档》：... 用户问题： 如何在系统里配置单点登录？

对应代码：

def build_prompt(question, context_with_sources): return ( "你是一个企业知识库问答助手。请根据下面提供的文档片段回答用户问题。\n\n" "要求：\n" "1. 只能使用文档片段中的信息，不要编造。\n" "2. 如果文档片段不足以回答问题，请直接说“根据现有资料无法回答”。\n" "3. 回答要简洁、准确，分点列出关键信息。\n" "4. 每个关键信息后面，用 [来源编号] 标注出自哪个片段。\n\n" "文档片段：\n" + context_with_sources + "\n\n" "用户问题：\n" + question )

流式输出这块，很多大模型API支持SSE流式返回，体验比等全部生成完更好。实现也不复杂，只需要把HTTP请求带上流式参数，然后逐段读取响应：

def stream_answer(question, context): prompt = build_prompt(question, context) resp = requests.post( "你的大模型API端点", headers={"Authorization": "Bearer your-api-key"}, json={ "model": "your-model-name", "messages": [{"role": "user", "content": prompt}], "stream": True }, stream=True ) for line in resp.iter_lines(): if not line: continue # 解析 SSE 数据 data = line.decode("utf-8") if data.startswith("data:"): chunk = json.loads(data[5:]) delta = chunk["choices"][0]["delta"].get("content", "") yield delta

用户端每拿到一段文字就渲染一段，体感上基本就是“打字机效果”，非常加分。

5. 效果调优与常见问题排查

这一章我想直接分享实测中高频踩到的问题，每一条都是我修过的真坑，你可以当速查表用。

5.1 检索不到相关内容怎么办

症状：用户问了一个问题，系统回答“根据现有资料无法回答”，但知识库里明明有答案。

优先排查这几个方面：

• 切分太碎：每块只有一两百字，语义信息不够，向量表示不准确。把chunk_size提到600到1000试试。
• 嵌入模型理解力不足：换一个更强或者专门为中文优化的Embedding模型，最直观。
• 问题表述和文档写法差异太大：用户说“怎么改密码”，文档里写的是“更新登录凭证”。试试做一次查询改写，让大模型先把口语化query转换成更书面化的多个检索子问题，再一一去查。
• 没有用混合检索：向量检索对“关键词精确匹配”不敏感，像型号、订单号这类内容，用BM25关键词匹配命中率更高。把向量检索和BM25结果用RRF（Reciprocal Rank Fusion）算法融合，效果立竿见影。

混合检索的伪代码：

def hybrid_search(query, chunks, alpha=0.5): vector_scores = vector_search(query) keyword_scores = bm25_search(query) final_scores = {} for doc in set(vector_scores) | set(keyword_scores): final_scores[doc] = alpha * vector_scores.get(doc, 0) + (1 - alpha) * keyword_scores.get(doc, 0) return sorted(final_scores.items(), key=lambda x: -x[1])[:top_k]

5.2 回答不准确、引用混乱怎么修

症状：模型答非所问，或者在回答里引用了和答案无关的文档。

这通常说明检索回来的片段里有“干扰项”。我有两个调整优先级很高的操作：

• 加重排序的筛选力度。把top_n从5降到3，把相关度分数低的片段直接过滤掉。宁可少给一点上下文，也别给错上下文。模型一旦被无关信息带偏，回答质量会断崖式下跌。
• 在Prompt里明确竞争信息时的处理方式。当多个文档片段内容冲突时，要求模型优先采用发布时间最新的文档；如果片段之间互相矛盾，要求模型在回答中明确指出不一致。

另一个原因是元数据没做好，导致引用串了。检查一下每个片段携带的url和title是否和文本内容一一对应，这个Bug我修过不止一次，问题往往出在向量库的metadata更新没跟上。

5.3 性能与成本控制

问答慢、成本高，是企业上线的常见拦路虎。几个实用手段：

另外提醒一句，日志一定要记录每次问答的：问题、检索到的片段列表、模型答案、耗时、token数。没有这些数据，出了问题全靠猜，根本没法优化。

我在实际项目里最后加的也是这套全链路日志。上线头两周就靠它捞出了三个问题：一个页面链接权重太高导致反复被爬、一个站点改了页面结构导致解析失败、一个客户提问的专有名词在文档里永远用了英文缩写导致检索命中率低。这三个问题没有一个能从代码逻辑里直接看出来，全是日志暴露的。这也是做知识库问答和做普通Web服务最大的不同：效果问题是一层层叠加出来的，日志就是唯一能让你剥洋葱的工具。

整套流程跑顺之后，我最大的体会是：RAG的地基在数据，灵魂在检索，最后那步生成反而是最不需要操心的。先把爬虫做规范了，把切分调好了，把重排序加上，哪怕换一个大模型，效果都不会差到哪去。如果你也在搭类似的东西，别急着上框架、上复杂组件，把我这套最小闭环先跑通，后面再按需加料。