庫(kù)問(wèn)答實(shí)戰(zhàn)：從零搭建本地RAG系統(tǒng))
最近在給團(tuán)隊(duì)搭建內(nèi)部知識(shí)庫(kù)問(wèn)答系統(tǒng)時(shí)遇到了一個(gè)很典型的問(wèn)題大模型對(duì)通用知識(shí)很擅長(zhǎng)但一碰到企業(yè)內(nèi)部的規(guī)章制度、產(chǎn)品文檔、歷史項(xiàng)目總結(jié)回答就開(kāi)始“一本正經(jīng)地胡說(shuō)八道”。嘗試過(guò)把資料直接拼進(jìn) Prompt很快又遇到 Token 長(zhǎng)度限制和成本問(wèn)題。后來(lái)把 RAG 體系完整落地了一遍才真正解決了“讓模型基于自有知識(shí)回答問(wèn)題”這個(gè)核心訴求。這篇文章會(huì)從 RAG 的原理講起然后完整搭建一個(gè)本地知識(shí)庫(kù)問(wèn)答系統(tǒng)涵蓋文檔加載、切塊、向量化、檢索、重排序和本地模型生成最后再聊一聊高級(jí)檢索的實(shí)戰(zhàn)思路。適合剛開(kāi)始接觸 AI 應(yīng)用開(kāi)發(fā)、想搞懂 RAG 全流程的讀者也適合已經(jīng)在做 LLM 應(yīng)用、想優(yōu)化召回效果的開(kāi)發(fā)者。1. 什么是 RAG先解決“模型不知道”的問(wèn)題1.1 大模型的知識(shí)局限大模型的知識(shí)來(lái)自訓(xùn)練時(shí)見(jiàn)過(guò)的數(shù)據(jù)所以它有兩個(gè)天然短板一是訓(xùn)練數(shù)據(jù)有截止時(shí)間新產(chǎn)生的信息它不知道二是私有知識(shí)它幾乎不可能學(xué)到比如公司內(nèi)部的客戶(hù)資料、設(shè)備手冊(cè)、項(xiàng)目復(fù)盤(pán)、專(zhuān)有規(guī)范這些數(shù)據(jù)不會(huì)出現(xiàn)在公開(kāi)語(yǔ)料里。如果直接把問(wèn)題拋給模型它只能依靠參數(shù)里記憶的“模糊印象”來(lái)回答結(jié)果就是一本正經(jīng)地編造。這不是模型不夠聰明而是它的知識(shí)邊界本身就固定了。RAG 的解決思路非常直接不要求模型記住所有知識(shí)而是在回答之前先從一個(gè)外部知識(shí)庫(kù)中檢索出相關(guān)的片段把這些片段作為上下文一起交給模型讓模型基于這些材料作答。1.2 RAG 的基本工作流程先給一個(gè)整體的流程印象用戶(hù)提問(wèn) - 查詢(xún)理解可選 - 檢索從知識(shí)庫(kù)中找回相關(guān)文檔片段 - 增強(qiáng)把檢索結(jié)果組裝進(jìn) Prompt - 生成大模型基于上下文生成答案 - 返回答案與引用來(lái)源整個(gè)過(guò)程可以拆成兩個(gè)離線階段和一個(gè)在線階段離線索引階段把文檔切塊對(duì)每個(gè)塊做向量化Embedding然后寫(xiě)入向量數(shù)據(jù)庫(kù)。在線檢索階段把用戶(hù)問(wèn)題向量化到向量數(shù)據(jù)庫(kù)里做相似度檢索取回 TopK 候選片段。在線生成階段把“用戶(hù)問(wèn)題 檢索片段 系統(tǒng)提示詞”拼在一起交給大模型生成最終回答。這里的核心思路是讓模型學(xué)會(huì)“查資料”而不是“背資料”。1.3 RAG 與微調(diào)的邊界很多初學(xué)者會(huì)把 RAG 和微調(diào)Fine-tuning搞混。簡(jiǎn)單區(qū)分一下維度RAG微調(diào)知識(shí)更新替換/新增文檔即可需要重新訓(xùn)練私有知識(shí)引入通過(guò)檢索注入上下文通過(guò)調(diào)整模型參數(shù)幻覺(jué)控制較好可引用來(lái)源有限仍可能編造成本主要是向量化和檢索服務(wù)訓(xùn)練和推理成本較高適合場(chǎng)景知識(shí)庫(kù)問(wèn)答、實(shí)時(shí)信息查詢(xún)指令遵循、領(lǐng)域風(fēng)格調(diào)整實(shí)際項(xiàng)目里兩者不是二選一而是可以結(jié)合使用。先用 RAG 解決知識(shí)的及時(shí)性和可追溯性再用微調(diào)優(yōu)化模型的表達(dá)方式和工具調(diào)用能力。2. 環(huán)境準(zhǔn)備與項(xiàng)目結(jié)構(gòu)2.1 運(yùn)行環(huán)境與版本說(shuō)明本文的實(shí)戰(zhàn)部分以本地環(huán)境為例避開(kāi)云服務(wù)依賴(lài)方便你完整復(fù)現(xiàn)。版本建議如下但需要根據(jù)你的實(shí)際環(huán)境調(diào)整操作系統(tǒng)Windows 10/11、Ubuntu 20.04 或 macOS 均可Python3.9 及以上本地模型推理llama.cpp Qwen2-7B-Instruct 量化模型向量模型BAAI/bge-small-zh-v1.5中文效果較好資源占用低向量數(shù)據(jù)庫(kù)FAISS開(kāi)發(fā)環(huán)境足夠生產(chǎn)可替換為 Milvus 或 pgvectorWeb 框架FastAPI分詞與關(guān)鍵詞檢索jieba BM25用于混合檢索注意不同版本之間接口可能有差異比如langchain的 API 改動(dòng)一直比較頻繁。本文盡量少依賴(lài)框架封裝用原生 Python 實(shí)現(xiàn)核心鏈路這樣你理解原理后換成任何框架都能快速上手。2.2 安裝依賴(lài)建議先創(chuàng)建虛擬環(huán)境python -m venv rag_env source rag_env/bin/activate # Windows 使用 rag_env\Scripts\activate接下來(lái)安裝必要依賴(lài)pip install fastapi uvicorn faiss-cpu sentence-transformers pip install llama-cpp-python pip install jieba rank-bm25如果你的機(jī)器沒(méi)有 NVIDIA GPUllama-cpp-python會(huì)走 CPU 推理速度會(huì)慢一些但勝在部署簡(jiǎn)單。如果要安裝支持 GPU 的版本可以自行查閱對(duì)應(yīng)平臺(tái)的編譯方式。2.3 項(xiàng)目目錄結(jié)構(gòu)為了方便維護(hù)我們把項(xiàng)目拆分成模塊rag_demo/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 入口 │ ├── config.py # 全局配置 │ ├── loader.py # 文檔加載與切塊 │ ├── embedder.py # Embedding 模型封裝 │ ├── store.py # 向量索引與檢索 │ ├── retriever.py # 混合檢索 重排序 │ └── llm.py # 本地模型加載與生成 ├── data/ │ └── docs/ # 原始知識(shí)文檔 ├── index/ # 向量索引持久化目錄 └── requirements.txt這個(gè)結(jié)構(gòu)不算復(fù)雜但模塊邊界清晰。索引構(gòu)建、檢索、生成是獨(dú)立的環(huán)節(jié)后續(xù)替換任意一個(gè)組件都不會(huì)影響整體流程。3. RAG 核心原理拆解3.1 索引階段切塊與向量化索引階段的第一步是切塊Chunking。為什么要切塊因?yàn)榇竽Ｐ洼斎腴L(zhǎng)度有限不可能把整本書(shū)塞進(jìn) Prompt。檢索的粒度越細(xì)能命中的片段越精準(zhǔn)。多個(gè)小片段可以組合出更豐富的上下文。切塊不是越短越好。切得太短語(yǔ)義不完整切得太長(zhǎng)噪音太多還浪費(fèi) Token。比較常見(jiàn)的策略是固定長(zhǎng)度切塊按字符數(shù)或 Token 數(shù)切比如每塊 300~500 字符。按段落切塊先按換行符拆成段落再把長(zhǎng)段落繼續(xù)拆。按語(yǔ)義切塊用分割模型識(shí)別語(yǔ)義邊界比如標(biāo)題、小節(jié)、句子邊界。父子切塊小片段用于檢索命中后返回對(duì)應(yīng)的父級(jí)大片段用于生成。切塊之后每一個(gè)塊會(huì)通過(guò) Embedding 模型轉(zhuǎn)換成一個(gè)高維向量。向量化的目標(biāo)是語(yǔ)義相近的文本在向量空間里的距離更近。這里有個(gè)容易踩的坑不要在檢索時(shí)才對(duì)文檔臨時(shí)向量化。必須提前把整份文檔庫(kù)向量化并建好索引用戶(hù)查詢(xún)時(shí)才能做到毫秒級(jí)檢索。3.2 檢索階段相似度計(jì)算與召回檢索階段做的事情是計(jì)算用戶(hù)問(wèn)題向量與知識(shí)庫(kù)中每個(gè)片段向量的相似度找出最相似的 TopK 片段。常見(jiàn)相似度計(jì)算方式余弦相似度Cosine最常用適合文本向量。內(nèi)積Dot Product在歸一化向量上等價(jià)于余弦相似度。歐氏距離值越小越相似。以 FAISS 為例IndexFlatIP就是內(nèi)積索引。如果向量都做了歸一化內(nèi)積結(jié)果等價(jià)于余弦相似度。import numpy as np from sentence_transformers import SentenceTransformer model SentenceTransformer(BAAI/bge-small-zh-v1.5) query 如何申請(qǐng)?jiān)O(shè)備維修 query_vector model.encode(query, normalize_embeddingsTrue) doc_vectors model.encode( [設(shè)備維修需要填寫(xiě)申請(qǐng)單, 請(qǐng)假流程說(shuō)明, 采購(gòu)審批權(quán)限], normalize_embeddingsTrue ) scores np.dot(doc_vectors, query_vector) print(scores)輸出結(jié)果中得分最高的就是與問(wèn)題最相關(guān)的片段。不過(guò)單純的向量檢索有一個(gè)問(wèn)題它對(duì)關(guān)鍵詞匹配不敏感。比如用戶(hù)搜索“維修”文檔里寫(xiě)的是“故障處理”雖然語(yǔ)義相近但向量召回可能不夠精確。這就是后面要講的混合檢索解決的問(wèn)題。3.3 生成階段Prompt 增強(qiáng)與上下文約束檢索到相關(guān)片段后不是直接把片段丟給模型而是組裝成一個(gè)結(jié)構(gòu)清晰的 Prompt。一個(gè)典型的 RAG Prompt 結(jié)構(gòu)如下你是一個(gè)智能問(wèn)答助手。請(qǐng)嚴(yán)格根據(jù)下面提供的參考文檔回答問(wèn)題。 如果參考文檔中沒(méi)有足夠信息請(qǐng)直接回答“當(dāng)前知識(shí)庫(kù)中沒(méi)有相關(guān)內(nèi)容”不要編造。 參考文檔 [1] 設(shè)備維修需要填寫(xiě)維修申請(qǐng)單交由部門(mén)主管審批。 [2] 緊急故障可以電話(huà)聯(lián)系運(yùn)維中心熱線12345。 用戶(hù)問(wèn)題如何申請(qǐng)?jiān)O(shè)備維修這里有幾個(gè)關(guān)鍵點(diǎn)明確告訴模型“只能依據(jù)參考文檔回答”減少幻覺(jué)。要求“沒(méi)有信息時(shí)承認(rèn)不知道”避免模型強(qiáng)行編造。給文檔編號(hào)方便模型在回答中引用。在 Prompt 中保留來(lái)源信息便于前端展示引用。Prompt 的設(shè)計(jì)直接決定了 RAG 回答質(zhì)量的下限。同一個(gè)檢索結(jié)果用不同的 Prompt 模板生成出的答案可能差別很大。4. 從 0 搭建本地 RAG 知識(shí)庫(kù)問(wèn)答系統(tǒng)接下來(lái)進(jìn)入實(shí)戰(zhàn)環(huán)節(jié)。我們會(huì)基于 llama.cpp Qwen2-7B FastAPI 構(gòu)建一個(gè)完整的本地 RAG 知識(shí)庫(kù)問(wèn)答系統(tǒng)。核心鏈路是加載文檔 - 切塊 - 向量化 - 建立索引 - 檢索 - 增強(qiáng) - 生成。4.1 文檔加載與切塊實(shí)現(xiàn)先實(shí)現(xiàn)文檔加載與切塊的模塊。這里以讀取.txt和.md文件為例你也可以擴(kuò)展為 PDF、Word 等格式。文件路徑app/loader.pyimport os from typing import List def load_documents(docs_dir: str) - List[str]: 加載目錄下的所有文本文件 docs [] for root, _, files in os.walk(docs_dir): for name in files: if name.endswith((.txt, .md)): file_path os.path.join(root, name) with open(file_path, r, encodingutf-8) as f: content f.read() docs.append({source: file_path, content: content}) return docs def split_text(text: str, chunk_size: int 400, overlap: int 80) - List[str]: 按固定長(zhǎng)度切塊并保留重疊區(qū)間避免切斷語(yǔ)義 chunks [] start 0 while start len(text): end start chunk_size chunk text[start:end] if chunk.strip(): chunks.append(chunk) if end len(text): break start end - overlap return chunks def build_chunks(docs_dir: str) - List[dict]: 加載文檔并返回帶有元信息的切塊列表 all_chunks [] docs load_documents(docs_dir) for doc in docs: chunks split_text(doc[content]) for idx, chunk in enumerate(chunks): all_chunks.append({ id: f{doc[source]}#{idx}, source: doc[source], text: chunk, }) return all_chunks這里切塊時(shí)增加了overlap參數(shù)目的是讓相鄰片段之間保留一部分上下文減少因?yàn)橛睬袑?dǎo)致的語(yǔ)義斷裂。比如一段話(huà)在 400 字符處正好被切斷重疊部分能保證下一塊的開(kāi)頭有足夠信息。4.2 向量化與索引構(gòu)建接下來(lái)實(shí)現(xiàn) Embedding 封裝和 FAISS 索引構(gòu)建。文件路徑app/embedder.pyfrom sentence_transformers import SentenceTransformer class Embedder: def __init__(self, model_name: str BAAI/bge-small-zh-v1.5): self.model SentenceTransformer(model_name) def embed_documents(self, texts): return self.model.encode(texts, normalize_embeddingsTrue) def embed_query(self, text): return self.model.encode(text, normalize_embeddingsTrue)文件路徑app/store.pyimport json import os import numpy as np import faiss class VectorStore: def __init__(self, embedder, index_dir: str index): self.embedder embedder self.index_dir index_dir self.index None self.metadata [] def build(self, chunks: list): 根據(jù)切塊列表構(gòu)建 FAISS 索引 texts [c[text] for c in chunks] vectors self.embedder.embed_documents(texts) dim vectors.shape[1] self.index faiss.IndexFlatIP(dim) self.index.add(vectors) self.metadata chunks os.makedirs(self.index_dir, exist_okTrue) faiss.write_index(self.index, os.path.join(self.index_dir, faiss.index)) with open(os.path.join(self.index_dir, metadata.json), w, encodingutf-8) as f: json.dump(self.metadata, f, ensure_asciiFalse, indent2) def load(self): 加載已有的索引和元數(shù)據(jù) index_path os.path.join(self.index_dir, faiss.index) meta_path os.path.join(self.index_dir, metadata.json) if not os.path.exists(index_path) or not os.path.exists(meta_path): raise FileNotFoundError(索引文件不存在請(qǐng)先執(zhí)行 build()) self.index faiss.read_index(index_path) with open(meta_path, r, encodingutf-8) as f: self.metadata json.load(f) def search(self, query: str, top_k: int 5): 向量相似度檢索 query_vector self.embedder.embed_query(query).reshape(1, -1) scores, indices self.index.search(query_vector, top_k) results [] for score, idx in zip(scores[0], indices[0]): if idx 0: continue results.append({ score: float(score), text: self.metadata[idx][text], source: self.metadata[idx][source], }) return results這里選用IndexFlatIP作為索引結(jié)構(gòu)。它是暴力精確檢索數(shù)據(jù)量不大時(shí)速度足夠快而且效果最準(zhǔn)確。等數(shù)據(jù)量到了百萬(wàn)級(jí)再換成 IVF 或 HNSW 也不遲。4.3 基于 FastAPI 實(shí)現(xiàn)檢索接口現(xiàn)在我們把檢索能力包裝成 HTTP 接口便于前端或其他服務(wù)調(diào)用。文件路徑app/main.pyfrom fastapi import FastAPI from pydantic import BaseModel from app.embedder import Embedder from app.store import VectorStore app FastAPI() embedder Embedder() store VectorStore(embedder) store.load() class SearchRequest(BaseModel): query: str top_k: int 5 app.post(/search) def search(req: SearchRequest): results store.search(req.query, req.top_k) return {query: req.query, results: results}啟動(dòng)服務(wù)uvicorn app.main:app --host 0.0.0.0 --port 8000調(diào)用接口curl -X POST http://localhost:8000/search \ -H Content-Type: application/json \ -d {query: 如何申請(qǐng)?jiān)O(shè)備維修, top_k: 3}返回結(jié)果里包含相似度得分、命中的文本片段和來(lái)源文件路徑方便后續(xù)拼接 Prompt 和展示引用。4.4 接入 Qwen2 本地模型生成答案檢索只是拿到了資料真正回答用戶(hù)問(wèn)題還需要大模型。我們通過(guò) llama.cpp 加載量化后的 Qwen2-7B 模型在本地進(jìn)行推理。文件路徑app/llm.pyfrom llama_cpp import Llama class LocalLLM: def __init__(self, model_path: str, n_ctx: int 4096): self.llm Llama( model_pathmodel_path, n_ctxn_ctx, n_threads8, verboseFalse, ) def generate(self, prompt: str): output self.llm( prompt, max_tokens1024, temperature0.2, top_p0.9, stop[/s, Human:, 用戶(hù):], ) return output[choices][0][text].strip()這里把temperature設(shè)置得比較低0.2是為了讓模型更嚴(yán)格地依據(jù)參考文檔作答減少隨機(jī)編造。接下來(lái)把檢索、Prompt 增強(qiáng)和生成串成一個(gè)完整接口。更新app/main.pyfrom fastapi import FastAPI from pydantic import BaseModel from app.embedder import Embedder from app.store import VectorStore from app.llm import LocalLLM app FastAPI() embedder Embedder() store VectorStore(embedder) store.load() # 根據(jù)實(shí)際模型路徑調(diào)整 llm LocalLLM(model_path./models/qwen2-7b-instruct-q4_k_m.gguf) class AskRequest(BaseModel): question: str top_k: int 5 def build_prompt(question: str, contexts: list) - str: docs_text \n\n.join( [f[{i1}] {item[text]}來(lái)源{item[source]} for i, item in enumerate(contexts)] ) prompt f你是一個(gè)智能問(wèn)答助手。請(qǐng)嚴(yán)格根據(jù)下面提供的參考文檔回答問(wèn)題。 如果參考文檔中沒(méi)有足夠信息請(qǐng)直接回答“當(dāng)前知識(shí)庫(kù)中沒(méi)有相關(guān)內(nèi)容”不要編造。 參考文檔 {docs_text} 用戶(hù)問(wèn)題{question} 請(qǐng)給出準(zhǔn)確、簡(jiǎn)潔的回答并標(biāo)注答案主要參考了哪個(gè)文檔編號(hào)。 return prompt app.post(/ask) def ask(req: AskRequest): # 1. 檢索 contexts store.search(req.question, req.top_k) # 2. 構(gòu)造 Prompt prompt build_prompt(req.question, contexts) # 3. 生成 answer llm.generate(prompt) return { question: req.question, answer: answer, references: [ {text: c[text], source: c[source]} for c in contexts ], }整個(gè)流程非常清晰先檢索再增強(qiáng)最后生成。前端拿到answer展示答案拿到references生成引用來(lái)源這樣用戶(hù)可以看到模型回答的依據(jù)。4.5 運(yùn)行與驗(yàn)證完整執(zhí)行流程如下。第一步準(zhǔn)備文檔。在data/docs目錄下放入你的知識(shí)文檔比如device_repair.md# 設(shè)備維修流程 設(shè)備出現(xiàn)故障后使用人需要在 OA 系統(tǒng)中填寫(xiě)維修申請(qǐng)單。 申請(qǐng)單需要注明設(shè)備編號(hào)、故障描述和期望維修時(shí)間。 部門(mén)主管審批通過(guò)后由行政部統(tǒng)一對(duì)接維修廠商。 緊急故障可以直接撥打運(yùn)維中心電話(huà) 400-1234-567。第二步構(gòu)建索引。簡(jiǎn)單寫(xiě)一個(gè)構(gòu)建腳本python -c from app.loader import build_chunks; from app.embedder import Embedder; from app.store import VectorStore; chunks build_chunks(data/docs); print(chunks:, len(chunks)); store VectorStore(Embedder()); store.build(chunks); print(build done)第三步啟動(dòng)服務(wù)并提問(wèn)uvicorn app.main:app --host 0.0.0.0 --port 8000curl -X POST http://localhost:8000/ask \ -H Content-Type: application/json \ -d {question: 設(shè)備壞了怎么報(bào)修}預(yù)期輸出中answer會(huì)引用文檔里的設(shè)備維修流程references會(huì)包含命中的原文和來(lái)源路徑。到這里一個(gè)最小可用的本地 RAG 系統(tǒng)就跑通了。接下來(lái)我們繼續(xù)做高級(jí)檢索優(yōu)化因?yàn)榛A(chǔ)版在真實(shí)場(chǎng)景里往往不夠用。5. 高級(jí)檢索實(shí)戰(zhàn)讓 RAG 更聰明5.1 混合檢索向量 關(guān)鍵詞純向量檢索擅長(zhǎng)語(yǔ)義相似但有時(shí)會(huì)漏掉“關(guān)鍵詞完全一致”的文檔。比如用戶(hù)問(wèn)“維修電話(huà)”文檔里確實(shí)寫(xiě)了一個(gè)維修電話(huà)向量檢索可能因?yàn)榫渥诱w語(yǔ)義偏移而沒(méi)有召回。解決方案是混合檢索同時(shí)跑向量檢索和關(guān)鍵詞檢索如 BM25再把兩路結(jié)果合并去重。BM25 是一個(gè)經(jīng)典的關(guān)鍵詞相關(guān)性排序算法。簡(jiǎn)單說(shuō)它根據(jù)詞頻、文檔長(zhǎng)度等因素計(jì)算查詢(xún)?cè)~和文檔的相關(guān)性對(duì)“精確詞命中”非常敏感。文件路徑app/retriever.pyimport jieba from rank_bm25 import BM25Okapi from app.store import VectorStore class HybridRetriever: def __init__(self, vector_store: VectorStore, chunks: list): self.vector_store vector_store self.chunks chunks tokenized_docs [list(jieba.cut(c[text])) for c in chunks] self.bm25 BM25Okapi(tokenized_docs) def retrieve(self, query: str, top_k: int 5, vec_weight: float 0.5): # 向量檢索 vec_results self.vector_store.search(query, top_k * 2) # 關(guān)鍵詞檢索 tokenized_query list(jieba.cut(query)) bm25_scores self.bm25.get_scores(tokenized_query) bm25_top_ids sorted( range(len(bm25_scores)), keylambda i: bm25_scores[i], reverseTrue )[: top_k * 2] # 合并結(jié)果 score_map {} for item in vec_results: score_map[item[source] # item[text][:20]] { text: item[text], source: item[source], score: item[score] * vec_weight, } for idx in bm25_top_ids: key self.chunks[idx][source] # self.chunks[idx][text][:20] if key not in score_map: score_map[key] { text: self.chunks[idx][text], source: self.chunks[idx][source], score: 0, } score_map[key][score] bm25_scores[idx] * (1 - vec_weight) # 按融合分?jǐn)?shù)排序 sorted_items sorted( score_map.values(), keylambda x: x[score], reverseTrue ) return sorted_items[:top_k]融合策略用的是加權(quán)求和。vec_weight控制向量和關(guān)鍵詞的比重實(shí)際項(xiàng)目里可以通過(guò)評(píng)估集來(lái)調(diào)參。也可以使用 RRFReciprocal Rank Fusion等更穩(wěn)定的融合方法。5.2 重排序Rerank 提升精度混合檢索取回的 TopK 里依然可能存在“看著相關(guān)但實(shí)際不對(duì)”的片段。這時(shí)候可以加一個(gè)重排序Rerank環(huán)節(jié)先用輕量級(jí)檢索擴(kuò)大召回再用一個(gè)更精確的交叉編碼器模型逐對(duì)精排。重排序的本質(zhì)是把“給定問(wèn)題判斷文檔是否相關(guān)”變成一個(gè)二分類(lèi)問(wèn)題。交叉編碼器把問(wèn)題和文檔拼接后輸入模型輸出相關(guān)性分?jǐn)?shù)比單純向量相似度更準(zhǔn)。from sentence_transformers import CrossEncoder reranker CrossEncoder(BAAI/bge-reranker-base) def rerank(query: str, candidates: list, top_k: int 3): pairs [(query, c[text]) for c in candidates] scores reranker.predict(pairs) scored list(zip(candidates, scores)) scored.sort(keylambda x: x[1], reverseTrue) return [c for c, s in scored[:top_k]]重排序會(huì)增加一次模型推理的耗時(shí)但能明顯提升最終答案的準(zhǔn)確性。生產(chǎn)環(huán)境里通常把它用于 TopK 精細(xì)篩選比如先召回 20 條再用 Rerank 選 3 條進(jìn)入 Prompt。5.3 查詢(xún)改寫(xiě)與 HyDE用戶(hù)的問(wèn)題往往不夠精確比如“這個(gè)流程需要哪些人審批”如果沒(méi)有上下文檢索系統(tǒng)很難判斷“這個(gè)流程”指什么。這時(shí)候可以在檢索前做查詢(xún)改寫(xiě)Query Rewriting把模糊問(wèn)題轉(zhuǎn)換成更明確的檢索語(yǔ)句。一種簡(jiǎn)單的方法是讓大模型先把問(wèn)題改寫(xiě)為適合檢索的查詢(xún)?cè)~。你是一個(gè)搜索查詢(xún)優(yōu)化器。請(qǐng)根據(jù)用戶(hù)的原始問(wèn)題生成 3 個(gè)適合檢索的查詢(xún)?cè)~。 要求保留關(guān)鍵實(shí)體和業(yè)務(wù)術(shù)語(yǔ)不要修改事實(shí)信息。 原始問(wèn)題這個(gè)流程需要哪些人審批另一種思路是 HyDEHypothetical Document Embeddings。它的做法是先讓大模型根據(jù)問(wèn)題生成一個(gè)“假設(shè)性答案”然后用這個(gè)答案去檢索。因?yàn)榧僭O(shè)性答案與目標(biāo)文檔在語(yǔ)義上更接近往往比直接檢索問(wèn)題效果更好。請(qǐng)根據(jù)你的知識(shí)回答下面問(wèn)題。答案可以是猜測(cè)性的也可以是概括性的。 問(wèn)題是請(qǐng)?zhí)峁┰O(shè)備維修申請(qǐng)流程。然后把生成的答案向量化去向量庫(kù)搜索。這種方式很適合“問(wèn)題簡(jiǎn)短、文檔復(fù)雜”的場(chǎng)景。5.4 Agentic RAG 簡(jiǎn)介傳統(tǒng) RAG 是“一次檢索一次生成”。Agentic RAG 則把檢索過(guò)程交給 Agent 智能規(guī)劃可以多輪檢索、調(diào)用工具、逐步推理。比如用戶(hù)問(wèn)“對(duì)比一下 A 和 B 兩種方案的差異”Agent 可能會(huì)先檢索 A 方案相關(guān)文檔。再檢索 B 方案相關(guān)文檔。還可能檢索“對(duì)比框架”“評(píng)估指標(biāo)”等材料。最后匯總生成對(duì)比結(jié)論。這比一次性檢索做得更深但實(shí)現(xiàn)復(fù)雜度也高很多。適合從“單點(diǎn)問(wèn)答”升級(jí)到“復(fù)雜分析”的場(chǎng)景。可以先掌握基礎(chǔ) RAG再逐步引入 Agent 能力。6. 常見(jiàn)問(wèn)題與排查清單6.1 召回結(jié)果差檢索不到相關(guān)內(nèi)容常見(jiàn)原因解決思路切塊太大語(yǔ)義被稀釋縮小 chunk_size增加 overlap切塊太小上下文不完整增大 chunk_size或使用父子切塊Embedding 模型不適合中文換用 bge-small-zh 或 bge-large-zh問(wèn)題描述與文檔表述差異大引入查詢(xún)改寫(xiě)、HyDE 或混合檢索數(shù)據(jù)量太少檢查文檔是否完整加載是否有編碼問(wèn)題排查時(shí)最直接的方式是先打印檢索結(jié)果不經(jīng)過(guò)生成環(huán)節(jié)。看看返回的片段是否真的相關(guān)。如果片段本身不對(duì)問(wèn)題大概率在索引或檢索策略上而不是 Prompt 和模型的問(wèn)題。6.2 切塊不合理導(dǎo)致回答破碎固定長(zhǎng)度切塊雖然簡(jiǎn)單但經(jīng)常把完整段落攔腰截?cái)?。比如一句?huà)從第 390 個(gè)字符開(kāi)始到第 410 個(gè)字符結(jié)束切塊后這句話(huà)只出現(xiàn)一半模型自然讀不懂。建議優(yōu)先“按標(biāo)題層級(jí) 段落”切塊。比如 Markdown 文檔先按##拆成章節(jié)再處理單個(gè)章節(jié)里的長(zhǎng)段落。也可以用父子切塊子塊短小用于精確檢索命中的子塊關(guān)聯(lián)到父塊生成時(shí)把父塊內(nèi)容作為上下文。6.3 本地模型推理慢llama.cpp在 CPU 上跑 7B 模型速度確實(shí)不快。優(yōu)化思路有幾個(gè)使用更小的量化等級(jí)比如 Q4_K_M 比 Q8 更快。減少n_ctx上下文長(zhǎng)度降低顯存/內(nèi)存占用?？刂苖ax_tokens不要無(wú)腦生成長(zhǎng)答案。對(duì)檢索片段做摘要只把關(guān)鍵信息放進(jìn) Prompt。如果條件允許升級(jí) GPU 或使用 API 版模型。另外可以加一層緩存相同問(wèn)題在短時(shí)間內(nèi)直接返回歷史答案減少模型調(diào)用次數(shù)。6.4 中文 Embedding 效果不佳通用英文 Embedding 模型處理中文經(jīng)?！八敛环?。建議優(yōu)先選擇中文優(yōu)化的 BGE 系列模型比如BAAI/bge-small-zh-v1.5。這類(lèi)模型在中文語(yǔ)義匹配、短文本檢索上的表現(xiàn)更穩(wěn)定。同時(shí)要注意查詢(xún)側(cè)和文檔側(cè)盡量用同一個(gè) Embedding 模型不要混用否則向量空間不一致相似度計(jì)算沒(méi)有意義。7. 最佳實(shí)踐與工程建議7.1 數(shù)據(jù)質(zhì)量?jī)?yōu)先于模型參數(shù)RAG 的效果上限由知識(shí)庫(kù)質(zhì)量決定。文檔里有錯(cuò)別字、格式混亂、內(nèi)容過(guò)期后面做再多優(yōu)化都是事倍功半。建議在進(jìn)入 RAG 流程前做一輪數(shù)據(jù)清洗去除頁(yè)眉頁(yè)腳、水印、無(wú)關(guān)廣告。統(tǒng)一編碼格式避免亂碼。對(duì)表格數(shù)據(jù)優(yōu)先轉(zhuǎn)成 Markdown 表格或結(jié)構(gòu)化的文本描述。給文檔補(bǔ)充元信息比如來(lái)源部門(mén)、更新時(shí)間、版本號(hào)。這些元信息不僅能幫你跟蹤知識(shí)來(lái)源還可以在 Prompt 中加入“信息更新時(shí)間”讓模型自行判斷知識(shí)是否過(guò)時(shí)。7.2 建立評(píng)估集持續(xù)回歸沒(méi)有評(píng)估的 RAG 系統(tǒng)很難迭代。建議從業(yè)務(wù)問(wèn)題中整理 50~100 條典型的“問(wèn)題-答案-參考文檔編號(hào)”作為黃金測(cè)試集。每次調(diào)整切塊策略、Embedding 模型或重排序邏輯都跑一遍評(píng)估集對(duì)比答案質(zhì)量和召回指標(biāo)。常用的評(píng)估指標(biāo)包括召回率Recall正確文檔是否出現(xiàn)在檢索結(jié)果中。命中率Hit RateTopK 中是否包含正確答案。忠實(shí)度生成答案是否嚴(yán)格基于參考文檔。答案相關(guān)性用戶(hù)視角的答案滿(mǎn)意度。評(píng)估可以通過(guò) LLM 自動(dòng)打分也可以人工抽檢。重點(diǎn)不是追求某個(gè)指標(biāo)完美而是找到當(dāng)前業(yè)務(wù)場(chǎng)景里的瓶頸在哪個(gè)環(huán)節(jié)。7.3 性能與緩存策略生產(chǎn)環(huán)境里RAG 鏈路很長(zhǎng)每個(gè)環(huán)節(jié)都可能成為瓶頸。優(yōu)化優(yōu)先級(jí)一般是文檔數(shù)量上量后向量檢索從IndexFlatIP換成IndexHNSW或IVF。給檢索結(jié)果加 Redis 緩存同樣的查詢(xún)不再重復(fù)計(jì)算。Rerank 只在最后一步對(duì)少量候選做不要對(duì)全庫(kù)做。盡量流式輸出答案減少首字延遲感知。對(duì)高頻問(wèn)題做預(yù)生成答案命中后直接返回不走完整鏈路。7.4 安全與權(quán)限邊界企業(yè)知識(shí)庫(kù)往往包含敏感數(shù)據(jù)。RAG 系統(tǒng)上線前要重點(diǎn)做權(quán)限隔離按用戶(hù)角色過(guò)濾可檢索的文檔集合。文檔入庫(kù)時(shí)標(biāo)記可見(jiàn)范圍檢索時(shí)強(qiáng)制附加權(quán)限過(guò)濾條件。不在日志中記錄完整問(wèn)答內(nèi)容避免敏感信息泄露。對(duì)生成內(nèi)容做脫敏校驗(yàn)防止模型間接泄露訓(xùn)練數(shù)據(jù)。尤其要注意RAG 并不會(huì)自動(dòng)擁有權(quán)限意識(shí)。如果文檔庫(kù)里有機(jī)密文件而檢索層沒(méi)有過(guò)濾任何用戶(hù)都可能通過(guò)巧妙提問(wèn)拿到不該看的內(nèi)容。權(quán)限必須放進(jìn)檢索鏈路而不是指望模型自行判斷。8. 總結(jié)與下一步學(xué)習(xí)路線到這里你已經(jīng)完整走通了 RAG 的核心鏈路文檔加載、切塊、向量化、向量檢索、Prompt 增強(qiáng)、本地模型生成也了解了混合檢索、重排序、查詢(xún)改寫(xiě)、Agentic RAG 等高級(jí)優(yōu)化方向。下一步建議這樣安排學(xué)習(xí)第一步動(dòng)手跑通本文的項(xiàng)目替換成你自己的文檔感受完整流程。第二步搭建一個(gè)小規(guī)模評(píng)估集觀察不同切塊參數(shù)和 TopK 對(duì)答案質(zhì)量的影響。第三步引入混合檢索和 Rerank對(duì)比優(yōu)化前后的效果差異。第四步熟悉 Dify、LangChain、LlamaIndex 等框架了解它們?nèi)绾伟驯疚牡牡讓舆壿嫹庋b成可配置模塊。第五步把檢索鏈路的權(quán)限、緩存、日志監(jiān)控補(bǔ)齊再考慮 Agent 和復(fù)雜任務(wù)編排。RAG 不是一套固定的代碼而是一套可以持續(xù)優(yōu)化的方法論。把每個(gè)環(huán)節(jié)的原理吃透工具換代時(shí)你也能快速遷移。希望這篇文章能幫你少走一些彎路有疑問(wèn)歡迎在評(píng)論區(qū)一起交流。 RAG知識(shí)庫(kù)問(wèn)答實(shí)戰(zhàn)：從零搭建本地RAG系統(tǒng)
最近在給團(tuán)隊(duì)搭建內(nèi)部知識(shí)庫(kù)問(wèn)答系統(tǒng)時(shí)遇到了一個(gè)很典型的問(wèn)題大模型對(duì)通用知識(shí)很擅長(zhǎng)但一碰到企業(yè)內(nèi)部的規(guī)章制度、產(chǎn)品文檔、歷史項(xiàng)目總結(jié)回答就開(kāi)始“一本正經(jīng)地胡說(shuō)八道”。嘗試過(guò)把資料直接拼進(jìn) Prompt很快又遇到 Token 長(zhǎng)度限制和成本問(wèn)題。后來(lái)把 RAG 體系完整落地了一遍才真正解決了“讓模型基于自有知識(shí)回答問(wèn)題”這個(gè)核心訴求。這篇文章會(huì)從 RAG 的原理講起然后完整搭建一個(gè)本地知識(shí)庫(kù)問(wèn)答系統(tǒng)涵蓋文檔加載、切塊、向量化、檢索、重排序和本地模型生成最后再聊一聊高級(jí)檢索的實(shí)戰(zhàn)思路。適合剛開(kāi)始接觸 AI 應(yīng)用開(kāi)發(fā)、想搞懂 RAG 全流程的讀者也適合已經(jīng)在做 LLM 應(yīng)用、想優(yōu)化召回效果的開(kāi)發(fā)者。1. 什么是 RAG先解決“模型不知道”的問(wèn)題1.1 大模型的知識(shí)局限大模型的知識(shí)來(lái)自訓(xùn)練時(shí)見(jiàn)過(guò)的數(shù)據(jù)所以它有兩個(gè)天然短板一是訓(xùn)練數(shù)據(jù)有截止時(shí)間新產(chǎn)生的信息它不知道二是私有知識(shí)它幾乎不可能學(xué)到比如公司內(nèi)部的客戶(hù)資料、設(shè)備手冊(cè)、項(xiàng)目復(fù)盤(pán)、專(zhuān)有規(guī)范這些數(shù)據(jù)不會(huì)出現(xiàn)在公開(kāi)語(yǔ)料里。如果直接把問(wèn)題拋給模型它只能依靠參數(shù)里記憶的“模糊印象”來(lái)回答結(jié)果就是一本正經(jīng)地編造。這不是模型不夠聰明而是它的知識(shí)邊界本身就固定了。RAG 的解決思路非常直接不要求模型記住所有知識(shí)而是在回答之前先從一個(gè)外部知識(shí)庫(kù)中檢索出相關(guān)的片段把這些片段作為上下文一起交給模型讓模型基于這些材料作答。1.2 RAG 的基本工作流程先給一個(gè)整體的流程印象用戶(hù)提問(wèn) - 查詢(xún)理解可選 - 檢索從知識(shí)庫(kù)中找回相關(guān)文檔片段 - 增強(qiáng)把檢索結(jié)果組裝進(jìn) Prompt - 生成大模型基于上下文生成答案 - 返回答案與引用來(lái)源整個(gè)過(guò)程可以拆成兩個(gè)離線階段和一個(gè)在線階段離線索引階段把文檔切塊對(duì)每個(gè)塊做向量化Embedding然后寫(xiě)入向量數(shù)據(jù)庫(kù)。在線檢索階段把用戶(hù)問(wèn)題向量化到向量數(shù)據(jù)庫(kù)里做相似度檢索取回 TopK 候選片段。在線生成階段把“用戶(hù)問(wèn)題 檢索片段 系統(tǒng)提示詞”拼在一起交給大模型生成最終回答。這里的核心思路是讓模型學(xué)會(huì)“查資料”而不是“背資料”。1.3 RAG 與微調(diào)的邊界很多初學(xué)者會(huì)把 RAG 和微調(diào)Fine-tuning搞混。簡(jiǎn)單區(qū)分一下維度RAG微調(diào)知識(shí)更新替換/新增文檔即可需要重新訓(xùn)練私有知識(shí)引入通過(guò)檢索注入上下文通過(guò)調(diào)整模型參數(shù)幻覺(jué)控制較好可引用來(lái)源有限仍可能編造成本主要是向量化和檢索服務(wù)訓(xùn)練和推理成本較高適合場(chǎng)景知識(shí)庫(kù)問(wèn)答、實(shí)時(shí)信息查詢(xún)指令遵循、領(lǐng)域風(fēng)格調(diào)整實(shí)際項(xiàng)目里兩者不是二選一而是可以結(jié)合使用。先用 RAG 解決知識(shí)的及時(shí)性和可追溯性再用微調(diào)優(yōu)化模型的表達(dá)方式和工具調(diào)用能力。2. 環(huán)境準(zhǔn)備與項(xiàng)目結(jié)構(gòu)2.1 運(yùn)行環(huán)境與版本說(shuō)明本文的實(shí)戰(zhàn)部分以本地環(huán)境為例避開(kāi)云服務(wù)依賴(lài)方便你完整復(fù)現(xiàn)。版本建議如下但需要根據(jù)你的實(shí)際環(huán)境調(diào)整操作系統(tǒng)Windows 10/11、Ubuntu 20.04 或 macOS 均可Python3.9 及以上本地模型推理llama.cpp Qwen2-7B-Instruct 量化模型向量模型BAAI/bge-small-zh-v1.5中文效果較好資源占用低向量數(shù)據(jù)庫(kù)FAISS開(kāi)發(fā)環(huán)境足夠生產(chǎn)可替換為 Milvus 或 pgvectorWeb 框架FastAPI分詞與關(guān)鍵詞檢索jieba BM25用于混合檢索注意不同版本之間接口可能有差異比如langchain的 API 改動(dòng)一直比較頻繁。本文盡量少依賴(lài)框架封裝用原生 Python 實(shí)現(xiàn)核心鏈路這樣你理解原理后換成任何框架都能快速上手。2.2 安裝依賴(lài)建議先創(chuàng)建虛擬環(huán)境python -m venv rag_env source rag_env/bin/activate # Windows 使用 rag_env\Scripts\activate接下來(lái)安裝必要依賴(lài)pip install fastapi uvicorn faiss-cpu sentence-transformers pip install llama-cpp-python pip install jieba rank-bm25如果你的機(jī)器沒(méi)有 NVIDIA GPUllama-cpp-python會(huì)走 CPU 推理速度會(huì)慢一些但勝在部署簡(jiǎn)單。如果要安裝支持 GPU 的版本可以自行查閱對(duì)應(yīng)平臺(tái)的編譯方式。2.3 項(xiàng)目目錄結(jié)構(gòu)為了方便維護(hù)我們把項(xiàng)目拆分成模塊rag_demo/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 入口 │ ├── config.py # 全局配置 │ ├── loader.py # 文檔加載與切塊 │ ├── embedder.py # Embedding 模型封裝 │ ├── store.py # 向量索引與檢索 │ ├── retriever.py # 混合檢索 重排序 │ └── llm.py # 本地模型加載與生成 ├── data/ │ └── docs/ # 原始知識(shí)文檔 ├── index/ # 向量索引持久化目錄 └── requirements.txt這個(gè)結(jié)構(gòu)不算復(fù)雜但模塊邊界清晰。索引構(gòu)建、檢索、生成是獨(dú)立的環(huán)節(jié)后續(xù)替換任意一個(gè)組件都不會(huì)影響整體流程。3. RAG 核心原理拆解3.1 索引階段切塊與向量化索引階段的第一步是切塊Chunking。為什么要切塊因?yàn)榇竽Ｐ洼斎腴L(zhǎng)度有限不可能把整本書(shū)塞進(jìn) Prompt。檢索的粒度越細(xì)能命中的片段越精準(zhǔn)。多個(gè)小片段可以組合出更豐富的上下文。切塊不是越短越好。切得太短語(yǔ)義不完整切得太長(zhǎng)噪音太多還浪費(fèi) Token。比較常見(jiàn)的策略是固定長(zhǎng)度切塊按字符數(shù)或 Token 數(shù)切比如每塊 300~500 字符。按段落切塊先按換行符拆成段落再把長(zhǎng)段落繼續(xù)拆。按語(yǔ)義切塊用分割模型識(shí)別語(yǔ)義邊界比如標(biāo)題、小節(jié)、句子邊界。父子切塊小片段用于檢索命中后返回對(duì)應(yīng)的父級(jí)大片段用于生成。切塊之后每一個(gè)塊會(huì)通過(guò) Embedding 模型轉(zhuǎn)換成一個(gè)高維向量。向量化的目標(biāo)是語(yǔ)義相近的文本在向量空間里的距離更近。這里有個(gè)容易踩的坑不要在檢索時(shí)才對(duì)文檔臨時(shí)向量化。必須提前把整份文檔庫(kù)向量化并建好索引用戶(hù)查詢(xún)時(shí)才能做到毫秒級(jí)檢索。3.2 檢索階段相似度計(jì)算與召回檢索階段做的事情是計(jì)算用戶(hù)問(wèn)題向量與知識(shí)庫(kù)中每個(gè)片段向量的相似度找出最相似的 TopK 片段。常見(jiàn)相似度計(jì)算方式余弦相似度Cosine最常用適合文本向量。內(nèi)積Dot Product在歸一化向量上等價(jià)于余弦相似度。歐氏距離值越小越相似。以 FAISS 為例IndexFlatIP就是內(nèi)積索引。如果向量都做了歸一化內(nèi)積結(jié)果等價(jià)于余弦相似度。import numpy as np from sentence_transformers import SentenceTransformer model SentenceTransformer(BAAI/bge-small-zh-v1.5) query 如何申請(qǐng)?jiān)O(shè)備維修 query_vector model.encode(query, normalize_embeddingsTrue) doc_vectors model.encode( [設(shè)備維修需要填寫(xiě)申請(qǐng)單, 請(qǐng)假流程說(shuō)明, 采購(gòu)審批權(quán)限], normalize_embeddingsTrue ) scores np.dot(doc_vectors, query_vector) print(scores)輸出結(jié)果中得分最高的就是與問(wèn)題最相關(guān)的片段。不過(guò)單純的向量檢索有一個(gè)問(wèn)題它對(duì)關(guān)鍵詞匹配不敏感。比如用戶(hù)搜索“維修”文檔里寫(xiě)的是“故障處理”雖然語(yǔ)義相近但向量召回可能不夠精確。這就是后面要講的混合檢索解決的問(wèn)題。3.3 生成階段Prompt 增強(qiáng)與上下文約束檢索到相關(guān)片段后不是直接把片段丟給模型而是組裝成一個(gè)結(jié)構(gòu)清晰的 Prompt。一個(gè)典型的 RAG Prompt 結(jié)構(gòu)如下你是一個(gè)智能問(wèn)答助手。請(qǐng)嚴(yán)格根據(jù)下面提供的參考文檔回答問(wèn)題。 如果參考文檔中沒(méi)有足夠信息請(qǐng)直接回答“當(dāng)前知識(shí)庫(kù)中沒(méi)有相關(guān)內(nèi)容”不要編造。 參考文檔 [1] 設(shè)備維修需要填寫(xiě)維修申請(qǐng)單交由部門(mén)主管審批。 [2] 緊急故障可以電話(huà)聯(lián)系運(yùn)維中心熱線12345。 用戶(hù)問(wèn)題如何申請(qǐng)?jiān)O(shè)備維修這里有幾個(gè)關(guān)鍵點(diǎn)明確告訴模型“只能依據(jù)參考文檔回答”減少幻覺(jué)。要求“沒(méi)有信息時(shí)承認(rèn)不知道”避免模型強(qiáng)行編造。給文檔編號(hào)方便模型在回答中引用。在 Prompt 中保留來(lái)源信息便于前端展示引用。Prompt 的設(shè)計(jì)直接決定了 RAG 回答質(zhì)量的下限。同一個(gè)檢索結(jié)果用不同的 Prompt 模板生成出的答案可能差別很大。4. 從 0 搭建本地 RAG 知識(shí)庫(kù)問(wèn)答系統(tǒng)接下來(lái)進(jìn)入實(shí)戰(zhàn)環(huán)節(jié)。我們會(huì)基于 llama.cpp Qwen2-7B FastAPI 構(gòu)建一個(gè)完整的本地 RAG 知識(shí)庫(kù)問(wèn)答系統(tǒng)。核心鏈路是加載文檔 - 切塊 - 向量化 - 建立索引 - 檢索 - 增強(qiáng) - 生成。4.1 文檔加載與切塊實(shí)現(xiàn)先實(shí)現(xiàn)文檔加載與切塊的模塊。這里以讀取.txt和.md文件為例你也可以擴(kuò)展為 PDF、Word 等格式。文件路徑app/loader.pyimport os from typing import List def load_documents(docs_dir: str) - List[str]: 加載目錄下的所有文本文件 docs [] for root, _, files in os.walk(docs_dir): for name in files: if name.endswith((.txt, .md)): file_path os.path.join(root, name) with open(file_path, r, encodingutf-8) as f: content f.read() docs.append({source: file_path, content: content}) return docs def split_text(text: str, chunk_size: int 400, overlap: int 80) - List[str]: 按固定長(zhǎng)度切塊并保留重疊區(qū)間避免切斷語(yǔ)義 chunks [] start 0 while start len(text): end start chunk_size chunk text[start:end] if chunk.strip(): chunks.append(chunk) if end len(text): break start end - overlap return chunks def build_chunks(docs_dir: str) - List[dict]: 加載文檔并返回帶有元信息的切塊列表 all_chunks [] docs load_documents(docs_dir) for doc in docs: chunks split_text(doc[content]) for idx, chunk in enumerate(chunks): all_chunks.append({ id: f{doc[source]}#{idx}, source: doc[source], text: chunk, }) return all_chunks這里切塊時(shí)增加了overlap參數(shù)目的是讓相鄰片段之間保留一部分上下文減少因?yàn)橛睬袑?dǎo)致的語(yǔ)義斷裂。比如一段話(huà)在 400 字符處正好被切斷重疊部分能保證下一塊的開(kāi)頭有足夠信息。4.2 向量化與索引構(gòu)建接下來(lái)實(shí)現(xiàn) Embedding 封裝和 FAISS 索引構(gòu)建。文件路徑app/embedder.pyfrom sentence_transformers import SentenceTransformer class Embedder: def __init__(self, model_name: str BAAI/bge-small-zh-v1.5): self.model SentenceTransformer(model_name) def embed_documents(self, texts): return self.model.encode(texts, normalize_embeddingsTrue) def embed_query(self, text): return self.model.encode(text, normalize_embeddingsTrue)文件路徑app/store.pyimport json import os import numpy as np import faiss class VectorStore: def __init__(self, embedder, index_dir: str index): self.embedder embedder self.index_dir index_dir self.index None self.metadata [] def build(self, chunks: list): 根據(jù)切塊列表構(gòu)建 FAISS 索引 texts [c[text] for c in chunks] vectors self.embedder.embed_documents(texts) dim vectors.shape[1] self.index faiss.IndexFlatIP(dim) self.index.add(vectors) self.metadata chunks os.makedirs(self.index_dir, exist_okTrue) faiss.write_index(self.index, os.path.join(self.index_dir, faiss.index)) with open(os.path.join(self.index_dir, metadata.json), w, encodingutf-8) as f: json.dump(self.metadata, f, ensure_asciiFalse, indent2) def load(self): 加載已有的索引和元數(shù)據(jù) index_path os.path.join(self.index_dir, faiss.index) meta_path os.path.join(self.index_dir, metadata.json) if not os.path.exists(index_path) or not os.path.exists(meta_path): raise FileNotFoundError(索引文件不存在請(qǐng)先執(zhí)行 build()) self.index faiss.read_index(index_path) with open(meta_path, r, encodingutf-8) as f: self.metadata json.load(f) def search(self, query: str, top_k: int 5): 向量相似度檢索 query_vector self.embedder.embed_query(query).reshape(1, -1) scores, indices self.index.search(query_vector, top_k) results [] for score, idx in zip(scores[0], indices[0]): if idx 0: continue results.append({ score: float(score), text: self.metadata[idx][text], source: self.metadata[idx][source], }) return results這里選用IndexFlatIP作為索引結(jié)構(gòu)。它是暴力精確檢索數(shù)據(jù)量不大時(shí)速度足夠快而且效果最準(zhǔn)確。等數(shù)據(jù)量到了百萬(wàn)級(jí)再換成 IVF 或 HNSW 也不遲。4.3 基于 FastAPI 實(shí)現(xiàn)檢索接口現(xiàn)在我們把檢索能力包裝成 HTTP 接口便于前端或其他服務(wù)調(diào)用。文件路徑app/main.pyfrom fastapi import FastAPI from pydantic import BaseModel from app.embedder import Embedder from app.store import VectorStore app FastAPI() embedder Embedder() store VectorStore(embedder) store.load() class SearchRequest(BaseModel): query: str top_k: int 5 app.post(/search) def search(req: SearchRequest): results store.search(req.query, req.top_k) return {query: req.query, results: results}啟動(dòng)服務(wù)uvicorn app.main:app --host 0.0.0.0 --port 8000調(diào)用接口curl -X POST http://localhost:8000/search \ -H Content-Type: application/json \ -d {query: 如何申請(qǐng)?jiān)O(shè)備維修, top_k: 3}返回結(jié)果里包含相似度得分、命中的文本片段和來(lái)源文件路徑方便后續(xù)拼接 Prompt 和展示引用。4.4 接入 Qwen2 本地模型生成答案檢索只是拿到了資料真正回答用戶(hù)問(wèn)題還需要大模型。我們通過(guò) llama.cpp 加載量化后的 Qwen2-7B 模型在本地進(jìn)行推理。文件路徑app/llm.pyfrom llama_cpp import Llama class LocalLLM: def __init__(self, model_path: str, n_ctx: int 4096): self.llm Llama( model_pathmodel_path, n_ctxn_ctx, n_threads8, verboseFalse, ) def generate(self, prompt: str): output self.llm( prompt, max_tokens1024, temperature0.2, top_p0.9, stop[/s, Human:, 用戶(hù):], ) return output[choices][0][text].strip()這里把temperature設(shè)置得比較低0.2是為了讓模型更嚴(yán)格地依據(jù)參考文檔作答減少隨機(jī)編造。接下來(lái)把檢索、Prompt 增強(qiáng)和生成串成一個(gè)完整接口。更新app/main.pyfrom fastapi import FastAPI from pydantic import BaseModel from app.embedder import Embedder from app.store import VectorStore from app.llm import LocalLLM app FastAPI() embedder Embedder() store VectorStore(embedder) store.load() # 根據(jù)實(shí)際模型路徑調(diào)整 llm LocalLLM(model_path./models/qwen2-7b-instruct-q4_k_m.gguf) class AskRequest(BaseModel): question: str top_k: int 5 def build_prompt(question: str, contexts: list) - str: docs_text \n\n.join( [f[{i1}] {item[text]}來(lái)源{item[source]} for i, item in enumerate(contexts)] ) prompt f你是一個(gè)智能問(wèn)答助手。請(qǐng)嚴(yán)格根據(jù)下面提供的參考文檔回答問(wèn)題。 如果參考文檔中沒(méi)有足夠信息請(qǐng)直接回答“當(dāng)前知識(shí)庫(kù)中沒(méi)有相關(guān)內(nèi)容”不要編造。 參考文檔 {docs_text} 用戶(hù)問(wèn)題{question} 請(qǐng)給出準(zhǔn)確、簡(jiǎn)潔的回答并標(biāo)注答案主要參考了哪個(gè)文檔編號(hào)。 return prompt app.post(/ask) def ask(req: AskRequest): # 1. 檢索 contexts store.search(req.question, req.top_k) # 2. 構(gòu)造 Prompt prompt build_prompt(req.question, contexts) # 3. 生成 answer llm.generate(prompt) return { question: req.question, answer: answer, references: [ {text: c[text], source: c[source]} for c in contexts ], }整個(gè)流程非常清晰先檢索再增強(qiáng)最后生成。前端拿到answer展示答案拿到references生成引用來(lái)源這樣用戶(hù)可以看到模型回答的依據(jù)。4.5 運(yùn)行與驗(yàn)證完整執(zhí)行流程如下。第一步準(zhǔn)備文檔。在data/docs目錄下放入你的知識(shí)文檔比如device_repair.md# 設(shè)備維修流程 設(shè)備出現(xiàn)故障后使用人需要在 OA 系統(tǒng)中填寫(xiě)維修申請(qǐng)單。 申請(qǐng)單需要注明設(shè)備編號(hào)、故障描述和期望維修時(shí)間。 部門(mén)主管審批通過(guò)后由行政部統(tǒng)一對(duì)接維修廠商。 緊急故障可以直接撥打運(yùn)維中心電話(huà) 400-1234-567。第二步構(gòu)建索引。簡(jiǎn)單寫(xiě)一個(gè)構(gòu)建腳本python -c from app.loader import build_chunks; from app.embedder import Embedder; from app.store import VectorStore; chunks build_chunks(data/docs); print(chunks:, len(chunks)); store VectorStore(Embedder()); store.build(chunks); print(build done)第三步啟動(dòng)服務(wù)并提問(wèn)uvicorn app.main:app --host 0.0.0.0 --port 8000curl -X POST http://localhost:8000/ask \ -H Content-Type: application/json \ -d {question: 設(shè)備壞了怎么報(bào)修}預(yù)期輸出中answer會(huì)引用文檔里的設(shè)備維修流程references會(huì)包含命中的原文和來(lái)源路徑。到這里一個(gè)最小可用的本地 RAG 系統(tǒng)就跑通了。接下來(lái)我們繼續(xù)做高級(jí)檢索優(yōu)化因?yàn)榛A(chǔ)版在真實(shí)場(chǎng)景里往往不夠用。5. 高級(jí)檢索實(shí)戰(zhàn)讓 RAG 更聰明5.1 混合檢索向量 關(guān)鍵詞純向量檢索擅長(zhǎng)語(yǔ)義相似但有時(shí)會(huì)漏掉“關(guān)鍵詞完全一致”的文檔。比如用戶(hù)問(wèn)“維修電話(huà)”文檔里確實(shí)寫(xiě)了一個(gè)維修電話(huà)向量檢索可能因?yàn)榫渥诱w語(yǔ)義偏移而沒(méi)有召回。解決方案是混合檢索同時(shí)跑向量檢索和關(guān)鍵詞檢索如 BM25再把兩路結(jié)果合并去重。BM25 是一個(gè)經(jīng)典的關(guān)鍵詞相關(guān)性排序算法。簡(jiǎn)單說(shuō)它根據(jù)詞頻、文檔長(zhǎng)度等因素計(jì)算查詢(xún)?cè)~和文檔的相關(guān)性對(duì)“精確詞命中”非常敏感。文件路徑app/retriever.pyimport jieba from rank_bm25 import BM25Okapi from app.store import VectorStore class HybridRetriever: def __init__(self, vector_store: VectorStore, chunks: list): self.vector_store vector_store self.chunks chunks tokenized_docs [list(jieba.cut(c[text])) for c in chunks] self.bm25 BM25Okapi(tokenized_docs) def retrieve(self, query: str, top_k: int 5, vec_weight: float 0.5): # 向量檢索 vec_results self.vector_store.search(query, top_k * 2) # 關(guān)鍵詞檢索 tokenized_query list(jieba.cut(query)) bm25_scores self.bm25.get_scores(tokenized_query) bm25_top_ids sorted( range(len(bm25_scores)), keylambda i: bm25_scores[i], reverseTrue )[: top_k * 2] # 合并結(jié)果 score_map {} for item in vec_results: score_map[item[source] # item[text][:20]] { text: item[text], source: item[source], score: item[score] * vec_weight, } for idx in bm25_top_ids: key self.chunks[idx][source] # self.chunks[idx][text][:20] if key not in score_map: score_map[key] { text: self.chunks[idx][text], source: self.chunks[idx][source], score: 0, } score_map[key][score] bm25_scores[idx] * (1 - vec_weight) # 按融合分?jǐn)?shù)排序 sorted_items sorted( score_map.values(), keylambda x: x[score], reverseTrue ) return sorted_items[:top_k]融合策略用的是加權(quán)求和。vec_weight控制向量和關(guān)鍵詞的比重實(shí)際項(xiàng)目里可以通過(guò)評(píng)估集來(lái)調(diào)參。也可以使用 RRFReciprocal Rank Fusion等更穩(wěn)定的融合方法。5.2 重排序Rerank 提升精度混合檢索取回的 TopK 里依然可能存在“看著相關(guān)但實(shí)際不對(duì)”的片段。這時(shí)候可以加一個(gè)重排序Rerank環(huán)節(jié)先用輕量級(jí)檢索擴(kuò)大召回再用一個(gè)更精確的交叉編碼器模型逐對(duì)精排。重排序的本質(zhì)是把“給定問(wèn)題判斷文檔是否相關(guān)”變成一個(gè)二分類(lèi)問(wèn)題。交叉編碼器把問(wèn)題和文檔拼接后輸入模型輸出相關(guān)性分?jǐn)?shù)比單純向量相似度更準(zhǔn)。from sentence_transformers import CrossEncoder reranker CrossEncoder(BAAI/bge-reranker-base) def rerank(query: str, candidates: list, top_k: int 3): pairs [(query, c[text]) for c in candidates] scores reranker.predict(pairs) scored list(zip(candidates, scores)) scored.sort(keylambda x: x[1], reverseTrue) return [c for c, s in scored[:top_k]]重排序會(huì)增加一次模型推理的耗時(shí)但能明顯提升最終答案的準(zhǔn)確性。生產(chǎn)環(huán)境里通常把它用于 TopK 精細(xì)篩選比如先召回 20 條再用 Rerank 選 3 條進(jìn)入 Prompt。5.3 查詢(xún)改寫(xiě)與 HyDE用戶(hù)的問(wèn)題往往不夠精確比如“這個(gè)流程需要哪些人審批”如果沒(méi)有上下文檢索系統(tǒng)很難判斷“這個(gè)流程”指什么。這時(shí)候可以在檢索前做查詢(xún)改寫(xiě)Query Rewriting把模糊問(wèn)題轉(zhuǎn)換成更明確的檢索語(yǔ)句。一種簡(jiǎn)單的方法是讓大模型先把問(wèn)題改寫(xiě)為適合檢索的查詢(xún)?cè)~。你是一個(gè)搜索查詢(xún)優(yōu)化器。請(qǐng)根據(jù)用戶(hù)的原始問(wèn)題生成 3 個(gè)適合檢索的查詢(xún)?cè)~。 要求保留關(guān)鍵實(shí)體和業(yè)務(wù)術(shù)語(yǔ)不要修改事實(shí)信息。 原始問(wèn)題這個(gè)流程需要哪些人審批另一種思路是 HyDEHypothetical Document Embeddings。它的做法是先讓大模型根據(jù)問(wèn)題生成一個(gè)“假設(shè)性答案”然后用這個(gè)答案去檢索。因?yàn)榧僭O(shè)性答案與目標(biāo)文檔在語(yǔ)義上更接近往往比直接檢索問(wèn)題效果更好。請(qǐng)根據(jù)你的知識(shí)回答下面問(wèn)題。答案可以是猜測(cè)性的也可以是概括性的。 問(wèn)題是請(qǐng)?zhí)峁┰O(shè)備維修申請(qǐng)流程。然后把生成的答案向量化去向量庫(kù)搜索。這種方式很適合“問(wèn)題簡(jiǎn)短、文檔復(fù)雜”的場(chǎng)景。5.4 Agentic RAG 簡(jiǎn)介傳統(tǒng) RAG 是“一次檢索一次生成”。Agentic RAG 則把檢索過(guò)程交給 Agent 智能規(guī)劃可以多輪檢索、調(diào)用工具、逐步推理。比如用戶(hù)問(wèn)“對(duì)比一下 A 和 B 兩種方案的差異”Agent 可能會(huì)先檢索 A 方案相關(guān)文檔。再檢索 B 方案相關(guān)文檔。還可能檢索“對(duì)比框架”“評(píng)估指標(biāo)”等材料。最后匯總生成對(duì)比結(jié)論。這比一次性檢索做得更深但實(shí)現(xiàn)復(fù)雜度也高很多。適合從“單點(diǎn)問(wèn)答”升級(jí)到“復(fù)雜分析”的場(chǎng)景。可以先掌握基礎(chǔ) RAG再逐步引入 Agent 能力。6. 常見(jiàn)問(wèn)題與排查清單6.1 召回結(jié)果差檢索不到相關(guān)內(nèi)容常見(jiàn)原因解決思路切塊太大語(yǔ)義被稀釋縮小 chunk_size增加 overlap切塊太小上下文不完整增大 chunk_size或使用父子切塊Embedding 模型不適合中文換用 bge-small-zh 或 bge-large-zh問(wèn)題描述與文檔表述差異大引入查詢(xún)改寫(xiě)、HyDE 或混合檢索數(shù)據(jù)量太少檢查文檔是否完整加載是否有編碼問(wèn)題排查時(shí)最直接的方式是先打印檢索結(jié)果不經(jīng)過(guò)生成環(huán)節(jié)。看看返回的片段是否真的相關(guān)。如果片段本身不對(duì)問(wèn)題大概率在索引或檢索策略上而不是 Prompt 和模型的問(wèn)題。6.2 切塊不合理導(dǎo)致回答破碎固定長(zhǎng)度切塊雖然簡(jiǎn)單但經(jīng)常把完整段落攔腰截?cái)?。比如一句?huà)從第 390 個(gè)字符開(kāi)始到第 410 個(gè)字符結(jié)束切塊后這句話(huà)只出現(xiàn)一半模型自然讀不懂。建議優(yōu)先“按標(biāo)題層級(jí) 段落”切塊。比如 Markdown 文檔先按##拆成章節(jié)再處理單個(gè)章節(jié)里的長(zhǎng)段落。也可以用父子切塊子塊短小用于精確檢索命中的子塊關(guān)聯(lián)到父塊生成時(shí)把父塊內(nèi)容作為上下文。6.3 本地模型推理慢llama.cpp在 CPU 上跑 7B 模型速度確實(shí)不快。優(yōu)化思路有幾個(gè)使用更小的量化等級(jí)比如 Q4_K_M 比 Q8 更快。減少n_ctx上下文長(zhǎng)度降低顯存/內(nèi)存占用?？刂苖ax_tokens不要無(wú)腦生成長(zhǎng)答案。對(duì)檢索片段做摘要只把關(guān)鍵信息放進(jìn) Prompt。如果條件允許升級(jí) GPU 或使用 API 版模型。另外可以加一層緩存相同問(wèn)題在短時(shí)間內(nèi)直接返回歷史答案減少模型調(diào)用次數(shù)。6.4 中文 Embedding 效果不佳通用英文 Embedding 模型處理中文經(jīng)?！八敛环?。建議優(yōu)先選擇中文優(yōu)化的 BGE 系列模型比如BAAI/bge-small-zh-v1.5。這類(lèi)模型在中文語(yǔ)義匹配、短文本檢索上的表現(xiàn)更穩(wěn)定。同時(shí)要注意查詢(xún)側(cè)和文檔側(cè)盡量用同一個(gè) Embedding 模型不要混用否則向量空間不一致相似度計(jì)算沒(méi)有意義。7. 最佳實(shí)踐與工程建議7.1 數(shù)據(jù)質(zhì)量?jī)?yōu)先于模型參數(shù)RAG 的效果上限由知識(shí)庫(kù)質(zhì)量決定。文檔里有錯(cuò)別字、格式混亂、內(nèi)容過(guò)期后面做再多優(yōu)化都是事倍功半。建議在進(jìn)入 RAG 流程前做一輪數(shù)據(jù)清洗去除頁(yè)眉頁(yè)腳、水印、無(wú)關(guān)廣告。統(tǒng)一編碼格式避免亂碼。對(duì)表格數(shù)據(jù)優(yōu)先轉(zhuǎn)成 Markdown 表格或結(jié)構(gòu)化的文本描述。給文檔補(bǔ)充元信息比如來(lái)源部門(mén)、更新時(shí)間、版本號(hào)。這些元信息不僅能幫你跟蹤知識(shí)來(lái)源還可以在 Prompt 中加入“信息更新時(shí)間”讓模型自行判斷知識(shí)是否過(guò)時(shí)。7.2 建立評(píng)估集持續(xù)回歸沒(méi)有評(píng)估的 RAG 系統(tǒng)很難迭代。建議從業(yè)務(wù)問(wèn)題中整理 50~100 條典型的“問(wèn)題-答案-參考文檔編號(hào)”作為黃金測(cè)試集。每次調(diào)整切塊策略、Embedding 模型或重排序邏輯都跑一遍評(píng)估集對(duì)比答案質(zhì)量和召回指標(biāo)。常用的評(píng)估指標(biāo)包括召回率Recall正確文檔是否出現(xiàn)在檢索結(jié)果中。命中率Hit RateTopK 中是否包含正確答案。忠實(shí)度生成答案是否嚴(yán)格基于參考文檔。答案相關(guān)性用戶(hù)視角的答案滿(mǎn)意度。評(píng)估可以通過(guò) LLM 自動(dòng)打分也可以人工抽檢。重點(diǎn)不是追求某個(gè)指標(biāo)完美而是找到當(dāng)前業(yè)務(wù)場(chǎng)景里的瓶頸在哪個(gè)環(huán)節(jié)。7.3 性能與緩存策略生產(chǎn)環(huán)境里RAG 鏈路很長(zhǎng)每個(gè)環(huán)節(jié)都可能成為瓶頸。優(yōu)化優(yōu)先級(jí)一般是文檔數(shù)量上量后向量檢索從IndexFlatIP換成IndexHNSW或IVF。給檢索結(jié)果加 Redis 緩存同樣的查詢(xún)不再重復(fù)計(jì)算。Rerank 只在最后一步對(duì)少量候選做不要對(duì)全庫(kù)做。盡量流式輸出答案減少首字延遲感知。對(duì)高頻問(wèn)題做預(yù)生成答案命中后直接返回不走完整鏈路。7.4 安全與權(quán)限邊界企業(yè)知識(shí)庫(kù)往往包含敏感數(shù)據(jù)。RAG 系統(tǒng)上線前要重點(diǎn)做權(quán)限隔離按用戶(hù)角色過(guò)濾可檢索的文檔集合。文檔入庫(kù)時(shí)標(biāo)記可見(jiàn)范圍檢索時(shí)強(qiáng)制附加權(quán)限過(guò)濾條件。不在日志中記錄完整問(wèn)答內(nèi)容避免敏感信息泄露。對(duì)生成內(nèi)容做脫敏校驗(yàn)防止模型間接泄露訓(xùn)練數(shù)據(jù)。尤其要注意RAG 并不會(huì)自動(dòng)擁有權(quán)限意識(shí)。如果文檔庫(kù)里有機(jī)密文件而檢索層沒(méi)有過(guò)濾任何用戶(hù)都可能通過(guò)巧妙提問(wèn)拿到不該看的內(nèi)容。權(quán)限必須放進(jìn)檢索鏈路而不是指望模型自行判斷。8. 總結(jié)與下一步學(xué)習(xí)路線到這里你已經(jīng)完整走通了 RAG 的核心鏈路文檔加載、切塊、向量化、向量檢索、Prompt 增強(qiáng)、本地模型生成也了解了混合檢索、重排序、查詢(xún)改寫(xiě)、Agentic RAG 等高級(jí)優(yōu)化方向。下一步建議這樣安排學(xué)習(xí)第一步動(dòng)手跑通本文的項(xiàng)目替換成你自己的文檔感受完整流程。第二步搭建一個(gè)小規(guī)模評(píng)估集觀察不同切塊參數(shù)和 TopK 對(duì)答案質(zhì)量的影響。第三步引入混合檢索和 Rerank對(duì)比優(yōu)化前后的效果差異。第四步熟悉 Dify、LangChain、LlamaIndex 等框架了解它們?nèi)绾伟驯疚牡牡讓舆壿嫹庋b成可配置模塊。第五步把檢索鏈路的權(quán)限、緩存、日志監(jiān)控補(bǔ)齊再考慮 Agent 和復(fù)雜任務(wù)編排。RAG 不是一套固定的代碼而是一套可以持續(xù)優(yōu)化的方法論。把每個(gè)環(huán)節(jié)的原理吃透工具換代時(shí)你也能快速遷移。希望這篇文章能幫你少走一些彎路有疑問(wèn)歡迎在評(píng)論區(qū)一起交流。