RAG 基本原理

为什么需要RAG

先有痛，再看药方

LLM到底是什么

LLM 的训练过程本质上是一个压缩过程：

LLM = 压缩机

训练过程：通过学习如何预测"下一个 token", 把万亿 token 的文本，压缩进几百亿个浮点数参数
        
推理（生成）过程：给一段提示词，从参数里解压出最可能的后续，token逐词解压，就形成了流畅文本

结果：知识被"固化"在参数里，训练完成 = 参数固化 = 知识冻结

理解了这个，我们可以推论出来LLM 的四个局限：

压缩 + 固化
    ↓
固化  → 新知识进不来（知识截止）
压缩  → 细节丢失，只能插值（幻觉）
有限  → 上下文装不下全部知识
封闭  → 私有数据从未见过

结论：LLM 解决”怎么说话”，RAG 解决”说的内容从哪来”。两者不是竞争关系，而是互补关系。

RAG是什么

rag核心思路

检索增强生成（RAG，Retrieval-Augmented Generation）最早是由Facebook(现Meta) AI Research在2020 年发表的论文《Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks》（ https://arxiv\.org/abs/2005\.11401 ）中正式提出，这种方法的核心思想是借助一些文本检索策略，让大模型每次问答前都带入相关文本，以此来改善大模型回答时的准确性。

为什么 2022–2023 年才爆发

RAG 论文 2020 年就发表了，为什么等到 2022–2023 年才成为主流？四个条件同时到位：

RAG vs 微调

微调（Fine-tuning）：
  把新知识固化到模型参数
  改变的是模型本身
  就像给人注射知识血清，让他变成领域专家

RAG：
  检索时临时把知识注入上下文
  不改变模型本身
  就像给人配一个随时可查阅的参考书架

为什么不用微调注入知识

原因一：灾难性遗忘

微调新知识时，模型会部分遗忘旧知识
就像人学了新技能反而忘了旧技能

实验证明：
  微调前：模型知道 ET9 续航 720 公里
  微调后（注入其他新车型知识）：可能把 ET9 的数据搞错

原因二：知识更新代价高

产品价格变了 → 要重新训练一遍模型？ → 不现实
活动规则更新 → 要重新训练？ → 不现实

RAG：更新知识库文件 → 立即生效 → 零训练成本

原因三：事实注入效果差

微调注入的"事实"往往不稳定：
  某个问法能答出来 → 换个问法又忘了
  远不如 RAG 的检索来得可靠

微调适合什么

适合微调的场景：
  ✓ 输出固定格式（永远输出 JSON）
  ✓ 说话风格符合品牌调性（专业/活泼）
  ✓ 学会领域术语和表达习惯
  ✓ 拒绝回答某类问题（安全对齐）
  ✓ 更好地完成特定任务（代码生成）

不适合微调的场景：
  ✗ 会变化的事实性知识（价格、活动、新闻）
  ✗ 大量私有文档（成本高、效果差、有泄露风险）
  ✗ 需要实时更新的信息

最佳实践：两者结合

微调负责：让模型"说话方式"符合业务要求
  → 品牌语气、输出格式、领域术语

RAG 负责：给模型提供"说话内容"的事实依据
  → 最新价格、私有文档、实时数据

RAG 范式演变

Naive RAG — 先把它跑起来

理解 RAG 最小可运行单元

import faiss
import numpy as np
import ollama
import os
from langchain_text_splitters import RecursiveCharacterTextSplitter

EMBED_MODEL = "bge-m3"
CHAT_MODEL = "qwen2.5"
client = ollama.Client(host="http://localhost:11434")

def chunk_documents(documents: list[str],chunk_size:int=200,chunk_overlap:int=40,separators:list=["\n\n", "\n", "。", "；", "，", " ", ""]) -> list[str]:
    """用 RecursiveCharacterTextSplitter 递归切分
    优先按段落/换行/句号边界,其次才硬切到 CHUNK_SIZE,语义更完整。
    """
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        separators=separators
    )
    return splitter.split_text("\n\n".join(documents))

def _embed(texts: list[str]) -> np.ndarray:
    """Ollama 的 embeddings 一次只能传一条,这里循环 + 拼成矩阵"""
    vectors = [client.embeddings(model=EMBED_MODEL, prompt=t)["embedding"] for t in texts]
    return np.array(vectors, dtype="float32")

# ── Step1:索引期(离线) ────────────────────────────────────
def build_index(documents: list[str]):
    """文档向量化,存入 FAISS"""
    embeddings = _embed(documents)

    # 归一化后用内积索引 == 余弦相似度(FAISS 没有 IndexFlatCosine)
    faiss.normalize_L2(embeddings)
    dim = embeddings.shape[1]               # nomic-embed-text 维度 768
    index = faiss.IndexFlatIP(dim)
    index.add(embeddings)

    return index, embeddings

# ── Step2:检索期(在线) ────────────────────────────────────
def retrieve(query: str, index, documents: list[str], top_k: int = 3):
    """向量检索 Top-K 文档"""
    query_vec = _embed([query])
    faiss.normalize_L2(query_vec)

    distances, indices = index.search(query_vec, top_k)
    return [documents[i] for i in indices[0]]

# ── Step3:生成期 ────────────────────────────────────────────
def generate(query: str, context: list[str]) -> str:
    """基于检索结果生成答案"""
    context_str = "\n".join(context)
    response = client.chat(
        model=CHAT_MODEL,
        messages=[
            {"role": "system",
             "content": "你是蔚来汽车的智能客服,基于提供的资料回答问题。"},
            {"role": "user",
             "content": f"参考资料:\n{context_str}\n\n问题:{query}"},
        ],
        options={"temperature": 0.2},
    )
    return response["message"]["content"]

# ── 主流程 ────────────────────────────────────────────────────
if __name__ == "__main__":
    # 示例语料(实际使用时替换为你的文档)
    documents = [
        "2026年1月，蔚来公司交付新车27,182台，同比增长96.1%。截至目前，蔚来公司已累计交付新车1,024,774台。今天下午，蔚来全新ES8在广州完成第60,000台新车交付，历时134天。纯电大三排SUV的黄金时代已经到来。在「Banyan 榕 3.3.0」版本推送后，新ET5、ET5T、ES6和EC6以及全新ES8和ET9的智能系统也将迭代至1.4.0版本，欢迎大家体验最新的世界模型辅助驾驶版本。今天，蔚来品牌针对新ET5、ET5T、ES6和EC6、乐道品牌针对乐道L60与L90推出7年超低息购车方案，7年0.49%年化费率，0金融服务费，提前还款0违约金。firefly萤火虫品牌推出7年超低息购车方案，锁单赠送马年新春奇遇礼包。蔚来换电次数已突破9,900万次，还有几天，1亿次换电就将达成。新春佳节即将到来，提前预祝大家马年大吉，万事顺遂。新的一年我们继续努力，一起加电！",
        "2026年2月，蔚来公司交付新车20,797台，同比增长57.6%。截至目前，蔚来公司已累计交付新车1,045,571台。2月27日，全新ES8在温州完成第70,000台新车交付，历时160天。纯电大三排SUV的黄金时代已经到来。2月6日22:33:18，蔚来1亿次换电达成；2026春节期间，蔚来累计提供换电服务2,073,500次，日均服务量较去年春节日均增长29.4%。2月15日至2月23日，蔚来能源在高速公路充换电量累计超2528万kWh，占全国高速公路充换电总量15%。自2月18日（大年初二）起，蔚来换电连续5日刷新单日服务纪录。2月26日，蔚来芯片子公司安徽神玑技术有限公司完成首轮股权融资协议签署，融资金额超22亿元人民币，投后估值近百亿。此次融资将有利于神玑公司持续地研发和推广高端、高竞争力的芯片产品，支撑蔚来在自动驾驶、具身智能等领域的长远布局。今天，蔚来品牌针对全新ES8提供购置税补贴10,000元，针对新ET5、ET5T、ES6和EC6推出7年超低息购车方案，首付比例20%起，7年0.49%年化费率。乐道品牌继7年超低息金融方案后，加推购置税全兜底政策，全系至高补贴10,262元。全国门店优选现车充足，专属礼遇享更多优惠。firefly萤火虫品牌在7年超低息购车方案上推出3月限时购车礼，购车即享3,000元选装基金，与2,000元购置税补贴或7kW交流家充桩（30米套包），另有其他多项购车优惠好礼。一年之计在于春，我们继续努力，一起加电！",
        "2026年3月，蔚来公司交付新车35,486台，同比增长136.0%，环比增长70.6%。其中，蔚来品牌交付新车22,490台，同比增长120.1%，环比增长48.4%；乐道品牌交付新车6,877台，同比增长42.7%，环比增长130.7%；firefly萤火虫品牌交付新车6,119台，环比增长130.3%。2026年一季度，蔚来公司共交付新车83,465台，同比增长98.3%。截至目前，蔚来公司已累计交付新车1,081,057台。3月20日，蔚来全新ES8完成第80,000台新车交付，历时181天。全新ES8已连续三个月斩获「大型 SUV」不分能源形式、不分价格区间、不分座位数销量冠军。乐道L90单月交付环比增长155.1%，推动乐道品牌累计交付量突破14万台。蔚来全新ES8和乐道L90共同引领纯电大三排SUV进入黄金时代。3月29日，firefly萤火虫迎来第50,000台用户交付，品牌迈入全新发展阶段。明天起，蔚来公司将进入密集的新品发布阶段。2026款ES6、EC6、ET5、ET5T将于4月2日正式上市，年款车型在设计、配置和智驾体验上均有升级。科技行政旗舰SUV蔚来ES9产品技术发布会将于4月9日正式召开。乐道L90 2026款将在4月下旬发布，新增激光雷达旗舰智驾版，乐道L80也即将上市。3月10日，蔚来公司发布2025年四季度及全年财报，宣布四季度实现经营利润12.5亿元，公司首次实现季度盈利。四季度现金储备459亿元，环比大幅增长近百亿。今天公布的一季度交付数超出了2026年一季度交付指引（80,000台-83,000台）。暖春四月，万象更新。我们继续努力，一起加电！",
        "2026年4月，蔚来公司交付新车29,356台，同比增长22.8%。其中，蔚来品牌交付新车19,024台；乐道品牌交付新车5,352台，同比增长21.6%；firefly萤火虫品牌交付新车4,980台。今年前四个月，蔚来公司共交付新车112,821台，同比增长71.0%。截至目前，蔚来公司已累计交付新车1,110,413台。从4月份开始，公司进入新产品密集发布和交付期，生产制造、市场营销等团队围绕新产品积极开展工作。2026款ES6、EC6、ET5、ET5T在四月初上市并交付；新款萤火虫已经开始交付；2026款乐道L90正式上市，将于5月9日开启交付；重磅新车乐道L80已经发布，将于5月15日上市交付；科技行政旗舰SUV蔚来ES9也将于5月底上市并交付。相信大家能选到适合自己的产品。历时215天完成第十万台交付，刷新中国乘用车市场40万元以上高端车型的销售和交付纪录的全新ES8，也在4月24日开幕的北京车展上发布了玄金特别版。这次北京车展，蔚来公司以「大有新境 A Vast New Realm」为主题，携NIO蔚来、ONVO乐道、firefly萤火虫三大品牌首次同台亮相，目前北京车展正迎来观展高峰，欢迎朋友们五一假期到蔚来公司展台做客。祝大家节日愉快，出行平安。我们继续努力，一起加电！",
        "2026年4月，蔚来全新ES8完成13,020台新车交付。截至目前，全新ES8已经拥有超过100,000名用户，并且刷新了中国40万元以上高端车型交付十万台最快纪录。此外，全新ES8已经连续四个月斩获大型SUV和40万元以上车型销量冠军，纯电大三排SUV的黄金时代已经到来！,即日起至2026年5月31日（含），购车用户可申请蔚来指定合作机构提供的5年60期分期金融方案，0首付起，前3年36期年化费率0%，后2年24期年化费率为3%。同时，订购全新ES8并成功锁单的用户，可享限时购置税补贴10,000元，5年「全域领航辅助 NOP+」免费使用权，以及飞航头等舱五件套。"      
    ]

    # 先切 chunk,再建索引
    chunks = chunk_documents(documents)
    print(f"原始文档 {len(documents)} 条 → 切分后 {len(chunks)} 个 chunk")

    # 建索引(检索单位从整篇文档变为 chunk)
    index, _ = build_index(chunks)

    # 问答
    query = "蔚来26年3月份销量怎么样"
    context = retrieve(query, index, chunks, top_k=4)
    answer = generate(query, context)

    print(f"问题:{query}")
    print(f"召回文档:{context}")
    print(f"答案:{answer}")

Naive RAG 的四个核心问题

定位：能跑通，但不好用。适合 Demo 验证和快速原型，生产环境需要 Advanced RAG 级别的优化。

Advanced RAG — 工程上怎么把它做好

没有明确的提出者，在Navie RAG架构之后由多方逐步演进得到，如微软研究院提出HyDE，谷歌引入Rerank，以及如Elastic和Cohere公司提出的各种优化技术等，由多方共同构成高级 RAG技术体系. 这些优化技术提出的动因，主要是由于Navie RAG仍存在一系列问题，如

检索精度不足：Navie RAG依赖关键词匹配（如BM25）或简单向量检索，导致语义偏差问题。
多上下文割裂：Navie RAG直接将检索到的上下文直接拼接Prompt输入LLM，没有做post-retrieval处理，文档容易出现冗余或矛盾。
无法处理复杂场景：Navie RAG工作流比较简单，没有针对性优化，系统表现上限有限。

三阶段框架

Pre-Retrieval  → 让问题更好问
Retrieval      → 让检索更准
Post-Retrieval → 把最好的喂给 LLM

#!/usr/bin/env python3
# -*- encoding: utf-8 -*-

import os
from langchain_chroma import Chroma
from langchain_ollama import ChatOllama, OllamaEmbeddings
from langchain_community.retrievers import BM25Retriever
from langchain_classic.retrievers import EnsembleRetriever
from langchain_core.documents import Document
from langchain_text_splitters import RecursiveCharacterTextSplitter
from sentence_transformers import CrossEncoder


OLLAMA_HOST = "http://localhost:11434"
CHAT_MODEL = "qwen2.5"
EMBED_MODEL = "bge-m3"

# ── 初始化 ──────────────────────────────────────────────────
embeddings = OllamaEmbeddings(model=EMBED_MODEL, base_url=OLLAMA_HOST)
llm = ChatOllama(model=CHAT_MODEL, base_url=OLLAMA_HOST, temperature=0)
reranker = CrossEncoder("BAAI/bge-reranker-base")

# ── 切分 chunk ────────────────────────────────────────────
def chunk_documents(documents: list[str],
                    chunk_size:int=200,
                    chunk_overlap:int=40,
                    separators:list=["\n\n", "\n", "。", "；", "，", " ", ""]) -> list[Document]:
    """递归字符切分:优先按段落/句子边界,其次才硬切到 chunk_size"""
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        separators=separators
    )
    return splitter.split_documents(documents)
# ── 索引期:建双路索引(向量 + BM25) ─────────────────────────
def build_retriever(docs: list[Document],
                    embeddings,
                    dense_k: int = 5,
                    bm25_k: int = 5,
                    weights: tuple[float, float] = (0.4, 0.6)) -> EnsembleRetriever:
    """根据文档构建 BM25 + 向量 的混合检索器。
    weights = (bm25_weight, dense_weight)
    """
    # 向量检索:Chroma
    vectorstore = Chroma.from_documents(docs, embeddings)
    dense_retriever = vectorstore.as_retriever(search_kwargs={"k": dense_k})

    # 关键词检索:BM25
    bm25_retriever = BM25Retriever.from_documents(docs)
    bm25_retriever.k = bm25_k

    # 混合检索:加权融合
    return EnsembleRetriever(
        retrievers=[bm25_retriever, dense_retriever],
        weights=list(weights),
    )

# ── Pre-Retrieval:Query 改写 ────────────────────────────────
def rewrite_query(query: str) -> str:
    prompt = f"将以下口语化问题改写为适合检索的精确语句,只输出改写结果:\n{query}"
    return llm.invoke(prompt).content.strip()

# ── Retrieval:混合检索 ──────────────────────────────────────
def hybrid_search(retriever: EnsembleRetriever, query: str) -> list[Document]:
    return retriever.invoke(query)

# ── Post-Retrieval:Reranking ────────────────────────────────
def rerank(query: str, docs: list[Document], top_k: int = 3) -> list[Document]:
    pairs = [(query, d.page_content) for d in docs]
    scores = reranker.predict(pairs)
    ranked = sorted(zip(scores, docs), key=lambda x: x[0], reverse=True)
    return [d for _, d in ranked[:top_k]]

# ── 生成 ─────────────────────────────────────────────────────
def generate(query: str, docs: list[Document]) -> str:
    context = "\n".join(d.page_content for d in docs)
    return llm.invoke(
        f"基于以下资料回答问题:\n{context}\n\n问题:{query}"
    ).content

# ── 主流程 ────────────────────────────────────────────────────
if __name__ == "__main__":
    # 示例语料
    documents = [
            "2026年1月，蔚来公司交付新车27,182台，同比增长96.1%。截至目前，蔚来公司已累计交付新车1,024,774台。今天下午，蔚来全新ES8在广州完成第60,000台新车交付，历时134天。纯电大三排SUV的黄金时代已经到来。在「Banyan 榕 3.3.0」版本推送后，新ET5、ET5T、ES6和EC6以及全新ES8和ET9的智能系统也将迭代至1.4.0版本，欢迎大家体验最新的世界模型辅助驾驶版本。今天，蔚来品牌针对新ET5、ET5T、ES6和EC6、乐道品牌针对乐道L60与L90推出7年超低息购车方案，7年0.49%年化费率，0金融服务费，提前还款0违约金。firefly萤火虫品牌推出7年超低息购车方案，锁单赠送马年新春奇遇礼包。蔚来换电次数已突破9,900万次，还有几天，1亿次换电就将达成。新春佳节即将到来，提前预祝大家马年大吉，万事顺遂。新的一年我们继续努力，一起加电！",
            "2026年2月，蔚来公司交付新车20,797台，同比增长57.6%。截至目前，蔚来公司已累计交付新车1,045,571台。2月27日，全新ES8在温州完成第70,000台新车交付，历时160天。纯电大三排SUV的黄金时代已经到来。2月6日22:33:18，蔚来1亿次换电达成；2026春节期间，蔚来累计提供换电服务2,073,500次，日均服务量较去年春节日均增长29.4%。2月15日至2月23日，蔚来能源在高速公路充换电量累计超2528万kWh，占全国高速公路充换电总量15%。自2月18日（大年初二）起，蔚来换电连续5日刷新单日服务纪录。2月26日，蔚来芯片子公司安徽神玑技术有限公司完成首轮股权融资协议签署，融资金额超22亿元人民币，投后估值近百亿。此次融资将有利于神玑公司持续地研发和推广高端、高竞争力的芯片产品，支撑蔚来在自动驾驶、具身智能等领域的长远布局。今天，蔚来品牌针对全新ES8提供购置税补贴10,000元，针对新ET5、ET5T、ES6和EC6推出7年超低息购车方案，首付比例20%起，7年0.49%年化费率。乐道品牌继7年超低息金融方案后，加推购置税全兜底政策，全系至高补贴10,262元。全国门店优选现车充足，专属礼遇享更多优惠。firefly萤火虫品牌在7年超低息购车方案上推出3月限时购车礼，购车即享3,000元选装基金，与2,000元购置税补贴或7kW交流家充桩（30米套包），另有其他多项购车优惠好礼。一年之计在于春，我们继续努力，一起加电！",
            "2026年3月，蔚来公司交付新车35,486台，同比增长136.0%，环比增长70.6%。其中，蔚来品牌交付新车22,490台，同比增长120.1%，环比增长48.4%；乐道品牌交付新车6,877台，同比增长42.7%，环比增长130.7%；firefly萤火虫品牌交付新车6,119台，环比增长130.3%。2026年一季度，蔚来公司共交付新车83,465台，同比增长98.3%。截至目前，蔚来公司已累计交付新车1,081,057台。3月20日，蔚来全新ES8完成第80,000台新车交付，历时181天。全新ES8已连续三个月斩获「大型 SUV」不分能源形式、不分价格区间、不分座位数销量冠军。乐道L90单月交付环比增长155.1%，推动乐道品牌累计交付量突破14万台。蔚来全新ES8和乐道L90共同引领纯电大三排SUV进入黄金时代。3月29日，firefly萤火虫迎来第50,000台用户交付，品牌迈入全新发展阶段。明天起，蔚来公司将进入密集的新品发布阶段。2026款ES6、EC6、ET5、ET5T将于4月2日正式上市，年款车型在设计、配置和智驾体验上均有升级。科技行政旗舰SUV蔚来ES9产品技术发布会将于4月9日正式召开。乐道L90 2026款将在4月下旬发布，新增激光雷达旗舰智驾版，乐道L80也即将上市。3月10日，蔚来公司发布2025年四季度及全年财报，宣布四季度实现经营利润12.5亿元，公司首次实现季度盈利。四季度现金储备459亿元，环比大幅增长近百亿。今天公布的一季度交付数超出了2026年一季度交付指引（80,000台-83,000台）。暖春四月，万象更新。我们继续努力，一起加电！",
            "2026年4月，蔚来公司交付新车29,356台，同比增长22.8%。其中，蔚来品牌交付新车19,024台；乐道品牌交付新车5,352台，同比增长21.6%；firefly萤火虫品牌交付新车4,980台。今年前四个月，蔚来公司共交付新车112,821台，同比增长71.0%。截至目前，蔚来公司已累计交付新车1,110,413台。从4月份开始，公司进入新产品密集发布和交付期，生产制造、市场营销等团队围绕新产品积极开展工作。2026款ES6、EC6、ET5、ET5T在四月初上市并交付；新款萤火虫已经开始交付；2026款乐道L90正式上市，将于5月9日开启交付；重磅新车乐道L80已经发布，将于5月15日上市交付；科技行政旗舰SUV蔚来ES9也将于5月底上市并交付。相信大家能选到适合自己的产品。历时215天完成第十万台交付，刷新中国乘用车市场40万元以上高端车型的销售和交付纪录的全新ES8，也在4月24日开幕的北京车展上发布了玄金特别版。这次北京车展，蔚来公司以「大有新境 A Vast New Realm」为主题，携NIO蔚来、ONVO乐道、firefly萤火虫三大品牌首次同台亮相，目前北京车展正迎来观展高峰，欢迎朋友们五一假期到蔚来公司展台做客。祝大家节日愉快，出行平安。我们继续努力，一起加电！",
            "2026年4月，蔚来全新ES8完成13,020台新车交付。截至目前，全新ES8已经拥有超过100,000名用户，并且刷新了中国40万元以上高端车型交付十万台最快纪录。此外，全新ES8已经连续四个月斩获大型SUV和40万元以上车型销量冠军，纯电大三排SUV的黄金时代已经到来！,即日起至2026年5月31日（含），购车用户可申请蔚来指定合作机构提供的5年60期分期金融方案，0首付起，前3年36期年化费率0%，后2年24期年化费率为3%。同时，订购全新ES8并成功锁单的用户，可享限时购置税补贴10,000元，5年「全域领航辅助 NOP+」免费使用权，以及飞航头等舱五件套。"      
        ]
    docs = [Document(page_content=text) for text in documents]

    # 1. 切分 chunk
    chunks = chunk_documents(docs)
    print(f"原始文档 {len(docs)} 条 → 切分后 {len(chunks)} 个 chunk")

    # 2. 构建混合检索器(检索单位是 chunk)
    hybrid_retriever = build_retriever(chunks, embeddings)

    query = "蔚来三月份销量?"

    # Pre-Retrieval
    rewritten = rewrite_query(query)
    print(f"改写后:{rewritten}")

    # Retrieval
    candidates = hybrid_search(hybrid_retriever, rewritten)
    print(f"召回数量:{len(candidates)}")

    # Post-Retrieval
    final_docs = rerank(query, candidates,3)
    print(f"Rerank 后:{[d.page_content[:30] for d in final_docs]}")

    # 生成
    answer = generate(query, final_docs)
    print(f"答案:{answer}")

Pre-Retrieval：优化输入

目标：让 Query 更准确地命中知识库

Query 改写

把用户口语化的问题改写成更适合检索的精确语句：

rewrite_prompt = """
将以下用户问题改写为更适合文档检索的精确查询语句：
用户问题：{question}
改写后：
"""

# 示例
# 输入："ET9 贵不贵"
# 输出："蔚来 ET9 官方指导价 售价"
# 结合**历史对话 + 用户画像**个性化改写
rewritten = llm(rewrite_prompt.format(question=user_query))

Query 扩展

一个问题生成多个角度的查询，扩大召回范围：

expansion_prompt = """
为以下问题生成 3 个不同角度的相关查询语句：
原始问题：{question}
输出 JSON 数组：
"""

# 输入："ET9 续航怎么样"
# 输出：[
#   "ET9 CLTC 续航里程",
#   "ET9 100kWh 电池包续航表现",
#   "ET9 实际用车冬季续航"
# ]

HyDE（假设文档）

先让 LLM 生成”假设答案”，用答案的向量检索，比用问题向量命中率更高：

# 直觉：文档空间里，答案比问题更像答案
hyde_prompt = "请为以下问题写一个简短的假设性答案：{question}"
hypothetical_answer = llm(hyde_prompt.format(question=query))

# 用假设答案的向量去检索，而非原始问题
docs = vectorstore.similarity_search(hypothetical_answer, k=5)

Chunking 优化策略

Chunking 是 RAG 质量的地基，切得不好，后面的优化都是在烂地基上盖楼。

为什么 Chunking 很重要

切得太小：
  "ET9 续航" 和 "720公里CLTC" 被切成两块
  → 检索到前半段，LLM 拿到残缺信息，答非所问

切得太大：
  一块里包含续航、价格、配置、服务……
  → 向量被稀释，相似度计算不准，召回质量差

好的 Chunking：语义完整 + 粒度合适 + 边界清晰

策略一：固定长度切块（不推荐）

这是最早期的做法，按固定字符数切割，加一点重叠防止边界信息丢失。

优点是实现简单，不需要任何语言理解能力，速度很快。

缺点是完全不管语义。它不知道这 512 个字符到底说完了一件事没有，很可能把一句话从中间切断。这个策略只适合快速做原型验证，或者内容本来就很均质、每段都讲一件事的纯文本。生产环境不建议使用。

from langchain.text_splitter import CharacterTextSplitter

splitter = CharacterTextSplitter(
    chunk_size=512, 
    chunk_overlap=10
)
chunks = splitter.split_documents(documents)

策略二：语义切块

语义切块的核心思想是：让相似度来决定在哪里切，而不是靠固定字符数。

具体做法是把文档按句子拆开，对相邻的句子计算 Embedding 相似度。如果两句话语义接近，说明它们在讲同一件事，不切；如果相似度突然下降，说明话题发生了转换，这里就是切割点。

代码里的 breakpoint_percentile_threshold=95 这个参数非常关键。它的意思是：只有当相邻两句的相似度低于所有相邻句对中最低的 5%，才判定为切割点。

这个值越大，切割越保守，块越大、语义越完整，但有时会把不同话题的内容放在一起；越小，切割越激进，块更小更聚焦，但可能把有关联的句子切开。

from llama_index.core.node_parser import SemanticSplitterNodeParser
from llama_index.embeddings.openai import OpenAIEmbedding

splitter = SemanticSplitterNodeParser(
    buffer_size=1,                      # 考虑前后各1句
    breakpoint_percentile_threshold=95, # 相似度低于95%分位就切
    embed_model=OpenAIEmbedding()
)
nodes = splitter.get_nodes_from_documents(documents)

# 原理示意：
# 句1：ET9 搭载 100kWh 电池包         ┐
# 句2：CLTC 续航 720 公里             ├ 语义相似 → 不切
# 句3：支持 500kW 超充                ┘
# 句4：ET9 采用全铝车身设计           ← 相似度突降 → 切！
# 句5：外观由知名设计师操刀

#breakpoint_percentile_threshold
#值越大（如 99）：
#  切割点更少，块更大，语义更完整
#  但可能把不相关内容放在一块

#值越小（如 80）：
#  切割点更多，块更小，语义更聚焦
#  但可能切断有关联的句子

#推荐：90~95，根据文档类型微调

策略三：Small-to-Big

核心思想：用小块检索，用大块生成

这个策略解决了一个很现实的矛盾：检索需要小块（精准），但生成需要大块（上下文完整）。

光靠一个粒度很难两头兼顾。Small-to-Big 的方案是建立层级结构——把文档切成大中小三种粒度，大块包含小块，形成父子关系。

检索阶段：只拿最小的块（比如 128 token）去做向量搜索，粒度小、语义聚焦，命中精准。

返回阶段：命中了小块之后，系统自动往上找它的父块（512 token），把完整的上下文交给 LLM。

代码里的 simple_ratio_thresh=0.4 是一个合并阈值：如果同一个父块下超过 40% 的子块都被命中了，就直接把整个父块返回，而不是把零散的子块拼在一起。这样能避免 LLM 拿到的是断断续续的碎片。

一句话总结：小块负责找得准，大块负责说得全。

from llama_index.core.node_parser import HierarchicalNodeParser
from llama_index.core.retrievers import AutoMergingRetriever
from llama_index.core import VectorStoreIndex, StorageContext

# 建立三层粒度（LlamaIndex 约定：level 越大粒度越小）
parser = HierarchicalNodeParser.from_defaults(
    chunk_sizes=[2048, 512, 128]
    # chunk_sizes[0]=2048 → level 1（祖父块，必要时进一步扩展）
    # chunk_sizes[1]=512  → level 2（父块，召回后扩展给 LLM）
    # chunk_sizes[2]=128  → level 3（叶子块，粒度小，用于检索）
)
nodes = parser.get_nodes_from_documents(documents)

# 存储层级关系（父子指针保存在 docstore）
storage_context = StorageContext.from_defaults()
storage_context.docstore.add_documents(nodes)

# 建索引：只把最细粒度的叶子节点放进向量库
from llama_index.core.node_parser import get_leaf_nodes
leaf_nodes   = get_leaf_nodes(nodes)
vector_index = VectorStoreIndex(leaf_nodes, storage_context=storage_context)

# 检索时自动合并到父块
retriever = AutoMergingRetriever(
    vector_index.as_retriever(similarity_top_k=10),
    storage_context,
    simple_ratio_thresh=0.4  # 同一父块下 >40% 的子块被命中就合并到父块
)

策略四：结构化切块

前面几个策略都是把文档当作”流水文本”来处理。但很多文档本身是有结构的——比如 Markdown 有标题层级，代码有函数和类的边界。结构化切块就是利用这些天然的语义边界来切割。

对于 Markdown 文档，按 H1/H2/H3 标题切割，每块自动带上层级元数据（比如”车型介绍 > ET9 > 续航”）。这样检索的时候，元数据可以作为过滤条件，进一步提升精准度。

from langchain.text_splitter import MarkdownHeaderTextSplitter

headers_to_split_on = [
    ("#",   "H1"),
    ("##",  "H2"),
    ("###", "H3"),
]
splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=headers_to_split_on
)
chunks = splitter.split_text(markdown_text)
# 每块自动带上层级元数据：
# chunk.metadata = {"H1": "车型介绍", "H2": "ET9", "H3": "续航"}

对于代码文档，按函数和类的边界切割。这里用到了 AST（抽象语法树）解析——把代码解析成语法树，按函数定义节点来切，保证每一块都是一个完整的函数或类，不会把函数体从中间截断。这对代码问答场景非常重要，因为截断的代码对 LLM 来说几乎没有参考价值。

from tree_sitter import Language, Parser

def split_code_by_ast(code: str, language: str):
    parser = get_parser(language)
    tree   = parser.parse(bytes(code, "utf-8"))
    chunks = []
    for node in tree.root_node.children:
        if node.type in ["function_definition", "class_definition"]:
            chunks.append({
                "content": code[node.start_byte:node.end_byte],
                "type":    node.type,
                "name":    get_node_name(node)
            })
    return chunks

这个策略的前提是文档本身结构良好。如果文档格式混乱、标题不规范，效果就会大打折扣。

策略五：情境化切块（Contextual Chunking）（Anthropic 提出）

它解决的问题是：很多 chunk 脱离了原文之后会变得”失忆”。比如一个 chunk 里只有”支持最快 500kW 超充”，单独看完全不知道这是在说哪款车、哪个配置，向量化之后信息量很低。

Contextual Chunking 的做法是：在存储每个 chunk 之前，先用 LLM 为它写一段情境说明，告诉它”你在整篇文档的哪个位置、你在讲什么”。然后把情境说明和原始内容拼在一起，作为一个整体去做 Embedding。

这样同样的内容向量就携带了足够的语境信息，即使用户问的问题措辞模糊，也能精准命中。

代价是：每个 chunk 在入库前都要调用一次 LLM，预处理成本会显著增加。适合对质量要求很高、文档库相对稳定（不是每天大量更新）的场景。

def contextual_chunking(document, chunks, llm):
    contextualized = []
    for chunk in chunks:
        context = llm(f"""
文档内容：{document.content}

以下是文档中的一个片段：
{chunk.content}

请用1-2句话说明这个片段在整篇文档中的位置和主要内容：
""")
        # 情境说明 + 原始内容一起向量化
        contextualized.append(
            Document(
                content=f"{context}\n\n{chunk.content}",
                metadata=chunk.metadata
            )
        )
    return contextualized

# 效果示例：
# 原始 chunk："支持最快 500kW 超充"
# 加了情境说明：
#   "本段描述蔚来 ET9 的充电性能规格，属于动力系统参数介绍。
#    支持最快 500kW 超充"
# → 即使用模糊问题也能精准命中

Chunking 策略对比

Retrieval：优化检索本身

目标：找到更准、更全的相关片段

BM25 是什么

BM25（Best Matching 25）是传统信息检索领域最经典的关键词匹配算法，TF-IDF 的改进版本。

Robertson & Zaragoza (2009). The Probabilistic Relevance Framework: BM25 and Beyond. Foundations and Trends in Information Retrieval.

https://www\.researchgate\.net/publication/220613776\_The\_Probabilistic\_Relevance\_Framework\_BM25\_and\_Beyond

BM25 vs 向量检索对比：

混合检索：为什么必须两路并行

向量检索的盲区：
  查"ET9 VIN 码" → 找到"车辆信息查询"（语义相近）
                   但可能找不到精确包含"VIN"的文档

BM25 的盲区：
  查"补能方式" → 找不到"充电和换电服务介绍"
                （不包含"补能"这个词）

混合检索 = 向量 + BM25 并行，RRF 融合，各取所长

RRF 融合算法

为什么不直接加权平均分数？

向量相似度：0.92（范围 0~1）
BM25 分数：  23.4（范围可以很大）

直接加权：0.92×0.6 + 23.4×0.4 = 9.912
→ BM25 的绝对值远大于余弦相似度
→ BM25 会完全主导结果，权重设置完全失效

RRF 的解法：不看分数，只看排名

公式：RRF_score(d) = Σ 1 / (k + rank_i(d))
k = 60（平滑参数，防止第1名权重过大）

k=60 的含义：
  不加 k：第1名 1/1=1.000，第2名 1/2=0.500，差距 2 倍
  加 k=60：第1名 1/61=0.0164，第2名 1/62=0.0161，差距 1.6%
  → 排名差距对最终得分影响更平滑

手工算一遍（k=60）：
  向量检索：文档A排1，文档B排2，文档C排3
  BM25：    文档C排1，文档A排2，文档D排3

  文档A：1/(60+1) + 1/(60+2) = 0.01639 + 0.01613 = 0.03252
  文档B：1/(60+2) + 0         = 0.01613
  文档C：1/(60+3) + 1/(60+1)  = 0.01587 + 0.01639 = 0.03226
  文档D：0         + 1/(60+3)  = 0.01587

  最终排名：文档A > 文档C > 文档B > 文档D
  → 文档A 两路都靠前，融合后排第一，符合直觉 ✓

from langchain.retrievers import EnsembleRetriever, BM25Retriever

bm25   = BM25Retriever.from_documents(documents)
bm25.k = 10
dense  = vectorstore.as_retriever(search_kwargs={"k": 10})

# 内部使用 RRF 融合
hybrid = EnsembleRetriever(
    retrievers=[bm25, dense],
    weights=[0.4, 0.6]   # 根据场景调整
)

权重参考：

大量专有名词/型号/数字 → BM25 权重调高（0.6/0.4）
问题口语化/模糊 → Dense 权重调高（0.3/0.7）
均衡场景（默认）→ 0.4/0.6

Post-Retrieval：优化检索结果

目标：把最有价值的片段喂给 LLM，去掉噪音

Reranking（重排序）

为什么需要 Reranking

在生产级 RAG 里，检索几乎总是分两个阶段：

全量文档（百万级）
   │ ① 召回（Retrieval）：快速从海量中拉出 Top-K（K=20~100）
   ▼
候选集（几十条）
   │ ② 重排（Reranking）：精确打分，挑出真正相关的 Top-N（N=3~5）
   ▼
最终送给 LLM 的上下文

第一阶段必须快（毫秒级 + 海量数据），所以用 Bi-Encoder（粗排），精度有限

LLM 对噪声极其敏感，塞 50 条进去效果反而下降，必须挑出最相关的几条

中间用一个慢但准的 Reranker 来把关

召回保”全”（Recall），重排保”准”（Precision）。

Bi-Encoder vs Cross-Encoder —— 架构差异是关键

Bi-Encoder（双塔模型）—— 检索阶段用

Query  ──► [Encoder] ──► query 向量
Doc    ──► [Encoder] ──► doc 向量
                              │
                              ▼
                         计算余弦相似度

特点：
- Query 和 Doc **分别独立编码**，互不影响
- Doc 向量可以**离线预计算并存入向量库**
- 在线只需算一次 query 向量 + 向量检索 → **极快**
- 但 Query 和 Doc 之间没有直接交互，**精度受限**

Cross-Encoder（交叉编码器）—— 重排阶段用

[CLS] Query [SEP] Doc [SEP] ──► [Transformer 联合编码] ──► 相关性分数

特点：
- Query 和 Doc **拼在一起**喂进同一个 Transformer
- 模型内部的 注意力 能**逐 token 对齐** Query 和 Doc 的语义
- 必须在线计算，**无法预计算**
- **精度高出一截**，但每对 (query, doc) 都要过一次模型 → **慢**

对比表：

	Bi-Encoder（向量检索）	Cross-Encoder（Reranker）
编码方式	Query / Doc 分开	Query + Doc 拼接联合编码
离线预计算	✅ 可以	❌ 不行
速度	快（毫秒级，百万文档）	慢（百毫秒，几十文档）
精度	中	高
适用阶段	召回	精排

所以两者不是替代，而是配合：粗排用 Bi-Encoder 把范围缩小，精排用 Cross-Encoder 精挑细选。

标准代码：召回 + 精排两阶段

from sentence_transformers import CrossEncoder

# 第一阶段：召回 Top-50（快）
query = "蔚来 ET9 的智驾芯片是什么？"
candidates = vectorstore.similarity_search(query, k=50)

# 第二阶段：Reranker 精排（准）
# BGE-Reranker 系列对中文友好，可私有化
reranker = CrossEncoder("BAAI/bge-reranker-large")

# 把 (query, doc) 成对输入，每对给一个分数
pairs = [(query, doc.page_content) for doc in candidates]
scores = reranker.predict(pairs)

# 按分数重排，取 Top-3
ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)
top_docs = [doc for doc, _ in ranked[:3]]

输出大致是这样：

召回的 Top-50 里：
  doc_5  (向量相似度 0.82) → Reranker 打分 0.95 ✅ 真正相关
  doc_1  (向量相似度 0.88) → Reranker 打分 0.41   ← 字面像但语义不对
  doc_12 (向量相似度 0.79) → Reranker 打分 0.92 ✅
  ...

向量检索排第 1 的，重排后未必是最相关的。这就是 Reranking 的价值。

上下文压缩

检索到的文档往往包含大量无关内容，只提取相关句子：

from langchain.retrievers.document_compressors import LLMChainExtractor

compressor = LLMChainExtractor.from_llm(llm)
# 输入：整段文档 + 用户问题
# 输出：只保留和问题相关的句子
compressed = compressor.compress_documents(docs, query)

Lost in the Middle

LLM 对上下文首尾注意力强、中间弱，重要文档放首尾：

def reorder_for_llm(docs):
    """重要文档放首尾，次要文档放中间"""
    reordered = []
    for i, doc in enumerate(docs):
        if i % 2 == 0:
            reordered.insert(0, doc)  # 放最前
        else:
            reordered.append(doc)     # 放最后
    return reordered

优化手段优先级

按性价比排序，指导实施顺序：

先上 Chunking + 混合检索 + Reranking，这三个投入产出比最高，能解决 80% 的质量问题。

Modular RAG — 怎么做到灵活组合

理解为什么要把流水线拆成积木

概览

论文原话：Advanced RAG 是 Modular RAG 的一个特例，Naive RAG 是 Advanced RAG 的一个特例。

论文：Modular RAG: Transforming RAG Systems into LEGO-like Reconfigurable Frameworks

https://arxiv\.org/abs/2407\.21059

三层架构

L1 Module Level      六大模块：Indexing / Pre-Retrieval / Retrieval /
                               Post-Retrieval / Generation / Orchestration

L2 Sub-module Level  每个模块内的具体操作（可插拔替换）
                     比如 Retrieval 下有：向量检索、BM25、知识图谱

L3 Operator Level    最细粒度的算法实现
                     比如向量检索下有：余弦相似度、点积、HNSW 索引

#!/usr/bin/env python3
# -*- encoding: utf-8 -*-

import os
from dataclasses import dataclass, field
from typing import Callable, List

os.environ["NO_PROXY"] = "127.0.0.1,localhost"

from langchain_chroma import Chroma
from langchain_ollama import ChatOllama, OllamaEmbeddings
from langchain_community.retrievers import BM25Retriever
from langchain_classic.retrievers import EnsembleRetriever
from langchain_core.documents import Document
from langchain_text_splitters import RecursiveCharacterTextSplitter
from sentence_transformers import CrossEncoder

# ---------------------------------------------------------------------------
# 全局组件 (一次性初始化,所有模块共享)
# ---------------------------------------------------------------------------
OLLAMA_HOST = "http://localhost:11434"
CHAT_MODEL = "qwen2.5"
EMBED_MODEL = "bge-m3"

embeddings = OllamaEmbeddings(model=EMBED_MODEL, base_url=OLLAMA_HOST)
llm = ChatOllama(model=CHAT_MODEL, base_url=OLLAMA_HOST, temperature=0)
reranker = CrossEncoder("BAAI/bge-reranker-base")  # 也可换更小的 ms-marco-MiniLM-L-6-v2

# ---------------------------------------------------------------------------
# Pre-Retrieval 模块: query -> [query, ...]
# 输入一个原始问题,输出一个或多个用于检索的查询
# ---------------------------------------------------------------------------
def pre_passthrough(query: str) -> List[str]:
    """直接透传,不做任何处理"""
    return [query]

def pre_rewrite(query: str) -> List[str]:
    """LLM 改写:口语 -> 检索友好"""
    prompt = f"将以下口语化问题改写为适合检索的精确语句,只输出改写结果:\n{query}"
    return [llm.invoke(prompt).content.strip()]

def pre_decompose(query: str) -> List[str]:
    """LLM 分解:复杂问题 -> 多个子问题"""
    prompt = (
        "请把下面的问题分解为 2~3 个相互独立的子问题,每行一个,不要编号、不要多余文字:\n"
        f"{query}"
    )
    raw = llm.invoke(prompt).content.strip()
    subs = [s.strip("-•　 \t") for s in raw.splitlines() if s.strip()]
    return subs or [query]

# ---------------------------------------------------------------------------
# Retriever 模块: List[query] -> List[Document]
# ---------------------------------------------------------------------------
def make_dense_retriever(docs: List[Document], k: int = 5) -> Callable[[List[str]], List[Document]]:
    vs = Chroma.from_documents(docs, embeddings)
    base = vs.as_retriever(search_kwargs={"k": k})

    def _run(queries: List[str]) -> List[Document]:
        seen, out = set(), []
        for q in queries:
            for d in base.invoke(q):
                if d.page_content not in seen:
                    seen.add(d.page_content)
                    out.append(d)
        return out
    return _run

def make_hybrid_retriever(docs: List[Document], k: int = 10,
                          weights=(0.4, 0.6)) -> Callable[[List[str]], List[Document]]:
    vs = Chroma.from_documents(docs, embeddings)
    dense = vs.as_retriever(search_kwargs={"k": k})
    bm25 = BM25Retriever.from_documents(docs)
    bm25.k = k
    hybrid = EnsembleRetriever(retrievers=[bm25, dense], weights=list(weights))

    def _run(queries: List[str]) -> List[Document]:
        seen, out = set(), []
        for q in queries:
            for d in hybrid.invoke(q):
                if d.page_content not in seen:
                    seen.add(d.page_content)
                    out.append(d)
        return out
    return _run

# ---------------------------------------------------------------------------
# Post-Retrieval 模块: (query, docs) -> docs
# ---------------------------------------------------------------------------
def post_topk(query: str, docs: List[Document], top_k: int = 5) -> List[Document]:
    """直接截取前 top_k(检索器自带顺序)"""
    return docs[:top_k]

def post_rerank(query: str, docs: List[Document], top_k: int = 3) -> List[Document]:
    """CrossEncoder 重排"""
    if not docs:
        return docs
    pairs = [(query, d.page_content) for d in docs]
    scores = reranker.predict(pairs)
    ranked = sorted(zip(scores, docs), key=lambda x: x[0], reverse=True)
    return [d for _, d in ranked[:top_k]]

# ---------------------------------------------------------------------------
# Generator 模块: (query, docs) -> answer
# ---------------------------------------------------------------------------
def gen_simple(query: str, docs: List[Document]) -> str:
    context = "\n".join(d.page_content for d in docs)
    return llm.invoke(
        f"基于以下资料回答问题,若资料不足请直说:\n{context}\n\n问题: {query}"
    ).content.strip()

# ---------------------------------------------------------------------------
# Pipeline: 用配置串起所有模块
# ---------------------------------------------------------------------------
@dataclass
class RAGPipeline:
    name: str
    pre: Callable[[str], List[str]]
    retriever: Callable[[List[str]], List[Document]]
    post: Callable[[str, List[Document]], List[Document]]
    generator: Callable[[str, List[Document]], str] = gen_simple
    post_kwargs: dict = field(default_factory=dict)

    def run(self, query: str) -> str:
        print(f"\n========== [{self.name}] ==========")
        print(f"Q: {query}")

        # 1) Pre-Retrieval
        queries = self.pre(query)
        print(f"  pre-retrieval -> {queries}")

        # 2) Retrieval
        candidates = self.retriever(queries)
        print(f"  retrieval     -> {len(candidates)} docs")

        # 3) Post-Retrieval
        final_docs = self.post(query, candidates, **self.post_kwargs)
        print(f"  post          -> {len(final_docs)} docs")
        for i, d in enumerate(final_docs, 1):
            print(f"      [{i}] {d.page_content[:50].replace(chr(10),' ')}...")

        # 4) Generation
        answer = self.generator(query, final_docs)
        print(f"A: {answer}")
        return answer

# ---------------------------------------------------------------------------
# 数据
# ---------------------------------------------------------------------------
# ── 切分 chunk ────────────────────────────────────────────
def chunk_documents(documents: list[str],
                    chunk_size:int=200,
                    chunk_overlap:int=40,
                    separators:list=["\n\n", "\n", "。", "；", "，", " ", ""]) -> list[Document]:
    """递归字符切分:优先按段落/句子边界,其次才硬切到 chunk_size"""
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        separators=separators
    )
    return splitter.split_documents(docs)

# ---------------------------------------------------------------------------
# 三套预设配置 —— 演示"换模块组合 = 换范式"
# ---------------------------------------------------------------------------
def build_presets(chunks: List[Document]) -> List[RAGPipeline]:
    dense = make_dense_retriever(chunks, k=5)
    hybrid = make_hybrid_retriever(chunks, k=10)

    return [
        RAGPipeline(
            name="naive (dense + topk)",
            pre=pre_passthrough,
            retriever=dense,
            post=post_topk,
            post_kwargs={"top_k": 3},
        ),
        RAGPipeline(
            name="hybrid + rewrite + rerank",
            pre=pre_rewrite,
            retriever=hybrid,
            post=post_rerank,
            post_kwargs={"top_k": 3},
        ),
        RAGPipeline(
            name="decompose + hybrid + rerank",
            pre=pre_decompose,
            retriever=hybrid,
            post=post_rerank,
            post_kwargs={"top_k": 4},
        ),
    ]

# ---------------------------------------------------------------------------
# 入口: 同一个问题跑三套配置,直观对比
# ---------------------------------------------------------------------------
if __name__ == "__main__":
    documents = [
        "2026年1月，蔚来公司交付新车27,182台，同比增长96.1%。截至目前，蔚来公司已累计交付新车1,024,774台。今天下午，蔚来全新ES8在广州完成第60,000台新车交付，历时134天。纯电大三排SUV的黄金时代已经到来。在「Banyan 榕 3.3.0」版本推送后，新ET5、ET5T、ES6和EC6以及全新ES8和ET9的智能系统也将迭代至1.4.0版本，欢迎大家体验最新的世界模型辅助驾驶版本。今天，蔚来品牌针对新ET5、ET5T、ES6和EC6、乐道品牌针对乐道L60与L90推出7年超低息购车方案，7年0.49%年化费率，0金融服务费，提前还款0违约金。firefly萤火虫品牌推出7年超低息购车方案，锁单赠送马年新春奇遇礼包。蔚来换电次数已突破9,900万次，还有几天，1亿次换电就将达成。新春佳节即将到来，提前预祝大家马年大吉，万事顺遂。新的一年我们继续努力，一起加电！",
        "2026年2月，蔚来公司交付新车20,797台，同比增长57.6%。截至目前，蔚来公司已累计交付新车1,045,571台。2月27日，全新ES8在温州完成第70,000台新车交付，历时160天。纯电大三排SUV的黄金时代已经到来。2月6日22:33:18，蔚来1亿次换电达成；2026春节期间，蔚来累计提供换电服务2,073,500次，日均服务量较去年春节日均增长29.4%。2月15日至2月23日，蔚来能源在高速公路充换电量累计超2528万kWh，占全国高速公路充换电总量15%。自2月18日（大年初二）起，蔚来换电连续5日刷新单日服务纪录。2月26日，蔚来芯片子公司安徽神玑技术有限公司完成首轮股权融资协议签署，融资金额超22亿元人民币，投后估值近百亿。此次融资将有利于神玑公司持续地研发和推广高端、高竞争力的芯片产品，支撑蔚来在自动驾驶、具身智能等领域的长远布局。今天，蔚来品牌针对全新ES8提供购置税补贴10,000元，针对新ET5、ET5T、ES6和EC6推出7年超低息购车方案，首付比例20%起，7年0.49%年化费率。乐道品牌继7年超低息金融方案后，加推购置税全兜底政策，全系至高补贴10,262元。全国门店优选现车充足，专属礼遇享更多优惠。firefly萤火虫品牌在7年超低息购车方案上推出3月限时购车礼，购车即享3,000元选装基金，与2,000元购置税补贴或7kW交流家充桩（30米套包），另有其他多项购车优惠好礼。一年之计在于春，我们继续努力，一起加电！",
        "2026年3月，蔚来公司交付新车35,486台，同比增长136.0%，环比增长70.6%。其中，蔚来品牌交付新车22,490台，同比增长120.1%，环比增长48.4%；乐道品牌交付新车6,877台，同比增长42.7%，环比增长130.7%；firefly萤火虫品牌交付新车6,119台，环比增长130.3%。2026年一季度，蔚来公司共交付新车83,465台，同比增长98.3%。截至目前，蔚来公司已累计交付新车1,081,057台。3月20日，蔚来全新ES8完成第80,000台新车交付，历时181天。全新ES8已连续三个月斩获「大型 SUV」不分能源形式、不分价格区间、不分座位数销量冠军。乐道L90单月交付环比增长155.1%，推动乐道品牌累计交付量突破14万台。蔚来全新ES8和乐道L90共同引领纯电大三排SUV进入黄金时代。3月29日，firefly萤火虫迎来第50,000台用户交付，品牌迈入全新发展阶段。明天起，蔚来公司将进入密集的新品发布阶段。2026款ES6、EC6、ET5、ET5T将于4月2日正式上市，年款车型在设计、配置和智驾体验上均有升级。科技行政旗舰SUV蔚来ES9产品技术发布会将于4月9日正式召开。乐道L90 2026款将在4月下旬发布，新增激光雷达旗舰智驾版，乐道L80也即将上市。3月10日，蔚来公司发布2025年四季度及全年财报，宣布四季度实现经营利润12.5亿元，公司首次实现季度盈利。四季度现金储备459亿元，环比大幅增长近百亿。今天公布的一季度交付数超出了2026年一季度交付指引（80,000台-83,000台）。暖春四月，万象更新。我们继续努力，一起加电！",
        "2026年4月，蔚来公司交付新车29,356台，同比增长22.8%。其中，蔚来品牌交付新车19,024台；乐道品牌交付新车5,352台，同比增长21.6%；firefly萤火虫品牌交付新车4,980台。今年前四个月，蔚来公司共交付新车112,821台，同比增长71.0%。截至目前，蔚来公司已累计交付新车1,110,413台。从4月份开始，公司进入新产品密集发布和交付期，生产制造、市场营销等团队围绕新产品积极开展工作。2026款ES6、EC6、ET5、ET5T在四月初上市并交付；新款萤火虫已经开始交付；2026款乐道L90正式上市，将于5月9日开启交付；重磅新车乐道L80已经发布，将于5月15日上市交付；科技行政旗舰SUV蔚来ES9也将于5月底上市并交付。相信大家能选到适合自己的产品。历时215天完成第十万台交付，刷新中国乘用车市场40万元以上高端车型的销售和交付纪录的全新ES8，也在4月24日开幕的北京车展上发布了玄金特别版。这次北京车展，蔚来公司以「大有新境 A Vast New Realm」为主题，携NIO蔚来、ONVO乐道、firefly萤火虫三大品牌首次同台亮相，目前北京车展正迎来观展高峰，欢迎朋友们五一假期到蔚来公司展台做客。祝大家节日愉快，出行平安。我们继续努力，一起加电！",
        "2026年4月，蔚来全新ES8完成13,020台新车交付。截至目前，全新ES8已经拥有超过100,000名用户，并且刷新了中国40万元以上高端车型交付十万台最快纪录。此外，全新ES8已经连续四个月斩获大型SUV和40万元以上车型销量冠军，纯电大三排SUV的黄金时代已经到来！,即日起至2026年5月31日（含），购车用户可申请蔚来指定合作机构提供的5年60期分期金融方案，0首付起，前3年36期年化费率0%，后2年24期年化费率为3%。同时，订购全新ES8并成功锁单的用户，可享限时购置税补贴10,000元，5年「全域领航辅助 NOP+」免费使用权，以及飞航头等舱五件套。"      
    ]
    docs = [Document(page_content=text) for text in documents]

    # 1. 切分 chunk
    chunks = chunk_documents(docs)
    print(f"原始文档 {len(docs)} 条 → 切分后 {len(chunks)} 个 chunk")

    pipelines = build_presets(chunks)
    query = "蔚来三月销量怎么样?"

    for p in pipelines:
        p.run(query)

编排层（Orchestration)

Orchestration 是控制 RAG 整体流程的模块。与传统固定流程不同，Modular RAG 在关键节点引入决策机制，根据前一步的输出动态选择后续操作。这种自适应能力正是 Modular RAG 区别于 Naive RAG 和 Advanced RAG 的核心标志。

Orchestration
  ├── Routing（路由）    → 决定走哪条路
  ├── Scheduling（调度） → 决定怎么走、按什么顺序
  └── Fusion（融合）     → 多路结果如何合并

路由（Routing）

router_prompt = """
判断以下问题需要哪种处理方式，返回 JSON：
- "direct": 简单事实，直接回答
- "simple_rag": 需要单次检索
- "multi_hop": 需要多步推理
- "iterative": 需要迭代检索

问题：{question}
"""

# "今天几号"            → {"type": "direct"}
# "ET9 续航多少"        → {"type": "simple_rag"}
# "ET9 和 L9 哪个更好" → {"type": "multi_hop"}
route = json.loads(llm(router_prompt.format(question=query)))

面对不同 query，RAG 系统会将其路由到针对不同场景定制的特定管道。这是一个多功能 RAG 架构的核心能力。系统需要根据模型输入或元数据，判断应当调用哪些模块。

我的理解：Routing 本质上是一个”分流器”或”总调度入口”。当一条 query 进来时，系统首先要判断这条 query 属于哪种类型、需要什么样的处理管道。

具体实现上有几种典型做法：

基于 LLM 的语义判断：把 query 交给 LLM，让它判断应走结构化数据查询路径（如 Text-to-SQL）还是非结构化向量检索路径。
基于元数据/规则的匹配：例如，query 中包含时间词就走知识图谱；含有代码就走代码检索库。
query 分类分桶：预先训练一个轻量分类器，将 query 归入几个桶（factoid / reasoning / open-ended），不同桶激活不同模块组合。

调度（Scheduling）

各组件可能需要按照特定顺序执行，或在某些条件下并行处理，甚至需要 LLM 根据不同输出进行判断。合理规划工作流对于提升系统效率和实现预期结果至关重要。

我的理解：如果说 Routing 决定的是”走哪条路”，那 Scheduling 决定的就是”怎么走这条路”。它管理的是流程的执行策略，包括：

顺序执行：最基础的线性调度，按步骤依次执行。
迭代调度：循环检索，每轮根据前一轮结果追加检索（对应 Loop Pattern）。
递归调度：对复杂 query 进行拆分，子问题递归检索汇总。
自适应调度：LLM 在每个决策节点主动判断”当前知识够不够，要不要再检索”（如 FLARE、Self-RAG 的思路）。

调度的精髓在于将 LLM 从”纯生成器”升格为”规划器”，LLM 不只是最后输出答案，而是全程参与任务的执行计划制定。

Scheduling 支持迭代或递归控制——例如重复检索-生成循环直到调度器或阈值条件满足，涵盖主动/自适应检索、多跳或树状结构查询。

融合（Fusion）

我的理解：Fusion 是 Branching 模式的”收口”阶段。当系统并行执行了多条路径之后，需要一种机制把这些分散的结果整合成一个高质量的最终输出。常见的融合策略有：

RRF（Reciprocal Rank Fusion，倒数排名融合）：对多路检索结果按排名加权合并，是混合检索中最常用的方案。
投票与加权合并：多个模型或多条子查询各自生成候选答案，少数服从多数或按置信度加权。
蒸馏/摘要压缩：多路文档先由 LLM 压缩为精要，再统一生成，避免上下文爆炸。
答案聚合生成：每个子查询分别生成局部回答，Fusion 环节用 LLM 把这些局部回答综合为完整答案。

三种编排实现方式

生产推荐：三层防御

第一层：规则（<1ms）
  极短问题 / 闲聊 / 敏感词 → 直接处理

第二层：轻量分类器（~100ms）
  判断 simple / multi_hop / iterative
  决定调哪些模块

第三层：执行中动态调整
  Loop Control 评估质量
  不够好就再来一轮，最多 N 次防止死循环

四种 Flow 模式

Agentic RAG

Agentic RAG 是 RAG 演进中一个质的飞跃——从”被动回答”变成”主动规划”。Agentic RAG 则把 LLM 变成一个主动推理的 Agent：它可以自主决定检索什么、何时检索、是否需要多步推理，并在多次工具调用之间保持状态。

和前几代的本质升级

Naive/Advanced RAG：
  用户问 → 系统检索 → LLM 回答（被动，流程写死）

Modular RAG：
  用户问 → 编排层路由 → 动态选模块（规则还是人写的）

Agentic RAG：
  用户问 → Agent 自主规划 → 决定用哪些工具
          → 执行 → 反思结果 → 决定是否继续
          → 迭代优化 → 回答（主动，规则由模型决定）

核心升级：控制权从代码转移到模型

核心框架：ReAct

Agentic RAG 最常用的推理框架是 ReAct（Reason + Act），每一步都遵循三个动作交替进行：

Thought（我需要什么）→ Action（去检索/调用工具）→ Observation（拿到了什么）→ 重复，直到 Final Answer

Thought：我需要 ET9 和 L9 的数据，先查 ET9
Action：search_kb("ET9 续航 价格 配置")
Observation：ET9 售价 75 万，续航 720 公里

Thought：还需要 L9 的数据
Action：search_kb("理想 L9 续航 价格")
Observation：L9 售价 45 万，纯电续航 215 公里

Thought：两个数据都有了，可以对比了
Action：answer("综合对比分析...")

from langchain.agents import create_react_agent, AgentExecutor

tools = [
    Tool("search_kb",  kb_search,  "搜索内部知识库"),
    Tool("web_search", web_search, "搜索互联网"),
    Tool("sql_query",  sql_query,  "查询结构化数据"),
    Tool("calculate",  calculate,  "数学计算"),
]

agent    = create_react_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, max_iterations=5)
result   = executor.invoke({"input": "ET9 和 L9 哪个更适合长途出行？"})

#!/usr/bin/env python3
# -*- encoding: utf-8 -*-

import json
import os
import re
from typing import Any, Callable, Dict, List

os.environ["NO_PROXY"] = "127.0.0.1,localhost"

from langchain_chroma import Chroma
from langchain_ollama import ChatOllama, OllamaEmbeddings
from langchain_core.documents import Document
from langchain_text_splitters import RecursiveCharacterTextSplitter

# ---------------------------------------------------------------------------
# 配置 & 全局组件
# ---------------------------------------------------------------------------
OLLAMA_HOST = "http://localhost:11434"
CHAT_MODEL = "qwen3"
EMBED_MODEL = "bge-m3"
MAX_STEPS = 5            # Agent 最多循环几轮
TOP_K = 3                # 每次 search 召回几条

embeddings = OllamaEmbeddings(model=EMBED_MODEL, base_url=OLLAMA_HOST)
llm = ChatOllama(model=CHAT_MODEL, base_url=OLLAMA_HOST, temperature=0)

# ---------------------------------------------------------------------------
# Query 改写: 把 Agent 给的 query 改成检索友好的形式
#   - 去口语化、聚焦关键词
#   - 不改变意图
# 对 Agent 是"透明"的: 它仍然只看到一个 search 工具
# ---------------------------------------------------------------------------
REWRITE_PROMPT = (
    "你是检索查询优化器。把下面的查询改写为更适合向量/关键词检索的简短陈述句, "
    "保留所有关键实体(品牌、时间、数字、专有名词等), 去掉口语化词、疑问语气、多余修饰。"
    "只输出改写后的查询本身, 不要解释:\n查询: {q}"
)

def rewrite_query(query: str) -> str:
    out = llm.invoke(REWRITE_PROMPT.format(q=query)).content.strip()
    # 去掉可能的引号 / 前后缀
    out = out.strip().strip("\"'“”‘’`")
    return out or query

# ---------------------------------------------------------------------------
# 工具:这里只内置 search + finish 两个
#   - search 内部会先做 query 改写, 然后再去向量库检索
# ---------------------------------------------------------------------------
def make_search_tool(docs: List[Document]) -> Callable[[str], str]:
    """构建向量检索工具,返回一个 (query)->观察文本 的函数"""
    vs = Chroma.from_documents(docs, embeddings)
    retriever = vs.as_retriever(search_kwargs={"k": TOP_K})

    def _search(query: str) -> str:
        rewritten = rewrite_query(query)
        print(f"    [rewrite] {query!r} -> {rewritten!r}")
        hits = retriever.invoke(rewritten)
        if not hits:
            return "(no results)"
        return "\n".join(f"[{i}] {d.page_content}" for i, d in enumerate(hits, 1))
    return _search

# ---------------------------------------------------------------------------
# Agent 提示词 —— 强制 LLM 用 JSON 表达"想什么/做什么"
# ---------------------------------------------------------------------------
SYSTEM_PROMPT = """你是一个能调用工具的 RAG Agent。

你能使用的工具:
  - search(query: str): 在本地知识库中检索相关片段
  - finish(answer: str): 你已经能回答问题时,用这个工具给出最终答案

每一步你必须只输出一个 JSON 对象 (不要 markdown 代码块, 不要多余文字),
格式严格如下:
{"thought": "你的推理过程(简短)", "action": "search" 或 "finish", "args": {"query": "..."} 或 {"answer": "..."}}

策略:
- 如果信息不足, 用 search 多次查不同的子问题 (比如把"A 和 B"拆成两次 search)
- 检索到足够信息后, 立即用 finish 给出答案
- 若资料中没有相关信息, finish 的 answer 写"资料中没有提到"
"""

def build_user_prompt(question: str, history: List[Dict[str, Any]]) -> str:
    """把对话历史拼成一个 prompt"""
    lines = [f"Question: {question}\n", "Trace:"]
    if not history:
        lines.append("(empty, this is the first step)")
    for i, step in enumerate(history, 1):
        lines.append(f"  Step {i}:")
        lines.append(f"    thought: {step['thought']}")
        lines.append(f"    action : {step['action']}({json.dumps(step['args'], ensure_ascii=False)})")
        lines.append(f"    observation: {step['observation']}")
    lines.append("\n现在请输出下一步的 JSON。")
    return "\n".join(lines)

def parse_action(raw: str) -> Dict[str, Any]:
    """从 LLM 输出里抽出 JSON。容忍模型加了 ```json fence``` 或前后废话。"""
    # 优先匹配第一段 {...} JSON
    m = re.search(r"\{.*\}", raw, re.DOTALL)
    if not m:
        raise ValueError(f"no JSON found in LLM output:\n{raw}")
    return json.loads(m.group(0))

# ---------------------------------------------------------------------------
# Agent 主循环
# ---------------------------------------------------------------------------
def run_agent(question: str, tools: Dict[str, Callable[..., str]], max_steps: int = MAX_STEPS) -> str:
    history: List[Dict[str, Any]] = []

    for step in range(1, max_steps + 1):
        print(f"\n--- step {step} ---")

        # 1) 让 LLM 决定下一步
        resp = llm.invoke([
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": build_user_prompt(question, history)},
        ]).content

        print("build_user_prompt:",build_user_prompt(question, history))
        print("resp:",resp)
        try:
            decision = parse_action(resp)
            thought = decision.get("thought", "")
            action = decision["action"]
            args = decision.get("args", {})
        except Exception as e:
            print(f"  [warn] failed to parse LLM output ({e}), forcing finish")
            return f"(parse error) raw output: {resp}"

        print(f"  thought: {thought}")
        print(f"  action : {action}({args})")

        # 2) 终止条件
        if action == "finish":
            return args.get("answer", "(no answer)")

        # 3) 执行工具
        if action not in tools:
            observation = f"(error) unknown tool: {action}"
        else:
            try:
                observation = tools[action](**args)
            except TypeError as e:
                observation = f"(error) bad args: {e}"

        print(f"  observation: {observation[:200]}{'...' if len(observation) > 200 else ''}")
        history.append({
            "thought": thought,
            "action": action,
            "args": args,
            "observation": observation,
        })

    return "(stopped: max_steps reached)"

# ── 切分 chunk ────────────────────────────────────────────
def chunk_documents(documents: list[str],
                    chunk_size:int=200,
                    chunk_overlap:int=40,
                    separators:list=["\n\n", "\n", "。", "；", "，", " ", ""]) -> list[Document]:
    """递归字符切分:优先按段落/句子边界,其次才硬切到 chunk_size"""
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        separators=separators
    )
    return splitter.split_documents(documents)

# ---------------------------------------------------------------------------
# 入口
# ---------------------------------------------------------------------------
if __name__ == "__main__":
    documents = [
        "2026年1月，蔚来公司交付新车27,182台，同比增长96.1%。截至目前，蔚来公司已累计交付新车1,024,774台。今天下午，蔚来全新ES8在广州完成第60,000台新车交付，历时134天。纯电大三排SUV的黄金时代已经到来。在「Banyan 榕 3.3.0」版本推送后，新ET5、ET5T、ES6和EC6以及全新ES8和ET9的智能系统也将迭代至1.4.0版本，欢迎大家体验最新的世界模型辅助驾驶版本。今天，蔚来品牌针对新ET5、ET5T、ES6和EC6、乐道品牌针对乐道L60与L90推出7年超低息购车方案，7年0.49%年化费率，0金融服务费，提前还款0违约金。firefly萤火虫品牌推出7年超低息购车方案，锁单赠送马年新春奇遇礼包。蔚来换电次数已突破9,900万次，还有几天，1亿次换电就将达成。新春佳节即将到来，提前预祝大家马年大吉，万事顺遂。新的一年我们继续努力，一起加电！",
        "2026年2月，蔚来公司交付新车20,797台，同比增长57.6%。截至目前，蔚来公司已累计交付新车1,045,571台。2月27日，全新ES8在温州完成第70,000台新车交付，历时160天。纯电大三排SUV的黄金时代已经到来。2月6日22:33:18，蔚来1亿次换电达成；2026春节期间，蔚来累计提供换电服务2,073,500次，日均服务量较去年春节日均增长29.4%。2月15日至2月23日，蔚来能源在高速公路充换电量累计超2528万kWh，占全国高速公路充换电总量15%。自2月18日（大年初二）起，蔚来换电连续5日刷新单日服务纪录。2月26日，蔚来芯片子公司安徽神玑技术有限公司完成首轮股权融资协议签署，融资金额超22亿元人民币，投后估值近百亿。此次融资将有利于神玑公司持续地研发和推广高端、高竞争力的芯片产品，支撑蔚来在自动驾驶、具身智能等领域的长远布局。今天，蔚来品牌针对全新ES8提供购置税补贴10,000元，针对新ET5、ET5T、ES6和EC6推出7年超低息购车方案，首付比例20%起，7年0.49%年化费率。乐道品牌继7年超低息金融方案后，加推购置税全兜底政策，全系至高补贴10,262元。全国门店优选现车充足，专属礼遇享更多优惠。firefly萤火虫品牌在7年超低息购车方案上推出3月限时购车礼，购车即享3,000元选装基金，与2,000元购置税补贴或7kW交流家充桩（30米套包），另有其他多项购车优惠好礼。一年之计在于春，我们继续努力，一起加电！",
        "2026年3月，蔚来公司交付新车35,486台，同比增长136.0%，环比增长70.6%。其中，蔚来品牌交付新车22,490台，同比增长120.1%，环比增长48.4%；乐道品牌交付新车6,877台，同比增长42.7%，环比增长130.7%；firefly萤火虫品牌交付新车6,119台，环比增长130.3%。2026年一季度，蔚来公司共交付新车83,465台，同比增长98.3%。截至目前，蔚来公司已累计交付新车1,081,057台。3月20日，蔚来全新ES8完成第80,000台新车交付，历时181天。全新ES8已连续三个月斩获「大型 SUV」不分能源形式、不分价格区间、不分座位数销量冠军。乐道L90单月交付环比增长155.1%，推动乐道品牌累计交付量突破14万台。蔚来全新ES8和乐道L90共同引领纯电大三排SUV进入黄金时代。3月29日，firefly萤火虫迎来第50,000台用户交付，品牌迈入全新发展阶段。明天起，蔚来公司将进入密集的新品发布阶段。2026款ES6、EC6、ET5、ET5T将于4月2日正式上市，年款车型在设计、配置和智驾体验上均有升级。科技行政旗舰SUV蔚来ES9产品技术发布会将于4月9日正式召开。乐道L90 2026款将在4月下旬发布，新增激光雷达旗舰智驾版，乐道L80也即将上市。3月10日，蔚来公司发布2025年四季度及全年财报，宣布四季度实现经营利润12.5亿元，公司首次实现季度盈利。四季度现金储备459亿元，环比大幅增长近百亿。今天公布的一季度交付数超出了2026年一季度交付指引（80,000台-83,000台）。暖春四月，万象更新。我们继续努力，一起加电！",
        "2026年4月，蔚来公司交付新车29,356台，同比增长22.8%。其中，蔚来品牌交付新车19,024台；乐道品牌交付新车5,352台，同比增长21.6%；firefly萤火虫品牌交付新车4,980台。今年前四个月，蔚来公司共交付新车112,821台，同比增长71.0%。截至目前，蔚来公司已累计交付新车1,110,413台。从4月份开始，公司进入新产品密集发布和交付期，生产制造、市场营销等团队围绕新产品积极开展工作。2026款ES6、EC6、ET5、ET5T在四月初上市并交付；新款萤火虫已经开始交付；2026款乐道L90正式上市，将于5月9日开启交付；重磅新车乐道L80已经发布，将于5月15日上市交付；科技行政旗舰SUV蔚来ES9也将于5月底上市并交付。相信大家能选到适合自己的产品。历时215天完成第十万台交付，刷新中国乘用车市场40万元以上高端车型的销售和交付纪录的全新ES8，也在4月24日开幕的北京车展上发布了玄金特别版。这次北京车展，蔚来公司以「大有新境 A Vast New Realm」为主题，携NIO蔚来、ONVO乐道、firefly萤火虫三大品牌首次同台亮相，目前北京车展正迎来观展高峰，欢迎朋友们五一假期到蔚来公司展台做客。祝大家节日愉快，出行平安。我们继续努力，一起加电！",
        "2026年4月，蔚来全新ES8完成13,020台新车交付。截至目前，全新ES8已经拥有超过100,000名用户，并且刷新了中国40万元以上高端车型交付十万台最快纪录。此外，全新ES8已经连续四个月斩获大型SUV和40万元以上车型销量冠军，纯电大三排SUV的黄金时代已经到来！,即日起至2026年5月31日（含），购车用户可申请蔚来指定合作机构提供的5年60期分期金融方案，0首付起，前3年36期年化费率0%，后2年24期年化费率为3%。同时，订购全新ES8并成功锁单的用户，可享限时购置税补贴10,000元，5年「全域领航辅助 NOP+」免费使用权，以及飞航头等舱五件套。"      
    ]

    # 切 chunk
    raw_docs = [Document(page_content=t) for t in documents]
    chunks = chunk_documents(raw_docs)
    print(f"chunks: {len(chunks)}")

    # 注册工具
    tools = {"search": make_search_tool(chunks)}

    # 故意挑一个 multi-hop 问题:naive RAG 一次检索很难同时覆盖
    question = "2026年1-4月蔚来公司一共交付了多少辆车"
    print(f"\nQuestion: {question}")

    answer = run_agent(question, tools)
    print(f"\n========== Final Answer ==========\n{answer}")

两个重要变体

Self-RAG

论文：Asai et al. (2023). Self-RAG: Learning to Retrieve, Generate, and Critique through Self-Reflection. ICLR 2024. arXiv:2310.11511

让模型自己决定：要不要检索？检索到的东西好不好？生成的答案靠不靠谱？

通过在模型训练时引入四种特殊的反思 Token，让模型在生成过程中自主做出判断。

Self-RAG 的核心机制是四个特殊 Token，每个相当于一个小裁判：

生成流程

判断检索必要性：模型根据 Retrieve 标记决定是否检索。若已有足够知识，直接生成，跳过检索。

并行候选生成：若需检索，获取多个候选文档段落，并对每个段落并行生成续写内容。

相关性评估：用 ISREL 标记过滤掉不相关的候选文档。

支撑性评估：用 ISSUP 标记检测生成内容是否存在幻觉。

质量打分与选优：用 ISUSE 标记对多条候选输出打分，选出最优答案。

特点

无需外部组件，反思能力内化于模型参数
需要对基础模型进行微调训练
多候选并行生成带来较大计算开销
强调模型的自主判断与推理能力

CRAG

论文：Yan et al. (2024). Corrective Retrieval Augmented Generation. arXiv:2401.15884

核心思想：在标准 RAG 的检索步骤之后增加「评估-纠正」模块，根据检索文档的置信度动态选择不同的知识补充策略。

CRAG 使用一个轻量级评估器对检索文档打分，并根据置信度选择以下三条路径：

正确（Correct）— 高置信度

检索文档高度相关，执行知识精炼（Knowledge Refinement）：

将文档切分为细粒度知识条（knowledge strips）
对每条知识打相关性分数
过滤低分片段，拼接高分内容作为最终上下文

模糊（Ambiguous）— 中置信度

评估器对文档相关性不确定，同时执行两条策略：

对本地检索文档进行知识精炼
触发网络搜索补充信息

将两者结合后用于生成答案，取长补短。

错误（Incorrect）— 低置信度

检索文档与查询无关，完全放弃本地检索结果，转向网络搜索（如 Google）重新获取相关信息。

特点

无需重训基础模型，评估器可单独训练，即插即用
始终先检索，再评估质量，流程清晰
低置信度时自动降级到网络搜索，知识覆盖更广

什么时候用 Agentic RAG

框架选型

GraphRAG — 检索向量看不到的东西

论文： Edge et al. (2024). From Local to Global: A Graph RAG Approach to Query-Focused Summarization. arXiv:2404.16130. Microsoft Research.

https://arxiv\.org/abs/2404\.16130

Graphrag 详细介绍【调研】GraphRAG

向量检索的根本盲区

向量检索的逻辑：找和问题语义最相似的文本片段

在两类问题上天然失效：

失效一：全局归纳问题
  "这篇文档的核心主题是什么？"
  → 答案不在任何片段里，向量找不到"最相似的片段"

失效二：多跳关系问题
  "蔚来汽车创始人的出生地在哪里？"
  → 需要：蔚来 → 创始人→ 李斌→ 出生地→ 安徽
  → 向量只能找相似文本，不能沿关系链推理

GraphRAG 的解法

普通 RAG：
  文档 → 片段 → 向量（只有"点"）
  只知道相似，不知道关联

GraphRAG：
  文档 → 实体 + 关系 → 知识图谱 → 社区摘要
  有"点"（实体），有"线"（关系），有"面"（社区摘要）

两个阶段

阶段一：索引期

原始文档
   ↓
1. 切分为 TextUnit（类似 Chunking，基本分析单元）
   ↓
2. LLM 抽取三要素：
   实体（Entity）：蔚来、ET9、李斌、换电站
     → 每个实体有：名称、类型、描述
   关系（Relationship）：
     → 李斌 --[创立]--> 蔚来
     → ET9  --[属于]--> 蔚来产品线
     → ET9  --[采用]--> 空气悬架技术
   声明（Claim）：
     → "ET9 CLTC 续航 720 公里"
     → "蔚来 2024 年交付量超过 22 万辆"
   ↓
3. 构建知识图谱（节点=实体，边=关系）
   ↓
4. 分层 Leiden 算法社区检测（Hierarchical Leiden，逐层细化划分）
   Level 0（最粗）：[蔚来, 理想, 小鹏]      → 新能源汽车社区
   Level 1（中等）：[蔚来全系车型]          → 蔚来产品社区
   Level 2（最细）：[ET9, ES8, EC6, ET7]   → 蔚来轿车社区
   ↓
5. LLM 为每个社区生成摘要报告
   → 包含：社区主题、关键实体、重要关系、核心声明
   → 这些摘要是支撑 Global Search 的核心
   ↓
6. 对实体、关系、社区报告分别向量化存储

为什么社区摘要是关键：

传统 RAG：每次只看到几个片段（局部视角）

GraphRAG：预先为每个社区生成摘要（全局视角）
  → Global Search：只需遍历社区摘要就能回答全局问题
  → 不需要扫描所有原始文档
  → 信息已经被结构化和压缩过

阶段二：查询期（四种检索方式）

Local Search

适合：具体实体/关系的精确查询

流程：
  1. 识别问题中的关键实体（ET9、设计师）
  2. 在图谱中找到对应节点
  3. 沿边检索相关实体和关系（1-2 跳）
  4. 找到相关 TextUnit（原文片段）
  5. 组合上下文生成答案

适合问题：
  "ET9 的设计师是谁？"
  "蔚来换电站覆盖哪些城市？"
  "李斌在蔚来担任什么职位？"
  "蔚来汽车创始人的出生地在哪里？"

Global Search

适合：需要理解整个语料库的宏观问题

流程（Map-Reduce）：
  1. 对所有社区摘要并行提问（Map）
     每个社区各自生成一个局部答案
  2. 汇总所有局部答案（Reduce）
     综合生成最终答案

适合问题：
  "这批文档的核心主题是什么？"
  "蔚来最常被讨论的问题有哪些？"
  "整个语料库反映了什么趋势？"

注意：需要遍历所有社区摘要，LLM 调用次数多，成本较高

DRIFT Search

适合：需要全局视野 + 局部细节

三阶段流程：
  第一阶段（Primer）：
    检索 Top-K 社区摘要 → 生成初步宽泛答案 + 追问列表
    "蔚来在高端市场..." + 追问：["换电网络覆盖？", "ET9配置？", "竞品？"]

  第二阶段（Follow-up）：
    并行执行每个追问的 Local Search
    → 检索具体实体、关系、原文片段
    → 生成中间答案 + 新追问（通常迭代 2 轮）

  第三阶段（Answer）：
    汇总所有中间答案 → 合成最终完整回答

LazyGraphRAG

核心思路：不预生成社区摘要，查询时按需计算

普通 GraphRAG 索引期：
  文档 → 实体抽取 → 图谱 → 社区检测 → LLM生成摘要（最贵步骤）

LazyGraphRAG 索引期：
  文档 → NLP 词频分析 → 简单图结构（不调 LLM，极快）

查询期：按需检索图结构 → 临时生成相关摘要 → 生成

效果（数据来自 Microsoft Research 官方博客，主要对比 Global Search 场景）：
  索引成本 = 接近普通向量 RAG（约完整 GraphRAG 的 0.1%）
  查询质量 = 在多数评测任务上接近完整 GraphRAG
  查询成本 = Global Search 场景下约为完整 GraphRAG 的 1/700

代码示例

# pip install graphrag

# 初始化 + 索引
# graphrag init  --root ./project
# graphrag index --root ./project

import asyncio
from graphrag.query.cli import run_local_search, run_global_search

# 局部查询：适合具体实体关系问题
local = asyncio.run(run_local_search(
    config_filepath="./project/settings.yaml",
    data_dir="./project/output",
    query="蔚来汽车创始人的出生地在哪里？"
))

# 全局查询：适合归纳总结类问题
global_ = asyncio.run(run_global_search(
    config_filepath="./project/settings.yaml",
    data_dir="./project/output",
    query="蔚来的整体战略方向是什么？"
))

适合 vs 不适合

RAG 是过渡性技术吗

质疑一：上下文窗口越来越大

GPT-3.5（2022）：     4K token
GPT-4（2023）：       128K token
Claude 3 / 3.5（2024）：200K token
Claude 4 / GPT-4.1 / Gemini 2.5+（2025~2026）：1M token（已 GA）

"窗口够大了，直接塞进去不就行了？"

质疑二：Generation-Driven Retrieval 出现

Claude Code、Cursor 等工具在代码场景下不依赖向量数据库
而是让模型直接使用 Grep / Glob / Read 等工具自主检索

RAG 不会消失的五个理由

理由一：上下文大不等于便宜

100 万 token 上下文：
  成本：每次查询处理 100 万 token
  延迟：几十秒

RAG：
  每次只处理 3-5 个片段（几千 token）
  延迟：秒级

对于高并发生产系统，成本差距在 100 倍以上

理由二：私有数据实时性问题永远存在

不管模型上下文多大：
  今天刚更新的价格 → 模型不知道
  昨天的服务记录   → 模型不知道
  刚签的合同       → 模型不知道

RAG 是解决"实时私有数据"的唯一正确方式

理由三：精确检索 vs 模糊记忆

长上下文：把 10 万份文档塞进 Prompt
  → 模型"看过"但不一定"记准"关键细节
  → 数字、日期、专有名词精确性下降
  → 所有主流模型在长上下文下，都有可测量的性能衰减（context rot），并非只有接近上限时才退化

RAG：精准检索，原文注入
  → 可溯源，可验证
  → 法律、医疗、金融等高精度场景不可替代所有主流模型在长上下文下都有可测量的性能衰减

理由四：RAG 在持续演进，不是消亡

Naive RAG → Advanced RAG → Modular RAG
→ Agentic RAG → GraphRAG

RAG 是一个持续演进的范式
Generation-Driven Retrieval 本身就是 Agentic RAG 的一种形态

理由五：企业落地的现实约束

大多数企业需要：
  ✓ 答案可追溯（合规要求）
  ✓ 数据不能给模型训练（数据安全）
  ✓ 实时更新知识（业务数据每天在变）
  ✓ 多数据源（文档、数据库、API）
这些约束在可预见的未来不会消失

怎么选

选型决策树

需求评估
   ↓
问题类型单一、场景固定？
   → YES → Advanced RAG，别过度设计
   → NO  ↓
需要跨文档推理 / 实体关系分析？
   → YES → GraphRAG（预算够）/ LazyGraphRAG（预算有限）
   → NO  ↓
需要多数据源 / 复杂业务编排？
   → YES → Modular RAG
   → NO  ↓
需要自主执行任务 / 多步骤推理？
   → YES → Agentic RAG
   → NO  ↓
快速验证 / 资源有限？
   → Naive RAG 起步，快速迭代

各代对应框架

评估指标

原则：先搭评估集，再谈优化。没有评估集的 RAG 优化都是玄学。建议先手工构造 50-100 条代表性 Q&A 作为评估集，每次优化后跑一遍指标对比，用数据说话。

使用 RAGAS 框架评估 RAG 质量，四个核心指标：

问题：

Claude code为什么不用向量数据库，而采用glob、grep这些最原始的工具？