准确率从60%到94%：构建RAG系统的核心策略（2026最新版）

一、朴素RAG的根本问题

# 朴素RAG方法def naive_rag(query: str) -> str:    # 1. 对查询进行向量化    query_embedding = embed(query)
    # 2. 查找相似片段    chunks = vector_db.search(query_embedding, top_k=5)
    # 3. 生成答案    context = "\n".join(chunks)    answer = llm.generate(f"Context: {context}\n\nQuestion: {query}")    return answer

二、真正有效的11个策略

from docling.chunking import HybridChunkerfrom transformers import AutoTokenizer
class SmartChunker:    def __init__(self, max_tokens=512):        # 使用实际的分词器，而不是字符计数        self.tokenizer = AutoTokenizer.from_pretrained(            "sentence-transformers/all-MiniLM-L6-v2"        )        self.chunker = HybridChunker(            tokenizer=self.tokenizer,            max_tokens=max_tokens,            merge_peers=True# 合并相邻的小分块        )
    def chunk_document(self, document):        # 分析文档结构（标题、段落、表格）        chunks = list(self.chunker.chunk(dl_doc=document))        # 每个分块包含标题上下文        contextualized_chunks = []        for chunk in chunks:            # 添加分层标题信息            contextualized_text = self.chunker.contextualize(chunk=chunk)            contextualized_chunks.append(contextualized_text)        return contextualized_chunks

async def enrich_chunk(chunk: str, document: str, title: str) -> str:    """使用LLM添加上下文前缀"""    prompt = f"""标题：{title}{document[:4000]}{chunk}
提供简要上下文（1-2句话）解释此分块  与完整文档的关系。格式："此分块来自[标题]，讨论[解释]。" """    response = await client.chat.completions.create(        model="gpt-4o-mini",        messages=[{"role": "user", "content": prompt}],        temperature=0,        max_tokens=150    )    context = response.choices[0].message.content.strip()    # 嵌入带上下文的版本    return f"{context}\n\n{chunk}"

之前："收入增长40%至3.14亿美元，利润率提高。"
之后："此分块来自ACME公司2024年第二季度SEC文件，讨论季度  财务表现与2024年第一季度的比较。收入增长40%至3.14亿美元，利润率提高。"

from sentence_transformers import CrossEncoder
# 初始化一次reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')async def search_with_reranking(query: str, limit: int = 5) -> list:    # 阶段1：快速向量检索（获取4倍候选）    candidate_limit = min(limit * 4, 20)    query_embedding = await embedder.embed_query(query)    candidates = await db.query(        "SELECT content, metadata FROM chunks ORDER BY embedding  $1 LIMIT $2",        query_embedding, candidate_limit    )    # 阶段2：使用交叉编码器重新排序    pairs = [[query, row['content']] for row in candidates]    scores = reranker.predict(pairs)
    # 按重排序分数排序并返回前N个    reranked = sorted(        zip(candidates, scores),          key=lambda x: x[1],          reverse=True    )[:limit]    return [doc for doc, score in reranked]

查询："第二季度收入增长因素是什么？"
纯向量（相似度分数）：1. "第二季度收入为3.14亿美元" (0.82)2. "增长因素包括..." (0.78)3. "第一季度，收入为..." (0.76)
重排序后（相关性分数）：1. "增长因素包括..." (0.94)2. "第二季度收入为3.14亿美元" (0.89)3. "第二季度表现的关键驱动因素..." (0.85)

async def expand_query(query: str) -> str:    """将简短查询扩展为详细版本"""    system_prompt = """你是查询扩展助手。  接受简短的用户查询并将其扩展为更详细的版本：1. 添加相关上下文和澄清2. 包含相关术语和概念3. 指定应涵盖的方面4. 保持原始意图5. 保持为单个连贯问题将查询扩展为详细2-3倍，同时保持专注。"""    response = await client.chat.completions.create(        model="gpt-4o-mini",        messages=[            {"role": "system", "content": system_prompt},            {"role": "user", "content": f"扩展此查询：{query}"}        ],        temperature=0.3    )    return response.choices[0].message.content.strip()

输入："什么是RAG？"输出："什么是检索增强生成（RAG），它如何将  信息检索与语言生成相结合，其关键组件和架构是什么，  以及它为问答系统提供了哪些优势？"

async def search_with_multi_query(query: str, limit: int = 5) -> list:    # 生成查询变体    variations_prompt = f"""生成此查询的3种不同表述：    "{query}"
    仅返回3个查询，每行一个。"""    response = await client.chat.completions.create(        model="gpt-4o-mini",        messages=[{"role": "user", "content": variations_prompt}],        temperature=0.7    )
    queries = [query] + response.choices[0].message.content.strip().split('\n')    # 并行执行所有搜索    search_tasks = []    for q in queries:        query_embedding = await embedder.embed_query(q)        task = db.fetch(            "SELECT * FROM match_chunks($1::vector, $2)",              query_embedding, limit        )        search_tasks.append(task)    results_lists = await asyncio.gather(*search_tasks)    # 按分块ID去重，保留最高相似度    seen = {}    for results in results_lists:        for row in results:            chunk_id = row['chunk_id']            if chunk_id notin seen or row['similarity'] > seen[chunk_id]['similarity']:                seen[chunk_id] = row    # 返回前N个唯一结果    return sorted(        seen.values(),          key=lambda x: x['similarity'],          reverse=True    )[:limit]

原始："如何部署ML模型？"变体1："将机器学习模型部署到生产环境的步骤是什么？"变体2："ML模型部署基础设施的最佳实践"变体3："训练模型的生产部署选项"

from pydantic_ai import Agentagent = Agent(    'openai:gpt-4o',    system_prompt='你是具有多个检索工具的RAG助手。为每个查询选择合适的工具。')@agent.toolasync def search_knowledge_base(query: str, limit: int = 5) -> str:    """文档分块的语义搜索"""    query_embedding = await embedder.embed_query(query)    results = await db.match_chunks(query_embedding, limit)    return format_results(results)@agent.toolasync def retrieve_full_document(document_title: str) -> str:    """当分块缺乏上下文时检索完整文档"""    result = await db.query(        "SELECT title, content FROM documents WHERE title ILIKE %s",        f"%{document_title}%"    )    return f"**{result['title']}**\n\n{result['content']}"@agent.toolasync def sql_query(question: str) -> str:    """查询结构化数据库获取特定数据"""    # 智能体可以为结构化数据编写SQL查询    # （在生产环境中，使用具有安全检查的适当SQL生成）    return execute_safe_sql(question)

用户："完整的退款政策是什么？"智能体推理：1. 调用search_knowledge_base("退款政策")   → 找到提及"refund_policy.pdf"的分块
2. 意识到分块没有完整政策
3. 调用retrieve_full_document("退款政策")   → 返回完整文档4. 从完整文档生成全面答案

async def search_with_self_reflection(query: str, limit: int = 5, max_iterations: int = 2) -> dict:    """自校正搜索循环"""    for iteration in range(max_iterations):        # 执行搜索        results = await vector_search(query, limit)        # 评分相关性        grade_prompt = f"""查询：{query}
检索到的文档：{format_docs_for_grading(results)}按1-5分评估这些文档与查询的相关性。仅用数字回答。"""        grade_response = await client.chat.completions.create(            model="gpt-4o-mini",            messages=[{"role": "user", "content": grade_prompt}],            temperature=0        )        grade = int(grade_response.choices[0].message.content.strip().split()[0])        # 如果结果好，返回它们        if grade >= 3:            return {                "results": results,                "iterations": iteration + 1,                "final_query": query            }        # 如果结果差且不是最后一次迭代，优化查询        if iteration < max_iterations - 1:            refine_prompt = f"""查询"{query}"返回了低相关性结果。
建议一个可能找到更好文档的改进查询。仅用改进后的查询回答。"""            refined_response = await client.chat.completions.create(                model="gpt-4o-mini",                messages=[{"role": "user", "content": refine_prompt}],                temperature=0.5            )            query = refined_response.choices[0].message.content.strip()    # 返回最佳尝试    return {        "results": results,        "iterations": max_iterations,        "final_query": query    }

迭代1：查询："部署"评分：2/5（太模糊）
迭代2：优化查询："机器学习模型部署到生产环境"评分：4/5（良好结果）

from graphiti_core import Graphitifrom graphiti_core.nodes import EpisodeType
# 初始化Graphiti（连接到Neo4j）graphiti = Graphiti("neo4j://localhost:7687", "neo4j", "password")async def ingest_document(text: str, source: str):    """摄取到知识图谱"""    # Graphiti自动提取实体和关系    await graphiti.add_episode(        name=source,        episode_body=text,        source=EpisodeType.text,        source_description=f"文档：{source}"    )async def search_knowledge_graph(query: str) -> str:    """混合搜索：语义 + 关键词 + 图"""    # Graphiti结合：    # - 语义相似性（嵌入）    # - BM25关键词搜索    # - 图结构遍历    # - 时间上下文    results = await graphiti.search(query=query, num_results=5)    # 格式化图结果    formatted = []    for result in results:        formatted.append(            f"实体：{result.node.name}\n"            f"类型：{result.node.type}\n"            f"关系：{result.relationships}"        )    return "\n---\n".join(formatted)

查询："谁运营ACME公司，第二季度发生了什么变化？"
纯向量搜索：- "ACME公司CEO信息..."- "第二季度变化包括..."知识图谱搜索：ACME公司（公司）  ├─ HAS_CEO → Jane Smith（人物）  ├─ REPORTED_REVENUE → $314M（财务）  │   └─ PERIOD → Q2 2024  └─ LOCATED_IN → California（地点）结果：可以回答 "Jane Smith运营ACME公司，收入在第二季度增加到3.14亿美元"

def ingest_hierarchical(document: str, title: str):    """创建父-子结构"""    # 父级：大部分（2000字符）    parent_chunks = [document[i:i+2000] for i in range(0, len(document), 2000)]    for parent_id, parent in enumerate(parent_chunks):        # 存储父级        metadata = {"heading": f"{title} - 部分 {parent_id}"}        db.execute(            "INSERT INTO parent_chunks (id, content, metadata) VALUES (%s, %s, %s)",            (parent_id, parent, json.dumps(metadata))        )        # 子级：小子分块（500字符）        child_chunks = [parent[j:j+500] for j in range(0, len(parent), 500)]        for child in child_chunks:            embedding = get_embedding(child)            db.execute(                "INSERT INTO child_chunks (content, embedding, parent_id) VALUES (%s, %s, %s)",                (child, embedding, parent_id)            )
async def hierarchical_search(query: str) -> str:    """搜索子级，返回父级"""    query_emb = get_embedding(query)    # 搜索小子级以提高精度    results = await db.query(        """SELECT p.content, p.metadata           FROM child_chunks c           JOIN parent_chunks p ON c.parent_id = p.id           ORDER BY c.embedding  %s LIMIT 3""",        query_emb    )    # 返回大父级以获取上下文    formatted = []    for content, metadata in results:        meta = json.loads(metadata)        formatted.append(f"[{meta['heading']}]\n{content}")    return "\n\n".join(formatted)

def late_chunk(text: str, chunk_size=512) -> list:    """分块前嵌入完整文档"""    # 步骤1：嵌入整个文档（最多8192个标记）    full_doc_token_embeddings = transformer_embed(text)  # 标记级别    # 步骤2：定义分块边界    tokens = tokenize(text)    chunk_boundaries = range(0, len(tokens), chunk_size)    # 步骤3：为每个分块池化标记嵌入    chunks_with_embeddings = []    for start in chunk_boundaries:        end = start + chunk_size        chunk_text = detokenize(tokens[start:end])        # 平均池化标记嵌入（保留完整文档上下文！）        chunk_embedding = mean_pool(full_doc_token_embeddings[start:end])        chunks_with_embeddings.append((chunk_text, chunk_embedding))    return chunks_with_embeddings

from sentence_transformers import SentenceTransformer, lossesfrom torch.utils.data import DataLoader
def prepare_training_data():    """领域特定的查询-文档对"""    return [        ("什么是EBITDA？", "EBITDA（利息、税项、折旧及摊销前利润..."),        ("解释资本支出", "资本支出（CapEx）指的是..."),        # ... 数千对更多    ]def fine_tune_model():    """在领域数据上微调"""    # 加载基础模型    model = SentenceTransformer('all-MiniLM-L6-v2')    # 准备训练数据    train_examples = prepare_training_data()    train_dataloader = DataLoader(train_examples, shuffle=True, batch_size=16)    # 定义损失函数    train_loss = losses.MultipleNegativesRankingLoss(model)
    # 训练    model.fit(        train_objectives=[(train_dataloader, train_loss)],        epochs=3,        warmup_steps=100    )    model.save('./fine_tuned_financial_model')    return model# 使用微调模型embedding_model = SentenceTransformer('./fine_tuned_financial_model')

查询："什么是营运资金？"
通用嵌入：1. "营运资金包括..." (0.72)2. "资本市场提供..." (0.68) ← 错误！3. "工作条件在..." (0.65) ← 错误！
微调嵌入：1. "营运资金包括..." (0.89)2. "营运资金比率计算..." (0.84)3. "有效管理营运资金..." (0.81)

三、组合策略的力量：实现94%准确率的关键

四、实施路线图：从简单开始，智能扩展

五、实际应用结果

六、常见错误避免

七、RAG的未来趋势

写在最后

• GitHub仓库：高级RAG策略: https://github.com/coleam00/ottomator-agents/tree/main/all-rag-strategies

• Anthropic的上下文检索研究: https://www.anthropic.com/news/contextual-retrieval
  

• Pinecone：重排序指南: https://www.pinecone.io/learn/series/rag/rerankers/
  

• Docling：上下文感知分块: https://github.com/DS4SD/docling
  

• Graphiti：RAG知识图谱: https://github.com/graphiti-ai/graphiti