
使用方式
关于
通过分块文档、生成嵌入向量、配置向量存储、构建混合搜索管道、应用重排序以及评估检索质量,设计和实现生产级 RAG 系统。用于构建 RAG 系统、向量数据库或知识增强 AI 应用。
RAG 架构师
核心工作流程
- 需求分析 — 识别检索需求、延迟约束、准确性要求和规模
- 向量存储设计 — 选择数据库、模式设计、索引策略、分片方案
- 分块策略 — 文档拆分、重叠、语义边界、元数据丰富
- 检索流水线 — 嵌入选择、查询转换、混合搜索、重排序
- 评估与迭代 — 指标跟踪、检索调试、持续优化
每个步骤在继续前进行验证(见下方检查点)。
参考指南
根据上下文加载详细指导:
| 主题 | 参考 | 加载时机 |
|------|------|----------|
| 向量数据库 | references/vector-databases.md | 比较 Pinecone、Weaviate、Chroma、pgvector、Qdrant |
| 嵌入模型 | references/embedding-models.md | 选择嵌入、微调、维度权衡 |
| 分块策略 | references/chunking-strategies.md | 文档拆分、重叠、语义分块 |
| 检索优化 | references/retrieval-optimization.md | 混合搜索、重排序、查询扩展、过滤 |
| RAG 评估 | references/rag-evaluation.md | 指标、评估框架、检索调试 |
实现示例
1. 文档分块
from langchain.text_splitter import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter(
chunk_size=800,
chunk_overlap=100,
separators=["\n\n", "\n", ". ", " "],
)
chunks = splitter.create_documents(
texts=[doc.page_content for doc in raw_docs],
metadatas=[{"source": doc.metadata["source"], "timestamp": doc.metadata.get("timestamp")} for doc in raw_docs],
)
检查点: assert all(c.metadata.get("source") for c in chunks), "Missing source metadata"
2. 生成嵌入与索引
from openai import OpenAI
import qdrant_client
from qdrant_client.models import VectorParams, Distance, PointStruct
client = OpenAI()
qdrant = qdrant_client.QdrantClient("localhost", port=6333)
qdrant.recreate_collection(
collection_name="knowledge_base",
vectors_config=VectorParams(size=1536, distance=Distance.COSINE),
)
def embed_chunks(chunks: list[str], model: str = "text-embedding-3-small") -> list[list[float]]:
response = client.embeddings.create(input=chunks, model=model)
return [r.embedding for r in response.data]
import hashlib, uuid
points = []
for i, chunk in enumerate(chunks):
doc_id = str(uuid.UUID(hashlib.md5(chunk.page_content.encode()).hexdigest()))
embedding = embed_chunks([chunk.page_content])[0]
points.append(PointStruct(id=doc_id, vector=embedding, payload=chunk.metadata))
qdrant.upsert(collection_name="knowledge_base", points=points)
检查点: assert qdrant.count("knowledge_base").count == len(set(p.id for p in points)), "Deduplication failed"
3. 混合搜索(向量 + BM25)
from qdrant_client.models import Filter, FieldCondition, MatchValue
from rank_bm25 import BM25Okapi
def hybrid_search(query: str, tenant_id: str, top_k: int = 20) -> list:
query_embedding = embed_chunks([query])[0]
tenant_filter = Filter(must=[FieldCondition(key="tenant_id", match=MatchValue(value=tenant_id))])
dense_results = qdrant.search(
collection_name="knowledge_base",
query_vector=query_embedding,
query_filter=tenant_filter,
limit=top_k,
)
corpus = [r.payload.get("text", "") for r in dense_results]
bm25 = BM25Okapi([doc.split() for doc in corpus])
bm25_scores = bm25.get_scores(query.split())
# Reciprocal Rank Fusion
ranked = sorted(
zip(dense_results, bm25_scores),
key=lambda x: 0.6 * x[0].score + 0.4 * x[1],
reverse=True,
)
return [r for r, _ in ranked[:top_k]]
4. 重排序 Top-K 结果
import cohere
co = cohere.Client("YOUR_API_KEY")
def rerank(query: str, results: list, top_n: int = 5) -> list:
docs = [r.payload.get("text", "") for r in results]
reranked = co.rerank(query=query, documents=docs, top_n=top_n, model="rerank-english-v3.0")
return [results[r.index] for r in reranked.results]
5. 检索评估
from ragas import evaluate
from ragas.metrics import context_precision, context_recall, faithfulness, answer_relevancy
from datasets import Dataset
eval_dataset = Dataset.from_dict({
"question": questions,
"contexts": retrieved_contexts,
"answer": generated_answers,
"ground_truth": ground_truths,
})
results = evaluate(eval_dataset, metrics=[context_precision, context_recall, faithfulness, answer_relevancy])
print(results)
约束
必须做
- 在领域数据上评估 chunk_size — 绝不盲目使用 512
- 使用确定性 ID 实现幂等 upsert 以去重
- 每个分块包含源元数据
- 验证检索结果非空后再传递给 LLM
- 使用评估框架(RAGAS 等)量化检索质量
- 多租户场景实施过滤
禁止做
- 未评估就选择嵌入模型
- 跳过数据验证检查点
- 在代码中硬编码 API 密钥
- 忽略检索延迟要求
- 未经重排序就将大量结果传递给 LLM
知识参考
RAG、向量数据库、Pinecone、Weaviate、Chroma、pgvector、Qdrant、LangChain、嵌入模型、BM25、混合搜索、重排序、RAGAS、Cohere Rerank
