跳转至

RAG 技术

1 RAG 核心流程

离线索引与在线检索生成

🔍 1. RAG 是什么？它为什么能让 LLM 回答训练数据之外的问题？

RAG（检索增强生成）是一种给 LLM 装上“外挂知识库”的技术。 在回答问题前，它先从外部文档中检索相关信息，然后把这些信息和问题一起喂给模型，让模型“参考着回答”，而不是只依赖记忆里的训练数据。

纯 LLM： 问题 → [LLM 的内部参数记忆] → 答案（可能过时、或完全编造）
RAG：    问题 → [检索相关文档] → [文档+问题] → LLM → 有据可查的答案

为什么能回答训练数据之外的问题？

因为 LLM 的答案不再来自它固化在参数里的旧知识，而是来自你实时塞给它的外部资料。这些资料可以是今天的新闻、公司内部文档、产品手册——任何模型训练时没见过的东西。模型只是做了一个“阅读理解”，从你给的上下文中提取答案，而不是调用它那可能过时、可能错乱的记忆。

用代码感受一下差别：

# 纯 LLM：可能产生幻觉
answer = llm("公司最新的休假政策是什么？")
# → "根据我的训练数据（截止2024年），公司的休假政策是..."

# RAG：检索到真实政策文档，再回答
docs = retrieve("公司最新的休假政策")  # 从向量库找到相关文档
answer = llm(f"根据以下资料回答：\n{docs}\n\n问题：公司最新的休假政策？")
# → "根据《2026年员工手册》，公司年假为15天，并可累积..."

这就是 RAG 的本质：把 LLM 从“记忆力有限的考生”变成了“随时可以翻参考书的开卷考生”。

⚙️ 2. RAG 系统的完整工作流程与工程踩坑点

一个生产级 RAG 系统分两大步：离线索引（把知识灌进向量库）和在线检索生成（回答用户问题）。每一步都有容易踩的坑。

image.png

2.1 离线索引阶段

工作流：

  1. 文档清洗：去掉页眉页脚、特殊字符、多余空行，统一编码。

  2. 文本切分：将长文档切成适合 LLM 上下文窗口的“chunk”。常用策略是基于固定长度（如 512 token）但有重叠（overlap）的滑动窗口，或按语义边界（如段落）切分。

  3. 向量化：用 Embedding 模型把每个 chunk 转成向量。

  4. 存入向量数据库：把向量 + 元数据（来源、页码）写入 Chroma / Milvus / Pinecone 等。

示例代码：文档索引

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma

# 1. 切分
splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
chunks = splitter.create_documents([raw_text])

# 2. 向量化并存储
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(chunks, embeddings, persist_directory="./db")
vectorstore.persist()

常见踩坑点：

  • 切分太粗：一个 chunk 包含多个话题，检索到的内容里只有一句话相关，其余全是噪音，LLM 容易“看花眼”。

  • 切分太细：关键信息被切断，比如“根据《公司法》第三十八条……”和“……公司可以……”被分在了两个 chunk，检索只命中一半，导致回答不完整。

  • 元数据丢失：没有存下文档标题、页码、发布日期等，后续无法给用户引用来源，也无法做时效性过滤。

  • Embedding 模型不匹配：用英文模型给中文做向量化，语义检索准确率会打折扣。

2.2 在线检索生成阶段

工作流：

  1. 用户问题 → 用同一套 Embedding 模型向量化。

  2. 在向量库中做相似度搜索，取 top-K（通常 K=3~5）。

  3. 把检索到的文档片段拼接到 Prompt 中，明确告诉 LLM “基于以下资料回答，不知道就说不知道”。

  4. LLM 生成答案，可选地附上引用来源。

示例代码：在线问答

def ask(question):
    # 检索
    docs = vectorstore.similarity_search(question, k=4)
    context = "\n\n".join([f"【来源{i+1}】{d.page_content}" for i,d in enumerate(docs)])

    # 生成
    prompt = f"""根据以下参考资料回答用户问题。如果资料中没有答案，请如实告知。
---
{context}
---
问题：{question}"""
    return llm(prompt)

常见踩坑点：

  • 检索不准：用户问“苹果股价”，检索回来一堆“苹果水果”的文档。解决方案是引入 Query Rewriting（让 LLM 先改写用户问题为更精准的检索词），或使用混合检索（向量 + BM25 关键词）。

  • 上下文超窗口：K 设得太大，拼出来的 Prompt 超过了模型上下文限制。需要对检索结果做重排序（Rerank），只保留最相关的片段。

  • LLM 不听话：明明上下文有答案，模型还是自己编。这需要在 Prompt 中加强约束，比如“你必须引用来源编号”、“如果资料中没有，回答‘未找到’”。

  • 文档过时：索引没及时更新，用户问到最新政策却查不到。需要建立定时或事件驱动的索引更新管道。

🐞 3. “文档明明有答案，LLM 却回答不知道”排查指南

这是一个经典的“RAG 失灵”场景，原因通常出在 检索阶段没召回，或 召回后 LLM 没能正确理解 这两个环节上。排查思路是沿着数据流分段定位。

排查路线图：

image.png

第一步：检索侧排查

动手检查：把用户问题放进检索函数，直接看返回的 docs 内容，确认答案是否在 top-K 文档中，以及在第几位（排名越靠后，越容易被 LLM 忽略）。

常见原因与修复：

查看内嵌表格

检索诊断代码示例：

# 快速检查检索质量
docs = vectorstore.similarity_search("怎么退钱", k=5)
for i, doc in enumerate(docs):
    print(f"--- 排名 {i+1} ---")
    print(doc.page_content[:200])
# 肉眼确认答案是否出现、排第几

第二步：生成侧排查

如果确认答案已经在检索到的文档中，但 LLM 仍然说“不知道”，问题就出在 Prompt 组装 或 LLM 自身。

常见原因与修复：

查看内嵌表格

生成侧验证方法：把检索到的文档和拼好的完整 Prompt 打印出来，自己当一次 LLM——如果连人都能从上下文里轻易找到答案，模型却不行，那大概率是模型本身的能力问题或 Prompt 有误导。如果人也找不到，那就是检索的问题。

一个端到端排查脚本的骨架：

def diagnose_rag_failure(question, retriever, llm):
    # 1. 检查检索结果
    docs = retriever(question)
    answer_in_docs = any("退款" in d.page_content or "退钱" in d.page_content for d in docs)
    if not answer_in_docs:
        return "根因：检索未召回。建议：加 Query Rewriting 或混合检索。"

    # 2. 检查 Prompt 和生成
    context = "\n".join([d.page_content for d in docs])
    prompt = f"基于以下资料回答：\n{context}\n\n问题：{question}"
    response = llm(prompt)

    if "未找到" in response or "不知道" in response:
        return "根因：LLM 生成侧问题。建议：优化 Prompt 指令，或升级模型。"

    return "暂时工作正常，可能是偶发。"

收束： RAG 系统出问题，不要瞎猜，一定是先查检索，再查生成。80% 的“回答不知道”都是检索没召回，剩下 20% 是 Prompt 没写好或模型太弱。顺着数据流分段排查，十几分钟就能定位到根因。

文档分块（Chunking）策略的选型原则

🔪 1. 为什么 RAG 需要对文档做分块？

两个硬约束，逼着我们切：

  1. 模型上下文窗口有限：即使现在有了 128K 的超长窗口，但塞入一整本 500 页的手册既不经济（Token 成本爆炸），也会让模型“注意力涣散”——真正有用的段落被埋在海量无关文字里，检索精度和生成质量都会断崖下降。

  2. 检索需要合适粒度：Embedding 模型是为短文本优化的。如果你把整章内容编码成一个向量，它只会学到“这一章大致关于什么”，而无法精细区分“第三段讲的是退款条件，第五段讲的是退货流程”。检索时命中整章，但里面 90% 的内容跟用户问题无关，模型读着读着就乱了。

直白比喻：

不分块的 RAG 就像去图书馆找资料，管理员不给你某一页，而是把整个书架搬到你面前，让你自己翻。分块就是把书拆成目录卡片，每张卡片摘录一段内容，你能快速抽出最相关的那几张，精准且省力。

📐 2. 文档分块策略有哪些？如何根据场景选型？经验值怎么定？

常见分块策略，从简单到智能：

固定长度切分               递归字符切分              语义切分                 结构切分
(Fixed-size)             (Recursive)              (Sentence/Semantic)      (Markdown/HTML)
──────────────────      ──────────────────       ────────────────────     ───────────────────
按字符数硬切              按优先级分隔符递归切       尽量按句子边界切          基于文档原生结构切
简单粗暴，可能切断句子     语言感知，最常用          语义完整，但成本高        适合富文档（PDF/MD）

① 固定长度切分

设置一个 chunk_size（如 500 字符），直接切。容易在句子中间拦腰截断，检索到的片段读起来莫名其妙。仅适合对质量要求极低的原型验证。

② 递归字符切分（Recursive Character Splitter）

这是生产环境最通用的方案。它按 ["\n\n", "\n", "。", ".", " "] 的优先级，尝试在更“自然”的分隔处断开——先按段落切，段落还太长就按句子，最后才按字符。绝大部分中英文混合文档都能用。

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,         # 每个 chunk 的目标最大长度 (字符/Token)
    chunk_overlap=50,       # 相邻 chunk 的重叠长度，防止语义断裂
    separators=["\n\n", "\n", "。", ".", " ", ""],
    length_function=len,
)
chunks = splitter.create_documents([full_text])

③ 语义/句子切分

先用 NLP 模型（如 spaCy）分句，再把句子组合成 chunk，确保每个 chunk 结尾都是完整的句子。对法律、合同等对语义完整性要求极高的场景非常有用，但性能开销大，对中英文混合场景需要语言模型支持。

④ 结构感知切分

针对 Markdown、HTML、JSON 等有原生结构的文档，按标题、表格、代码块等逻辑单元切分。例如 MarkdownHeaderTextSplitter 可以按 #、## 层级切，每个 chunk 自动带上章节标题作为元数据，极大提升检索定位能力。

场景选型速查：

查看内嵌表格

Chunk Size 和 Overlap 经验值：

这两个参数没有银弹，但可以根据你的 Embedding 模型和业务反复实验，以下给出经过大量项目验证的起步值。

查看内嵌表格

调优方向：如果检索到的片段“话只说一半”，增大 overlap 或改用语义切分；如果检索到的片段“啥都沾点但不精”，适当减小 chunk_size。

🔧 3. 检索到的内容经常语义不完整（只有半句话），怎么解决？

“只有半句话”的本质是：答案所需的信息跨越了两个或多个 chunk，但检索时只抽出了其中一片。

原因诊断与分步修复：

查看内嵌表格

三种工程解决方案，由易到难：

方案一：句子窗口检索（Sentence Window Retrieval）

检索时用小 chunk（保证精准度），但喂给 LLM 时，自动“展开窗口”，把命中小 chunk 的前后若干句子也一并送入上下文。

# 原理示意：检索返回 id=57 的 chunk
def expand_window(chunk_id, all_chunks, window=1):
    start = max(0, chunk_id - window)
    end = min(len(all_chunks), chunk_id + window + 1)
    return "".join(all_chunks[start:end])

很多框架直接支持。例如在 LlamaIndex 中，可以使用 SentenceWindowNodeParser，它会自动构建句子级别的节点，并在检索后展开窗口。

from llama_index.core.node_parser import SentenceWindowNodeParser
parser = SentenceWindowNodeParser(
    sentence_window_size=3,  # 向前后各取3句
    window_metadata_key="window",
)
nodes = parser.get_nodes_from_documents(docs)

方案二：父子文档索引（Parent-Child Indexing）

推荐用于生产环境。核心思想是：用小的子 chunk 做索引（检索精准），用大的父 chunk 做上下文（回答完整）。

          父文档 (Parent) - 完整的一段内容
          ├── 子 chunk 1
          ├── 子 chunk 2
          └── 子 chunk 3
检索命中子 chunk 2 → 返回整个父文档作为上下文

在 LangChain 中可以用 ParentDocumentRetriever 实现：

from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore

# 创建父子文档检索器
retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=InMemoryStore(),
    child_splitter=RecursiveCharacterTextSplitter(chunk_size=200),  # 子片段，用于检索
    parent_splitter=RecursiveCharacterTextSplitter(chunk_size=1000), # 父文档，用于喂给LLM
)
retriever.add_documents(docs)

方案三：自动合并检索 + Rerank

先正常检索多个候选 chunk，然后用一个 Rerank 模型（如 Cohere Rerank、BGE-Reranker）对这些候选重新排序。Rerank 模型通常计算 Query 与整个文档片段的相关性时更精确，能把那个真正包含完整信息的 chunk 排到最前面。配合邻段扩展，进一步补齐缺失的上下文。

排查脚本：验证修复效果

def check_completeness(question, retriever, llm):
    docs = retriever(question)
    for i, doc in enumerate(docs):
        first_char = doc.page_content.strip()[0] if doc.page_content else ""
        last_char = doc.page_content.strip()[-1] if doc.page_content else ""
        # 简单判断：结尾不是句号/问号/感叹号，可能被切断
        if last_char not in "。！？.!?":
            print(f"⚠️  Chunk {i+1} 结尾不完整: ...{doc.page_content[-50:]}")
        if first_char not in "“‘「（(【" and not first_char.isalpha():
            print(f"⚠️  Chunk {i+1} 开头不完整: {doc.page_content[:50]}...")

收束： 语义不完整的问题，不能只靠调 overlap 来解决。最稳健的方案是采用父子文档索引——用精准的小片段检索，用完整的大上下文回答。这相当于把“寻找答案”和“理解答案”两件事分开优化，各自做到极致。

4.2 检索质量优化

🧬 1. 什么是稀疏向量和稠密向量？它们在检索场景中各有什么特点？

在 RAG 检索中，把文本变成向量有两种截然不同的哲学：稀疏向量 和 稠密向量。它们就像查字典和找相似，各有所长。

image.png

稀疏向量

像一个极其详细的词典索引。每个维度对应词典里的一个词，向量里绝大部分位置是 0，只有文档中真正出现的词才有非零值。它不会“猜”，你搜“ERP”，它绝对不会去找“企业资源计划”。优势在于精确匹配专有名词、缩写、产品型号，但缺点也很明显：不懂同义词，也理解不了语义，搜“便宜”返回不了“实惠”。

稠密向量

像一个语义压缩器。用 Embedding 模型（如 text-embedding-3、BGE）把整段文本压缩成几百到几千维的实数向量，每个维度都承载着深层语义。搜“性价比高的手机”，它可能返回“千元机评测”的文档，即使原文里没有“性价比”这个词。但它对冷冰冰的专有名词（如 “AZ-900”、“RFC 6749”）不够敏感，可能被语义相近的通用词带偏。

两者的互补关系：

  • 用户问“怎么退款”，稠密向量能理解“退款”和“退钱”是同义，找到相关条款。

  • 用户搜“错误码 E10042”，稀疏向量能精准定位到那一段，而稠密向量可能被一堆包含“错误码”但无关的文档淹没。

⚖️ 2. 什么是混合检索？BM25 与稠密向量各有什么优劣？如何通过 RRF 融合？

混合检索 就是把稀疏检索和稠密检索的结果“取长补短”，融合出一份既精准又相关的结果列表。

2.1 BM25 与稠密向量的优劣势对比

查看内嵌表格

2.2 RRF（倒数排名融合）：无痛的混合检索

混合检索的关键是融合算法。RRF (Reciprocal Rank Fusion) 是最简单但也最有效的方案之一。它不关心每个检索源的具体分数（因为稀疏和稠密的分数含义完全不同，不可直接加权），只看每条结果在各自列表中的排名。

RRF 公式： 对于每条文档 dd，在多个检索器 r∈Rr∈R 中的排名为 rankr(d)rankr(d)，其融合得分为

image.png

其中 kk 是平滑常数，通常取 60。排名越靠前，得分越高。

代码示例：用 Python 实现 RRF 融合

from typing import List, Dict

def reciprocal_rank_fusion(
    bm25_results: Dict[str, float],  # doc_id -> BM25 score
    dense_results: Dict[str, float], # doc_id -> Dense score
    k: int = 60
) -> List[str]:
    """
    根据两路检索的分数（或排名）进行RRF融合，返回按融合得分降序的文档ID列表。
    这里假设输入已经是排序好的排名，1为最高排名。
    """
    # 将分数转换为排名（1-based）
    bm25_ranked = sorted(bm25_results.items(), key=lambda x: x[1], reverse=True)
    dense_ranked = sorted(dense_results.items(), key=lambda x: x[1], reverse=True)

    bm25_ranks = {doc_id: i+1 for i, (doc_id, _) in enumerate(bm25_ranked)}
    dense_ranks = {doc_id: i+1 for i, (doc_id, _) in enumerate(dense_ranked)}

    # 融合
    all_docs = set(bm25_ranks.keys()) | set(dense_ranks.keys())
    rrf_scores = {}
    for doc in all_docs:
        rrf = 0
        if doc in bm25_ranks:
            rrf += 1 / (k + bm25_ranks[doc])
        if doc in dense_ranks:
            rrf += 1 / (k + dense_ranks[doc])
        rrf_scores[doc] = rrf

    # 按RRF得分降序排序
    return sorted(rrf_scores, key=rrf_scores.get, reverse=True)

在 LangChain 中，可以直接使用 EnsembleRetriever 封装多种检索器，内部支持权重或 RRF 融合。例如：

from langchain.retrievers import EnsembleRetriever
ensemble = EnsembleRetriever(
    retrievers=[bm25_retriever, dense_retriever],
    weights=[0.4, 0.6],  # 权重分配，也可使用RRF
)

调优经验：通常先给稠密检索稍高权重（0.6-0.7），因为它对自然语言更友好。但如果用户反馈“专有名词搜不到”，就适当提高 BM25 权重，或改用 RRF 不依赖分数的方式。

🎯 3. 专有名词召回效果很差，怎么优化？

当用户搜索“请解释 X-9000 的校准流程”时，理想结果是精确命中产品手册的那一页，现实却是返回一堆泛泛的“校准说明”。专有名词召回差，通常是稠密检索为主导的架构出了问题。

解决路径：从单一向量到多重保障

① 升级到混合检索（最快见效）

立刻引入 BM25 检索器，并与现有稠密检索融合。BM25 对于“X-9000”这种独特字符串极其敏感，只要文档里有，几乎一定能排到前面。这是最低成本、最高收益的一步。

② 增加静态词典 + 精确匹配查询 对于极高频的专有名词（如产品型号、API 名称、错误码），在检索前做一次简单的字符串匹配扫描。如果用户问题中匹配到预设的专有名词，直接构造一个 term 查询，到向量库的元数据中精确匹配，然后把精确匹配的文档排在结果最前面。

# 简单示意：基于元数据过滤精确搜索
def search_with_synonyms(query: str, vectorstore, synonyms_map: dict):
    # 对已知专有名词，同时做精确元数据过滤
    exact_docs = []
    for term, field in synonyms_map.items():
        if term.lower() in query.lower():
            # 精确匹配元数据中的字段
            exact_docs.extend(vectorstore.filter({field: term}))
    # 常规向量检索
    semantic_docs = vectorstore.similarity_search(query, k=5)
    # 将精确结果去重后放在最前面
    final_docs = exact_docs + [d for d in semantic_docs if d not in exact_docs]
    return final_docs[:5]

③ 微调 Embedding 模型（长期方案） 如果你的领域有大量特有术语（如医药、法律、芯片设计），用通用 Embedding 模型效果肯定打折扣。收集一份领域相关的“术语-同义词-上下文”数据集，对 Embedding 模型做对比学习微调（如使用 sentence-transformers 的 MultipleNegativesRankingLoss），让模型学会将术语与其正确文档拉近，与无关文档推远。这是成本最高但效果最彻底的方案。

④ Query 改写 + 同义词扩展

在检索前，用 LLM 或规则把用户问题里的专有名词进行扩展。例如将“X9K”扩展为“X9K (产品全称: X9-Krypton, 型号代码: X9K-2026)”。扩展后的查询同时送入 BM25 和稠密检索，增加命中概率。

# 利用小模型或规则扩展查询
def expand_query(query: str) -> str:
    expansions = {
        "X9K": "X9K 产品全称 X9-Krypton 型号代码 X9K-2026",
        "API v2": "API v2 也叫 REST v2 旧版接口",
    }
    for term, expansion in expansions.items():
        if term in query:
            query += " " + expansion
    return query

⑤ 预处理索引，注入元数据 在文档入库时，就从内容中自动抽取专有名词（可用正则、NER 模型或关键词提取器），并将它们作为独立的元数据字段（如 product_id, error_code）存入。这样在检索时，可以针对这些字段做精确过滤或提升权重。

Reranker 的原理与效果提升

🔍 1. 为什么要有召回和精排两个阶段？只用向量检索直接返回 Top-5 不行吗？

简短回答：向量检索（召回）追求“快”和“全”，精排追求“准”。直接返回向量检索的 Top-5，相当于让一个短跑选手去下围棋——速度够快，但精度远远不够。

image.png

为什么向量检索的 Top-5 不够准？

向量检索用的是 Bi-Encoder（双塔模型）：查询和文档分别编码成向量，再计算余弦相似度。为了追求毫秒级检索，文档向量通常预先计算好，查询向量即时生成。这带来一个致命弱点：查询和文档之间没有任何直接的 token 级交互。模型只能通过一个固定的向量来“记住”查询的所有意图，当查询复杂时，信息必然丢失。

举个例子：查询“苹果的最新笔记本性能如何？”

  • 稠密向量很可能被“苹果”带偏，把一堆关于“苹果水果”或“苹果公司手机”的文档排在前面，因为它们和“苹果”语义最接近。

  • 真正相关的 MacBook Pro 评测，因为“苹果”的语义权重被稀释，可能排到第 20 名以后。

如果你只取 Top-5，大概率拿到的是一堆“苹果”相关的噪音，真正的答案被截断在 5 名之外。召回阶段必须取 Top-K（K 通常取 50~200），保证答案在候选集里；精排阶段再用更精准的模型把答案从中“挑出来”，送到 LLM 面前。

⚖️ 2. 为什么要引入 Reranker？Bi-Encoder 和 Cross-Encoder 有什么区别？两阶段架构怎么设计？

2.1 Bi-Encoder vs Cross-Encoder：本质区别

查看内嵌表格

直观比喻：

  • Bi-Encoder 像招聘助理，快速浏览一万份简历，筛出 100 份可能合适的。

  • Cross-Encoder 像部门主管，把这 100 份简历仔细读一遍，挑出最匹配的 5 份。

2.2 两阶段架构设计

image.png

代码示例：用 sentence-transformers 的 Cross-Encoder 做精排

from sentence_transformers import CrossEncoder

# 加载 Cross-Encoder 模型（通常比 Bi-Encoder 大，但更准）
reranker = CrossEncoder('BAAI/bge-reranker-large')

# 召回阶段已经拿到了候选文档 (top-100)
candidates = [
    "太阳能电池的基本原理",
    "光伏电站运维手册",
    "单晶硅效率提升的研究与实践",
    "如何清洗太阳能板",
    "钙钛矿材料的最新进展",
]
query = "如何提高光伏板的转换效率？"

# 构造 (query, doc) 对
pairs = [(query, doc) for doc in candidates]

# 精排：Cross-Encoder 对每一对打分
scores = reranker.predict(pairs)

# 按分数降序排列，取 Top-3 给 LLM
ranked_docs = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)[:3]
# 结果：单晶硅效率提升研究 > 钙钛矿材料 > 太阳能电池原理

为什么不能直接用 Cross-Encoder 做全量检索？

假设你有 1000 万篇文档，Cross-Encoder 需要对每一篇都和查询拼在一起跑一遍完整的 Transformer 前向。这一轮下来可能要几分钟甚至几小时，延迟根本无法接受。两阶段架构的本质是 “用廉价的粗筛换昂贵的精挑”——100 次 Cross-Encoder 调用是可行的，100 万次则不行。

🐢 3. 线上 RAG 延迟过高，Reranker 是瓶颈，怎么优化？

当你发现 RAG 系统端到端延迟超过 5 秒，而 Profile 显示 Reranker 占了其中 3 秒以上时，问题通常出在：候选太多、模型太重、调用太频繁。优化方向对应着“减量、提速、复用”三个字。

优化方案组合拳

查看内嵌表格

实际优化案例：

优化前：

  • 召回 100 条，用 bge-reranker-large（1.3GB 模型，CPU 推理）

  • 对 100 条逐个打分，总耗时约 3.5 秒。

优化后：

  1. 召回数降至 30，延迟降至 1.1 秒。

  2. 换用 bge-reranker-base 并转为 ONNX 格式，单次推理速度翻倍。

  3. 引入多查询批处理：把同时到达的 3 个查询的候选拼成 batch 一起推理，GPU 利用率提升。

  4. 对于“退货流程”等高频查询，直接缓存排序结果，命中后延迟忽略不计。

最终端到端延迟从 5.2 秒降至 1.8 秒，Reranker 耗时占比降至 0.5 秒以内。

代码示例：ONNX 加速 + 输入截断 + 批处理

from optimum.onnxruntime import ORTModelForSequenceClassification
from transformers import AutoTokenizer
import torch

class FastReranker:
    def __init__(self, model_path: str):
        # 使用 ONNX Runtime 加速推理
        self.model = ORTModelForSequenceClassification.from_pretrained(model_path, export=True)
        self.tokenizer = AutoTokenizer.from_pretrained(model_path)

    def rerank(self, query: str, candidates: list, max_len: int = 256) -> list:
        pairs = [(query, doc[:max_len]) for doc in candidates]
        inputs = self.tokenizer(pairs, padding=True, truncation=True,
                                max_length=max_len, return_tensors="pt")
        with torch.no_grad():
            scores = self.model(**inputs).logits.squeeze(-1)
        # 按分数降序
        ranked_indices = scores.argsort(descending=True)
        return [candidates[i] for i in ranked_indices]

# 使用
reranker = FastReranker("BAAI/bge-reranker-base")
top_docs = reranker.rerank("怎么退钱", candidate_docs)

关键权衡：减少召回候选数是最直接的手段，但必须用离线评估确认 Top-30 的召回率依然满足业务要求。如果发现一些重要答案被截断在 30 名之外，就需要配合更精准的召回策略（如混合检索），而不是盲目增大候选数。

一句话收束：Reranker 的优化从来不是单一维度，而是“减候选、轻模型、上缓存、批处理”的组合拳。在精度和延迟之间找到那个让用户觉得“秒回且靠谱”的平衡点，才是 RAG 工程师的核心手艺。

Query 改写（Query Rewriting）提升召回质量

❓ 1. 为什么用户的原始 Query 直接去检索效果往往不好？

一句话解释：用户的自然语言问题和知识库中存储的文档语言之间存在“表达鸿沟”——用户用口语问，文档用书面语写，用户用泛指，文档用精确术语。

这种鸿沟具体表现在三个层面：

image.png

① 语言风格鸿沟（口语 vs 书面语）

用户习惯于口语化、省略句、甚至带错别字的提问。而知识库里的文档通常是正式、书面化的表达。向量检索背后是语义模型，它对口语和书面语的匹配天然存在偏差。

② 信息密度鸿沟（泛指 vs 精确）

用户常说“那个东西”、“上次提到的”，这些指代在单次检索中完全无法理解。此外，复杂问题可能包含多个意图，比如“这个 API 和那个 API 有什么区别，性能如何？”——如果直接向量化，语义被平均，很可能哪个意图都没检索准。

③ 结构鸿沟（碎片 vs 完整逻辑）

文档中信息是层次化、结构化的，而用户提问往往是碎片化的。比如用户问“怎么解决内存泄漏”，文档里可能分布在“常见问题”章节和“最佳实践”章节，直接检索可能只召回其中一部分。

直观验证：

你可以做个简单实验：把用户 Query 向量化后检索，看返回结果的余弦相似度分数。你会发现很多看似相关的分数并不高，而真正相关的文档分数却被一堆噪音淹没。这是因为 Embedding 模型本身也是“折中”的，它无法从一句口语中精准捕捉到文档中对应的那个精确短语。

🧪 2. RAG 中有哪些 Query 改写策略？Multi-Query、HyDE、Step-Back Prompting 分别解决什么问题，原理是什么？

为了解决上述鸿沟，业界演化出了多种 Query 改写策略。它们不是在检索算法上修修补补，而是从源头把“人类的问题”转译成“更适合检索的查询”。

策略一：Multi-Query（多查询改写）

解决什么问题：用户提问角度单一，召回面太窄。

原理：用 LLM 把用户原始 Query 改写成多个不同角度、不同表达方式的查询，并行检索，再合并去重。这就像你让三个不同背景的人去图书馆找同一问题的资料，他们可能会用不同的关键词、不同的问题，最终拼出更全面的信息。

def multi_query_rewrite(original_query: str, llm) -> list:
    prompt = f"""将以下用户问题改写成3个不同的查询，每个从不同角度或使用不同关键词，以便更全面地检索相关文档。
原始问题：{original_query}
改写查询："""
    response = llm(prompt)
    # 解析返回的3个查询
    queries = [q.strip() for q in response.split('\n') if q.strip()]
    return queries

# 示例
original = "怎么提高模型性能？"
queries = multi_query_rewrite(original, llm)
# 可能返回：
# 1. 模型推理速度优化方法
# 2. 提高LLM准确率的调优技巧
# 3. 大模型吞吐量和延迟优化

注意：多查询会增加 3~5 倍的检索开销，所以通常在检索后合并时用 RRF（倒数排名融合）或直接拼接，而不是简单叠加。适合对召回率要求极高的场景。

策略二：HyDE（假设文档嵌入）

解决什么问题：用户提问太短、太模糊，无法有效检索。

原理：让 LLM 先“脑补”一篇可能包含答案的假设文档，然后把这篇假设文档向量化去检索。 这听起来反直觉，但非常有效。因为 LLM 生成的假设文档会模仿知识库文档的写作风格、用词和结构，从而拉近了用户意图和文档之间的语义距离。

用户问：“怎么缓解过拟合？”
  → LLM 生成假设文档：“过拟合是指模型在训练集上表现好，在测试集上表现差。缓解方法有：1. 增加数据量 2. 正则化（L1/L2） 3. Dropout 4. 早停法...”
  → 用这个假设文档去检索，比直接用“怎么缓解过拟合？”更可能召回机器学习教材里的相关章节。

def hyde_rewrite(original_query: str, llm) -> str:
    prompt = f"""请写一篇简短文档（约100字），回答以下问题。不需要完全正确，只需用专业、正式的语言描述可能的内容。
问题：{original_query}
假设文档："""
    hypothetical_doc = llm(prompt)
    return hypothetical_doc

适用场景：用户问题非常简短、泛泛，或者领域专业性很强，用户不会用专业术语提问时。HyDE 尤其适合 FAQ 类知识库。

策略三：Step-Back Prompting（后退提示）

解决什么问题：用户问的具体问题太细节，但检索需要更上层的背景知识。

原理：当用户问“GPT-4o 的注意力头数是多少？”这类具体参数时，直接检索可能找不到精确数字。但如果“后退一步”，先检索“GPT-4o 模型架构概述”，再从里面找头数就容易多了。Step-Back 就是让 LLM 把用户的具体问题抽象成一个更上层的、更宽泛的“背景问题”，然后用这个背景问题去检索，最后在检索到的文档中抽取出具体答案。

def step_back_rewrite(original_query: str, llm) -> str:
    prompt = f"""你是一个帮助检索的助手。对于用户的具体问题，请生成一个更通用、更上层的背景问题，以便检索相关文档。
具体问题：{original_query}
背景问题："""
    return llm(prompt)

# 示例
step_back_rewrite("GPT-4o 的注意力头数是多少？", llm)
# → "GPT-4o 的模型架构和参数配置"

适用场景：用户问题非常具体、需要特定细节，且这些细节可能在文档中不是直接以问答形式存在，而是隐藏在概述章节中。

三种策略对比总结：

查看内嵌表格

🎯 3. 用户反馈对“太专业”和“太宽泛”的问题都检索不好，你怎么分别应对？

这是典型的“检索粒度悖论”：对窄问题要精，对宽问题要全。解决方案不是一刀切，而是动态路由 + 自适应改写。

核心思路：在收到用户 Query 后，先由一个小模型或规则判断问题类型，然后走不同的改写和检索管道。

image.png

3.1 对“太专业”的问题（如查某个具体 API 的参数说明）

痛点：专业术语、缩写、代码符号在语义检索中容易被“淹没”，稠密向量对这些冷冰冰的字符串不敏感。

应对组合拳：

① 混合检索前置：在检索入口处就同时走向量检索和 BM25 关键词检索，并给 BM25 更高权重。专业术语几乎总能被 BM25 精准命中。

② 元数据过滤：在索引文档时，就把 API 名称、参数名等作为独立字段存储。当识别出问题包含特定实体时，直接用过滤器筛选，而不是纯粹靠相似度排序。

③ 改写优化：对于特别长的 API 描述查询，先用 LLM 提取核心关键词（如 API 名、参数名），再用这些关键词做 BM25 精确搜索。

def handle_professional_query(query: str, vectorstore, bm25_retriever) -> str:
    # 1. 提取核心实体（简单规则或小模型）
    entities = extract_technical_terms(query)  # 如 ["X-9000", "calibrate"]

    # 2. 如果有明确实体，先精确过滤
    exact_docs = []
    for entity in entities:
        exact_docs.extend(vectorstore.filter(field="product_id", value=entity))

    # 3. 同时做混合检索
    semantic_docs = vectorstore.similarity_search(query, k=10)
    keyword_docs = bm25_retriever.search(query, k=10)

    # 4. 融合：精确匹配的排最前，其余 RRF 融合
    all_docs = exact_docs + rrf_fusion(semantic_docs, keyword_docs)
    return all_docs[:5]

3.2 对“太宽泛”的问题（如问某个框架的整体架构）

痛点：宽泛问题需要的答案可能分散在多个章节，单次检索只能抓到局部片段，缺乏全局视角。

应对组合拳：

① Step-Back 改写：把“FastAPI 的整体架构是什么？”后退为“FastAPI 架构概述”或“FastAPI 核心组件”，先用这个宽泛查询检索到综述性文档。

② 子问题分解（Sub-Question Decomposition）：对于明显的复合问题，用 LLM 拆成多个子问题，分别检索，再汇总。例如“整体架构”可拆为“路由机制”、“依赖注入”、“中间件”、“请求处理流程”等子问题。

③ 大块文档召回：针对宽泛问题，增大 chunk_size 或直接使用父文档索引（检索时返回更大的上下文片段），因为宽泛问题需要更完整的上下文来支撑回答。

def handle_broad_query(query: str, llm, retriever) -> str:
    # 1. 判断是否为宽泛问题（如包含"架构"、"原理"、"概述"等词）
    if is_broad_query(query):
        # 2. 使用 Step-Back 生成抽象问题
        background_query = step_back_rewrite(query, llm)
        # 3. 用抽象问题检索，同时增加检索数
        docs = retriever.retrieve(background_query, top_k=20)
        # 4. 如果文档较长，可以考虑用 LLM 做子文档摘要或生成大纲
        return docs
    # 否则走默认流程
    return retriever.retrieve(query, top_k=5)

更智能的做法：动态路由

可以在网关层部署一个轻量级分类器（如 DistilBERT fine-tune 或直接用 LLM 判断），根据问题类型自动选择管道。这样对用户透明，整体系统鲁棒性更强。

4.3 向量数据库选型

数据库选型

🔷 1. 向量数据库和关系型数据库有什么本质区别？

把这两种数据库放在一起比，不是比谁更好，而是看它们各自解决的根本问题不一样。一句话：关系型数据库管“精确”，向量数据库管“模糊”。

关系型数据库（PostgreSQL / MySQL）
  数据形态：结构化表格，字段严格定义
  查询方式：精确匹配、范围查询、JOIN
  索引结构：B-Tree / Hash，支持等值和排序
  擅长：事务、多表关联、复杂业务逻辑

向量数据库（Milvus / Qdrant / Chroma）
  数据形态：非结构化文本、图片、音频 → 高维向量
  查询方式：相似度搜索（余弦、欧氏距离）
  索引结构：ANN 近似最近邻（HNSW / IVF）
  擅长：语义搜索、推荐、去重

本质差异，我用一个排查问题的经历来说明： 有一次用户反馈“搜‘退款政策’找不到东西”，SQL 里用的是 WHERE content LIKE '%退款%'，但文档里写的是“退货流程”，一个字都对不上。这就是结构化检索的局限——它没有“理解”能力。而向量数据库把“退款政策”和“退货流程”映射到空间里很接近的两个点，因为它们语义相近。这就是本质：关系型查的是“字符串”，向量型查的是“含义”。

代码层面区别：

-- 关系型：精确查找，一字不差sql
SELECT * FROM docs WHERE content LIKE '%退款%';   -- 查不到“退货流程”

# 向量型：语义查找，理解含义
results = collection.query(
    query_embeddings=[embed("退款政策")],
    n_results=5
)
# 第一条结果大概率就是“退货流程说明”

但向量数据库不适合管钱、管库存、管订单——那些需要 ACID 事务和精确计数的场景，仍然是关系型数据库的绝对领域。不要把向量数据库当成 SQL 的替代品，它是为非结构化数据的“语义搜索”而生的专业工具。

🔷 2. 主流向量数据库各有什么特点？不同场景怎么选？

目前市面上的向量数据库，按部署形态大致分三类：自托管开源、云托管、PG 插件。我把核心的几个拉出来做一个快速的权衡分析。

查看内嵌表格

场景选型决策树：

你们是创业小团队，只有后端 2 人？
  ├─ 是 → 首选 pgvector（复用现有 PG），或 Pinecone（不想运维）
  └─ 否 → 数据量超过 5000 万，需要极致性能？
          ├─ 是 → Milvus 或 Qdrant（分布式部署）
          └─ 否 → 已用 PG？ pgvector。想单独服务？ Qdrant / Chroma

一个具体判断：如果你们已经重度依赖 PostgreSQL，且向量数据量在千万级以下，pgvector 往往是最划算的选择——它把向量检索变成了 SELECT 语句的一部分，团队不用学新数据库，权限、备份、监控全都能沿用现成的。

🔷 3. 公司 500 万条文档接入 RAG，要权限过滤，2 人后端团队，怎么选型？

这个场景有三个硬约束：中等数据量（500 万）、必须有权限隔离（部门 A 不能搜部门 B 的文档）、运维人力极度有限（2 人）。它们共同指向一个选择：pgvector。

为什么不是 Milvus？

Milvus 性能确实强，但部署需要 etcd、MinIO、Coordinator、Proxy 等一整套组件。2 个人维护一套分布式系统，出问题半夜起来修，风险太高。在 500 万这个量级，pgvector 配合 HNSW 索引完全够用，不必上重型武器。

权限过滤怎么实现？ 这是选 pgvector 最关键的理由。PostgreSQL 原生支持 Row-Level Security（RLS），你可以在文档表上创建策略，让每个部门的用户只能看到自己部门的行。向量检索时，SQL 的 WHERE 子句会自动过滤掉无权限的数据，然后再在过滤后的集合里做相似度搜索——权限和搜索在同一个查询里完成，不需要应用层再拼结果。

我们的落地设计：

-- 1. 文档表，包含向量字段和部门字段
CREATE TABLE docs (
    id SERIAL PRIMARY KEY,
    content TEXT,
    department VARCHAR(50),
    embedding VECTOR(1536)  -- OpenAI embedding 维度
);

-- 2. HNSW 索引（比默认 IVFFlat 查询更快）
CREATE INDEX ON docs USING hnsw (embedding vector_cosine_ops);

-- 3. 行级安全策略：用户只能看自己部门的文档
ALTER TABLE docs ENABLE ROW LEVEL SECURITY;
CREATE POLICY dept_isolation ON docs
    FOR SELECT
    USING (department = current_setting('app.current_department'));

-- 4. 查询时，应用先设置当前部门，然后执行向量检索
SET app.current_department = 'engineering';
SELECT content, 1 - (embedding <=> query_embedding) AS similarity
FROM docs
ORDER BY embedding <=> query_embedding
LIMIT 10;

代码示例：应用层查询（Python）

import psycopg2

def search_docs(query_text: str, department: str, embedding_model):
    query_emb = embedding_model.encode(query_text)
    conn = psycopg2.connect(...)
    cur = conn.cursor()
    # 设置当前请求的部门上下文
    cur.execute("SET app.current_department = %s", (department,))
    cur.execute("""
        SELECT content, 1 - (embedding <=> %s::vector) AS similarity
        FROM docs
        ORDER BY embedding <=> %s::vector
        LIMIT 10
    """, (query_emb, query_emb))
    rows = cur.fetchall()
    cur.close()
    return rows

为什么这个方案适合 2 人团队？

  • 没有引入新数据库，运维负担几乎为零。

  • 权限模型完全复用 PG 的 RLS，不需要在应用层写复杂的权限逻辑。

  • 向量检索就是一句 SQL，团队里任何一个后端都能看懂、维护。

  • 500 万条数据，HNSW 索引的查询延迟在 10ms 以内，完全满足业务需求。

如果未来数据涨到 5000 万？ 那时候团队也大概率扩张了，可以把向量部分迁移到 Milvus 或 Qdrant，而权限数据仍然留在 PG，在应用层做一次结果过滤。但当下，pgvector 是这个场景下最务实、最稳健的选择。

4.4 RAG 评估

RAG 指标

🎯 1. 什么是 RAG 系统评估中的 Faithfulness 指标？

Faithfulness（忠实度）衡量的是：LLM 生成的答案，是不是严格基于你塞给它的“参考资料”（检索到的上下文），而不是凭自己训练时的记忆凭空捏造。

换句话说，它追问的是：“你对我从知识库里搬来的那几页纸，有没有老老实实照着回答？”

image.png

一个高 Faithfulness 的系统，答案里的每一个事实主张，都能在检索到的上下文中找到出处。

而低 Faithfulness 则意味着模型在“自由发挥”，可能产生幻觉、加入过时训练数据、或曲解文档。

为什么 Faithfulness 是 RAG 的“高压线”？

因为 RAG 的初衷就是“用外部知识对抗幻觉”。如果模型连你递给它的资料都不尊重，那 RAG 就失去了存在的意义。在医疗、法律、金融等领域，一个低 Faithfulness 的回答可能导致严重后果，所以它通常是 RAG 评估体系中最敏感的指标。

一个简单的示例判断：

# 简单的基于规则的 Faithfulness 检查（示意）
def simple_faithfulness_check(answer, context):
    # 将答案拆分为独立的事实主张（简化：按句号拆分）
    claims = answer.split("。")
    unfaithful_claims = []
    for claim in claims:
        if not any(claim.strip() in ctx for ctx in context):
            unfaithful_claims.append(claim)
    return unfaithful_claims  # 返回无法在上下文中找到的句子

实际生产中当然不会这么简单，但核心思想一样：逐条核对。更准确的做法是用 LLM-as-Judge 来做这件事。

📊 2. RAGAS 框架的核心指标体系，以及 LLM-as-Judge 的常见偏差

RAGAS (Retrieval Augmented Generation Assessment) 是目前 RAG 评估的事实标准之一。它不依赖人工标注，而是用 LLM 作为裁判，自动对 RAG 系统的关键维度进行打分。

2.1 RAGAS 核心指标速览

RAGAS 指标体系
├── Generation 指标 (评估生成答案)
│   ├── Faithfulness (忠实度)   —— 答案是否完全基于上下文
│   └── Answer Relevancy (答案相关性) —— 答案是否紧扣问题
│
└── Retrieval 指标 (评估检索质量)
    ├── Context Precision (上下文精确度) —— 检索到的文档里有多少是真正有用的
    └── Context Recall (上下文召回率)   —— 回答问题所需的信息，上下文都包含了吗

  • Faithfulness：将答案拆解为若干“陈述”，逐条判断能否在上下文中找到支撑。最终得分 = 能找到支撑的陈述数 / 总陈述数。

  • Answer Relevancy：判断答案是否切题、不冗余。

  • Context Precision：检索到的文档片段中，相关文档的排名是否靠前。

  • Context Recall：回答所需的所有关键信息，是否都被检索出来了。

代码示例：使用 RAGAS 评估一个 RAG 结果

from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision, context_recall
from datasets import Dataset

# 准备评估数据
eval_dataset = Dataset.from_dict({
    "question": ["公司最新的年假政策是什么？"],
    "answer": ["根据2026年员工手册，年假为15天，可累积。注意必须每年休完10天。"],
    "contexts": [["公司年假政策：每年15天，可累积至30天。"]],
    "ground_truth": ["公司年假为15天，可累积至30天，无强制休完要求。"]
})

# 运行评估
results = evaluate(eval_dataset, metrics=[faithfulness, answer_relevancy,
                                          context_precision, context_recall])
print(results)
# 输出: {'faithfulness': 0.5, ...}  因为 "必须每年休完10天" 找不到上下文支持

2.2 LLM-as-Judge 的常见偏差与工程规避

用 LLM 做裁判不是完美无瑕的，我踩过的几个坑：

  1. 位置偏差：裁判 LLM 倾向于认为第一个答案更好，或者在长文档中更关注开头和结尾，忽略中间。

  2. 规避：成对比较时，交换两个答案的顺序各评一次，取平均分；或改为绝对分项打分，不直接对比。

  3. 长度偏差：裁判容易给更长的回答打高分，即使内容是冗余的。

  4. 规避：在 Prompt 中明确要求“简洁性也是质量的一部分”，或在量规里惩罚冗余。

  5. 自我偏好：用 GPT-4 评估 GPT-4 生成的答案，可能偏高。

  6. 规避：引入其他模型家族（如 Claude）作为第二裁判，或定期用人工标注校准自动评分的系统偏差。

  7. 评分不一致：同一份答案在不同时间打分可能不同。

  8. 规避：在 Prompt 中附带 2-3 个标准评分示例（锚点），让裁判每次都有同一个参考系。

示例：减少位置偏差的 Pairwise Judge

def pairwise_judge_with_position_correction(answer_a, answer_b, question):
    prompt = f"问题: {question}\n回答 A: {answer_a}\n回答 B: {answer_b}\n请选出更好的回答。"
    choice_ab = llm(prompt)  # A在前
    prompt_ba = f"问题: {question}\n回答 A: {answer_b}\n回答 B: {answer_a}\n请选出更好的回答。"
    choice_ba = llm(prompt_ba) # B在前
    # 综合判断
    if choice_ab == "A" and choice_ba == "B":
        return "A更好"   # 顺序交换后仍一致
    elif choice_ab == "B" and choice_ba == "A":
        return "B更好"
    else:
        return "位置偏差严重，需人工复核"

🔍 3. Faithfulness 持续偏低，但 Context Recall 很高，怎么排查和解决？

这是一个“找到了资料，但没老实抄”的典型场景。Context Recall 高，说明你要的信息基本都检索出来了。Faithfulness 低，说明 LLM 在生成答案时，没有忠实地使用这些信息，而是自己添油加醋，或者干脆绕开上下文编造答案。

排查路线图：

image.png

具体排查操作：

第一步，捞 20 条 Faithfulness 为 0 的案例，把它们的 Question、Context、Answer 和 Ground Truth（如果有）打印出来，逐条人工分析。看 LLM 是哪句话“出圈”了。

第二步，分析出圈的类型：

  • 是凭空加了数据（比如“必须休完10天”）？

  • 是把训练数据里的旧政策搬出来了？

  • 是对上下文的错误总结/夸大？

第三步，根据类型下药。

如果是“凭空加数据”或“自己编”，通常是 Prompt 约束不够，或 LLM 指令遵循能力弱。可以这样改进 Prompt：

你只能根据以下参考信息回答问题。如果信息不足以回答，请直接说“根据现有资料无法确定”。
严格禁止使用你的训练知识或进行任何推测。
参考信息：
{context}
问题：{question}

同时换用指令遵循更强的模型（如从 Qwen-7B 换成 GPT-4o-mini），或者降低 Temperature（如 0.1）以减少随机性。

如果是“模型更相信自己的训练记忆”，比如用户问“现在的 CEO 是谁？”，上下文里明明写着“当前 CEO 是张三”，模型却坚持回答“李四”（因为训练数据里李四占多数）。这需要在 Prompt 中明确强调上下文的时间权威性：

注意：参考信息中的事实可能与你训练数据中的信息冲突。你必须无条件优先相信参考信息，因为它是更新的、权威的。

如果是“上下文太长，模型注意力涣散”，则需要在检索后加一个压缩或重排步骤，确保喂给 LLM 的上下文是高度精炼的 Top-2 或 Top-3 片段，而不是海量碎片的大杂烩。例如使用 LLMChainExtractor 或 Reranker 只保留核心段落。

一个端到端修复后的验证脚本思路：

def diagnose_and_fix(question, context, answer, llm):
    # 1. 诊断：用 LLM 判断幻觉的类型
    diagnose_prompt = f"""
    上下文：{context}
    答案：{answer}
    请判断答案中哪些句子无法在上下文中找到依据，并分析为什么会这样（上下文冲突？凭空捏造？过度总结？）。
    """
    diagnosis = llm(diagnose_prompt)

    # 2. 根据诊断结果动态调整 Prompt 或上下文
    if "凭空捏造" in diagnosis:
        new_prompt = f"你只能根据以下信息回答，不得添加任何外部知识：\n{context}\n问题：{question}"
    elif "上下文冲突" in diagnosis:
        new_prompt = f"注意：以下信息是最新的权威资料。如果与你已知的信息冲突，必须遵循以下信息。\n{context}\n问题：{question}"
    else:
        # 默认：精简上下文
        new_context = re_rank_and_truncate(question, context, top_k=2)
        new_prompt = f"根据以下信息回答：\n{new_context}\n问题：{question}"

    # 3. 用新 Prompt 重新生成
    new_answer = llm(new_prompt)
    return new_answer, diagnosis

收束： 高 Context Recall 是地基，低 Faithfulness 是墙上的裂缝。裂缝不在于没找到砖，而在于砌墙的手法出了问题。我们的任务就是仔细看每一道裂缝的形状，然后去强化那几条砌墙的规则——Prompt 的指令、模型的纪律、上下文的精炼度。修好了这些，RAG 才能真的做到“言之有据，据必可查”。

Graph RAG

🧩 1. 知识图谱中的三元组是什么？

三元组是知识图谱存储和表示知识的最基本单位，可以理解为一句“主谓宾”式的陈述句，只不过它被高度结构化，变成了机器能直接处理的格式。

三元组 = (主体, 关系, 客体)

  • 主体：通常是一个实体，比如一个人、一个地点、一个产品。

  • 关系：描述主体和客体之间的联系，通常是动词或介词短语。

  • 客体：可以是另一个实体，也可以是一个具体的属性值（比如数字、日期、字符串）。

直观图示与对应关系：

image.png

这组三元组就可以表达为：

  • (张三, 担任, 产品经理)

  • (张三, 属于, 研发部)

  • (张三, 开发, 产品 Alpha)

  • (产品 Alpha, 上线于, 2025-06-01)

把这些三元组串起来，就构成了一张巨大的、可推理的网络，这就是知识图谱。

代码示例：用 Python 字典存储三元组并进行简单推理

# 定义三元组列表
triples = [
    ("张三", "担任", "产品经理"),
    ("张三", "属于", "研发部"),
    ("张三", "开发", "产品 Alpha"),
    ("产品 Alpha", "上线于", "2025-06-01")
]

# 查询“产品 Alpha 是由哪个部门的谁开发的？”
def find_developer(product):
    results = []
    for s, r, o in triples:
        if r == "开发" and o == product:
            person = s
            # 继续查找这个人的部门
            for s2, r2, o2 in triples:
                if s2 == person and r2 == "属于":
                    results.append((person, o2))
    return results

print(find_developer("产品 Alpha"))  # 输出: [('张三', '研发部')]

这种基于关系链的多跳查询，正是向量检索难以直接做到的。

🌐 2. Graph RAG 的核心原理，解决了传统向量 RAG 的哪些局限性？微软 GraphRAG 的实现思路是什么？

传统向量 RAG 的局限性：

传统 RAG 把文档切成片段，用向量相似度匹配问题与片段。这非常擅长“大海捞针”——从海量文档中找出和问题语义最相关的那个段落。但当问题需要理解整体结构、总结多个实体间的关系、或进行多跳推理时，它就捉襟见肘了。比如你问“张三在项目里做了哪些关键决策？”，向量检索可能只返回包含“张三”和“决策”的碎片，无法把分散在不同文档中的决策点串联成一个时间线或逻辑链。

Graph RAG 的核心原理：

Graph RAG 正是为了解决这类“全局性”问题而生。它不把文档看作独立的碎片，而是先利用 LLM 从文档中抽取实体和关系，构建一个知识图谱。然后在回答问题时，不是去匹配原始文本，而是在图谱上进行遍历、搜索或社区发现，找到与问题相关的子图，最后再把这些结构化信息喂给 LLM 生成答案。

传统向量 RAG： 问题 → 向量相似度 → 若干文档片段 → LLM 总结
Graph RAG：    问题 → 图搜索/遍历 → 相关知识子图 → 将子图序列化为文本 → LLM 生成答案

它解决的三大局限：

  1. 多跳推理：向量检索无法回答“张三的直属上级的邮箱是什么”，因为它需要先找到张三，再通过“汇报给”关系找到上级，最后取属性。Graph RAG 可以通过关系遍历直接完成。

  2. 全局归纳：传统 RAG 难以回答“公司有哪些产品线？”因为相关信息分散在几十个文档里，没有一个片段能完整回答。Graph RAG 却可以通过寻找图中“产品”类型的所有实体，直接归纳出来。

  3. 逻辑一致性：对于需要梳理因果关系、时间线的问题，Graph RAG 可以在图上基于边（关系）构建路径，形成一条清晰的叙事链，而不是把一堆相关的句子堆砌给 LLM。

微软 GraphRAG 的实现思路：

微软开源的 GraphRAG 走的是“全局知识图谱 + 社区分层总结”的路线。它不针对单个问题进行实时图遍历，而是做了一件更彻底的预处理工作：

  1. 实体和关系抽取：对全部源文档，用 LLM 抽取三元组 (主体, 关系, 客体)，构建知识图谱。

  2. 社区发现：使用 Leiden 等社区发现算法，将大规模图切分成不同层级的社区（Communities），这些社区代表了某个主题或某群紧密关联的实体。

  3. 社区总结：对每一个社区，用 LLM 生成一个社区摘要。这个摘要描述了该社区内主要实体、它们的关系和关键信息。这相当于给图谱的每一部分都生成了“导读”。

  4. 回答问题：当用户提问时，GraphRAG 会定位到相关的社区摘要，然后将这些摘要作为上下文，连同原始问题一起送给 LLM 生成答案。这使得 LLM 能够轻易地获得一个主题的全局性概述，从而回答归纳总结类问题。

示例代码：从文档中抽取三元组并构建简单图

import openai
import networkx as nx

def extract_triples(text):
    """用 LLM 从文本中抽取三元组"""
    prompt = f"""从以下文本中抽取 (主体, 关系, 客体) 三元组。
    文本: {text}
    三元组:"""
    response = openai.ChatCompletion.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}]
    )
    # 假设返回格式为 "(张三, 担任, 产品经理)\n(张三, 开发, 产品 Alpha)"
    triples = []
    for line in response.choices[0].message.content.split("\n"):
        line = line.strip("() ")
        if "," in line:
            s, r, o = line.split(",", 2)
            triples.append((s.strip(), r.strip(), o.strip()))
    return triples

# 构建 NetworkX 有向图
G = nx.DiGraph()
for text in documents:
    for s, r, o in extract_triples(text):
        G.add_edge(s, o, relation=r)

# 现在可以执行图遍历回答多跳问题
def get_decision_timeline(product):
    """查询某产品的关键决策时间线"""
    decisions = []
    for node in G.predecessors(product):
        if G[node][product].get("relation") == "决策":
            decisions.append((node, G.nodes[node].get("date")))
    return sorted(decisions, key=lambda x: x[1])

🔍 3. 用户反馈系统无法回答“某产品从立项到上线，经历了哪些关键决策”这类问题，现有向量 RAG 如何改造？

问题诊断：

用户要的不是某个具体片段，而是一条完整的时间线/叙事链。这类问题的答案通常散落在很多文档中：立项在邮件里，关键决策在会议纪要里，上线在发布文档里。向量 RAG 的“取 Top-K 片段”模式，无法自动将这些分散的点串成一条线。

改造方案：从“片段检索”走向“图 + 时间线”的混合架构

不需要推倒重来，我们可以在现有向量 RAG 的基础上，引入轻量级的图结构和预处理机制，专门优化对这类叙述性问题的回答。

架构蓝图：

image.png

具体实施步骤：

  1. 离线预处理：事件抽取与图谱构建 对所有相关文档，用 LLM 抽取出事件节点，每个事件包含：(时间, 主体, 动作, 细节)。然后按产品、项目等主体，构建一个带有时间维度的有向图。
# 抽取事件的 prompt 模板
event_prompt = """
从以下文本中抽取与产品决策相关的事件，每个事件输出一行：
格式：日期 | 主体 | 动作 | 简述
文本：{text}
"""

  1. 将这些事件按产品存入数据库，并在产品内部按时间排序。

  2. 在线查询：时序路径检索 当用户问“产品 Alpha 从立项到上线的关键决策”时，系统识别到这是“历程”类问题，直接查询产品 Alpha 对应的事件列表，按时间排序后，取出一系列事件节点。

def query_product_timeline(product_name):
    events = db.query("SELECT * FROM events WHERE product=? ORDER BY date", product_name)
    # 将事件序列化为描述性文本
    timeline_text = "关键决策历程如下：\n"
    for i, evt in enumerate(events, 1):
        timeline_text += f"{i}. {evt.date}: {evt.subject} 执行了 {evt.action}，详情：{evt.detail}\n"
    return timeline_text

生成答案

把这条时序文本作为上下文，与用户原始问题一起送给 LLM，让它整合成自然、连贯的回答。由于我们提供了完整的时间线，LLM 就能自然地总结出“从立项到上线的关键决策”，而不会产生幻觉或遗漏。

prompt = f"""根据以下事件时间线，回答用户问题。
时间线：
{timeline_text}
问题：{user_question}
"""
answer = llm(prompt)

为什么这个方案有效？

  • 保留了向量 RAG：对于常规的、片段式的问答，依然走高效的向量检索。

  • 为特定问题提供定制索引：通过为产品、项目等实体创建“事件时间线”索引，解决了传统 RAG 做不好全局归纳和多点串联的问题。

  • 成本可控：不需要构建全量知识图谱，只需要对每个产品抽取出关键决策事件，数据量不大，LLM 抽取的成本也有限。

从 0 到 1 的 MVP 实现思路：

先用一个简单的 SQLite 数据库存事件，写一个 Python 脚本定时对新增文档做事件抽取，再改造现有 RAG 的入口，增加一个意图分类器。当意图是“历程/关键决策”时，走图查询；其他情况走向量检索。整个改造两周内就能上线验证效果。

收束：

对于这类需要梳理时间线和因果链的问题，核心矛盾不在于模型理解力不够，而在于我们没有把“被拆散的知识”重新“串起来”交到模型手里。通过引入一个轻量级的事件时间线索引，我们就能在维持向量 RAG 优点的同时，弥补它无法回答全局性、序列性问题的短板。这比直接上全套 Graph RAG 更务实，也更易于落地。

Self-RAG

🧠 1. 传统 RAG 和 Self-RAG 在检索触发机制上的核心区别是什么？

一句话概括：传统 RAG 的检索是“强制触发”，无论问题是什么，先查再说。Self-RAG 的检索是“按需触发”，模型自己决定要不要查、查什么、查完怎么用。

传统 RAG：
  用户提问 → 【必定检索】→ 把检索结果拼进 Prompt → LLM 生成答案

Self-RAG：
  用户提问 → LLM 先“思考”要不要检索？
              ├─ 不需要（比如“你好”）→ 直接生成答案
              └─ 需要 → 生成检索查询 → 检索 →
                        LLM 批判每个检索片段是否相关 →
                        基于相关片段生成答案，并自我批判答案是否忠实于片段

本质区别：

传统 RAG 把“检索”和“生成”硬性分离，检索是一个独立的前置步骤，无论问题是什么，都强制执行。这导致大量简单问题（如“你叫什么名字？”、“今天天气怎么样？”）也走了一遍昂贵的检索流程，增加了延迟和成本。

Self-RAG 则把“要不要检索”的决策权交给了模型本身。它在训练时被灌输了特殊的“反思令牌”（Reflection Tokens），使得模型可以在推理时动态判断是否需要检索，并对检索回来的每一段内容进行批判性评估——哪些是相关的，哪些是废话，生成的答案是否忠实于检索到的信息。

一个直观的代码对比：

# 传统 RAG：无差别检索
def traditional_rag(question):
    docs = vectorstore.similarity_search(question, k=5)  # 不管问题是什么，都去查
    prompt = f"参考资料：\n{docs}\n\n问题：{question}\n回答："
    return llm(prompt)

# Self-RAG：模型自主决定是否检索
def self_rag(question, llm):
    # 模型生成过程可以输出特殊的 token 来表明意图
    response = llm.generate_with_tokens(question)  # 能输出 <RETRIEVE> 或 <NO_RETRIEVE>
    if "<RETRIEVE>" in response:
        query = llm.extract_query(response)  # 模型自己写的检索 query
        docs = vectorstore.search(query)
        final = llm.generate_with_critique(question, docs)  # 生成 + 批判
        return final
    else:
        return llm.extract_answer(response)  # 直接返回答案，不走检索

这就是 Self-RAG 最核心的设计哲学：不是所有问题都需要检索，检索应该是一个可以被模型自主调用的“工具”，而不是一个强制的前置流程。

🔬 2. Self-RAG 的核心机制：如何通过 Reflection Token 实现自我批判和按需检索？

Self-RAG 的核心创新在于在模型的词表中引入了一套特殊的“反思令牌”，并让模型在生成过程中使用这些令牌来规划、监控和评估自己的行为。这相当于给模型安装了一个“元认知系统”。

常用的 Reflection Token 类型：

查看内嵌表格

工作流程可以拆解为四个步骤：

步骤一：按需检索 (Retrieve on Demand)
  模型首先生成 <RETRIEVE> 或 <NO_RETRIEVE>。
  如果判定需要检索，模型会继续生成一个或多个检索查询。

步骤二：并行检索并批判
  对检索返回的每个片段，模型输出 <ISREL>（相关）或 <IRREL>（不相关）。
  对相关片段，进一步输出 <ISSUP>（支持答案）或 <NOTSUP>。

步骤三：选择性生成
  模型基于被判定为相关且有用的片段，生成答案。
  生成过程中，可以随时输出 <FULLY>（完全忠实）、<PARTIALLY>（部分忠实）或 <NO>（不忠实）来监控自己。

步骤四：自我修正与最终输出
  如果模型在某一步发现自己不忠实或不正确，它可以回溯、重新检索或重新生成，直到输出 <FULLY> 满意为止。

代码示例：模拟 Self-RAG 的推理循环

def self_rag_inference(question, llm, retriever):
    """
    模拟 Self-RAG 的推理过程。实际 Self-RAG 模型的生成是端到端的，
    这里为了演示逻辑，拆解为步骤。
    """
    # 步骤一：生成决策 token
    decision_prompt = f"问题: {question}\n请判断是否需要检索外部信息。输出 <RETRIEVE> 或 <NO_RETRIEVE>"
    decision = llm(decision_prompt)

    if "<NO_RETRIEVE>" in decision:
        answer_prompt = f"直接回答以下问题: {question}"
        return llm(answer_prompt)

    # 步骤二：生成检索查询
    query_prompt = f"问题: {question}\n请生成一个检索查询词："
    search_query = llm(query_prompt)
    docs = retriever(search_query, top_k=5)

    # 步骤三：对每个文档片段进行批判
    relevant_docs = []
    for i, doc in enumerate(docs):
        critique_prompt = f"问题: {question}\n文档片段: {doc}\n判断该片段是否与问题相关，输出 <ISREL> 或 <IRREL>"
        judgement = llm(critique_prompt)
        if "<ISREL>" in judgement:
            relevant_docs.append(doc)

    if not relevant_docs:
        return "抱歉，我未找到相关信息。"

    # 步骤四：基于相关文档生成答案，并自我批判
    context = "\n".join(relevant_docs)
    generate_prompt = f"""参考资料：
{context}

问题：{question}

请基于参考资料回答问题。回答完后，用 <FULLY>、<PARTIALLY> 或 <NO> 评估你的答案是否忠实于参考资料。"""
    final_output = llm(generate_prompt)

    # 如果模型自评为不忠实，可以触发重试或降级
    if "<NO>" in final_output:
        # 简化处理：重新生成一次
        final_output = llm(f"请更仔细地基于以下资料回答：\n{context}\n问题：{question}")

    # 移除 token 返回给用户
    return final_output.replace("<FULLY>", "").replace("<PARTIALLY>", "").replace("<NO>", "")

核心价值：

这套机制让模型从“无脑复读机”变成了“谨慎的研究员”——它会先问自己“我到底需不需要查资料？”，拿到资料后会逐一审视“这段有用吗？”，写完答案后还会自检“我说的这些，资料里都能找到吗？”。这正是从 RAG 到 Self-RAG 的质变。

🚦 3. 大量“今天天气怎么样”这类问题也在走检索，导致延迟高、成本大，如何优化？

问题根因非常明确：系统没有对“是否需要检索”进行判断，所有问题都无差别地走了一遍检索流水线。 而实际业务中，可能有 20%-40% 的请求是闲聊、通用知识或简单指令，根本不需要查知识库。

优化方案：在检索之前，增加一个轻量级的“门控”层。

优化前：
  用户提问 → [向量化 + 检索 + LLM] → 回答

优化后：
  用户提问 → [轻量级意图分类] →
              ├─ 需要检索 → [检索 + RAG 生成] → 回答
              └─ 无需检索 → 直接回答 / 缓存答案 → 回答

具体实现策略，由简到繁：

方案一：基于规则的快速过滤（最快上线）

用一组简单的规则，拦截那些明显不需要检索的问题。

def needs_retrieval(question: str) -> bool:
    # 规则1：闲聊/问候
    greeting_patterns = ["你好", "你是谁", "你叫什么", "再见", "谢谢", "今天天气"]
    if any(pattern in question for pattern in greeting_patterns):
        return False

    # 规则2：通用知识（模型训练数据里大概率有）
    common_knowledge_keywords = ["什么是", "如何写", "Python 基础", "公式", "翻译"]
    if any(keyword in question for keyword in common_knowledge_keywords):
        return False

    # 规则3：太短的问题（5个字以下，大概率是闲聊）
    if len(question) < 5:
        return False

    # 默认需要检索
    return True

方案二：用轻量级分类模型做意图识别

训练一个 BERT 级别的小模型，对用户问题做二分类：“需要检索” vs “无需检索”。训练数据可以从线上日志里人工标注 500-1000 条得到。

from transformers import pipeline

# 加载预先 fine-tune 好的分类器
classifier = pipeline("text-classification", model="your-retrieval-intent-classifier")

def needs_retrieval_ml(question: str) -> bool:
    result = classifier(question)[0]
    return result["label"] == "NEEDS_RETRIEVAL" and result["score"] > 0.8

方案三：Self-RAG 风格的推理（效果最好，但模型需改造）

如果你的模型支持（如使用了经过微调的 Self-RAG 模型），可以直接让模型在生成的第一阶段输出是否需要检索的令牌，完全省掉额外的分类模型。

def self_rag_gate(question, llm):
    decision = llm.generate_special_tokens(question, max_tokens=1)
    return "<RETRIEVE>" in decision

方案四：结合缓存，进一步减少无谓调用

对于“今天天气怎么样”这类高频且答案相对固定的问题，即使在无检索路径里，也可以引入语义缓存。用向量相似度匹配历史问题，如果命中且未过期，直接返回缓存答案。

import hashlib, redis

cache = redis.Redis()

def cached_or_generate(question, generate_func):
    # 简单的精确缓存（用哈希键）
    key = hashlib.md5(question.encode()).hexdigest()
    cached = cache.get(key)
    if cached:
        return cached.decode()
    answer = generate_func(question)
    cache.setex(key, 3600, answer)  # 1小时过期
    return answer

完整的优化后流程：

def optimized_rag(question):
    # 第一步：门控判断
    if not needs_retrieval(question):
        # 走直接生成或缓存
        return direct_answer(question)

    # 第二步：需要检索的，再走检索+生成
    docs = vectorstore.similarity_search(question, k=5)
    return rag_generate(question, docs)

效果预估：

通过引入门控机制，通常可以将检索调用量降低 30%-50%，P50 延迟下降 40% 以上，同时大大节省 Embedding 和向量检索的计算成本。这对于高频、低时延要求的在线服务尤为重要。

收束：

Self-RAG 的思想，并不一定需要完整的 Self-RAG 模型才能落地。理解它“按需检索”和“自我批判”的精髓之后，我们可以先用规则和轻量模型逐步接近这个目标。让系统学会判断“什么时候不该查”，比让它学会“怎么查得更准”有时候更能立竿见影地提升用户体验。

Agentic RAG

🤖 1. Agentic RAG 和传统 RAG Pipeline 最核心的区别是什么？

一句话：传统 RAG 是“死”的流水线，而 Agentic RAG 是“活”的智能体。前者只负责执行固定步骤，后者能自主决定做什么、怎么做、何时停。

传统 RAG Pipeline：
  问题 → [固定: 检索] → [固定: 拼提示词] → [固定: LLM 生成] → 答案

Agentic RAG：
  问题 → [Agent 大脑] → 需要检索吗？
           ├─ 否 → 直接回答
           └─ 是 → 去哪里检索？(选知识库/工具)
                   → 检索后：结果够好吗？
                       ├─ 否 → 改写查询，再检索，或切换策略
                       └─ 是 → 生成答案 → 检查答案忠实度/完整性
                                  ├─ 通过 → 输出
                                  └─ 未通过 → 修正或补充检索

五个维度的本质差异：

查看内嵌表格

核心区别在于“决策权”：传统 RAG 的每一步都是代码写死的，而 Agentic RAG 引入了一个 Agent 大脑（通常是 LLM 加上工具使用能力），让它能在一个 while 循环里不断问自己：“我现在知道什么？还缺什么？接下来做什么？”。这个大脑赋予 RAG 前所未有的灵活性和鲁棒性。

🧭 2. 什么是 Agentic RAG？在多知识库场景下，Agent 如何实现动态路由和迭代检索？

Agentic RAG 就是将 RAG 的检索和生成能力封装成工具，交给一个能够推理和规划的 Agent 来调度。 这个 Agent 不仅能回答问题，还能主动分析问题、制定检索计划、并在执行过程中根据反馈调整策略。

在多知识库场景下（例如，公司内部有“产品文档库”、“HR 政策库”、“技术规范库”），Agent 需要解决两个核心问题：去哪里查 和 怎么查到底。

2.1 动态路由：去哪查？

Agent 首先需要根据用户意图，自动选择正确的知识库或工具。这通常由一个 Router Agent 或分类器实现。

image.png

实现方式：

  • LLM 路由：在 System Prompt 中描述各个知识库的职责，让 LLM 根据问题输出选择。例如：“如果问题是 HR 相关，输出 HR；如果是产品相关，输出 PRODUCT；否则输出 GENERAL。”

  • 语义路由：将各个知识库的“描述”向量化，计算用户问题的向量，选择最匹配的知识库。

示例代码：用 LLM 实现动态路由

def route_to_knowledge_base(question: str, llm) -> str:
    prompt = f"""
你是一个路由器。根据用户问题，决定应该查询哪个知识库。
知识库列表：
- HR_DB: 包含员工手册、休假政策、福利待遇等HR相关信息。
- PRODUCT_DB: 包含产品功能、API文档、版本发布说明等。
- TECH_DB: 包含内部技术规范、架构设计、代码标准等。
如果问题不涉及任何特定知识库，回复 "GENERAL"。

用户问题: {question}
请只回复知识库名称（HR_DB, PRODUCT_DB, TECH_DB, GENERAL）:
"""
    return llm(prompt).strip()

2.2 迭代检索：怎么查到底？

复杂问题往往不是一次检索就能解决的。Agentic RAG 允许 Agent 进行 多轮检索，每一轮都基于上一轮的结果调整策略。这包括：

  1. 子问题分解：将复杂问题拆成多个子问题，逐一检索再汇总。

  2. 查询改写：如果第一次检索结果不理想，让 LLM 改写查询词，用同义词或更精确的术语重新检索。

  3. 多源验证：从一个知识库查到部分信息，再从另一个知识库补充或交叉验证。

示例代码：一个简单的迭代 Agent 循环

def agentic_rag(question: str, llm, tools: dict, max_iter=5):
    """
    tools: {'HR_DB': retriever_func, 'PRODUCT_DB': retriever_func, ...}
    """
    context = []
    for i in range(max_iter):
        # 1. Agent 思考下一步行动
        action_prompt = f"""
你是一个研究助手。你的目标是回答用户问题。你可以使用以下工具：
- search_hr(query): 搜索HR知识库
- search_product(query): 搜索产品知识库
- search_tech(query): 搜索技术知识库
- finish(answer): 当你确定能完整回答时，调用此函数。

当前已收集的信息：
{context if context else "无"}

用户问题：{question}

请决定下一步行动。只输出函数名和参数，例如: search_hr("年假政策")
或者 finish("你的最终答案")
"""
        action = llm(action_prompt).strip()

        if action.startswith("finish"):
            return action.split("finish(")[1].rstrip(")").strip('"')

        # 2. 解析并执行工具调用
        tool_name, query = parse_action(action)
        if tool_name in tools:
            result = tools[tool_name](query)
            context.append(f"[{tool_name}] 查询: {query}\n结果: {result[:500]}")
        else:
            context.append(f"错误: 未知工具 {tool_name}")

    # 超过最大迭代，强制生成答案
    force_prompt = f"基于以下信息，尽最大努力回答：{context}\n问题：{question}"
    return llm(force_prompt)

在这个循环里，Agent 可以看到自己已经知道的信息（context），并决定是继续查、换个库查、还是已经够了。这就是 迭代检索 的核心——Agent 在信息积累的每一步都重新评估局势，而不是一条路走到黑。

🔁 3. 用户反馈 Agent 在回答复杂问题时经常“原地打转”，同样的知识库查了 5 遍还没有答案，如何解决？

“原地打转”的本质是：Agent 缺乏对“检索有效性”的评估能力和“终止条件”的刚性约束。 它像一只被关在房间里的蜜蜂，不停撞向同一扇玻璃窗，以为下一次就能出去。

排查与解决路线图：

image.png

具体优化措施：

① 给 Agent 装上“无用探测器”

在每一次检索后，强制 Agent 回答一个问题：“这次返回的信息，是否让我离最终答案更近了？”如果连续两次答案都是“否”，就强制切换策略（如改写查询、换知识库、或承认无法回答）。

def evaluate_retrieval_utility(previous_context, new_info, llm):
    prompt = f"""
之前的上下文：{previous_context}
新检索到的信息：{new_info}
请判断新信息是否提供了之前没有的、有助于回答问题的内容。只回答 "有用" 或 "无用"。
"""
    return llm(prompt).strip()

# 在 Agent 循环中
if evaluate_retrieval_utility(old_context, new_result) == "无用":
    useless_count += 1
    if useless_count >= 2:
        # 强制切换策略，如进行查询改写或直接回答“信息不足”
        action = "switch_strategy"
else:
    useless_count = 0

② 引入“防重复”记忆机制

记录每一次检索的查询词。如果 Agent 试图用完全相同或高度相似的查询再次搜索，直接拦截并提示它改写。

past_queries = set()

def is_redundant(query):
    # 简单去重，也可以用余弦相似度判断
    if query in past_queries:
        return True
    # 相似度判断（可选）
    for past_q in past_queries:
        if similarity(query, past_q) > 0.9:
            return True
    past_queries.add(query)
    return False

③ 设置硬性界限，优雅退出

除了最大步数，还可以设置“无进展终止”：如果连续 N 轮检索的信息都没有使上下文长度或关键实体数量增加，就强制结束，让 Agent 基于现有信息生成答案并坦白“部分信息缺失”。

④ 从“直给检索”升级为“子问题分解 + 递归检索”

许多“原地打转”是因为问题太复杂，单个查询无法命中。Agent 应该先将问题分解为可分别检索的子问题，然后逐一解决。如果子问题 A 的检索结果为空，Agent 应该尝试改写查询或承认子问题无法回答，而不是反复去撞南墙。

示例代码：带反思和防重机制的增强 Agent

def enhanced_agentic_rag(question, llm, tools, max_iter=8):
    context = []
    past_queries = set()
    useless_streak = 0

    for i in range(max_iter):
        # 生成下一步行动，要求 Agent 同时输出思考过程
        prompt = f"""
你是一个研究助手。回答以下问题：{question}
你可以使用工具：search_hr, search_product, search_tech, finish(answer)
规则：
- 不要重复之前失败的查询。过去的查询：{past_queries}
- 如果连续两次检索无用，请换策略或考虑直接回答。
当前上下文：{context}

请输出下一步：思考：<你的思考> 行动：<函数名(参数)>"""
        response = llm(prompt)
        think, action = parse_think_and_action(response)

        if action.startswith("finish"):
            return action.split("finish(")[1].rstrip(")").strip('"')

        tool_name, query = parse_action(action)
        if is_redundant(query, past_queries):
            # 拦截重复查询，强制反思
            context.append("系统提示：该查询已执行过，请改写查询。")
            continue

        past_queries.add(query)
        result = tools[tool_name](query)

        # 评估有用性
        utility = evaluate_retrieval_utility(context, result[:500], llm)
        if utility == "无用":
            useless_streak += 1
            if useless_streak >= 2:
                context.append("连续检索无进展，请考虑更换策略或直接回答。")
                useless_streak = 0  # 重置以便给一次机会
        else:
            useless_streak = 0
            context.append(result[:500])

    # 强制结束，生成最佳努力答案
    return llm(f"基于以下信息回答，如果无法回答请说明：\n{context}\n问题：{question}")

收束：

Agent 原地打转，不是它太蠢，而是我们没给它画好“停”的线——没有评价机制、没有去重、没有切换策略的勇气。给 Agent 加上“自我反思”、“去重记忆”和“刚性退出”这三道保险，它就能从一只无头苍蝇，变成一位知道何时坚持、何时放弃、何时换条路走的成熟研究员。这正是 Agentic RAG 工程化的核心所在：用代码的确定性，去驾驭智能体的不确定性。