前几天接了个需求把散落在官网、产品手册、内部FAQ里的内容做成一个能用自然语言直接提问的企业私有知识库。需求一句话能说完真正落地却牵扯出爬取、清洗、切分、向量化、检索、重排、生成一长串环节。我用的技术栈也简单粗暴——Python跑全流程RAG检索增强生成做问答核心。这篇就把我在项目里的完整链路、踩过的坑和可以直接复用的代码一起整理出来给正在做同类需求的同学参考。如果你想给自己的文档库加一个“AI问答”这套东西完全可以拿来改。1. 先拆需求私有知识库问答到底在做什么在做任何开发之前我习惯先把“看起来高大上”的需求翻译成具体问题。所谓企业私有知识库问答本质上就是三个子问题内容从哪里来可能是官网页面、产品手册、售后文档、工单记录甚至是一堆PDF。内容怎么存要支持快速检索还不能泄露到外部。问题怎么答用户提问之后系统能不能给出有依据、可溯源的回答。这三个问题对应三条技术线采集清洗、索引构建、问答生成。串起来就是一套标准的RAG流程。1.1 为什么是RAG而不是微调很多人第一反应是把文档拿去微调大模型我直接否掉这个选项。微调适合的场景是“教会模型某种能力”比如让它模仿某种写作风格、学会特定格式的输出。但知识库问答需要的是“准确回忆某条知识”而且知识还在频繁更新——今天上线的新产品参数明天就要能查到。微调一次成本高、周期长知识更新还得重新训练完全不划算。RAG的思路是把“记忆”外置文档先被切块、向量化、存入向量数据库用户提问时先检索出相关片段再把片段和问题一起交给大模型组织答案。模型不需要“记住”你的文档它只需要“读懂”检索回来的内容。这套机制决定了它天然适合私有知识库这类动态、可追溯的需求。另外RAG还有两个很现实的优点一是所有答案都能标注来源出了问题能倒查二是如果发现答得不对直接改文档、改检索结果就行不用动模型。1.2 全链路模块划分整个系统我分成了四层每一层的职责和产出都很明确层级核心任务产出物采集层爬取授权页面、解析正文、清洗去重干净的纯文本/结构化文档索引层切分文档、向量化、写入向量库可检索的向量索引问答层检索相关片段、构造Prompt、调用大模型带引用的回答评估调优层检查检索质量、回答准确性持续修问题可量化的效果报告这里要特别强调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}, timeout15) 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, hrefTrue): 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, stripTrue) # 兜底取body下所有文本 body soup.find(body) return body.get_text(\n, stripTrue) if body else 如果页面结构不统一这套办法会有点吃力。这时候我一般用trafilatura这个库它是专门做网页正文抽取的准确率比我手写规则高不少import trafilatura def extract_with_trafilatura(html): downloaded trafilatura.extract(html, include_commentsFalse, include_tablesTrue) return downloaded or 两者怎么选页面结构规整、量也不大的场景BeautifulSoup完全够用页面来源五花八门比如既有帮助中心又有博客又有论坛直接上trafilatura省心。2.3 数据清洗与“脏数据”处理正文提取完还需要做一轮清洗。这一步的目标很单纯把文本变成干净的、语义完整的自然语言段落。我常用的清洗操作包括去除不可见字符和乱码统一换行符去掉多余空行、表格的线框符|---|---|---这类处理编码问题latin-1读出来变乱码的强制转utf-8时要用errorsreplace避免崩溃对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_size700, overlap150): # 先按段落拆段落足够大的再按句子切 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单机、数据量百万以内内存索引简单直接入门首选Qdrant需要水平扩展、过滤查询Rust写的部署容易配套丰富Milvus千万级以上超大规模功能全但运维成本偏高pgvector已有PostgreSQL业务库能复用SQL生态适合小团队我这次用的是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_k10): 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_k50第二步用一个重排序模型Reranker对这50个候选按相关性精确打分只保留最相关的5到10个片段再送给大模型。def rerank(query, candidates, top_n5): # candidates: list of {text: ...} pairs [(query, c[text]) for c in candidates] scores reranker.score(pairs) # 返回相关性分数 sorted_pairs sorted(zip(scores, candidates), keylambda 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_k50) texts [meta[chunk_text] for score, meta in candidates] # 2. 重排序 final_chunks rerank(question, texts, top_n5) # 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 }, streamTrue ) 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结果用RRFReciprocal Rank Fusion算法融合效果立竿见影。混合检索的伪代码def hybrid_search(query, chunks, alpha0.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(), keylambda x: -x[1])[:top_k]5.2 回答不准确、引用混乱怎么修症状模型答非所问或者在回答里引用了和答案无关的文档。这通常说明检索回来的片段里有“干扰项”。我有两个调整优先级很高的操作加重排序的筛选力度。把top_n从5降到3把相关度分数低的片段直接过滤掉。宁可少给一点上下文也别给错上下文。模型一旦被无关信息带偏回答质量会断崖式下跌。在Prompt里明确竞争信息时的处理方式。当多个文档片段内容冲突时要求模型优先采用发布时间最新的文档如果片段之间互相矛盾要求模型在回答中明确指出不一致。另一个原因是元数据没做好导致引用串了。检查一下每个片段携带的url和title是否和文本内容一一对应这个Bug我修过不止一次问题往往出在向量库的metadata更新没跟上。5.3 性能与成本控制问答慢、成本高是企业上线的常见拦路虎。几个实用手段手段解决的问题说明缓存高频问题重复提问消耗tokenkey问题答案命中直接返回缩短上下文大模型计费按token重排序后只保留2~4个片段配合检索足够流式输出首字延迟体验差边生成边返回用户感知更快向量检索降级并发量大时CPU高加一层布隆过滤先按标签粗筛掉无关数据Embedding模型本地化数据外泄风险、API费用开源的Embedding模型在CPU上都能跑另外提醒一句日志一定要记录每次问答的问题、检索到的片段列表、模型答案、耗时、token数。没有这些数据出了问题全靠猜根本没法优化。我在实际项目里最后加的也是这套全链路日志。上线头两周就靠它捞出了三个问题一个页面链接权重太高导致反复被爬、一个站点改了页面结构导致解析失败、一个客户提问的专有名词在文档里永远用了英文缩写导致检索命中率低。这三个问题没有一个能从代码逻辑里直接看出来全是日志暴露的。这也是做知识库问答和做普通Web服务最大的不同效果问题是一层层叠加出来的日志就是唯一能让你剥洋葱的工具。整套流程跑顺之后我最大的体会是RAG的地基在数据灵魂在检索最后那步生成反而是最不需要操心的。先把爬虫做规范了把切分调好了把重排序加上哪怕换一个大模型效果都不会差到哪去。如果你也在搭类似的东西别急着上框架、上复杂组件把我这套最小闭环先跑通后面再按需加料。
