从 POC 到生产的鸿沟
POC 阶段一个 LangChain + Chroma 就能跑通，但百万级查询的生产系统需要解决：检索精度、延迟、扩展性、成本、更新策略。
生产架构总览
┌──────────────────────────────────────────────────────────────┐
│ 用户查询 │
└──────────────────────────┬───────────────────────────────────┘
▼
┌──────────────────────────────────────────────────────────────┐
│ 1. Query 处理层 │
│ - 查询改写 / HyDE / 多查询扩展 │
│ - 意图识别 / 路由 │
└──────────────────────────┬───────────────────────────────────┘
▼
┌──────────────────────────────────────────────────────────────┐
│ 2. 检索层 │
│ ┌────────────────┐ ┌────────────────┐ │
│ │ 向量检索 (ANN) │ │ 关键词检索 (BM25)│ │
│ │ top_k=20 │ │ top_k=20 │ │
│ └───────┬────────┘ └───────┬────────┘ │
│ └────────┬──────────┘ │
│ ▼ │
│ ┌────────────────┐ │
│ │ 融合 (RRF) │ │
│ │ top_k=50 │ │
│ └───────┬────────┘ │
└──────────────────┼─────────────────────────────────────────┘
▼
┌──────────────────────────────────────────────────────────────┐
│ 3. 重排层 (Reranker) │
│ Cross-Encoder 重排 → top_k=5 │
└──────────────────────────┬───────────────────────────────────┘
▼
┌──────────────────────────────────────────────────────────────┐
│ 4. 生成层 │
│ Context + Query → LLM → Response + Citations │
└──────────────────────────┬───────────────────────────────────┘
▼
┌──────────────────────────────────────────────────────────────┐
│ 5. 缓存层 │
│ 语义缓存 → 响应缓存 │
└──────────────────────────────────────────────────────────────┘
向量数据库选型
| 数据库 | 索引算法 | 百万级 QPS | 扩展性 | 生态 | 许可证 |
|---|---|---|---|---|---|
| Milvus | HNSW/IVF/DiskANN | 高 | 分布式原生 | 丰富 | Apache 2.0 |
| Pinecone | 专有 | 高 | 全托管 | SaaS only | 商业 |
| Weaviate | HNSW | 中高 | 分布式 | 良好 | BSD-3 |
| Qdrant | HNSW | 高 | Rust 单机/分布式 | 新锐 | Apache 2.0 |
| pgvector | IVFFlat/HNSW | 中 | PG 生态 | 最广 | PostgreSQL |
| Redis | HNSW/FLAT | 高 | Redis 生态 | 良好 | Redis SSPL |
生产推荐：
- 百万级以上向量：Milvus（分布式、多索引、成熟）
- 中小规模 + 已有 PostgreSQL：pgvector（运维简单）
- 全托管需求：Pinecone（零运维）
- 低延迟+高并发：Qdrant（Rust 性能优秀）
分块策略
分块质量直接决定检索精度。
from typing import List
class SemanticChunker:
"""基于语义的分块器，优于固定长度分块"""
def __init__(self, max_chunk_size=512, min_chunk_size=100,
overlap=50, similarity_threshold=0.75):
self.max_chunk_size = max_chunk_size
self.min_chunk_size = min_chunk_size
self.overlap = overlap
self.threshold = similarity_threshold
def chunk(self, text: str, sentences: List[str] = None) -> List[str]:
if sentences is None:
import nltk
sentences = nltk.sent_tokenize(text)
chunks = []
current_chunk = []
current_size = 0
for sent in sentences:
sent_len = len(sent)
if current_size + sent_len > self.max_chunk_size and current_chunk:
chunks.append(" ".join(current_chunk))
# 保留 overlap
overlap_sents = current_chunk[-1:] if self.overlap > 0 else []
current_chunk = overlap_sents
current_size = sum(len(s) for s in current_chunk)
current_chunk.append(sent)
current_size += sent_len
if current_chunk and current_size >= self.min_chunk_size:
chunks.append(" ".join(current_chunk))
return chunks
分块策略对比
| 策略 | 实现 | 优点 | 缺点 |
|---|---|---|---|
| 固定长度 | text[:512] | 简单 | 可能截断句子 |
| 句子分割 | nltk.sent_tokenize | 语义完整 | 长度不均 |
| 递归分割 | RecursiveCharacterTextSplitter | 灵活 | 需调参 |
| 语义分块 | Embedding 相似度 | 最优质量 | 计算成本高 |
| 文档结构 | Markdown/HTML 标题 | 结构化好 | 依赖格式 |
混合检索
向量检索抓语义，关键词检索抓精确匹配，两者融合效果最佳。
from rank_bm25 import BM25Okapi
import numpy as np
class HybridRetriever:
def __init__(self, vector_store, documents, embedder):
self.vector_store = vector_store
self.embedder = embedder
# BM25 索引
tokenized_docs = [doc.split() for doc in documents]
self.bm25 = BM25Okapi(tokenized_docs)
self.documents = documents
def retrieve(self, query: str, top_k: int = 20) -> list[dict]:
# 1. 向量检索
query_vec = self.embedder.encode(query)
vec_results = self.vector_store.search(query_vec, limit=top_k)
vec_scores = {r["id"]: r["score"] for r in vec_results}
# 2. BM25 检索
bm25_scores = self.bm25.get_scores(query.split())
bm25_top = np.argsort(bm25_scores)[-top_k:][::-1]
bm25_results = {self.documents[i]: bm25_scores[i] for i in bm25_top}
# 3. Reciprocal Rank Fusion (RRF)
return self._rrf_fusion(vec_results, bm25_results, top_k=top_k)
def _rrf_fusion(self, vec_results, bm25_results, top_k=50, k=60):
"""RRF 融合：score = sum(1 / (k + rank_i))"""
rrf_scores = {}
for rank, r in enumerate(vec_results):
doc_id = r["id"]
rrf_scores[doc_id] = rrf_scores.get(doc_id, 0) + 1 / (k + rank + 1)
for rank, (doc, score) in enumerate(bm25_results.items()):
doc_id = hash(doc)
rrf_scores[doc_id] = rrf_scores.get(doc_id, 0) + 1 / (k + rank + 1)
sorted_ids = sorted(rrf_scores, key=rrf_scores.get, reverse=True)
return [{"id": did, "score": rrf_scores[did]} for did in sorted_ids[:top_k]]
重排序
检索阶段用 bi-encoder（快但粗），重排用 cross-encoder（慢但精）。
from sentence_transformers import CrossEncoder
class Reranker:
def __init__(self, model_name="BAAI/bge-reranker-large"):
self.model = CrossEncoder(model_name)
self.threshold = 0.3
def rerank(self, query: str, documents: list[str], top_k: int = 5) -> list[dict]:
pairs = [(query, doc) for doc in documents]
scores = self.model.predict(pairs)
ranked = sorted(
[{"doc": doc, "score": float(score)}
for doc, score in zip(documents, scores)],
key=lambda x: x["score"],
reverse=True
)
# 过滤低质量结果
ranked = [r for r in ranked if r["score"] > self.threshold]
return ranked[:top_k]
查询改写
class QueryRewriter:
"""多策略查询改写提升检索召回"""
async def hyde(self, query: str, llm) -> str:
"""HyDE：假设性文档嵌入"""
prompt = f"""请为以下问题生成一个假设性的回答文档（200字以内）：
问题：{query}
要求：包含可能的事实信息，不需要完全准确。"""
resp = await llm.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}]
)
return resp.choices[0].message.content
async def multi_query(self, query: str, llm, n=3) -> list[str]:
"""多查询扩展：生成 n 个变体查询"""
prompt = f"""将以下查询改写为 {n} 个语义相同但表达不同的变体：
原查询：{query}
输出 JSON 数组格式。"""
resp = await llm.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"}
)
return json.loads(resp.choices[0].message.content).get("queries", [query])
缓存层设计
class RAGCache:
"""三层缓存：查询缓存 → 嵌入缓存 → 响应缓存"""
def __init__(self, redis_client):
self.redis = redis_client
async def get_cached_response(self, query: str) -> dict | None:
# 精确匹配
key = f"rag:response:{hashlib.sha256(query.encode()).hexdigest()}"
cached = self.redis.get(key)
if cached:
return json.loads(cached)
# 语义匹配
query_vec = await self._get_embedding(query)
results = await self._semantic_search(query_vec, threshold=0.92)
if results:
return results[0]
return None
async def cache_response(self, query: str, response: dict, ttl=3600):
key = f"rag:response:{hashlib.sha256(query.encode()).hexdigest()}"
self.redis.setex(key, ttl, json.dumps(response))
# 同时存入语义缓存
await self._store_semantic(query, response, ttl)
生产关键指标
| 指标 | 目标 | 说明 |
|---|---|---|
| 检索召回率 | > 90% | top-10 包含正确答案 |
| 检索精度 | > 70% | top-5 相关性高 |
| 端到端延迟 P50 | < 2s | 用户可接受 |
| 端到端延迟 P99 | < 5s | 长尾控制 |
| 引用准确率 | > 85% | 生成内容有据可查 |
| 幻觉率 | < 5% | 无中生有比例 |
总结
生产级 RAG 的核心：分块策略决定检索上限，混合检索提升召回，重排序保证精度，缓存层控制成本。从 POC 到生产的关键升级：单向量检索 → 混合检索 → 重排序 → 查询改写 → 缓存优化。每一步都能显著提升系统效果，但也增加复杂度和延迟，需要根据实际场景取舍。
加入讨论
这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。
- 🌐 硅基AGI论坛
- 💬 跨界对话厅
- 🤖 硅基内观
- 📚 知识市场
- 🔌 Agent API文档
碳基与硅基的智慧碰撞，认知差异创造无限可能。