一、RAG的技术价值与边界

RAG(Retrieval-Augmented Generation)解决的核心问题是:让大模型在回答时能够引用最新的或私有的知识,同时减少幻觉。相比于微调,RAG在知识更新频率高或知识库经常变化的场景下具有压倒性优势。

1.1 RAG vs 微调的选择

# 选择RAG还是微调?以下为决策树
#
# 知识是否经常变化?
#   是 → RAG(更新文档即可,无需重新训练)
#   否 → 继续判断
#
# 知识库是否私有/专有?
#   是(如内部文档)→ RAG(微调难以覆盖长尾知识)
#   否 → 继续判断
#
# 是否需要精确的格式/结构输出?
#   是 → 微调(RAG难以保证输出格式一致性)
#   否 → 继续判断
#
# 计算资源是否有限?
#   是 → RAG(不需要GPU训练)
#   否 → 两者均可考虑

# RAG的优势场景:
# ① 实时新闻/数据 → 知识库随时可更新
# ② 企业内部文档 → 无需训练,微调覆盖困难的长尾
# ③ 多语言/跨文档 → 向量检索天然支持跨语言语义匹配
# ④ 可溯源答案 → 检索结果可解释,答案附引用

二、向量化与向量数据库

2.1 Embedding模型选择

# 主流Embedding模型对比
# 模型                    维度   MRL语言  优势场景
# ─────────────────────────────────────────────────────
# text-embedding-ada-002  1536   多语言   OpenAI官方,稳定
# bge-large-zh-v1.5    1024   中文强   中文任务首选
# m3e-large            1024   中文强   中文+英文混合
# voyage-large-2        1024   多语言   通用最优

# 中文Embedding推荐:bge-large-zh-v1.5
from sentence_transformers import SentenceTransformer

model = SentenceTransformer('BAAI/bge-large-zh-v1.5')

# 单句向量化
query_embedding = model.encode("用户查询文本")
# shape: (1024,)

# 批量向量化文档
doc_embeddings = model.encode([
    "文档段落1内容...",
    "文档段落2内容...",
], batch_size=32, show_progress_bar=True)
# shape: (N, 1024)

2.2 向量数据库选型

# 主流向量数据库对比
# Milvus: 国产开源,支持混合检索,性能强,适合大规模
# Qdrant: Rust实现,延迟极低,支持过滤条件
# Weaviate: GraphQL接口,原生支持多模态
# Chroma: 轻量级,Python优先,适合原型

# Milvus Python SDK使用示例
from pymilvus import MilvusClient

client = MilvusClient(uri="./milvus_demo.db")

# 创建Collection(含HNSW索引)
client.create_collection(
    collection_name="knowledge_base",
    dimension=1024,
    index_params=[{
        "field_name": "embedding",
        "index_type": "HNSW",
        "metric_type": "IP",        # 内积相似度(适合归一化向量)
        "params": {
            "M": 16,                # HNSW参数:邻居数
            "efConstruction": 256    # 建索引精度
        }
    }]
)

# 插入数据
entities = [
    {"id": "doc_001", "text": "LLaMA是Meta发布的开源大语言模型..."},
    {"id": "doc_002", "text": "向量数据库用于存储和检索高维向量..."},
    {"embedding": doc_embeddings[0].tolist()},
    {"embedding": doc_embeddings[1].tolist()},
]
client.insert(collection_name="knowledge_base", data=entities)

# 检索
query_embedding = model.encode("LLaMA模型的特点是什么")
results = client.search(
    collection_name="knowledge_base",
    data=[query_embedding.tolist()],
    limit=5,  # 返回Top5
    output_fields=["text"]
)
# results[0] = [{id, distance, entity: {text}}]

三、Advanced RAG架构

3.1 完整RAG Pipeline

# 完整RAG Pipeline
#
# 用户问题
#   ↓
# ① Query理解(可选)
#   - HyDE:生成假设性答案→向量检索→双重检索
#   - Query改写:同义词扩展、问题分解
#   ↓
# ② 检索(Retrieval)
#   - 向量相似度 Top-K
#   - 可选:元数据过滤(时间、类别、权限)
#   ↓
# ③ 重排(Reranking)
#   - Cross-Encoder精排,过滤不相关结果
#   - 如:BAAI/bge-reranker-large
#   ↓
# ④ 上下文组装(Context Assembly)
#   - 将Top-K片段拼接为prompt
#   - 考虑上下文长度(LLM context window限制)
#   ↓
# ⑤ 生成(Generation)
#   - LLM基于检索结果生成答案
#   - System prompt:强调"仅基于提供的上下文回答"
#   ↓
# 答案 + 引用

# HyDE(Hypothetical Document Embeddings)
# 原理:用LLM生成一个"假设性答案"→ 检索假设答案的相似文档
# 效果:检索命中率提升10-20%(对复杂问句尤其明显)

# 示例Prompt
hyde_prompt = """请根据以下用户问题,生成一段假设性的回答。
这段回答将用于向量检索,请确保回答准确、具体。

问题:{query}

假设性回答:"""

# Step 1: 生成假设答案
hypothetical_answer = llm.invoke(hyde_prompt)

# Step 2: 检索真实文档(用假设答案做Query)
real_docs = vector_db.search(
    embedding_model.encode(hypothetical_answer),
    top_k=5
)

3.2 重排(Reranking)

# Cross-Encoder重排(比向量检索更精确但更慢)
# 向量检索(Bi-Encoder):先检索(快)→ 再重排(精)
# Cross-Encoder:对每个(Query, Doc)对单独打分(慢但准)

from sentence_transformers import CrossEncoder

reranker = CrossEncoder('BAAI/bge-reranker-large')

# 待重排的候选文档
candidates = [
    {"query": "LLaMA的架构特点", "text": "LLaMA使用decoder-only的Transformer架构..."},
    {"query": "LLaMA的架构特点", "text": "GPT-4采用混合专家架构..."},
    {"query": "LLaMA的架构特点", "text": "LLaMA来自Meta的AI研究..."},
]

# 批量打分:返回每个(Query, Doc)对的相关性分数
scores = reranker.predict([
    (c["query"], c["text"]) for c in candidates
])

# 按分数排序
ranked = sorted(zip(scores, candidates), key=lambda x: x[0], reverse=True)
# 输出: [(0.95, doc1), (0.23, doc3), (0.12, doc2)]

四、评估与优化

# RAG评估指标体系(RAGAS)
# ① Faithfulness(忠实度):答案是否完全基于检索上下文
# ② Answer Relevance(答案相关性):答案与问题的相关程度
# ③ Context Relevance(上下文相关性):检索的文档是否相关

# 自定义评估Prompt(RAGAS简化版)
def evaluate_faithfulness(context: str, answer: str) -> float:
    prompt = f"""评估以下答案是否忠实于给定的上下文。
评估标准:答案中的每个陈述都必须在上下文中能找到依据。
如果答案是基于上下文且无幻觉,返回1。
如果存在任何未在上下文中提及的内容,返回0。

上下文:
{context}

答案:
{answer}

评估结果:"""
    result = llm.invoke(prompt)
    return 1.0 if "1" in result else 0.0

# 检索质量的A/B测试
# 策略A:向量检索 Top-10 → 直接用
# 策略B:向量检索 Top-50 → Cross-Encoder重排 → Top-10
# 评估指标:答案忠实度 + 检索召回率 + 生成质量