RAG 技术原理与知识库实践
作者整理日期：2026-06-09 技术栈：Python 3.11 / LangChain / LlamaIndex / Chroma / Qdrant / Ollama
1. RAG 基础概念
1.1 什么是 RAG
RAG（Retrieval-Augmented Generation，检索增强生成）是一种将外部知识检索与大型语言模型生成能力相结合的技术范式。它由 Meta AI Research 在 2020 年的论文《Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks》中正式提出。
核心思想：在生成答案之前，先从外部知识库中检索最相关的文档片段，将其作为上下文注入 Prompt，从而让 LLM 基于真实、最新、私有的知识进行回答。
用户问题 → [检索引擎] → 相关文档片段 → [LLM] → 基于文档的回答
1.2 为什么需要 RAG：LLM 的四大局限
局限一：知识截止日期（Knowledge Cutoff）
每个 LLM 都有训练数据截止日期。GPT-4 的截止日期是 2024 年初，Claude 3.5 是 2024 年中。任何截止日期之后发生的事件，模型都无法知晓。这对于需要实时信息的场景（技术文档、产品规格、法律条规）来说是致命缺陷。
局限二：私有数据无法访问
企业内部文档、个人笔记、代码仓库、数据库——这些从未出现在公开互联网上的数据，LLM 自然不知道。RAG 正是解决这个问题最主流的方案。
局限三：幻觉问题（Hallucination）
LLM 本质上是概率语言模型，它”倾向于生成听起来合理的内容”，而不是”只说自己知道的”。当模型对某个问题没有足够训练数据时，它会自信地编造答案——这就是幻觉。RAG 通过提供真实的检索证据，能显著抑制幻觉的发生。
局限四：Context Window 限制
GPT-4 Turbo 有 128k token 的上下文窗口，Gemini 1.5 Pro 有 1M token。但真实的企业知识库可能有几百万份文档、几十亿 token。RAG 通过精准检索，只把最相关的几千 token 送入模型，解决了”装不下”的问题。
1.3 RAG vs Fine-tuning vs In-context Learning
| 维度 | Fine-tuning（微调） | In-context Learning | RAG |
|---|---|---|---|
| 知识更新成本 | 高（需重新训练） | 中（需重写 Prompt） | 低（更新数据库即可） |
| 私有数据支持 | 需要大量标注数据 | 受 context 窗口限制 | 天然支持，无大小限制 |
| 推理延迟 | 低 | 低 | 中（需要检索步骤） |
| 幻觉抑制 | 中 | 低 | 高（有文档证据） |
| 可解释性 | 低 | 中 | 高（可引用来源） |
| 适用场景 | 改变模型风格/能力 | 少量示例引导 | 知识密集型问答 |
结论：三者不是互斥关系，而是互补的。最佳实践是：先用 Fine-tuning 让模型掌握领域语言风格，再用 RAG 提供实时知识，再用 In-context Learning 提供格式示例。
1.4 RAG 的三个核心阶段
阶段一：索引（Indexing） — 离线预处理
原始文档 → 加载解析 → 文本分块（Chunking）→ 向量化（Embedding）→ 存入向量数据库
阶段二：检索（Retrieval） — 在线查询
用户问题 → 向量化 → 向量相似度搜索 → Top-K 文档片段
阶段三：生成（Generation） — LLM 推理
[系统 Prompt] + [检索到的文档] + [用户问题] → LLM → 最终回答
2. 向量嵌入原理
2.1 Embedding 是什么：语义的数学表达
Embedding（嵌入）是将高维离散的文本信息映射到低维连续向量空间的过程。直觉上：语义相似的文本，其向量在空间中距离更近。
例如，以下三句话的向量关系：
- “苹果是一种水果” →
[0.82, -0.31, 0.54, ...]（1536 维） - “Apple is a fruit” →
[0.81, -0.29, 0.52, ...]（1536 维） ← 语义接近 - “特斯拉是一家汽车公司” →
[-0.41, 0.67, -0.23, ...]（1536 维） ← 语义远离
前两句的余弦相似度可能高达 0.97，而与第三句只有 0.12。
现代 Embedding 模型的工作原理是 Transformer 编码器（BERT 架构），通过大规模对比学习（Contrastive Learning）训练，让语义相近的句子对的向量靠近，语义不同的句子对的向量远离。
2.2 嵌入模型选型
OpenAI text-embedding-3-large
- 维度：3072（可压缩到 256）
- 价格：$0.13/1M tokens
- 质量：MTEB 排行榜前列，多语言支持优秀
- 缺点：需要联网，有隐私风险，按量付费
from openai import OpenAI
client = OpenAI()
response = client.embeddings.create(
model="text-embedding-3-large",
input="你好，世界",
dimensions=1024 # 可压缩以节省存储
)
vector = response.data[0].embedding # List[float], 长度 1024BGE-M3（中英文最佳开源模型）
由智源研究院（BAAI）推出，目前中文 RAG 场景的首选开源模型。
- 维度：1024
- 特点：支持中英文双语、支持多粒度检索（稠密+稀疏+多向量）
- 上下文长度：最大 8192 tokens
- 模型大小：约 2.3GB
- MTEB 中文榜：持续排名 Top 3
from FlagEmbedding import BGEM3FlagModel
model = BGEM3FlagModel('BAAI/bge-m3', use_fp16=True)
# 批量编码
sentences = ["什么是 RAG？", "RAG 全称检索增强生成"]
embeddings = model.encode(sentences, batch_size=12, max_length=512)['dense_vecs']
# embeddings.shape: (2, 1024)nomic-embed-text（本地轻量）
- 维度：768
- 特点：完全开源（Apache 2.0），可本地运行，速度极快
- 部署：通过 Ollama 一行命令即可使用
ollama pull nomic-embed-textimport ollama
response = ollama.embeddings(
model='nomic-embed-text',
prompt='这是一段需要嵌入的文本'
)
vector = response['embedding'] # List[float], 长度 768维度、速度、质量三角权衡
| 模型 | 维度 | 速度（CPU） | 中文质量 | 隐私 | 成本 |
|---|---|---|---|---|---|
| text-embedding-3-large | 3072 | 云端 | 极好 | 差 | 付费 |
| BGE-M3 | 1024 | 中 | 极好 | 好 | 免费 |
| BGE-small-zh | 512 | 快 | 好 | 好 | 免费 |
| nomic-embed-text | 768 | 快 | 中 | 好 | 免费 |
| all-MiniLM-L6-v2 | 384 | 极快 | 差 | 好 | 免费 |
选型建议：
- 个人知识库（中文为主）→ BGE-M3 或 BGE-small-zh
- 隐私要求极高 + 资源受限 → nomic-embed-text（Ollama）
- 商业产品 + 预算充足 → text-embedding-3-large
2.3 余弦相似度 vs 欧氏距离
余弦相似度（Cosine Similarity）：
cos(A, B) = (A · B) / (||A|| × ||B||)
范围 [-1, 1]，值越大越相似。不受向量长度影响，只关注方向。这是文本检索最常用的度量，因为文本的语义信息主要编码在方向上而非长度上。
欧氏距离（L2 Distance）：
d(A, B) = sqrt(Σ(Aᵢ - Bᵢ)²)
值越小越相似。对向量的绝对值敏感，适用于图像检索等场景。
对于归一化后的向量（单位向量），余弦相似度和 L2 距离是等价的（cos_sim = 1 - L2²/2），大多数 Embedding 模型输出的向量已经归一化，因此两种度量的检索结果相同。
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
# 余弦相似度
def cos_sim(a: list, b: list) -> float:
a, b = np.array(a), np.array(b)
return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))
# 实际使用中向量已归一化，可直接点积
def cos_sim_normalized(a: np.ndarray, b: np.ndarray) -> float:
return float(np.dot(a, b))2.4 Chunking（文本分块）策略
分块策略是 RAG 质量的关键，直接影响检索精度。
策略一：固定大小分块（Fixed-size Chunking）
from langchain.text_splitter import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter(
chunk_size=512, # 每块最大字符数
chunk_overlap=64, # 相邻块的重叠字符数，避免语义在边界截断
separators=["\n\n", "\n", "。", "！", "？", " ", ""]
)
chunks = splitter.split_text(document_text)优点：简单，速度快。缺点：可能在句子中间截断，损失语义完整性。
策略二：语义边界分块（Semantic Chunking）
from langchain_experimental.text_splitter import SemanticChunker
from langchain_community.embeddings import HuggingFaceEmbeddings
embedder = HuggingFaceEmbeddings(model_name="BAAI/bge-small-zh-v1.5")
splitter = SemanticChunker(
embedder,
breakpoint_threshold_type="percentile", # 用嵌入相似度变化检测语义边界
breakpoint_threshold_amount=95 # 第 95 百分位的相似度差异处分割
)
chunks = splitter.create_documents([document_text])优点：块内语义高度一致。缺点：计算开销大（需要对每个句子嵌入）。
策略三：Markdown 感知分块
from langchain.text_splitter import MarkdownHeaderTextSplitter
headers_to_split_on = [
("#", "H1"),
("##", "H2"),
("###", "H3"),
]
splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split_on)
md_chunks = splitter.split_text(markdown_text)
# 每个 chunk 自动携带 {'H1': '章节名', 'H2': '子节名'} 元数据最佳实践分块参数（针对中文技术文档）：
chunk_size = 400~600 字符
chunk_overlap = 50~100 字符（约 10~15%）
3. 向量数据库选型
3.1 主流向量数据库详细对比
| 数据库 | 语言 | 许可证 | 本地/云 | 持久化 | 全文搜索 | 元数据过滤 | 推荐场景 |
|---|---|---|---|---|---|---|---|
| Chroma | Python/Go | Apache 2.0 | 本地+云 | 是 | 否 | 是 | 原型/个人项目 |
| Qdrant | Rust | Apache 2.0 | 本地+云 | 是 | 否 | 是（强） | 生产环境 |
| Weaviate | Go | BSD 3 | 本地+云 | 是 | 是 | 是 | 混合检索 |
| FAISS | C++ | MIT | 本地 | 需手动 | 否 | 否 | 纯向量搜索研究 |
| LanceDB | Rust | Apache 2.0 | 本地 | 是（Arrow） | 是 | 是 | 多模态/数据分析 |
| Milvus | Go/C++ | Apache 2.0 | 本地+云 | 是 | 是 | 是 | 大规模企业 |
| pgvector | C | PostgreSQL | 本地+云 | 是 | 是（PG） | 是（SQL） | 已有 PostgreSQL |
3.2 选型建议
个人知识库 / 快速原型 → Chroma
零配置，纯 Python，持久化只需指定一个目录。适合 10 万以下的文档块。
生产环境 / 高并发 → Qdrant
Rust 编写，内存效率高，支持命名过滤（payload filter），支持多向量、稀疏向量。Docker 一行部署，REST + gRPC 双接口。
已有 PostgreSQL → pgvector
无需新增基础设施，SQL 语法直接操作向量。支持 IVFFlat 和 HNSW 索引。
多模态数据（图像+文本）→ LanceDB
原生 Apache Arrow 格式，与 Pandas/Polars 完美集成，支持图像向量。
4. 知识库 RAG Pipeline 完整实现
以下是一个完整、可运行的 RAG Pipeline，使用 LangChain + Chroma + BGE-M3（本地）或 OpenAI（云端）。
4.1 环境准备
pip install langchain langchain-community langchain-chroma \
chromadb FlagEmbedding sentence-transformers \
python-frontmatter tiktoken4.2 核心 Pipeline 代码
"""
rag_pipeline.py — 完整 RAG Pipeline
支持：Obsidian Markdown 加载 / BGE-M3 嵌入 / Chroma 存储 / 检索 / LLM 生成
"""
import os
import hashlib
from pathlib import Path
from typing import Optional
import chromadb
from chromadb.config import Settings
import frontmatter # python-frontmatter
from FlagEmbedding import BGEM3FlagModel
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_core.documents import Document
# ─────────────────────────────────────────────
# 1. 嵌入模型封装
# ─────────────────────────────────────────────
class BGEEmbedder:
"""本地 BGE-M3 嵌入模型，支持批量编码"""
def __init__(self, model_name: str = "BAAI/bge-m3", use_fp16: bool = True):
print(f"加载嵌入模型: {model_name} ...")
self.model = BGEM3FlagModel(model_name, use_fp16=use_fp16)
def embed_texts(self, texts: list[str], batch_size: int = 16) -> list[list[float]]:
"""批量将文本列表转为向量列表"""
result = self.model.encode(
texts,
batch_size=batch_size,
max_length=512,
return_dense=True,
return_sparse=False,
return_colbert_vecs=False
)
return result['dense_vecs'].tolist()
def embed_query(self, query: str) -> list[float]:
"""将单个查询文本转为向量（查询时使用指令前缀）"""
instructed = f"Represent this sentence for searching relevant passages: {query}"
return self.embed_texts([instructed])[0]
# ─────────────────────────────────────────────
# 2. 文档加载器（Obsidian Markdown）
# ─────────────────────────────────────────────
class ObsidianLoader:
"""加载 Obsidian Vault 中的所有 Markdown 文件，解析 frontmatter 元数据"""
def __init__(self, vault_path: str):
self.vault_path = Path(vault_path)
self.splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=60,
separators=["\n\n", "\n", "。", "！", "？", "；", " ", ""]
)
def _expand_wiki_links(self, text: str) -> str:
"""将 [[页面名]] 格式展开为 '页面名' 文本（简化处理）"""
import re
# [[Page|Alias]] → Alias；[[Page]] → Page
text = re.sub(r'\[\[([^\]|]+)\|([^\]]+)\]\]', r'\2', text)
text = re.sub(r'\[\[([^\]]+)\]\]', r'\1', text)
return text
def _file_hash(self, filepath: Path) -> str:
"""计算文件内容 MD5，用于增量更新判断"""
return hashlib.md5(filepath.read_bytes()).hexdigest()
def load_documents(self, exclude_dirs: list[str] | None = None) -> list[Document]:
"""加载所有 .md 文件并分块，返回 LangChain Document 列表"""
exclude_dirs = set(exclude_dirs or [".obsidian", ".trash", "templates"])
documents = []
for md_file in sorted(self.vault_path.rglob("*.md")):
# 跳过排除目录
parts = set(md_file.relative_to(self.vault_path).parts)
if parts & exclude_dirs:
continue
try:
post = frontmatter.load(md_file)
except Exception as e:
print(f"跳过 {md_file}: {e}")
continue
raw_content = self._expand_wiki_links(post.content)
if len(raw_content.strip()) < 50: # 过滤空文件
continue
# 提取 frontmatter 元数据
metadata_base = {
"source": str(md_file.relative_to(self.vault_path)),
"file_name": md_file.stem,
"file_hash": self._file_hash(md_file),
"tags": ",".join(post.metadata.get("tags", [])),
"created": str(post.metadata.get("created", "")),
"confidence": str(post.metadata.get("confidence", "medium")),
}
# 分块
chunks = self.splitter.create_documents(
texts=[raw_content],
metadatas=[metadata_base]
)
# 为每块添加序号
for i, chunk in enumerate(chunks):
chunk.metadata["chunk_index"] = i
chunk.metadata["total_chunks"] = len(chunks)
documents.extend(chunks)
print(f"加载完成：{len(documents)} 个文档块")
return documents
# ─────────────────────────────────────────────
# 3. 向量存储（Chroma）
# ─────────────────────────────────────────────
class VectorStore:
"""基于 Chroma 的向量存储，支持增量更新"""
def __init__(
self,
persist_dir: str,
embedder: BGEEmbedder,
collection_name: str = "knowledge_base"
):
self.embedder = embedder
self.client = chromadb.PersistentClient(
path=persist_dir,
settings=Settings(anonymized_telemetry=False)
)
self.collection = self.client.get_or_create_collection(
name=collection_name,
metadata={"hnsw:space": "cosine"} # 使用余弦相似度
)
print(f"向量库已就绪，当前文档数: {self.collection.count()}")
def add_documents(self, documents: list[Document], batch_size: int = 64):
"""批量添加文档，自动跳过已存在的（通过 file_hash + chunk_index 去重）"""
new_docs = []
for doc in documents:
doc_id = f"{doc.metadata['file_hash']}_{doc.metadata['chunk_index']}"
doc.metadata["_id"] = doc_id
new_docs.append(doc)
# 分批处理
for i in range(0, len(new_docs), batch_size):
batch = new_docs[i:i + batch_size]
texts = [d.page_content for d in batch]
ids = [d.metadata["_id"] for d in batch]
metadatas = [d.metadata for d in batch]
# 批量向量化
embeddings = self.embedder.embed_texts(texts)
# 使用 upsert（存在则更新，不存在则插入）
self.collection.upsert(
ids=ids,
embeddings=embeddings,
documents=texts,
metadatas=metadatas
)
print(f" 已处理 {min(i + batch_size, len(new_docs))}/{len(new_docs)} 块")
def query(
self,
query: str,
top_k: int = 5,
filter_dict: dict | None = None
) -> list[dict]:
"""
检索最相关的文档块
filter_dict 示例: {"tags": {"$contains": "python"}}
"""
query_vector = self.embedder.embed_query(query)
results = self.collection.query(
query_embeddings=[query_vector],
n_results=top_k,
where=filter_dict,
include=["documents", "metadatas", "distances"]
)
hits = []
for doc, meta, dist in zip(
results["documents"][0],
results["metadatas"][0],
results["distances"][0]
):
hits.append({
"content": doc,
"metadata": meta,
"score": 1 - dist, # 余弦距离转相似度
})
return hits
# ─────────────────────────────────────────────
# 4. RAG 问答引擎
# ─────────────────────────────────────────────
RAG_SYSTEM_PROMPT = """你是一个基于个人知识库的智能助手。
请严格根据以下检索到的知识片段回答用户问题。
规则：
1. 只使用提供的知识片段中的信息，不要使用外部知识
2. 如果知识片段中没有足够信息，请明确说"根据现有知识库，无法回答此问题"
3. 回答时标注来源文件，格式：（来源：[文件名]）
4. 置信度低的知识片段（confidence: low）请在引用时注明"此信息置信度较低"
已检索的知识片段：
{context}
请回答用户问题。"""
class RAGEngine:
"""RAG 问答引擎，支持本地 Ollama 和 OpenAI"""
def __init__(
self,
vector_store: VectorStore,
llm_backend: str = "ollama", # "ollama" 或 "openai"
model_name: str = "qwen2.5:7b",
top_k: int = 5
):
self.store = vector_store
self.top_k = top_k
self.backend = llm_backend
if llm_backend == "ollama":
import ollama as ollama_client
self._llm = ollama_client
self._model = model_name
elif llm_backend == "openai":
from openai import OpenAI
self._llm = OpenAI()
self._model = model_name or "gpt-4o-mini"
def _build_context(self, hits: list[dict]) -> str:
"""将检索结果格式化为上下文字符串"""
lines = []
for i, hit in enumerate(hits, 1):
meta = hit["metadata"]
confidence_note = " [置信度低]" if meta.get("confidence") == "low" else ""
lines.append(
f"[片段 {i}] 来源: {meta.get('file_name', '未知')}{confidence_note}\n"
f"相关度: {hit['score']:.3f}\n"
f"{hit['content']}\n"
)
return "\n---\n".join(lines)
def ask(self, question: str, filter_dict: dict | None = None) -> dict:
"""执行 RAG 问答，返回答案和引用来源"""
# 检索
hits = self.store.query(question, top_k=self.top_k, filter_dict=filter_dict)
if not hits or hits[0]["score"] < 0.4:
return {
"answer": "根据现有知识库，未找到与此问题相关的内容。",
"sources": [],
"retrieved_chunks": hits
}
context = self._build_context(hits)
prompt = RAG_SYSTEM_PROMPT.format(context=context)
messages = [
{"role": "system", "content": prompt},
{"role": "user", "content": question}
]
# 调用 LLM
if self.backend == "ollama":
response = self._llm.chat(model=self._model, messages=messages)
answer = response["message"]["content"]
else:
response = self._llm.chat.completions.create(
model=self._model, messages=messages
)
answer = response.choices[0].message.content
# 提取来源
sources = list({h["metadata"].get("source", "") for h in hits})
return {
"answer": answer,
"sources": sources,
"retrieved_chunks": hits
}
# ─────────────────────────────────────────────
# 5. 主程序入口
# ─────────────────────────────────────────────
if __name__ == "__main__":
VAULT_PATH = "/path/to/your/obsidian/vault"
PERSIST_DIR = "/path/to/chroma_db"
# 初始化组件
embedder = BGEEmbedder("BAAI/bge-m3")
store = VectorStore(PERSIST_DIR, embedder)
# 加载并索引文档（首次运行或需要更新时执行）
loader = ObsidianLoader(VAULT_PATH)
docs = loader.load_documents()
store.add_documents(docs)
# 初始化 RAG 引擎
engine = RAGEngine(store, llm_backend="ollama", model_name="qwen2.5:7b")
# 交互问答
while True:
q = input("\n请输入问题（q 退出）: ").strip()
if q.lower() == "q":
break
result = engine.ask(q)
print(f"\n回答：\n{result['answer']}")
print(f"\n来源文件：{', '.join(result['sources'])}")5. 高级 RAG 技术
5.1 HyDE（Hypothetical Document Embeddings）
核心思想：与其用”问题”的向量去匹配”答案”的向量（两者语义空间不对齐），不如先让 LLM 生成一个假设性答案，再用假设答案的向量去检索。
class HyDERetriever:
"""HyDE：用假设文档提升检索质量"""
HYDE_PROMPT = """请为以下问题生成一段假设性的详细回答。
这段回答将用于检索相关文档，因此请尽量包含该领域的专业术语。
不需要回答正确，只需生成语义相关的内容。
问题: {question}
假设回答:"""
def __init__(self, engine: RAGEngine):
self.engine = engine
def generate_hypothesis(self, question: str) -> str:
"""用 LLM 生成假设文档"""
if self.engine.backend == "ollama":
resp = self.engine._llm.chat(
model=self.engine._model,
messages=[{"role": "user", "content": self.HYDE_PROMPT.format(question=question)}]
)
return resp["message"]["content"]
# OpenAI 版本类似...
def ask(self, question: str) -> dict:
# 生成假设文档
hypothesis = self.generate_hypothesis(question)
# 用假设文档+原始问题混合向量检索
h_vec = self.engine.store.embedder.embed_query(hypothesis)
q_vec = self.engine.store.embedder.embed_query(question)
# 取两个向量的平均（简单融合）
import numpy as np
mixed_vec = ((np.array(h_vec) + np.array(q_vec)) / 2).tolist()
# 直接用混合向量检索
results = self.engine.store.collection.query(
query_embeddings=[mixed_vec],
n_results=self.engine.top_k
)
# ... 后续同正常 RAG 流程适用场景：问题和答案的表达方式差异较大时（如”什么是XX” → 知识库中存储的是描述性段落）。
5.2 Parent-Child Chunking（父子分块）
核心思想：用小块精确检索，用大块提供完整上下文。
from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryByteStore
from langchain_chroma import Chroma
# 子块：用于检索（小而精确）
child_splitter = RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=20)
# 父块：用于送入 LLM（大而完整）
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=800, chunk_overlap=100)
vectorstore = Chroma(embedding_function=embedder, persist_directory="./chroma_db")
docstore = InMemoryByteStore() # 存储父块（可替换为 Redis）
retriever = ParentDocumentRetriever(
vectorstore=vectorstore,
docstore=docstore,
child_splitter=child_splitter,
parent_splitter=parent_splitter,
)
retriever.add_documents(raw_documents)
# 检索时：自动用小块找到相关位置，返回对应的大块
parent_docs = retriever.invoke("你的问题")效果：检索精度不损失，同时 LLM 得到更完整的上下文。
5.3 Re-ranking（重排序）
向量检索的 Top-K 是基于近似最近邻算法的粗排结果，精度有上限。Cross-Encoder 重排器通过对（query, document）对进行精细的交叉注意力打分，显著提升精度。
from sentence_transformers import CrossEncoder
class Reranker:
"""使用 BGE Reranker 对检索结果重排序"""
def __init__(self, model_name: str = "BAAI/bge-reranker-v2-m3"):
self.model = CrossEncoder(model_name)
def rerank(self, query: str, hits: list[dict], top_n: int = 3) -> list[dict]:
"""对检索结果重排，返回前 top_n 个"""
pairs = [(query, hit["content"]) for hit in hits]
scores = self.model.predict(pairs)
# 按重排分数排序
for hit, score in zip(hits, scores):
hit["rerank_score"] = float(score)
reranked = sorted(hits, key=lambda x: x["rerank_score"], reverse=True)
return reranked[:top_n]
# 在 RAGEngine.ask() 中集成 Re-ranking
def ask_with_rerank(self, question: str, reranker: Reranker) -> dict:
# 先宽泛检索 20 个候选
candidates = self.store.query(question, top_k=20)
# 再精排取 Top 3
hits = reranker.rerank(question, candidates, top_n=3)
context = self._build_context(hits)
# ... 后续生成步骤不变重排器推荐：
BAAI/bge-reranker-v2-m3（中英双语，推荐）cross-encoder/ms-marco-MiniLM-L-6-v2（英文，轻量）
5.4 Self-query Retrieval（自查询检索）
让 LLM 自动从问题中提取结构化过滤条件，结合向量检索和元数据过滤。
from langchain.retrievers.self_query.base import SelfQueryRetriever
from langchain.chains.query_constructor.base import AttributeInfo
from langchain_openai import ChatOpenAI
# 定义元数据字段的含义
metadata_field_info = [
AttributeInfo(name="tags", description="文档标签，如 python, ai, productivity", type="string"),
AttributeInfo(name="confidence", description="知识置信度: high/medium/low", type="string"),
AttributeInfo(name="created", description="创建日期，YYYY-MM-DD 格式", type="string"),
]
retriever = SelfQueryRetriever.from_llm(
llm=ChatOpenAI(model="gpt-4o-mini"),
vectorstore=vectorstore,
document_contents="个人知识库笔记",
metadata_field_info=metadata_field_info,
verbose=True
)
# 用户问："2025年之后写的、关于 Python 的高置信度笔记有哪些？"
# LLM 自动生成过滤条件: {"tags": {"$contains": "python"}, "confidence": "high"}
docs = retriever.invoke("2025年之后写的关于 Python 的高置信度笔记")6. Obsidian Vault → RAG 完整方案
6.1 增量更新策略
核心思路：维护一个”文件指纹表”（哈希值），只重新索引发生变化的文件。
import json
from pathlib import Path
class IncrementalIndexer:
"""增量索引管理器：只重新索引变更的文件"""
HASH_DB_PATH = ".rag_index_hashes.json"
def __init__(self, vault_path: str, store: VectorStore):
self.vault_path = Path(vault_path)
self.store = store
self.hash_db_path = self.vault_path / self.HASH_DB_PATH
self._hashes = self._load_hashes()
def _load_hashes(self) -> dict:
if self.hash_db_path.exists():
return json.loads(self.hash_db_path.read_text())
return {}
def _save_hashes(self):
self.hash_db_path.write_text(json.dumps(self._hashes, indent=2, ensure_ascii=False))
def _file_hash(self, filepath: Path) -> str:
return hashlib.md5(filepath.read_bytes()).hexdigest()
def update(self, loader: ObsidianLoader):
"""扫描所有文件，只更新变更的部分"""
all_md_files = list(self.vault_path.rglob("*.md"))
changed_files, deleted_files = [], []
# 检测变更和新增
current_keys = set()
for f in all_md_files:
key = str(f.relative_to(self.vault_path))
current_keys.add(key)
new_hash = self._file_hash(f)
if self._hashes.get(key) != new_hash:
changed_files.append(f)
self._hashes[key] = new_hash
# 检测删除
for key in list(self._hashes.keys()):
if key not in current_keys:
deleted_files.append(key)
del self._hashes[key]
# 删除旧索引
if deleted_files:
print(f"删除 {len(deleted_files)} 个已删除文件的索引...")
for key in deleted_files:
# 根据文件名前缀删除相关块
results = self.store.collection.get(where={"source": key})
if results["ids"]:
self.store.collection.delete(ids=results["ids"])
# 重新索引变更文件
if changed_files:
print(f"重新索引 {len(changed_files)} 个变更文件...")
# 先删除旧块
for f in changed_files:
key = str(f.relative_to(self.vault_path))
results = self.store.collection.get(where={"source": key})
if results["ids"]:
self.store.collection.delete(ids=results["ids"])
# 加载并重新索引
new_docs = []
for f in changed_files:
file_docs = loader.load_documents_from_file(f)
new_docs.extend(file_docs)
if new_docs:
self.store.add_documents(new_docs)
self._save_hashes()
print(f"增量更新完成：变更 {len(changed_files)} 个文件，删除 {len(deleted_files)} 个文件")7. 知识库问答系统设计
7.1 Prompt Engineering for RAG
一个高质量的 RAG Prompt 需要包含以下元素：
RAG_PROMPT_TEMPLATE = """你是一个基于个人知识库的智能助手。
## 你的知识来源
以下是从知识库中检索到的相关内容（按相关度排序）：
{context}
## 回答规则
1. **严格依据上述内容回答**，不得使用知识库以外的信息
2. **引用来源**：每个关键陈述后注明来源，格式：`（来源：文件名）`
3. **置信度传递**：若某片段标注 [置信度低]，相关陈述后注明"此信息置信度较低，建议验证"
4. **无法回答时**：若知识库内容不足以回答，明确说明"当前知识库中没有足够信息回答此问题"，不得推测或捏造
5. **结构化输出**：对于复杂问题，使用标题和列表组织回答
## 用户问题
{question}
## 你的回答
"""7.2 置信度传递机制
Obsidian 笔记的 frontmatter 中可以标注置信度：
---
title: Python GIL 机制
tags: [python, concurrency]
confidence: high # high / medium / low
source: "Python 官方文档 3.12"
created: 2025-03-15
---RAG 系统在回答时将置信度传递给用户：
def format_hit_with_confidence(hit: dict) -> str:
meta = hit["metadata"]
confidence = meta.get("confidence", "medium")
confidence_label = {
"high": "",
"medium": " ⚠️ 中等置信度",
"low": " ❗ 低置信度，建议验证"
}.get(confidence, "")
return (
f"来源：{meta['file_name']}{confidence_label}\n"
f"相关度：{hit['score']:.2f}\n"
f"内容：{hit['content']}"
)7.3 无答案处理
def ask_with_guardrails(self, question: str) -> dict:
hits = self.store.query(question, top_k=5)
# 相关度过低：直接拒绝
max_score = max((h["score"] for h in hits), default=0)
if max_score < 0.45:
return {
"answer": f"当前知识库中没有与「{question}」相关的内容。\n"
f"最高相关度仅 {max_score:.2f}（阈值 0.45），建议：\n"
f"1. 换用其他关键词重试\n"
f"2. 在知识库中添加相关笔记",
"sources": [],
"refused": True
}
# 正常问答流程...8. 本地部署方案（隐私优先）
8.1 完整本地技术栈
Ollama（本地 LLM + 嵌入）
├── LLM: qwen2.5:7b 或 llama3.1:8b
└── Embedding: nomic-embed-text 或 BGE-M3（通过 HuggingFace 本地加载）
Chroma（本地向量数据库）
└── 持久化到本地目录
Python RAG Application
└── LangChain / 自定义 Pipeline
8.2 Docker Compose 一键部署
# docker-compose.yml
version: "3.9"
services:
ollama:
image: ollama/ollama:latest
container_name: rag_ollama
ports:
- "11434:11434"
volumes:
- ollama_data:/root/.ollama
deploy:
resources:
reservations:
devices:
- driver: nvidia
count: all
capabilities: [gpu]
restart: unless-stopped
chromadb:
image: chromadb/chroma:latest
container_name: rag_chroma
ports:
- "8000:8000"
volumes:
- chroma_data:/chroma/chroma
environment:
- CHROMA_SERVER_AUTH_PROVIDER=chromadb.auth.token_authn.TokenAuthenticationServerProvider
- CHROMA_SERVER_AUTH_CREDENTIALS=your-secret-token
- ANONYMIZED_TELEMETRY=False
restart: unless-stopped
rag_app:
build:
context: .
dockerfile: Dockerfile
container_name: rag_app
ports:
- "7860:7860" # Gradio UI
volumes:
- ./vault:/vault:ro # Obsidian Vault（只读挂载）
- ./index:/app/index # 索引缓存
environment:
- OLLAMA_BASE_URL=http://ollama:11434
- CHROMA_HOST=chromadb
- CHROMA_PORT=8000
- CHROMA_TOKEN=your-secret-token
- VAULT_PATH=/vault
depends_on:
- ollama
- chromadb
restart: unless-stopped
volumes:
ollama_data:
chroma_data:# Dockerfile
FROM python:3.11-slim
WORKDIR /app
RUN pip install --no-cache-dir \
langchain langchain-community langchain-chroma \
chromadb FlagEmbedding sentence-transformers \
python-frontmatter gradio
COPY . .
# 拉取 Ollama 模型（首次启动时执行）
CMD ["python", "app.py"]一键启动：
docker compose up -d
# 拉取需要的模型
docker exec rag_ollama ollama pull qwen2.5:7b
docker exec rag_ollama ollama pull nomic-embed-text8.3 带 Gradio UI 的完整应用
"""app.py — 带 Web UI 的本地 RAG 应用"""
import gradio as gr
from rag_pipeline import BGEEmbedder, VectorStore, RAGEngine, ObsidianLoader
import os
VAULT_PATH = os.environ.get("VAULT_PATH", "/vault")
PERSIST_DIR = os.environ.get("INDEX_PATH", "/app/index/chroma_db")
LLM_MODEL = os.environ.get("LLM_MODEL", "qwen2.5:7b")
# 初始化（模块级别，只执行一次）
print("初始化 RAG 系统...")
embedder = BGEEmbedder("BAAI/bge-m3")
store = VectorStore(PERSIST_DIR, embedder)
engine = RAGEngine(store, llm_backend="ollama", model_name=LLM_MODEL)
print("RAG 系统就绪")
def chat(message: str, history: list, show_sources: bool) -> tuple[str, list]:
result = engine.ask(message)
answer = result["answer"]
if show_sources and result["sources"]:
answer += f"\n\n---\n**引用来源**：\n" + "\n".join(f"- {s}" for s in result["sources"])
history.append((message, answer))
return "", history
def reindex():
loader = ObsidianLoader(VAULT_PATH)
docs = loader.load_documents()
store.add_documents(docs)
return f"索引完成：共 {len(docs)} 个文档块，数据库中共 {store.collection.count()} 条记录"
with gr.Blocks(title="Personal Knowledge RAG") as demo:
gr.Markdown("# 个人知识库问答系统")
with gr.Row():
chatbot = gr.Chatbot(height=500)
with gr.Row():
msg = gr.Textbox(placeholder="输入问题...", scale=4)
show_src = gr.Checkbox(label="显示来源", value=True, scale=1)
with gr.Row():
submit_btn = gr.Button("提问", variant="primary")
reindex_btn = gr.Button("重新索引知识库")
reindex_status = gr.Textbox(label="索引状态", interactive=False)
submit_btn.click(chat, [msg, chatbot, show_src], [msg, chatbot])
msg.submit(chat, [msg, chatbot, show_src], [msg, chatbot])
reindex_btn.click(reindex, outputs=reindex_status)
demo.launch(server_name="0.0.0.0", server_port=7860)8.4 性能基准参考
| 硬件配置 | 嵌入速度（BGE-M3） | LLM 推理（Qwen2.5:7b） | 检索延迟（10万块） |
|---|---|---|---|
| M2 Mac (16GB) | ~50 chunks/s | ~25 tokens/s | < 50ms |
| RTX 3090 (24GB) | ~200 chunks/s | ~60 tokens/s | < 30ms |
| CPU Only (16核) | ~15 chunks/s | ~5 tokens/s | < 100ms |
首次索引 1000 篇笔记（约 5000 个 chunk）所需时间：
- M2 Mac：约 100 秒
- RTX 3090：约 25 秒
- CPU：约 330 秒
增量更新（只处理变更文件）通常在 10 秒以内完成。
总结
RAG 是目前将 LLM 与私有知识结合的最实用方案。本文覆盖的技术路径：
- 基础 Pipeline：Obsidian Markdown → BGE-M3 嵌入 → Chroma 存储 → 检索 → Qwen2.5 生成
- 质量提升：HyDE 假设文档 + Parent-Child 分块 + BGE Reranker 重排序
- 生产化：增量更新、置信度传递、无答案拒绝、Docker 部署
对于个人知识库场景，推荐起步配置：
嵌入模型：BAAI/bge-small-zh-v1.5（速度快，中文质量好）
向量库：Chroma（零配置，本地持久化）
LLM：Ollama + qwen2.5:7b（本地，隐私安全）
分块：RecursiveCharacterTextSplitter(chunk_size=500, overlap=60)
随着知识库增长（>10 万条记录），再迁移到 Qdrant + BGE-M3 + 重排器的生产级方案。