RAG本地私有知识库构建实战指南

一、概述

1.1 什么是 RAG

RAG(Retrieval-Augmented Generation,检索增强生成)是一种将信息检索大语言模型生成相结合的技术架构。其核心理念是:当用户提问时,先从私有知识库中检索出最相关的内容,再将检索结果作为上下文提供给大模型,由模型结合具体内容生成答案。

RAG 相比纯微调方案的优势在于:

  • 成本低:无需重新训练模型
  • 实时性强:知识库更新后立即可查询
  • 可解释性:答案可溯源到具体文档
  • 私有化部署:数据不出本地,安全可控

1.2 RAG 架构总览

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
用户提问


┌─────────────────────┐
│ Query Processing │ 1. 查询处理(分词、意图识别)
└─────────┬───────────┘


┌─────────────────────┐
│ Vector Search │ 2. 向量检索(计算相似度)
│ (Embedding + ANN) │
└─────────┬───────────┘


┌─────────────────────┐
│ Knowledge Base │ 3. 知识库(文档集合 + 向量索引)
│ (Documents Store) │
└─────────┬───────────┘


┌─────────────────────┐
│ Context Assembly │ 4. 上下文组装(Prompt 构建)
└─────────┬───────────┘


┌─────────────────────┐
│ LLM Generation │ 5. 大模型生成答案
└─────────┬───────────┘


最终答案(含引用来源)

二、技术选型

2.1 文档解析工具

工具 适用格式 优点 缺点
PyMuPDF (fitz) PDF Python 原生,处理快 复杂表格/公式识别弱
pdfplumber PDF 表格提取效果好 慢,复杂 PDF 内存占用高
pypdf PDF 轻量简单 功能有限
python-docx DOCX 原生支持 无法处理扫描件
python-pptx PPT 幻灯片解析 需配合 OCR
Unstructured 通用 支持多种格式,统一接口 依赖多,复杂文档效果一般
Marker PDF PDF 转 Markdown,公式/表格优 速度慢,资源消耗大
PaddleOCR 图片/扫描件 中文识别好 需额外安装

2.2 文本分块策略

分块(Chunking)是将文档切成适合检索和上下文窗口的小段,是 RAG 效果的核心环节之一。

策略 方法 适用场景 块大小
固定大小分块 按 token 数或字符数硬切 通用场景,快速简单 256~512 tokens
滑动窗口 重叠分块,保留上下文 需要保留段落连续性 512 tokens / 64 overlap
语义分块 按句子/段落切分 保留完整语义单元 自动检测
递归分块 按层级结构递归切分 有明确结构的文档 层级化
基于标题分块 按文档结构(标题划分) Markdown / Word 结构化
Agentic Chunking LLM 自主判断切分点 高质量需求 成本较高
1
2
3
4
5
6
7
8
9
10
11
from langchain.text_splitter import RecursiveCharacterTextSplitter

# 滑动窗口分块(推荐通用场景)
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=512, # 块大小(token)
chunk_overlap=64, # 重叠 token 数,保留上下文连续性
separators=["\n\n", "\n", "。", "!", "?", " ", ""],
length_function=len, # 按字符数计算
)

chunks = text_splitter.split_text(document)

2.3 Embedding 模型选择

模型 维度 中文支持 特点 部署方式
BGE-large-zh 1024 优秀 中英文均优,开源 本地
BGE-base-zh 768 优秀 轻量级 本地
m3e-large 1024 优秀 向量质量好 本地
m3e-small 768 优秀 轻量快速 本地
GTE-large-zh 1024 优秀 阿里开源 本地
Jina-embeddings 1024 支持多语言 云/本地
text-embedding-ada-002 1536 一般 OpenAI API,效果好 云端
1
2
3
4
5
6
7
8
9
10
11
# 本地 Embedding 部署示例(使用 Sentence Transformers)
from sentence_transformers import SentenceTransformer

# 轻量级(推荐 CPU 或低显存环境)
model = SentenceTransformer('moka-ai/m3e-small')

# 高质量(推荐 GPU)
model = SentenceTransformer('BAAI/bge-large-zh-v1.5')

# 编码文本
embeddings = model.encode(["要嵌入的文本"], normalize_embeddings=True)

2.4 向量数据库选型

数据库 类型 优势 适用规模 部署难度
Milvus 专用向量 功能全面,性能强,国产 中大型 中等
Qdrant 专用向量 Rust 实现,高性能,API 友好 中大型 简单
Weaviate 矢量+结构化 混合检索,原生 GraphQL 中大型 简单
Chroma 矢量(嵌入式) 最简单,LangChain 集成 小型/实验 极简单
FAISS 矢量索引 Meta 开源,免费,高效 中型(单机) 简单
pgvector PostgreSQL 扩展 SQL 兼容,现有 PG 可用 中型 简单
Pinecone 专用向量 云原生,免运维 大型 极简单
Zilliz Cloud 专用向量(Milvus 云版) 全托管 大型 极简单

推荐路径:

  • 个人/小团队 → ChromaQdrant(docker 一键部署)
  • 企业级 → MilvusZilliz Cloud
  • 已有 PostgreSQL → pgvector

2.5 大语言模型选择

模型 场景 显存需求 特点
GPT-4o / o1 / o3 云端最强 效果最好,成本高
Claude 3.5 / 3.7 云端最强 长上下文优秀
Qwen2.5-72B-Instruct 本地最强 2× A100 80GB 中文优秀
Qwen2.5-32B-Instruct 本地主力 1× A100 80GB 性价比高
Qwen2.5-14B-Instruct 本地入门 1× A100 40GB 效果好
Qwen2.5-7B-Instruct 轻量 1× RTX 4090 速度快
DeepSeek-R1-Distill 系列 推理能力 各异 推理能力强

2.6 RAG 框架选型

框架 开发语言 特点 适用场景
LangChain Python 功能最全,生态最大 复杂 RAG,学术/生产
LlamaIndex Python 数据连接能力强 结构化数据
RAGFlow Python UI 好看,文档理解强 企业级知识库
Dify TypeScript 低代码,拖拽编排 快速构建应用
MaxKB Python 界面友好,支持多种 LLM 企业内知识库
AnythingLLM TypeScript 界面美观,连接器多 快速上手
FastGPT Go/TS 节点编排,工作流 复杂工作流
LangChain-Chatchat Python 中文友好,开箱即用 中文 RAG 快速部署

三、构建流程

3.1 环境准备

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# 基础环境
conda create -n rag python=3.11
conda activate rag

# 核心依赖
pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu121
pip install langchain langchain-community langchain-huggingface
pip install sentence-transformers
pip install qdrant-client # Qdrant 向量数据库客户端
pip install pymupdf # PDF 解析
pip install python-docx # Word 文档
pip install tiktoken # token 计算
pip install rank-bm25 # 关键词检索(可选,混合检索用)
pip install fastapi uvicorn # API 服务

3.2 文档解析

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
import fitz  # PyMuPDF
from pathlib import Path

class PDFParser:
"""PDF 文档解析器"""

def parse_pdf(self, file_path: str) -> list[dict]:
"""解析 PDF,返回页面列表"""
doc = fitz.open(file_path)
pages = []

for page_num, page in enumerate(doc):
text = page.get_text("text")
tables = page.find_tables()
pages.append({
"page_num": page_num + 1,
"text": text.strip(),
"tables": tables.extract() if tables else [],
"source": Path(file_path).name,
})
doc.close()
return pages

def parse_docx(self, file_path: str) -> list[dict]:
"""解析 Word 文档"""
import docx
doc = docx.Document(file_path)
chunks = []
current_chunk = []
current_len = 0

for para in doc.paragraphs:
text = para.text.strip()
if not text:
continue
if current_len + len(text) > 1000:
chunks.append({
"text": "\n".join(current_chunk),
"source": Path(file_path).name,
})
current_chunk = [text]
current_len = len(text)
else:
current_chunk.append(text)
current_len += len(text)

if current_chunk:
chunks.append({
"text": "\n".join(current_chunk),
"source": Path(file_path).name,
})
return chunks

def parse_markdown(self, file_path: str) -> list[dict]:
"""解析 Markdown 文档"""
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
import re
sections = re.split(r'\n(?=#+\s)', content)
chunks = []
for section in sections:
section = section.strip()
if len(section) < 50:
continue
chunks.append({
"text": section,
"source": Path(file_path).name,
})
return chunks

def parse_batch(self, folder_path: str) -> list[dict]:
"""批量解析文件夹中的文档"""
all_docs = []
folder = Path(folder_path)
for file_path in folder.rglob('*'):
if not file_path.is_file():
continue
suffix = file_path.suffix.lower()
try:
if suffix == '.pdf':
all_docs.extend(self.parse_pdf(str(file_path)))
elif suffix in ['.docx', '.doc']:
all_docs.extend(self.parse_docx(str(file_path)))
elif suffix in ['.md', '.txt']:
all_docs.extend(self.parse_markdown(str(file_path)))
except Exception as e:
print(f"Error parsing {file_path}: {e}")
continue
return all_docs

3.3 文本分块

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
from langchain.text_splitter import RecursiveCharacterTextSplitter

class TextChunker:
"""文本分块处理器"""

def __init__(self, chunk_size=512, chunk_overlap=64):
self.chunk_size = chunk_size
self.chunk_overlap = chunk_overlap

def chunk_documents(self, documents: list[dict]) -> list[dict]:
"""对文档列表进行分块"""
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=self.chunk_size,
chunk_overlap=self.chunk_overlap,
separators=["\n\n", "\n", "。", "!", "?", ". ", " ", ""],
length_function=self._token_len,
)
chunks = []
for doc in documents:
texts = text_splitter.split_text(doc['text'])
for i, text in enumerate(texts):
chunks.append({
"content": text,
"source": doc.get('source', ''),
"chunk_id": i,
})
return chunks

def _token_len(self, text: str) -> int:
"""估算 token 数(中文约 1.5 字符/token,英文约 4 字符/token)"""
import re
chinese_chars = len(re.findall(r'[\u4e00-\u9fff]', text))
other_chars = len(text) - chinese_chars
return int(chinese_chars * 1.5 + other_chars * 0.25)

3.4 向量化与存储

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
from sentence_transformers import SentenceTransformer
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
import uuid

class VectorStore:
"""向量数据库管理器"""

def __init__(self, collection_name="knowledge_base"):
self.collection_name = collection_name
self.embedding_model = SentenceTransformer('BAAI/bge-large-zh-v1.5')
self.qdrant = QdrantClient(host="localhost", port=6333)
self._create_collection()

def _create_collection(self):
"""创建 Collection"""
collections = [c.name for c in self.qdrant.get_collections().collections]
if self.collection_name not in collections:
self.qdrant.create_collection(
collection_name=self.collection_name,
vectors_config=VectorParams(
size=1024, # BGE-large 的维度
distance=Distance.COSINE,
),
)
print(f"Collection '{self.collection_name}' 创建成功")

def embed_texts(self, texts: list[str]) -> list[list[float]]:
"""将文本列表转为向量"""
embeddings = self.embedding_model.encode(
texts,
normalize_embeddings=True,
show_progress_bar=True,
)
return embeddings.tolist()

def add_chunks(self, chunks: list[dict]):
"""将分块数据写入向量数据库"""
texts = [chunk['content'] for chunk in chunks]
embeddings = self.embed_texts(texts)

points = []
for i, (chunk, embedding) in enumerate(zip(chunks, embeddings)):
point = PointStruct(
id=str(uuid.uuid4()),
vector=embedding,
payload={
"content": chunk['content'],
"source": chunk.get('source', ''),
"chunk_id": chunk.get('chunk_id', 0),
}
)
points.append(point)

batch_size = 100
for i in range(0, len(points), batch_size):
batch = points[i:i+batch_size]
self.qdrant.upsert(
collection_name=self.collection_name,
points=batch
)
print(f"全部写入完成,共 {len(points)} 条")

def search(self, query: str, top_k=5) -> list[dict]:
"""向量检索"""
query_embedding = self.embed_texts([query])[0]
results = self.qdrant.search(
collection_name=self.collection_name,
query_vector=query_embedding,
limit=top_k,
)
return [
{
"content": r.payload['content'],
"source": r.payload['source'],
"score": r.score,
}
for r in results
]

3.5 混合检索(推荐)

单一向量检索可能遗漏关键词匹配的重要文档,混合检索(向量 + BM25 关键词)可显著提升召回率:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
from rank_bm25 import BM25Okapi
import jieba

class HybridRetriever:
"""混合检索器(向量 + BM25)"""

def __init__(self, vector_store: VectorStore):
self.vector_store = vector_store
self.bm25_index = None
self.corpus = []

def build_bm25_index(self, chunks: list[dict]):
"""构建 BM25 索引"""
tokenized_corpus = [list(jieba.cut(chunk['content'])) for chunk in chunks]
self.bm25_index = BM25Okapi(tokenized_corpus)
self.corpus = chunks

def retrieve(self, query: str, top_k=5, vector_weight=0.7) -> list[dict]:
"""混合检索(RRF 融合)"""
# 1. 向量检索
vector_results = self.vector_store.search(query, top_k=top_k * 2)

# 2. BM25 检索
tokenized_query = list(jieba.cut(query))
bm25_scores = self.bm25_index.get_scores(tokenized_query)
top_indices = sorted(range(len(bm25_scores)), key=lambda i: bm25_scores[i], reverse=True)[:top_k * 2]
bm25_results = [self.corpus[i] for i in top_indices]

# 3. RRF 融合(Reciprocal Rank Fusion)
k = 60
rrf_scores = {}

for rank, result in enumerate(vector_results):
key = result['content']
rrf_scores[key] = result['score'] # 保留向量分数

for rank, result in enumerate(bm25_results):
key = result['content']
if key not in rrf_scores:
rrf_scores[key] = (1 / (k + rank + 1)) * 0.001 # 用归一化的 BM25 score
else:
rrf_scores[key] += (1 / (k + rank + 1))

fused_results = sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True)[:top_k]
content_to_meta = {r['content']: r for r in vector_results + bm25_results}

return [
{
"content": content,
"source": content_to_meta.get(content, {}).get('source', ''),
"score": score,
}
for content, score in fused_results
]

3.6 RAG 查询与生成

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
from transformers import AutoModelForCausalLM, AutoTokenizer
import torch

class RAGSystem:
"""RAG 检索增强生成系统"""

def __init__(self, vector_store: VectorStore, llm_model_name: str = "Qwen/Qwen2.5-14B-Instruct"):
self.vector_store = vector_store
print(f"加载 LLM: {llm_model_name}")
self.tokenizer = AutoTokenizer.from_pretrained(llm_model_name, trust_remote_code=True)
self.tokenizer.pad_token = self.tokenizer.eos_token
self.model = AutoModelForCausalLM.from_pretrained(
llm_model_name,
torch_dtype=torch.bfloat16,
device_map="auto",
trust_remote_code=True,
)

def build_prompt(self, query: str, retrieved_docs: list[dict], max_ctx_tokens=3000) -> str:
"""构建 Prompt"""
context_parts = []
total_len = 0

for i, doc in enumerate(retrieved_docs):
doc_text = f"[文档{i+1}] 来源:{doc.get('source', '未知')}\n{doc['content']}"
doc_len = len(doc_text)
if total_len + doc_len > max_ctx_tokens * 1.5:
break
context_parts.append(doc_text)
total_len += doc_len

context = "\n\n".join(context_parts)

prompt = f"""你是一个专业的问答助手。请根据以下参考文档回答用户的问题。

**要求:**
1. 仅根据参考文档回答,不要编造信息
2. 如果参考文档中没有相关信息,请明确说明
3. 在回答中引用参考文档的来源
4. 回答要准确、完整、有条理

**参考文档:**
{context}

**用户问题:**
{query}

**回答:**"""
return prompt

def generate(self, query: str, top_k=5, max_new_tokens=512) -> tuple[str, list[dict]]:
"""检索 + 生成"""
retrieved_docs = self.vector_store.search(query, top_k=top_k)
if not retrieved_docs:
return "抱歉,未在知识库中找到相关信息。", []

prompt = self.build_prompt(query, retrieved_docs)
inputs = self.tokenizer(prompt, return_tensors="pt").to(self.model.device)

outputs = self.model.generate(
**inputs,
max_new_tokens=max_new_tokens,
temperature=0.3,
top_p=0.9,
repetition_penalty=1.1,
)

response = self.tokenizer.decode(
outputs[0][inputs['input_ids'].shape[1]:],
skip_special_tokens=True
)
return response.strip(), retrieved_docs

3.7 API 服务封装

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional
import uvicorn

app = FastAPI(title="RAG API", version="1.0")

vector_store = None
rag_system = None

class QueryRequest(BaseModel):
query: str
top_k: Optional[int] = 5
max_new_tokens: Optional[int] = 512

class QueryResponse(BaseModel):
answer: str
sources: list[dict]

@app.post("/query", response_model=QueryResponse)
async def query(request: QueryRequest):
try:
answer, sources = rag_system.generate(
query=request.query,
top_k=request.top_k,
max_new_tokens=request.max_new_tokens,
)
return QueryResponse(
answer=answer,
sources=[
{"content": s['content'][:200], "source": s.get('source', ''), "score": s['score']}
for s in sources
]
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))

@app.post("/ingest")
async def ingest_documents(folder_path: str):
from pdf_parser import PDFParser
from text_chunker import TextChunker
parser = PDFParser()
chunks = parser.parse_batch(folder_path)
chunker = TextChunker()
chunked = chunker.chunk_documents(chunks)
vector_store.add_chunks(chunked)
return {"status": "ok", "chunks": len(chunked)}

@app.get("/health")
async def health():
return {"status": "ok"}

if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)

四、高级优化策略

4.1 Query 改写与扩展

用户提问往往不够精准,通过 Query 改写提升检索效果:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
def query_rewrite(self, query: str) -> list[str]:
"""生成多个查询改写版本"""
rewrite_prompt = f"""请对以下用户问题进行改写和扩展,生成 3-5 个不同的搜索查询。

要求:
1. 保持原意
2. 尝试不同的表达方式
3. 可以拆分复杂问题
4. 可以补充相关概念

原始问题:{query}

请以 JSON 数组格式返回,格式:[query1, query2, ...],每个 query 不超过 50 字"""
# 调用 LLM 生成改写后,返回多个查询
return [query, "改写版本1", "改写版本2", "扩展查询"]

4.2 重排序(Re-ranking)

初检向量检索后,使用更强的交叉编码器重排序,提升精度:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from sentence_transformers import CrossEncoder

class ReRanker:
"""文档重排序"""

def __init__(self, model_name="BAAI/bge-reranker-large"):
self.cross_encoder = CrossEncoder(model_name)

def rerank(self, query: str, documents: list[str], top_k=5) -> list[dict]:
"""对文档列表重排序"""
pairs = [[query, doc] for doc in documents]
scores = self.cross_encoder.predict(pairs)

scored_docs = sorted(
zip(documents, scores),
key=lambda x: x[1],
reverse=True
)[:top_k]

return [
{"content": doc, "rerank_score": float(score)}
for doc, score in scored_docs
]

4.3 Self-RAG(自反思 RAG)

引入 LLM 自我判断检索是否必要、生成是否事实一致:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
class SelfRAG:
"""Self-RAG 实现框架"""

ISREL_PROMPT = """请判断以下问题是否需要从外部知识库检索信息来回答。
问题:{query}
请只回答"需要"或"不需要"。"""

GROUNDNESS_PROMPT = """请判断以下回答是否与参考文档中的信息一致。
参考文档:{context}
回答:{response}
请回答"一致"或"不一致",并简要说明理由。"""

def __init__(self, llm):
self.llm = llm

def need_retrieval(self, query: str) -> bool:
"""判断是否需要检索"""
response = self.llm.generate(self.ISREL_PROMPT.format(query=query))
return "需要" in response

def generate(self, query: str, retrieved_docs: list[dict]) -> str:
"""带自反思的生成"""
context = "\n\n".join([doc['content'] for doc in retrieved_docs])
generation_prompt = f"""基于以下参考文档回答问题。
参考文档:
{context}
问题:{query}
回答:"""
response = self.llm.generate(generation_prompt)

# 检验事实一致性
groundness = self.llm.generate(
self.GROUNDNESS_PROMPT.format(context=context, response=response)
)

if "不一致" in groundness:
response += "\n\n注意:上述部分内容与文档信息存在不一致,请以上述参考文档为准。"

return response

4.4 常见 RAG 架构变体

架构 原理 适用场景
Naive RAG 直接检索 + 生成 简单问答
Query-Decomposition 将问题分解为多个子问题 复杂多跳问题
Tree of Thoughts 多路径推理 需要推理的知识问答
Self-RAG 模型自反思检索必要性 需要高准确率场景
RAG + Knowledge Graph 知识图谱补充结构化信息 复杂关系推理
HyDE 用 LLM 生成假设答案再检索 检索词与答案表述差异大时
Corrective RAG 检验检索结果质量,必要时重检 噪声多的知识库

五、部署方案

5.1 Docker 一键部署(RAGFlow)

RAGFlow 是中文友好的开源 RAG 平台,支持文档解析、嵌入、检索:

1
2
3
4
5
git clone https://github.com/infiniflow/ragflow.git
cd ragflow
# 配置 docker/.env 中的 API_KEY 和 Embedding 模型
docker compose up -d
# 访问 http://localhost:9380

5.2 Qdrant + 自建 RAG 服务部署

1
2
3
4
5
6
7
8
9
10
# 1. 启动 Qdrant 向量数据库
docker run -d --name qdrant \
-p 6333:6333 \
-p 6334:6334 \
-v $(pwd)/qdrant_storage:/qdrant/storage \
qdrant/qdrant

# 2. 启动 RAG API 服务
# (上面的 FastAPI 代码)
uvicorn main:app --host 0.0.0.0 --port 8000 --reload

5.3 企业级部署架构

1
2
3
4
5
6
7
8
9
10
11
12
用户请求 → Nginx(负载均衡)→ RAG API 集群(多实例)

┌─────────────────────┼─────────────────────┐
▼ ▼ ▼
┌──────────┐ ┌──────────┐ ┌──────────┐
│ Qdrant-1 │ ←─────→ │ Qdrant-2 │ ←───────→ │ Qdrant-3 │
│ (向量存储) │ │ (向量存储) │ │ (向量存储) │
└──────────┘ └──────────┘ └──────────┘
│ │ │
└───────────────────┴─────────────────────┘

MinIO / S3(文档存储)

六、效果评估

6.1 常用评估指标

指标 说明 评估方式
Precision@K Top-K 结果中相关文档的比例 人工标注
Recall@K 所有相关文档中被召回的比例 人工标注
MRR (Mean Reciprocal Rank) 第一个相关文档排名的倒数均值 人工标注
NDCG 考虑排序位置的相关性得分 人工标注
Answer Accuracy 生成答案的事实正确性 人工评估
Answer Relevance 生成答案与问题的相关性 LLM 评估
Faithfulness 生成答案与检索文档的一致性 LLM 评估

6.2 评估工具

工具 说明
RAGAS RAG 专用评估框架,支持 Faithfulness、Relevance 等指标
Trulens 支持追踪和评估 LangChain/LlamaIndex 应用
DeepEval 基于 LLM 的单元测试式评估
OpenCompass 通用大模型评测,支持 RAG 场景

七、常见问题排查

问题 原因 解决方案
检索结果不相关 Embedding 模型不合适 更换为领域适配的模型,如 BGE-large-zh
召回率低 分块过大/过小 调整 chunk_size 和 overlap
模型幻觉 Prompt 不够强调仅用文档 强化 Prompt 约束,增加”不确定请说不知道”
检索速度慢 向量维度过高/数据库大 使用 HNSW 索引降低延迟
上下文超长 检索块过多 设置 max_ctx_tokens 限制
多跳问题效果差 单次检索不够 使用 Query-Decomposition 分解问题
表格/图片内容无法理解 分块时破坏结构 使用 Marker 将 PDF 转为 Markdown 保留结构

八、参考资源


文档生成时间:2026-04-08
作者:J-tuzkibug