RAG 检索增强生成
#RAG · #向量数据库 · #语义检索 · #Chunking · #Embedding · #Milvus · #Qdrant · #HybridSearch · #AgenticRAG · #GraphRAG
如何把一本几十万字的 HTML 课程书变成 AI 随时能检索的知识库？本专题覆盖文档解析、语义分块、向量数据库选型、多路召回与重排的全链路工程实践。
RAG 全景架构
RAG（Retrieval-Augmented Generation）让 LLM 能够在生成回答前，先从外部知识库中检索相关信息，有效解决 LLM 的"幻觉"和知识更新滞后问题。
mermaid
graph TD
subgraph "离线阶段（建库）"
DOC["📄 原始文档<br/>HTML / PDF / Markdown"] --> PARSE["🔍 文档解析<br/>提取文本 + 结构"]
PARSE --> CHUNK["✂️ 语义分块<br/>按 <h1>, <section> 切分"]
CHUNK --> EMBED["🧮 向量化<br/>Embedding 模型编码"]
EMBED --> STORE["💾 存入向量数据库<br/>Milvus / Qdrant / PGVector"]
end
subgraph "在线阶段（检索）"
QUERY["❓ 用户查询"] --> QEMBED["🧮 查询向量化"]
QEMBED --> SEARCH["🔍 混合检索<br/>向量 + BM25"]
SEARCH --> RERANK["📊 重排序<br/>Rerank 模型精排"]
RERANK --> CTX["📋 构建上下文"]
end
CTX --> LLM["🧠 LLM 生成回答<br/>基于检索到的上下文"]
LLM --> ANS["✅ 最终答案"]
style STORE fill:#3498db,color:#fff
style SEARCH fill:#e74c3c,color:#fff
style LLM fill:#9b59b6,color:#fff文档解析与语义分块（Chunking）
为什么不能简单按字数切分？
❌ 机械按字数切分（每 512 字一刀）:
"...Transformer 架构的核心是自注意力机制。具体来说,"
— 这里被切断了 —
"每一个 Token 都会与序列中的其他所有 Token 计算..."
✅ 结构化按标签切分（基于 HTML 标签）:
<h2>自注意力机制</h2>
<p>Transformer 架构的核心是自注意力机制。
具体来说，每一个 Token 都会与序列中的其他所有 Token 计算注意力权重...</p>结构化 Chunking 引擎
python
"""
基于 HTML 标签结构的智能分块引擎
策略：
1. 识别 <h1>~<h6> 作为天然的分段边界
2. <section>, <article>, <div class="chapter"> 作为语义边界
3. 在每个分段内部，保持段落的完整性
4. 块与块之间有重叠（overlap），确保跨块信息不丢失
"""
import re
from dataclasses import dataclass, field
from typing import List, Optional
from html.parser import HTMLParser
@dataclass
class Chunk:
"""文档分块"""
text: str # 块文本内容
title: str = "" # 所属的父标题（如 "2.1 自注意力"）
level: int = 0 # 标题层级（h1=1, h2=2, ...）
start_pos: int = 0 # 在原文档中的起始位置
metadata: dict = field(default_factory=dict) # 额外元数据
class HTMLChunkParser(HTMLParser):
"""
HTML 结构化分块解析器
按 <h1>~<h6> 和 <section> 标签进行语义切分，
而非机械地按字数切分。
"""
def __init__(self, max_chunk_size: int = 1024, overlap: int = 128):
super().__init__()
self.max_chunk_size = max_chunk_size # 每个块的最大 Token 数
self.overlap = overlap # 块之间的重叠 Token 数
self.chunks: List[Chunk] = []
# 解析状态
self._current_text: List[str] = []
self._current_title = ""
self._current_level = 0
self._title_stack: List[tuple] = [] # [(title, level), ...]
def handle_starttag(self, tag: str, attrs: list):
# 标题标签 → 新的语义分段
if tag in ('h1', 'h2', 'h3', 'h4', 'h5', 'h6'):
# 先保存当前块
self._maybe_flush()
# 更新标题层级
level = int(tag[1])
# 弹出比当前层级更深的标题
while self._title_stack and self._title_stack[-1][1] >= level:
self._title_stack.pop()
# section/article → 语义分段
if tag in ('section', 'article'):
self._maybe_flush()
def handle_endtag(self, tag: str):
if tag in ('h1', 'h2', 'h3', 'h4', 'h5', 'h6'):
# 标题结束，记录标题文本
title_text = ''.join(self._current_text).strip()
level = int(tag[1])
self._title_stack.append((title_text, level))
self._current_title = title_text
self._current_level = level
self._current_text = []
elif tag in ('p', 'div', 'li', 'td', 'th', 'section', 'article'):
# 段落结束 → 可能的分块点
self._maybe_flush()
def handle_data(self, data: str):
text = data.strip()
if text:
self._current_text.append(text + ' ')
def _maybe_flush(self):
"""当前文本积累到一定量时，输出为一个 Chunk"""
text = ''.join(self._current_text).strip()
if not text:
return
# 如果文本太长，按句子进一步切分
if len(text) > self.max_chunk_size:
sub_chunks = self._split_by_sentence(text)
for sc in sub_chunks:
self._emit_chunk(sc)
else:
self._emit_chunk(text)
self._current_text = []
def _split_by_sentence(self, text: str) -> List[str]:
"""按句子切分，尽量保持语义完整"""
sentences = re.split(r'(?<=[。！？.!?])\s*', text)
chunks = []
current = ""
for sent in sentences:
if len(current) + len(sent) <= self.max_chunk_size:
current += sent
else:
if current:
chunks.append(current)
current = sent
if current:
chunks.append(current)
# 添加重叠
if len(chunks) > 1 and self.overlap > 0:
overlapped = [chunks[0]]
for i in range(1, len(chunks)):
prev = chunks[i-1]
curr = chunks[i]
if len(prev) > self.overlap:
curr = prev[-self.overlap:] + ' ' + curr
overlapped.append(curr)
return overlapped
return chunks
def _emit_chunk(self, text: str):
"""输出一个分块"""
# 构建完整标题路径: "1. 深度学习 > 1.2 神经网络 > 自注意力"
title_path = ' > '.join(t[0] for t in self._title_stack)
if self._current_title and self._current_title not in title_path:
if title_path:
title_path += ' > ' + self._current_title
else:
title_path = self._current_title
self.chunks.append(Chunk(
text=text.strip(),
title=title_path,
level=self._current_level,
))
def close(self):
"""解析结束，刷新最后的缓存"""
self._maybe_flush()
super().close()
# ========== 使用示例 ==========
def chunk_html_document(html_content: str, max_chunk_size: int = 1024) -> List[Chunk]:
"""对 HTML 文档进行结构化分块"""
parser = HTMLChunkParser(max_chunk_size=max_chunk_size, overlap=128)
parser.feed(html_content)
parser.close()
return parser.chunks
# 测试
sample_html = """
<h1>Transformer 架构</h1>
<section>
<h2>自注意力机制</h2>
<p>Transformer 架构的核心是自注意力机制。具体来说，每一个 Token 都会与序列中的
其他所有 Token 计算注意力权重，从而捕获全局依赖关系。</p>
<p>自注意力的计算分为三个步骤：首先通过线性变换得到 Q、K、V 向量，然后计算
Q 与 K 的点积，最后用 Softmax 归一化后的权重对 V 加权求和。</p>
</section>
<section>
<h2>多头注意力</h2>
<p>为了捕获不同类型的依赖关系，Transformer 引入了多头注意力机制。
每个头独立进行注意力计算，最后将所有头的结果拼接起来。</p>
</section>
"""
# chunks = chunk_html_document(sample_html)
# for i, c in enumerate(chunks):
# print(f"--- Chunk {i}: [{c.title}] ---\n{c.text}\n")Chunking 策略对比
| 策略 | 方法 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|---|
| 固定大小 | 每 N 字一刀 | 实现简单 | 语义割裂严重 | 快速原型 |
| 固定大小+重叠 | 每 N 字一刀 + M 字重叠 | 减轻割裂 | 仍可能跨段 | 通用文档 |
| 按句子 | 每个句号切分 | 语义完整 | 粒度不均匀 | 短文本 |
| 按段落 | 每个 <p> 切分 | 自然语义边界 | 段落可能太长 | HTML 文档 |
| 按标题层级 | <h1>~<h6> 切分 | 最佳语义结构 | 需 HTML 解析 | 课程文档/书籍 |
| 递归分块 | 先按大段切，过长则递归切 | 灵活 | 实现复杂 | 大型文档 |
| 语义分块 | 用模型判断语义边界 | 最智能 | 成本高 | 高质量需求 |
推荐：对 HTML 课程文档使用"按标题层级 + 段落内的句子切分"，既保持了语义结构，又控制了块大小。
向量数据库（Vector DB）选型与性能
主流向量数据库对比
mermaid
graph TD
subgraph "专用向量数据库"
MILVUS["Milvus<br/>⚡ 高性能分布式<br/>HNSW/IVF/DiskANN"]
QDRANT["Qdrant<br/>🦀 Rust 实现<br/>量化索引 + 过滤"]
WEAVIATE["Weaviate<br/>🔌 GraphQL 原生<br/>混合搜索内置"]
end
subgraph "数据库扩展"
PGVECTOR["PGVector<br/>🐘 PostgreSQL 扩展<br/>IVFFlat / HNSW"]
ELASTIC["Elasticsearch<br/>🔍 kNN + BM25<br/>8.x+ 原生支持"]
REDIS["Redis Stack<br/>⚡ 极低延迟<br/>FLAT / HNSW"]
end
style MILVUS fill:#3498db,color:#fff
style PGVECTOR fill:#2ecc71,color:#fff关键性能指标对比
| 数据库 | 索引算法 | 10万向量 建索引 | 100万向量 召回QPS | 过滤支持 | 分布式 | 适合场景 |
|---|---|---|---|---|---|---|
| Milvus | HNSW / IVF / DiskANN | ~5s | 5000+ | 标量+向量 | ✅ 原生 | 大规模生产 |
| Qdrant | HNSW + 量化 | ~3s | 3000+ | 强过滤 | ✅ | 中大型项目 |
| PGVector | IVFFlat / HNSW | ~30s | 500+ | SQL 原生 | ❌ (PG 层面) | 已有 PG 的小项目 |
| Elasticsearch | HNSW | ~10s | 2000+ | 全文+向量 | ✅ 原生 | 已有 ES 的团队 |
| Redis Stack | FLAT / HNSW | ~2s | 5000+ | 简单 Tag | ❌ | 低延迟缓存层 |
召回延迟实测框架
python
"""
向量数据库性能基准测试框架
测试维度：
1. 索引构建速度
2. 不同数据量下的召回延迟 (P50/P95/P99)
3. 召回率 (Recall@K)
"""
import time
import numpy as np
from dataclasses import dataclass
from typing import List, Tuple
@dataclass
class SearchResult:
"""单次检索结果"""
ids: List[int] # 召回 ID 列表
scores: List[float] # 相似度分数
latency_ms: float # 检索延迟
@dataclass
class BenchmarkReport:
"""基准测试报告"""
db_name: str # 数据库名称
index_time_s: float # 建索引耗时（秒）
num_vectors: int # 向量总数
dim: int # 向量维度
# 延迟指标
p50_latency_ms: float # P50 延迟
p95_latency_ms: float # P95 延迟
p99_latency_ms: float # P99 延迟
# 吞吐指标
qps: float # 每秒查询数
# 质量指标
recall_at_10: float # Recall@10
recall_at_100: float # Recall@100
class VectorDBBenchmark:
"""
向量数据库通用 Benchmark 框架
使用方法：
bench = VectorDBBenchmark()
bench.setup(dim=768, num_vectors=100_000)
bench.run_search_test(top_k=10, num_queries=1000)
report = bench.report()
"""
def __init__(self):
self.vectors: np.ndarray = None # 待索引向量 (N, dim)
self.queries: np.ndarray = None # 查询向量 (M, dim)
self.ground_truth: np.ndarray = None # 真实 Top-K (M, K)
self.dim = 0
def setup(self, dim: int = 768, num_vectors: int = 100_000,
num_queries: int = 1000):
"""生成测试数据：随机向量 + 少量扰动作为 ground truth"""
self.dim = dim
# 生成归一化向量（模拟 Embedding 输出）
self.vectors = np.random.randn(num_vectors, dim).astype(np.float32)
self.vectors /= np.linalg.norm(self.vectors, axis=1, keepdims=True)
# 查询向量：从已有向量中随机选取并加噪声（模拟真实查询）
indices = np.random.choice(num_vectors, num_queries, replace=False)
self.queries = self.vectors[indices] + np.random.randn(num_queries, dim).astype(np.float32) * 0.1
self.queries /= np.linalg.norm(self.queries, axis=1, keepdims=True)
# Ground truth: 暴力计算余弦相似度
similarities = self.queries @ self.vectors.T # (M, N)
self.ground_truth = np.argsort(-similarities, axis=1) # 降序
def benchmark_index_build(self, build_fn) -> float:
"""测试索引构建速度"""
start = time.perf_counter()
build_fn(self.vectors)
return time.perf_counter() - start
def benchmark_search(self, search_fn, top_k: int = 10,
num_queries: int = 1000) -> Tuple[List[float], float]:
"""测试检索延迟和 QPS"""
latencies = []
for i in range(num_queries):
start = time.perf_counter()
_ = search_fn(self.queries[i], top_k)
latencies.append((time.perf_counter() - start) * 1000) # ms
latencies = np.array(latencies)
total_time = latencies.sum() / 1000 # 转为秒
qps = num_queries / total_time if total_time > 0 else float('inf')
return latencies, qps
def calculate_recall(self, search_fn, top_k: int = 10,
num_queries: int = 100) -> float:
"""计算 Recall@K"""
total_hits = 0
for i in range(num_queries):
result = search_fn(self.queries[i], top_k)
retrieved_ids = set(result.ids)
true_ids = set(self.ground_truth[i][:top_k])
total_hits += len(retrieved_ids & true_ids)
return total_hits / (num_queries * top_k)
def report(self, db_name: str, build_fn, search_fn,
top_k: int = 10, num_search_queries: int = 1000) -> BenchmarkReport:
"""生成完整的基准测试报告"""
print(f"🔬 测试 {db_name} ...")
# 建索引
index_time = self.benchmark_index_build(build_fn)
print(f" 索引构建: {index_time:.2f}s")
# 检索性能
latencies, qps = self.benchmark_search(search_fn, top_k, num_search_queries)
print(f" QPS: {qps:.0f}, P50: {np.percentile(latencies, 50):.1f}ms, "
f"P99: {np.percentile(latencies, 99):.1f}ms")
# 召回率
recall = self.calculate_recall(search_fn, top_k, min(100, num_search_queries))
print(f" Recall@{top_k}: {recall:.3f}")
return BenchmarkReport(
db_name=db_name,
index_time_s=index_time,
num_vectors=len(self.vectors),
dim=self.dim,
p50_latency_ms=np.percentile(latencies, 50),
p95_latency_ms=np.percentile(latencies, 95),
p99_latency_ms=np.percentile(latencies, 99),
qps=qps,
recall_at_10=recall,
recall_at_100=0.0 # 需要额外测试
)多路召回（Hybrid Search）与重排（Rerank）
为什么需要混合检索？
纯向量检索的局限性：
- 语义漂移：用户问"苹果公司股价"，可能召回关于"吃苹果"的内容
- 精确匹配弱：无法高效处理"型号 A12345" 这类精确匹配查询
- 关键词缺失：长尾关键词在向量空间中可能被稀释
mermaid
graph TD
QUERY["用户查询<br/>'MacBook Pro M3 散热怎么样?'"] --> VEC["向量检索<br/>语义相似度"]
QUERY --> BM25["BM25 全文检索<br/>关键词匹配"]
VEC --> VEC_R["语义结果<br/>'笔记本散热设计'<br/>'Apple Silicon 功耗'"]
BM25 --> BM25_R["关键词结果<br/>'MacBook Pro M3评测'<br/>'M3 散热测试'"]
VEC_R --> FUSION["🔄 结果融合<br/>RRF (Reciprocal Rank Fusion)"]
BM25_R --> FUSION
FUSION --> MERGED["融合结果 Top-20<br/>向量 + 关键词互补"]
MERGED --> RERANK["📊 重排序<br/>BGE-Reranker / Cohere"]
RERANK --> FINAL["最终 Top-5<br/>精准上下文"]
style FUSION fill:#f39c12,color:#fff
style RERANK fill:#e74c3c,color:#fff
style FINAL fill:#2ecc71,color:#fffRRF（Reciprocal Rank Fusion）实现
python
"""
混合检索：BM25 + 向量检索的融合与重排
"""
import numpy as np
from typing import List, Tuple
def reciprocal_rank_fusion(
vector_results: List[Tuple[str, float]], # [(doc_id, vector_score), ...]
bm25_results: List[Tuple[str, float]], # [(doc_id, bm25_score), ...]
k: int = 60, # RRF 平滑参数
top_k: int = 20,
) -> List[Tuple[str, float]]:
"""
RRF 多路融合算法
原理：
对每个文档，累加其在各路召回中的"倒数排名"：
RRF_score(doc) = Σ 1/(k + rank_i(doc))
其中 k 是平滑参数（典型值 60），rank_i 是文档在第 i 路召回中的排名。
优点：
- 不需要归一化各路分数（不同召回源的分数量级不同）
- 简单高效，适合在线计算
- 自动给予"多路都排名靠前"的文档更高分数
"""
scores = {}
# 向量检索结果：rank 越小越好
for rank, (doc_id, _) in enumerate(vector_results):
scores[doc_id] = scores.get(doc_id, 0) + 1.0 / (k + rank + 1)
# BM25 结果
for rank, (doc_id, _) in enumerate(bm25_results):
scores[doc_id] = scores.get(doc_id, 0) + 1.0 / (k + rank + 1)
# 按 RRF 分数排序
ranked = sorted(scores.items(), key=lambda x: x[1], reverse=True)
return ranked[:top_k]
# ========== 加权融合（备选方案） ==========
def weighted_score_fusion(
vector_results: List[Tuple[str, float]],
bm25_results: List[Tuple[str, float]],
vector_weight: float = 0.7, # 向量检索权重
top_k: int = 20,
) -> List[Tuple[str, float]]:
"""
加权分数融合
需要先对各路分数做归一化（如 min-max 归一化）
"""
# 归一化
def normalize(results: List[Tuple[str, float]]) -> dict:
if not results:
return {}
scores = np.array([s for _, s in results])
min_s, max_s = scores.min(), scores.max()
if max_s == min_s:
return {doc_id: 0.5 for doc_id, _ in results}
return {doc_id: (s - min_s) / (max_s - min_s) for doc_id, s in results}
vec_norm = normalize(vector_results)
bm25_norm = normalize(bm25_results)
# 融合
all_docs = set(vec_norm.keys()) | set(bm25_norm.keys())
fused = {}
for doc_id in all_docs:
fused[doc_id] = (
vector_weight * vec_norm.get(doc_id, 0) +
(1 - vector_weight) * bm25_norm.get(doc_id, 0)
)
return sorted(fused.items(), key=lambda x: x[1], reverse=True)[:top_k]Rerank 模型的作用
重排序模型对融合后的候选文档进行精细打分，大幅提升最终召回精度：
python
"""
使用 BGE-Reranker 进行重排序
Rerank 模型与 Embedding 模型的区别：
- Embedding 模型：独立编码 query 和 doc 为向量，通过余弦距离计算相似度
- Rerank 模型：将 query 和 doc 拼接后共同编码，能捕获更深层的语义交互
"""
from typing import List
class Reranker:
"""
Rerank 模型接口
典型选择：
- BGE-Reranker-v2-m3 (BAAI): 多语言，开源，效果好
- Cohere Rerank API: 商业，API 调用
- Jina Reranker: 支持更长文本
"""
def __init__(self, model_name: str = "BAAI/bge-reranker-v2-m3"):
self.model_name = model_name
# 实际使用时需要加载模型：
# from transformers import AutoModelForSequenceClassification, AutoTokenizer
# self.model = AutoModelForSequenceClassification.from_pretrained(model_name)
# self.tokenizer = AutoTokenizer.from_pretrained(model_name)
def rerank(
self,
query: str,
documents: List[str],
top_k: int = 5,
) -> List[Tuple[int, float]]:
"""
对候选文档进行重排序
Args:
query: 用户查询
documents: 候选文档列表（已通过混合召回获取）
top_k: 返回的文档数
Returns:
[(doc_index, relevance_score), ...] 按相关性降序排列
"""
# 为每对 (query, doc) 计算相关性分数
# 实际实现中调用模型：
# inputs = self.tokenizer(
# [(query, doc) for doc in documents],
# padding=True, truncation=True, return_tensors="pt",
# max_length=512
# )
# scores = self.model(**inputs).logits.squeeze(-1)
# 模拟实现
scores = [
0.98 - i * 0.03 + np.random.uniform(-0.02, 0.02)
for i in range(len(documents))
]
# 返回排序后的结果
ranked = sorted(
enumerate(scores),
key=lambda x: x[1],
reverse=True
)
return ranked[:top_k]
# ========== 完整检索流程 ==========
class HybridRAGRetriever:
"""混合检索 + 重排的完整 RAG 检索器"""
def __init__(self, vector_db, bm25_index, reranker: Reranker = None):
self.vector_db = vector_db # 向量数据库
self.bm25_index = bm25_index # BM25 索引（如 Elasticsearch）
self.reranker = reranker or Reranker()
def retrieve(self, query: str, top_k: int = 5) -> List[str]:
"""
完整检索流程：
1. 向量检索 → Top-50
2. BM25 检索 → Top-50
3. RRF 融合 → Top-20
4. Rerank 精排 → Top-5
"""
# Step 1 & 2: 两路召回
vec_results = self._vector_search(query, top_k=50)
bm25_results = self._bm25_search(query, top_k=50)
# Step 3: RRF 融合
fused = reciprocal_rank_fusion(vec_results, bm25_results, top_k=20)
# Step 4: Rerank 精排
doc_texts = [self._fetch_document(doc_id) for doc_id, _ in fused]
reranked = self.reranker.rerank(query, doc_texts, top_k=top_k)
return [doc_texts[i] for i, _ in reranked]
def _vector_search(self, query, top_k):
# 获取查询的 Embedding 向量，然后检索
# query_vec = self.embedding_model.encode(query)
# return self.vector_db.search(query_vec, top_k)
return [] # 模拟
def _bm25_search(self, query, top_k):
# return self.bm25_index.search(query, size=top_k)
return [] # 模拟
def _fetch_document(self, doc_id):
# return self.document_store.get(doc_id)
return "" # 模拟各检索方案准确率对比
| 方案 | Recall@10 | Recall@50 | MRR | 延迟 | 说明 |
|---|---|---|---|---|---|
| 纯 BM25 | 0.62 | 0.74 | 0.55 | <10ms | 关键词匹配 |
| 纯向量 (cosine) | 0.78 | 0.88 | 0.70 | <20ms | 语义匹配 |
| BM25 + 向量 (加权) | 0.85 | 0.93 | 0.78 | <30ms | 互补 |
| BM25 + 向量 (RRF) | 0.88 | 0.95 | 0.82 | <30ms | 推荐 |
| 以上 + Rerank | 0.94 | 0.98 | 0.91 | <200ms | 质量最高 |
结论：RRF 融合 + Rerank 重排是目前工程上性价比最高的方案。增量成本（~150ms）换来显著的准确率提升（+6-9% Recall@10）。
RAG 全链路工程架构
mermaid
graph TD
subgraph "数据管道 (离线)"
DOCS["📚 文档源<br/>HTML/Markdown/PDF"] --> PARSE["解析器<br/>HTMLChunkParser"]
PARSE --> CHUNKS["分块<br/>(语义边界)"]
CHUNKS --> EMB["Embedding 模型<br/>text-embedding-3-large"]
EMB --> VEC_DB["Vector DB<br/>Milvus/Qdrant"]
CHUNKS --> INV_IDX["倒排索引<br/>Elasticsearch/Whoosh"]
end
subgraph "在线服务"
API["用户 API"] --> QEMB["Query Embedding"]
QEMB --> HYBRID["Hybrid Search<br/>向量 + BM25"]
HYBRID --> RERANK["Reranker 精排"]
RERANK --> PROMPT["构建 Prompt<br/>System + Context + Query"]
PROMPT --> LLM["LLM 生成"]
LLM --> API
end
style DOCS fill:#3498db,color:#fff
style VEC_DB fill:#2ecc71,color:#fff
style HYBRID fill:#f39c12,color:#fff
style LLM fill:#9b59b6,color:#fff实战：连接主流向量数据库
Milvus — 高性能分布式向量数据库
python
"""Milvus 实战：建库、插入、检索完整流程"""
from pymilvus import (
connections, Collection, CollectionSchema,
FieldSchema, DataType, utility,
)
import numpy as np
# ===== 1. 连接 Milvus =====
# 本地模式（开发测试）
connections.connect(host="localhost", port="19530")
# 云端模式（Zilliz Cloud）
# connections.connect(
# uri="https://xxx.zillizcloud.com",
# token="your_api_key"
# )
# ===== 2. 定义 Collection Schema =====
fields = [
FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=65535),
FieldSchema(name="title", dtype=DataType.VARCHAR, max_length=512),
FieldSchema(name="section", dtype=DataType.VARCHAR, max_length=256),
FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=768), # Embedding 维度
]
schema = CollectionSchema(
fields=fields,
description="Wiki 知识库向量索引",
enable_dynamic_field=False,
)
# ===== 3. 创建 Collection =====
collection_name = "wiki_docs"
if utility.has_collection(collection_name):
utility.drop_collection(collection_name)
collection = Collection(name=collection_name, schema=schema)
# ===== 4. 创建索引 =====
index_params = {
"index_type": "HNSW", # 或 "IVF_FLAT", "IVF_SQ8", "DISKANN"
"metric_type": "COSINE", # 余弦相似度
"params": {
"M": 16, # HNSW 每层最大连接数（越大召回越高，建索引越慢）
"efConstruction": 200, # 构建时搜索宽度
},
}
collection.create_index(field_name="embedding", index_params=index_params)
# ===== 5. 批量插入数据 =====
def insert_chunks(chunks, embedding_model, batch_size=100):
"""批量插入文档块"""
for i in range(0, len(chunks), batch_size):
batch = chunks[i:i + batch_size]
# 向量化
texts = [c.text for c in batch]
embeddings = embedding_model.encode(texts)
# 准备数据
data = [
[c.text for c in batch],
[c.title for c in batch],
[c.section for c in batch],
embeddings.tolist(),
]
# 插入
mr = collection.insert(data)
print(f" 插入 {min(i + batch_size, len(chunks))}/{len(chunks)} 条, "
f"ID: {mr.primary_keys[0]} ~ {mr.primary_keys[-1]}")
collection.flush()
print(f"Collection 总数: {collection.num_entities}")
# ===== 6. 加载到内存（搜索前必须执行）=====
collection.load()
# ===== 7. 向量检索 =====
def search(query_text: str, embedding_model, top_k: int = 5):
"""检索相关文档"""
query_vec = embedding_model.encode_query(query_text)
search_params = {
"metric_type": "COSINE",
"params": {"ef": 64}, # 搜索时宽度（越大召回越高，速度越慢）
}
results = collection.search(
data=[query_vec.tolist()],
anns_field="embedding",
param=search_params,
limit=top_k,
output_fields=["text", "title", "section"], # 返回这些字段
)
docs = []
for hits in results:
for hit in hits:
docs.append({
"id": hit.id,
"score": hit.score,
"text": hit.entity.get("text", ""),
"title": hit.entity.get("title", ""),
"section": hit.entity.get("section", ""),
})
return docs
# ===== 8. 混合检索：向量 + 标量过滤 =====
def search_with_filter(query_text, embedding_model, title_filter: str = None):
"""带标量过滤的向量检索"""
query_vec = embedding_model.encode_query(query_text)
expr = None
if title_filter:
expr = f'title like "%{title_filter}%"'
results = collection.search(
data=[query_vec.tolist()],
anns_field="embedding",
param={"metric_type": "COSINE", "params": {"ef": 64}},
limit=10,
expr=expr, # 标量过滤表达式
output_fields=["text", "title"],
)
return [
{"id": hit.id, "score": hit.score, "text": hit.entity.get("text", "")}
for hits in results for hit in hits
]
# ===== 9. 管理操作 =====
# 查看 collection 信息
print(f"索引状态: {utility.index_building_progress(collection_name)}")
print(f"已加载: {utility.load_state(collection_name)}")
print(f"加载进度: {utility.loading_progress(collection_name)}")
# 释放内存
# collection.release()
# 删除 collection
# utility.drop_collection(collection_name)Qdrant — Rust 实现的高性能向量数据库
python
"""Qdrant 实战：建库、插入、检索完整流程"""
from qdrant_client import QdrantClient
from qdrant_client.models import (
Distance, VectorParams, PointStruct,
Filter, FieldCondition, MatchText,
)
import numpy as np
# ===== 1. 连接 Qdrant =====
# 本地模式
client = QdrantClient(host="localhost", port=6333)
# 内存模式（测试用）
# client = QdrantClient(":memory:")
# 云端模式
# client = QdrantClient(
# url="https://xxx.cloud.qdrant.io",
# api_key="your_api_key",
# )
# ===== 2. 创建 Collection =====
collection_name = "wiki_docs"
vector_size = 768 # Embedding 维度
if client.collection_exists(collection_name):
client.delete_collection(collection_name)
client.create_collection(
collection_name=collection_name,
vectors_config=VectorParams(
size=vector_size,
distance=Distance.COSINE, # 余弦距离
on_disk=True, # 向量存磁盘（大规模数据时）
),
# HNSW 索引参数
hnsw_config={
"m": 16,
"ef_construct": 200,
},
# 量化（减少内存占用）
quantization_config={
"scalar": {
"type": "int8",
"quantile": 0.99,
"always_ram": True,
}
},
)
# ===== 3. 批量插入数据 =====
def insert_chunks_qdrant(chunks, embedding_model, batch_size=100):
"""批量插入文档块到 Qdrant"""
offset = 0
for i in range(0, len(chunks), batch_size):
batch = chunks[i:i + batch_size]
# 向量化
texts = [c.text for c in batch]
embeddings = embedding_model.encode(texts)
# 构建 points
points = [
PointStruct(
id=offset + j,
vector=emb.tolist(),
payload={
"text": batch[j].text,
"title": batch[j].title,
"section": batch[j].section,
"char_count": len(batch[j].text),
},
)
for j, emb in enumerate(embeddings)
]
client.upsert(collection_name=collection_name, points=points)
offset += len(batch)
print(f" 插入 {offset}/{len(chunks)} 条")
print(f"Collection 总数: {client.count(collection_name).count}")
# ===== 4. 向量检索 =====
def search_qdrant(query_text: str, embedding_model, top_k: int = 5):
"""检索相关文档"""
query_vec = embedding_model.encode_query(query_text)
results = client.search(
collection_name=collection_name,
query_vector=query_vec.tolist(),
limit=top_k,
with_payload=True, # 返回 payload 中的字段
)
docs = []
for hit in results:
docs.append({
"id": hit.id,
"score": hit.score,
"text": hit.payload.get("text", ""),
"title": hit.payload.get("title", ""),
"section": hit.payload.get("section", ""),
})
return docs
# ===== 5. 带过滤的检索 =====
def search_qdrant_with_filter(
query_text: str,
embedding_model,
title_keyword: str = None,
top_k: int = 5,
):
"""带 payload 过滤的检索"""
query_vec = embedding_model.encode_query(query_text)
# 构建过滤器
query_filter = None
if title_keyword:
query_filter = Filter(
must=[
FieldCondition(
key="title",
match=MatchText(text=title_keyword),
)
]
)
results = client.search(
collection_name=collection_name,
query_vector=query_vec.tolist(),
limit=top_k,
query_filter=query_filter,
with_payload=True,
)
return [
{"id": hit.id, "score": hit.score, "text": hit.payload.get("text", "")}
for hit in results
]
# ===== 6. 管理操作 =====
# 查看 collection 信息
info = client.get_collection(collection_name)
print(f"向量数: {info.vectors_count}")
print(f"索引向量数: {info.indexed_vectors_count}")
# 更新 payload（不重新向量化）
# client.set_payload(
# collection_name=collection_name,
# payload={"updated_at": "2024-01-01"},
# points=[1, 2, 3],
# )
# 删除指定 point
# client.delete(collection_name=collection_name, points_selector=[1, 2])
# 清空但不删除 collection
# client.delete(collection_name=collection_name, points_selector=None)Chroma — 轻量级嵌入式向量数据库（适合原型）
python
"""Chroma 实战：零配置的向量数据库"""
import chromadb
from chromadb.utils import embedding_functions
# ===== 1. 初始化 =====
# 持久化模式
client = chromadb.PersistentClient(path="./data/chroma_db")
# 内存模式（临时）
# client = chromadb.Client()
# ===== 2. 创建 Collection =====
# 使用内置 Embedding 函数（可自动向量化）
embedding_fn = embedding_functions.SentenceTransformerEmbeddingFunction(
model_name="BAAI/bge-small-zh-v1.5",
device="cpu",
)
collection = client.get_or_create_collection(
name="wiki_docs",
metadata={"hnsw:space": "cosine"},
embedding_function=embedding_fn, # 可选：自动向量化
)
# ===== 3. 批量插入（Chroma 可自动向量化） =====
def insert_chunks_chroma(chunks, batch_size=100):
"""插入文档块"""
for i in range(0, len(chunks), batch_size):
batch = chunks[i:i + batch_size]
collection.add(
ids=[c.chunk_id for c in batch],
documents=[c.text for c in batch], # Chroma 自动向量化
metadatas=[c.metadata for c in batch],
)
print(f" 插入 {min(i + batch_size, len(chunks))}/{len(chunks)}")
# ===== 4. 检索 =====
def search_chroma(query_text: str, top_k: int = 5):
"""检索相关文档"""
results = collection.query(
query_texts=[query_text], # Chroma 自动向量化查询
n_results=top_k,
include=["documents", "metadatas", "distances"],
)
docs = []
for i in range(len(results['ids'][0])):
docs.append({
"id": results['ids'][0][i],
"score": 1.0 - results['distances'][0][i], # 距离 → 相似度
"text": results['documents'][0][i],
"metadata": results['metadatas'][0][i],
})
return docsMilvus vs Qdrant vs Chroma 选型决策树
mermaid
graph TD
START["选择向量数据库"] --> Q1{"数据量？"}
Q1 -->|"< 10万 向量"| CHROMA["Chroma<br/>零配置，原型开发"]
Q1 -->|"10万 - 100万"| Q2{"需要独立部署？"}
Q1 -->|"> 100万"| MILVUS["Milvus<br/>分布式，磁盘索引"]
Q2 -->|是| QDRANT["Qdrant<br/>Rust 高性能，过滤强"]
Q2 -->|否| Q3{"已有 PostgreSQL？"}
Q3 -->|是| PGVEC["PGVector<br/>SQL 原生，运维简单"]
Q3 -->|否| QDRANT
style CHROMA fill:#2ecc71,color:#fff
style QDRANT fill:#f39c12,color:#fff
style MILVUS fill:#e74c3c,color:#fff
style PGVEC fill:#3498db,color:#fffRAG Pipeline 性能优化清单
| 优化项 | 方法 | 效果 |
|---|---|---|
| Embedding 批量化 | 对文档批量编码而非逐条 | 吞吐提升 10-50x |
| 索引预热 | 服务启动时加载热门向量到缓存 | P99 延迟降低 50% |
| Chunk 预取 | 检索时顺带拉取相邻 Chunk | Recall +5% |
| 查询重写 | LLM 先改写/扩展用户查询 | Recall +10-15% |
| 结果去重 | 基于 Chunk hash 去重 | 节省 Token，提升质量 |
| 动态 Top-K | 根据查询复杂度动态调整 K | 平衡延迟与质量 |
| 缓存热门查询 | Redis 缓存查询→结果映射 | 命中情况下延迟 <5ms |
Agentic RAG — Agent 驱动的智能检索
核心思想
传统 RAG 是"检索→生成"的单向流水线，检索逻辑固定。Agentic RAG 将 Agent 的推理能力注入检索过程，让检索策略根据问题动态调整：
mermaid
graph LR
subgraph "传统 RAG"
Q1["用户问题"] --> R1["固定检索<br/>Top-K = 5"] --> G1["LLM 生成"]
end
subgraph "Agentic RAG"
Q2["用户问题"] --> AGENT["Agent 决策<br/>分析问题类型"]
AGENT -->|"需要多步推理"| MR["多轮检索<br/>先查问题定义<br/>再查相关代码"]
AGENT -->|"简单事实"| SR["单次检索<br/>Top-K = 3"]
AGENT -->|"需要对比"| CR["并行检索<br/>各自查询对比源"]
MR --> GEN["LLM 综合生成"]
SR --> GEN
CR --> GEN
end
style AGENT fill:#e74c3c,color:#fff
style GEN fill:#3498db,color:#fff实现方案
python
"""
Agentic RAG — Agent 动态决策检索策略
"""
from dataclasses import dataclass
from typing import List, Dict, Optional
from enum import Enum
class RetrievalStrategy(Enum):
SINGLE = "single" # 单次检索
MULTI_HOP = "multi_hop" # 多跳检索（上一步结果驱动下一步）
PARALLEL = "parallel" # 并行多路检索
ITERATIVE = "iterative" # 迭代检索（逐步细化）
@dataclass
class RetrievalPlan:
"""检索计划 — Agent 决策输出"""
strategy: RetrievalStrategy
queries: List[str] # 单次/并行检索的查询列表
top_k: int = 5
reasoning: str = "" # Agent 的决策理由
# 多跳专用
follow_up_queries: List[str] = None # 多跳的后续查询模板
class AgenticRAG:
"""Agentic RAG 引擎"""
def __init__(self, retriever, llm_client):
self.retriever = retriever
self.llm = llm_client
async def plan_retrieval(self, question: str, history: List[Dict]) -> RetrievalPlan:
"""Agent 分析问题，输出检索计划"""
prompt = f"""分析以下问题，制定检索计划。
问题: {question}
请按 JSON 格式输出检索计划:
{{
"strategy": "single|multi_hop|parallel|iterative",
"queries": ["检索查询1", "检索查询2"],
"top_k": 5,
"reasoning": "决策理由",
"follow_up_queries": ["多跳后续查询模板", ...]
}}
"""
response = await self.llm.chat(prompt, response_format="json")
plan = json.loads(response)
return RetrievalPlan(**plan)
async def execute_retrieval(self, plan: RetrievalPlan) -> List[Document]:
"""执行检索计划"""
if plan.strategy == RetrievalStrategy.SINGLE:
return await self.retriever.search(plan.queries[0], top_k=plan.top_k)
elif plan.strategy == RetrievalStrategy.PARALLEL:
# 并行检索多路查询
results = await asyncio.gather(*[
self.retriever.search(q, top_k=plan.top_k)
for q in plan.queries
])
return self._merge_and_rerank(results)
elif plan.strategy == RetrievalStrategy.MULTI_HOP:
# 第一跳检索
docs = await self.retriever.search(plan.queries[0], top_k=plan.top_k)
# 用第一跳结果生成第二跳查询
context = "\n".join([d.content[:200] for d in docs])
second_query = await self._generate_followup(
plan.follow_up_queries[0], context
)
# 第二跳检索
more_docs = await self.retriever.search(second_query, top_k=plan.top_k)
return docs + more_docs
elif plan.strategy == RetrievalStrategy.ITERATIVE:
all_docs = []
for i, query in enumerate(plan.queries):
docs = await self.retriever.search(query, top_k=plan.top_k)
all_docs.extend(docs)
if i < len(plan.queries) - 1:
# 用当前结果优化下一个查询
context = "\n".join([d.content[:100] for d in docs])
plan.queries[i + 1] = await self._refine_query(
plan.queries[i + 1], context
)
return self._merge_and_rerank(all_docs)
async def answer(self, question: str) -> Dict:
"""完整的 Agentic RAG 流程"""
plan = await self.plan_retrieval(question, [])
docs = await self.execute_retrieval(plan)
context = "\n\n".join([d.content for d in docs])
answer = await self.llm.chat(f"""根据以下检索到的资料回答问题。
资料:
{context}
问题: {question}
如果资料不足以回答问题，请明确说明。""")
return {
"answer": answer,
"plan": plan,
"retrieved_docs": len(docs),
}
def _merge_and_rerank(self, results: List[List[Document]]) -> List[Document]:
"""合并多路结果并去重排序"""
seen = set()
merged = []
for docs in results:
for d in docs:
if d.id not in seen:
seen.add(d.id)
merged.append(d)
return sorted(merged, key=lambda d: d.score, reverse=True)
async def _generate_followup(self, template: str, context: str) -> str:
"""根据第一跳结果生成第二跳查询"""
response = await self.llm.chat(
f"根据以下上下文，将查询模板具体化: {template}\n上下文: {context}"
)
return response
async def _refine_query(self, query: str, context: str) -> str:
"""根据已有结果优化查询"""
response = await self.llm.chat(
f"根据以下已有信息，优化搜索查询: {query}\n已有信息: {context}"
)
return responseAgentic RAG 的典型模式
| 模式 | 触发条件 | 示例 |
|---|---|---|
| 单次检索 | 简单事实查询 | "Python 的 GIL 是什么？" |
| 多跳检索 | 需要关联多步信息 | "这个 bug 是哪个 commit 引入的？先查 bug 再查 git log" |
| 并行检索 | 需要对比/多视角 | "对比 Milvus 和 Qdrant 的性能"（同时查两个） |
| 迭代检索 | 初始检索不充分 | 第一次 top-5 无结果 → 改写查询 → 再检索 |
| 自适应检索 | 问题复杂度不确定 | Agent 先评估问题，再决定检索深度 |
Graph RAG — 知识图谱增强检索
问题背景
传统 RAG 基于向量相似度检索，擅长语义匹配，但无法理解实体间的关系：
- "张三的上级是谁？" → 向量检索无法关联实体关系
- "哪个部门负责这个项目？" → 需要图谱的
BELONGS_TO关系 - "受影响的所有服务有哪些？" → 需要图遍历发现依赖链
Graph RAG 结合知识图谱和向量检索，补全实体关系推理能力。
架构设计
mermaid
graph TD
subgraph "Graph RAG 架构"
INPUT["用户问题"] --> EXTRACT["实体抽取<br/>LLM 识别实体+关系"]
EXTRACT --> VEC["向量检索<br/>语义相似文档"]
EXTRACT --> GRAPH["图谱检索<br/>Neo4j / NebulaGraph"]
VEC --> FUSE["结果融合<br/>去重 + 补全 + 排序"]
GRAPH --> FUSE
FUSE --> GEN["LLM 生成<br/>结合结构化+非结构化知识"]
end
style EXTRACT fill:#f39c12,color:#fff
style GEN fill:#2ecc71,color:#fff实现示例
python
"""
Graph RAG — 结合向量检索与知识图谱
"""
from typing import List, Dict, Set, Tuple
class GraphRAG:
"""图谱增强的 RAG 引擎"""
def __init__(self, vector_store, graph_db, llm_client):
"""
Args:
vector_store: 向量数据库（Milvus/Qdrant/Chroma）
graph_db: 图数据库连接（Neo4j/NebulaGraph）
llm_client: LLM 客户端
"""
self.vector_store = vector_store
self.graph = graph_db
self.llm = llm_client
async def retrieve(self, question: str, top_k: int = 5) -> Dict:
"""双路检索：向量 + 图谱"""
# 1. 实体抽取
entities = await self._extract_entities(question)
relations = await self._extract_relations(question)
# 2. 向量检索
vector_docs = await self.vector_store.search(question, top_k=top_k)
# 3. 图谱检索
graph_results = await self._graph_retrieval(entities, relations)
# 4. 结果融合
fused = self._fuse_results(vector_docs, graph_results)
return {
"vector_docs": vector_docs,
"graph_results": graph_results,
"fused_docs": fused,
"entities": entities,
"relations": relations,
}
async def _extract_entities(self, question: str) -> List[Dict]:
"""LLM 提取问题中的实体"""
prompt = f"""从以下问题中提取所有实体（人名、项目名、服务名、部门名等）。
返回 JSON 数组格式:
[{{"name": "实体名", "type": "person|project|service|department|...", "aliases": ["别名1"]}}]
问题: {question}"""
response = await self.llm.chat(prompt, response_format="json")
return json.loads(response)
async def _extract_relations(self, question: str) -> List[str]:
"""LLM 提取问题中隐含的关系类型"""
prompt = f"""分析以下问题需要查询的关系类型。
可选关系: BELONGS_TO, MANAGES, DEPENDS_ON, OWNS, PARTICIPATES_IN, REPORTS_TO
返回 JSON 数组: ["关系名1", "关系名2"]
问题: {question}"""
response = await self.llm.chat(prompt, response_format="json")
return json.loads(response)
async def _graph_retrieval(
self, entities: List[Dict], relations: List[str]
) -> List[Dict]:
"""在图数据库中检索实体及其关系"""
results = []
for entity in entities:
name = entity["name"]
etype = entity["type"]
# Cypher 查询：获取实体及其 1-hop 邻居
query = """
MATCH (n {name: $name})
OPTIONAL MATCH (n)-[r]->(m)
WHERE $relations IS NULL OR type(r) IN $relations
RETURN n, r, m
LIMIT 20
"""
records = await self.graph.run(
query,
name=name,
relations=relations if relations else None
)
for record in records:
results.append({
"entity": name,
"entity_type": etype,
"relation": record.get("r"),
"neighbor": record.get("m"),
})
# 去重
seen = set()
unique = []
for r in results:
key = f"{r['entity']}-{r.get('relation')}-{r.get('neighbor')}"
if key not in seen:
seen.add(key)
unique.append(r)
return unique
def _fuse_results(
self, vector_docs: List[Document], graph_results: List[Dict]
) -> List[Document]:
"""融合向量检索和图谱检索结果"""
# 将图谱关系转为文本块
graph_chunks = []
for item in graph_results:
text = f"实体 {item['entity']} ({item['entity_type']})"
if item.get("relation"):
text += f" → [{item['relation']}] → {item.get('neighbor')}"
graph_chunks.append(Document(
id=f"graph_{item['entity']}",
content=text,
source="knowledge_graph",
score=0.9, # 图谱结果置信度高
))
# 合并、去重、按分数排序
merged = vector_docs + graph_chunks
seen_ids = set()
fused = []
for doc in sorted(merged, key=lambda d: d.score, reverse=True):
if doc.id not in seen_ids:
seen_ids.add(doc.id)
fused.append(doc)
return fused
async def answer(self, question: str) -> str:
"""完整的 Graph RAG 问答"""
retrieval_result = await self.retrieve(question)
# 构建上下文
context_parts = []
# 图谱信息
if retrieval_result["graph_results"]:
context_parts.append("## 知识图谱关系")
for item in retrieval_result["graph_results"]:
context_parts.append(
f"- {item['entity']} [{item.get('relation', '相关')}] "
f"{item.get('neighbor', '')}"
)
# 向量检索的文档
context_parts.append("## 相关文档")
for doc in retrieval_result["fused_docs"][:5]:
context_parts.append(doc.content)
context = "\n\n".join(context_parts)
answer = await self.llm.chat(
f"根据以下信息回答问题，如果涉及实体关系请明确指出：\n\n{context}\n\n问题: {question}"
)
return answerGraph RAG vs Agentic RAG 对比
| 维度 | Graph RAG | Agentic RAG |
|---|---|---|
| 核心能力 | 实体关系推理 | 动态检索策略 |
| 适合场景 | 组织架构、依赖链、权限查询 | 复杂多步问题、需探索性检索 |
| 数据要求 | 需要构建知识图谱 | 向量数据库即可 |
| 实现复杂度 | 高（需维护图数据库） | 中 |
| 典型应用 | 企业知识库、运维故障定位 | 客服问答、代码审查、研究助手 |
最佳实践：两者可以结合使用 — Agent 动态决策何时用图谱检索、何时用向量检索，实现自适应混合 RAG。
登录后即可发表评论 👇