检索增强生成(RAG)实战：如何用Embedding优化大模型Prompt Engineering

1次阅读

没有评论

共计 2558 个字符，预计需要花费 7 分钟才能阅读完成。

传统大模型存在两个致命伤：

知识固化问题：训练数据截止后，模型无法自动更新知识。例如用 GPT- 3 回答 2023 年后的事件时，要么拒绝回答，要么产生错误信息
幻觉风险：当模型遇到知识盲区时，倾向于编造看似合理实则错误的答案

最近处理客户咨询时，就遇到经典案例：某金融客户直接用 ChatGPT 回答最新监管政策，导致 30% 的答案引用已废止条文。

训练成本：100 万 tokens 约 $3（GPT-3.5），且需要专业标注团队
迭代周期：至少 2 周 / 次
效果局限：仍受限于基础模型的知识截止时间

实施成本：开源 Embedding 模型 +FAISS 可实现零 API 费用
响应速度：知识库更新实时生效
可解释性：可追溯生成结果的参考来源

实测对比：在医疗问答场景下，RAG 方案将准确率从 68% 提升至 92%，同时降低 60% 的幻觉率。

推荐组合策略：

通用领域：text-embedding-3-small（性价经实测最优）
专业领域：bge-small-en-v1.5（在法律 / 医疗等垂直领域表现更好）

关键参数测试结果：

# Embedding 维度对比
models = {
    'ada-002': 1536,
    'text-embedding-3-small': 512,  # 实测效果相当但速度快 2 倍
    'bge-base': 768
}

FAISS 优化技巧：

import faiss

# 关键配置项
d = 512  # 维度
index = faiss.IndexFlatIP(d)  # 内积比 L2 更适合语义搜索

# 专业领域建议添加 PCA 降维
pca = faiss.PCAMatrix(d, 256)
index = faiss.IndexPreTransform(pca, index)

银行业务文档的处理经验：

法律条款：按 200-300 字符 分块，保留完整条款
操作手册：500 字符+ 重叠 50 字符，保持步骤连贯
研究报告：先按章节分，再按 400 字符 切分

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=300,
    chunk_overlap=30,
    separators=["\n\n", "\n", ".", " "]  # 优先按段落分割
)

金融风控场景的模板设计：

template = """ 基于以下监管要求（引用自{source}）：{context}

请回答：{question} 
回答时务必：1. 指出具体条款编号
2. 如涉及数值要求必须精确到小数点后两位
3. 不确定时明确声明 "该问题不在当前知识库覆盖范围内"
"""

预填充策略：
行业白皮书
常见问题库
产品文档
混合检索：初期结合关键词搜索弥补 Embedding 不足

flowchart LR
    A[用户输入] --> B{是否 OOV?}
    B -- 是 --> C[同义词替换]
    B -- 否 --> D[向量化]
    C --> E[再次检查]

检索成功率：top_k 结果中有效文档占比
命中准确率：人工审核前 20% 结果的精确度
响应延迟：P99 控制在 200ms 内

# 数据预处理管道
import pandas as pd
from sentence_transformers import SentenceTransformer

# 1. 加载知识库
df = pd.read_csv('knowledge_base.csv')

# 2. 初始化 Embedding 模型（GPU 加速推荐）encoder = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2', 
                            device='cuda')

# 3. 批量生成向量
batch_size = 32  # 根据 GPU 内存调整
vectors = [encoder.encode(text) for text in batch_generator(df['text'], batch_size)]

# 4. 构建 FAISS 索引
import faiss
index = faiss.IndexFlatIP(encoder.get_sentence_embedding_dimension())
faiss.normalize_L2(vectors)  # 关键步骤！index.add(np.array(vectors))

# 5. 检索服务封装
class RAGEngine:
    def __init__(self, index, encoder):
        self.index = index
        self.encoder = encoder

    def query(self, question, top_k=3):
        # 向量化查询
        query_vec = self.encoder.encode(question)
        query_vec = query_vec.reshape(1, -1)
        faiss.normalize_L2(query_vec)

        # 相似度搜索
        distances, indices = self.index.search(query_vec, top_k)

        # 结果后处理
        return [{'text': df.iloc[i]['text'],
            'score': float(distances[0][j]),
            'source': df.iloc[i]['source']
        } for j, i in enumerate(indices[0])]