RAG 架构
一条最基础的 RAG 管道分三步：索引（把文档切块、向量化后存入向量库）、检索（把用户问题向量化，取最相近的若干块）、生成（把检索到的块拼进提示，让模型基于资料作答）。下图是一次完整问答的在线链路：
注意「索引」是离线阶段，与在线问答解耦：知识库更新时只需重跑索引，不用动线上服务。这也是 RAG 相比微调最大的工程优势：知识更新成本从「重训模型」降为「重建索引」。
关键认知：RAG 的效果上限定在「检索质量」。检索不到，生成再强也白搭。所以切分与召回策略优先于模型选择。
文档切分
切块（chunking）太大则噪声多，太小则语义断裂。常见做法：按段落或固定 token 数切分，并保留滑动重叠（overlap）避免截断句子；对代码、表格等结构化内容用专用切分器。
先看最朴素的固定长度切分，30 行原生 Python 就能实现，重点是理解 overlap 的作用：
# 固定长度切分：按字符数切，块间保留重叠避免截断关键句
def fixed_size_chunk(text, chunk_size=500, overlap=100):
"""chunk_size: 每块最大字符数; overlap: 相邻块的重叠字符数"""
chunks = []
start = 0
while start < len(text):
end = start + chunk_size
chunks.append(text[start:end])
# 下一块从 end - overlap 开始，保证跨块的句子在两块中都完整出现
start = end - overlap
return chunks
with open("handbook.txt", encoding="utf-8") as f:
chunks = fixed_size_chunk(f.read())
print(f"共切出 {len(chunks)} 块")固定长度实现简单但会在句子中间「一刀切」。更优解是语义切分（semantic chunking）：先按句子切，计算相邻句子的嵌入相似度，在相似度骤降的位置断开，让每块内部主题一致：
# 语义切分：在相邻句子相似度骤降处断块（pip install sentence-transformers）
from sentence_transformers import SentenceTransformer
import numpy as np
import re
model = SentenceTransformer("BAAI/bge-small-zh-v1.5") # 中文小模型, CPU 可跑
def semantic_chunk(text, threshold=0.55):
# 1. 按中文句号/问号/感叹号切成句子
sents = [s for s in re.split(r"(?<=[。！？])", text) if s.strip()]
# 2. 批量计算句向量并归一化, 点积即余弦相似度
embs = model.encode(sents, normalize_embeddings=True)
chunks, cur = [], [sents[0]]
for i in range(1, len(sents)):
sim = float(np.dot(embs[i - 1], embs[i]))
if sim < threshold: # 相似度骤降 = 主题切换, 断块
chunks.append("".join(cur))
cur = []
cur.append(sents[i])
chunks.append("".join(cur))
return chunks两种策略如何选？用这张对比表快速判断：
- 固定长度 + overlap：实现零成本、速度快、块大小可控（利于控制 token 预算）；缺点是可能切断语义。适合结构松散的日志、聊天记录、初版原型。
- 语义切分：块内主题一致、检索命中率更高；缺点是要多跑一遍嵌入、块长不可控。适合手册、论文、政策文档等长文。
- 结构感知切分：Markdown 按标题层级切、代码按函数切（LangChain 的
MarkdownHeaderTextSplitter、RecursiveCharacterTextSplitter.from_language），有结构信息时永远优先用结构。
经验值：中文场景 chunk 大小 300 至 800 字符、overlap 取 10% 至 20% 是常见起点。但没有万能参数，一定要用自己的查询集实测（见下文评估章节）。
切分策略对比实验
下面是一段完整可运行的对比脚本，在同一篇中文文章上运行三种切分策略 -- 固定长度、递归字符拆分（LangChain RecursiveCharacterTextSplitter） 和 语义拆分（SemanticChunker），对比块数、平均块长与前 3 块的示例输出：
# chunk_compare.py：三种切分策略对比（pip install langchain langchain-experimental）
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings
import re
# ---------- 准备数据 ----------
with open("article.txt", encoding="utf-8") as f:
text = f.read()
# ---------- 策略 1：固定长度切分 ----------
def fixed_size(text, size=500, overlap=100):
chunks, start = [], 0
while start < len(text):
chunks.append(text[start:start + size])
start += size - overlap
return chunks
fixed_chunks = fixed_size(text)
# ---------- 策略 2：递归字符拆分（优先按段落，再按句子） ----------
recursive_splitter = RecursiveCharacterTextSplitter(
separators=["\n\n", "\n", "。", "！", "？", " ", ""],
chunk_size=500, chunk_overlap=100
)
recursive_chunks = recursive_splitter.create_documents([text])
# ---------- 策略 3：语义切分（嵌入相似度骤降处断块） ----------
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
semantic_splitter = SemanticChunker(
embeddings, breakpoint_threshold_type="percentile",
breakpoint_threshold_amount=50 # 相似度低于 50 百分位时断开
)
semantic_docs = semantic_splitter.create_documents([text])
# ---------- 统计对比 ----------
strategies = [
("固定长度", fixed_chunks),
("递归字符", [d.page_content for d in recursive_chunks]),
("语义切分", [d.page_content for d in semantic_docs]),
]
for name, chunks in strategies:
lens = [len(c) for c in chunks]
print(f"\n=== {name} ===")
print(f"块数: {len(chunks)} 平均块长: {sum(lens)/len(lens):.0f} 字符 min: {min(lens)} max: {max(lens)}")
# 打印前 3 块的首句，肉眼判断语义完整性
for i, c in enumerate(chunks[:3]):
preview = re.sub(r"\s+", " ", c[:80].strip())
print(f" 块{i+1}: {preview}...")跑完后你会看到：固定切分在 500 字符处一刀切，可能把一句话断在两块里；递归切分优先在段落/句号处断开，块边界更自然；语义切分在主题切换位置断开，块内语义最连贯但块长波动最大。建议先用递归切分作为基线，如果 Hit Rate 不理想再尝试语义切分。
向量化与检索
嵌入模型（Embedding）把文本映射成向量，相似语义在向量空间里距离更近。检索阶段用近似最近邻（ANN）算法如 HNSW、IVF 快速召回 Top-K。先用 sentence-transformers 直观感受「语义相近 = 向量相近」：
# 嵌入与相似度：语义相近的句子, 余弦相似度更高
from sentence_transformers import SentenceTransformer
model = SentenceTransformer("BAAI/bge-small-zh-v1.5")
texts = ["如何申请退货退款", "我想退掉这个商品", "今天天气真好"]
embs = model.encode(texts, normalize_embeddings=True)
# 归一化后点积即余弦相似度
print(embs[0] @ embs[1]) # 约 0.7+, 语义相近
print(embs[0] @ embs[2]) # 约 0.2, 语义无关生产中不会手写最近邻循环，而是交给向量库。LangChain 的最简检索骨架如下：
# LangChain 最简 RAG 检索骨架
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
vectorstore = Chroma.from_documents(docs, OpenAIEmbeddings())
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})
docs = retriever.invoke(question)召回后建议接一层重排序（re-rank），用交叉编码器对候选块与问题做精细打分，提升送入模型的块质量。双编码器（bi-encoder）快但粗，交叉编码器（cross-encoder）慢但准，典型组合是「向量召回 50 条 → 交叉编码器重排取前 5」：
# 交叉编码器重排序：对召回的候选块精细打分
from sentence_transformers import CrossEncoder
reranker = CrossEncoder("BAAI/bge-reranker-base")
question = "退货需要多少天内申请?"
# candidates 是向量召回的 50 个文本块
scores = reranker.predict([(question, c) for c in candidates])
# 按分数降序, 只保留前 5 块送入生成阶段
top5 = [c for _, c in sorted(zip(scores, candidates), reverse=True)[:5]]端到端管线实战
把前面的零件组装起来：下面这段约 60 行的脚本是一条完整可运行的本地 RAG 管线，覆盖「切分 → 嵌入 → 写库 → 检索 → 拼上下文 → 生成」六步。只依赖三个库：pip install sentence-transformers chromadb openai。
# mini_rag.py：端到端 RAG 管线（本地嵌入 + Chroma + OpenAI 兼容接口）
from sentence_transformers import SentenceTransformer
import chromadb
from openai import OpenAI
# ---------- 第 1 步：切分 ----------
def chunk_text(text, size=500, overlap=100):
chunks, start = [], 0
while start < len(text):
chunks.append(text[start:start + size])
start += size - overlap
return chunks
with open("handbook.txt", encoding="utf-8") as f:
chunks = chunk_text(f.read())
# ---------- 第 2 步：嵌入（本地模型, 无需 API 费用） ----------
embedder = SentenceTransformer("BAAI/bge-small-zh-v1.5")
embeddings = embedder.encode(chunks, normalize_embeddings=True)
# ---------- 第 3 步：写入向量库（持久化到本地目录） ----------
client = chromadb.PersistentClient(path="./rag_db")
col = client.get_or_create_collection("handbook")
col.add(
ids=[f"chunk-{i}" for i in range(len(chunks))],
documents=chunks,
embeddings=embeddings.tolist(),
)
# ---------- 第 4 步：检索 ----------
def retrieve(question, k=4):
q_emb = embedder.encode([question], normalize_embeddings=True)
res = col.query(query_embeddings=q_emb.tolist(), n_results=k)
return res["documents"][0] # Top-K 文本块列表
# ---------- 第 5 步：拼装上下文 ----------
def build_prompt(question, ctx_chunks):
context = "\n\n".join(f"[资料{i+1}] {c}" for i, c in enumerate(ctx_chunks))
return f"""仅根据以下资料回答问题, 并标注引用的资料编号。
资料中找不到答案时, 回答"资料中未提及", 不要编造。
{context}
问题：{question}"""
# ---------- 第 6 步：生成（任意 OpenAI 兼容服务, 含本地 vLLM/Ollama） ----------
llm = OpenAI() # 读取环境变量 OPENAI_API_KEY / OPENAI_BASE_URL
question = "退货需要在几天内申请?"
resp = llm.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": build_prompt(question, retrieve(question))}],
temperature=0, # 事实型问答用低温度, 减少发挥
)
print(resp.choices[0].message.content)三处工程细节值得注意：提示词里显式要求「找不到就说未提及」是对抗幻觉的第一道闸；标注资料编号让回答可追溯；temperature=0 让事实型回答更稳定。若偏好框架，等价的 LangChain 版本只需十几行：
# LangChain 等价实现：LCEL 把检索与生成串成一条链
from langchain_chroma import Chroma
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
emb = HuggingFaceEmbeddings(model_name="BAAI/bge-small-zh-v1.5")
retriever = Chroma(persist_directory="./rag_db", embedding_function=emb).as_retriever(search_kwargs={"k": 4})
prompt = ChatPromptTemplate.from_template(
"仅根据资料回答, 找不到就说未提及。\n资料：{context}\n问题：{question}"
)
def fmt(docs):
return "\n\n".join(d.page_content for d in docs)
chain = ({"context": retriever | fmt, "question": RunnablePassthrough()}
| prompt | ChatOpenAI(model="gpt-4o-mini", temperature=0))
print(chain.invoke("退货需要在几天内申请?").content)高级 RAG
基础管线跑通后，多数效果问题出在召回环节。以下四个模式按投入产出比排序：
- 多路召回：向量检索 + 关键词（BM25）混合，兼顾语义与精确匹配。型号、人名、错误码这类字面匹配是向量检索的盲区。
- HyDE：先让模型生成「假设答案」，用假设答案去检索，缓解「问题短、文档长」的语义鸿沟。
- 自查询：让模型把问题拆成语义查询与元数据过滤（如时间、部门、文档类型），精准定位。
- 上下文压缩：检索后只保留与问题相关的句子，减少噪声与 token 消耗。
多路召回是性价比最高的一步，核心是用 RRF（倒数排名融合）合并两路结果，不需要调分数权重：
# 混合检索：BM25 + 向量召回, RRF 融合（pip install rank-bm25 jieba）
from rank_bm25 import BM25Okapi
import jieba
bm25 = BM25Okapi([list(jieba.cut(c)) for c in chunks])
def hybrid_search(question, k=4, rrf_k=60):
# 两路各自产生一个排名列表（元素为 chunk 下标）
bm25_scores = bm25.get_scores(list(jieba.cut(question)))
bm25_rank = sorted(range(len(chunks)), key=lambda i: -bm25_scores[i])
q = embedder.encode([question], normalize_embeddings=True)
ids = col.query(query_embeddings=q.tolist(), n_results=len(chunks))["ids"][0]
vec_rank = [int(i.split("-")[1]) for i in ids]
# RRF：得分 = Σ 1/(rrf_k + 该路中的名次), 名次越靠前贡献越大
fused = {}
for rank in (bm25_rank, vec_rank):
for pos, idx in enumerate(rank):
fused[idx] = fused.get(idx, 0) + 1 / (rrf_k + pos + 1)
best = sorted(fused, key=fused.get, reverse=True)[:k]
return [chunks[i] for i in best]调优顺序：先把切分与混合召回做好，再加重排序，最后才考虑 HyDE 等生成式技巧。每加一层都用同一份评测集验证收益，否则只是在叠加延迟。
混合检索完整实现
下面是最小可运行的混合检索 + 评估脚本，对比纯向量检索和 BM25+向量 RRF 混合检索在同一个询集上的 Hit@5 和 MRR。依赖：pip install sentence-transformers rank-bm25 chromadb。
# hybrid_retrieval_eval.py：混合检索 BM25 + 向量 RRF 融合，含评估
from sentence_transformers import SentenceTransformer
from rank_bm25 import BM25Okapi
import chromadb
import jieba
import numpy as np
# ---------- 1. 准备数据 ----------
chunks = ["退货需在签收后 7 天内申请，超过 7 天系统自动关闭退换通道。",
"商品存在质量问题的，运费由卖家承担；非质量原因的退货，运费由买家承担。",
"退款将在退货签收后 3 个工作日内原路返回支付账户。",
"换货需先提交申请，客服审核后 48 小时内给出处理结果。",
"本店所有电子产品提供一年整机保修，终身技术支持。",
"积分商城每周五上午 10 点更新，积分不可转让或提现。",
"VIP 会员享全年 9 折、优先配送、专属客服三重权益。"]
# ---------- 2. 构建 BM25 索引 ----------
bm25 = BM25Okapi([list(jieba.cut(c)) for c in chunks])
# ---------- 3. 构建向量库 ----------
embedder = SentenceTransformer("BAAI/bge-small-zh-v1.5")
embs = embedder.encode(chunks, normalize_embeddings=True)
client = chromadb.Client() # 内存模式，演示用
col = client.create_collection("demo")
col.add(ids=[f"c{i}" for i in range(len(chunks))],
documents=chunks, embeddings=embs.tolist())
# ---------- 4. RRF 融合函数 ----------
def rrf_fusion(rank_lists, rrf_k=60):
"""
多路排名列表 → RRF 融合得分排序
rank_lists: [[idx3, idx1, idx0, ...], [idx1, idx0, idx3, ...]]
每路内部按相关性从高到低排列
"""
scores = {}
for ranks in rank_lists:
for pos, idx in enumerate(ranks):
scores[idx] = scores.get(idx, 0) + 1 / (rrf_k + pos + 1)
return sorted(scores, key=scores.get, reverse=True)
# ---------- 5. 两路检索对比 ----------
questions = [
("退货期限是多少天?", "c0"), # 应命中 chunk 0(7 天)
("运费由谁承担?", "c1"), # 应命中 chunk 1(运费)
("保修政策是什么?", "c4"), # 应命中 chunk 4(保修)
("VIP 会员有什么权益?", "c6"), # 应命中 chunk 6(VIP)
("退款多久到账?", "c2"), # 应命中 chunk 2(退款)
]
all_ids = set(range(len(chunks)))
def evaluate(name, rank_fn, k=5):
"""rank_fn(question) returns sorted idx list"""
hit, rr_sum = 0, 0.0
for q, gold in questions:
ranked = rank_fn(q)
top_k = ranked[:k]
if gold in top_k:
hit += 1
rr_sum += 1 / (top_k.index(gold) + 1)
n = len(questions)
print(f"{name}: Hit@{k} = {hit}/{n} ({hit/n:.0%}), MRR = {rr_sum/n:.3f}")
# 纯向量检索
def vector_rank(q):
q_emb = embedder.encode([q], normalize_embeddings=True)
ids = col.query(query_embeddings=q_emb.tolist(), n_results=len(chunks))["ids"][0]
return [int(i.strip("c")) for i in ids]
# 混合检索：BM25 + 向量 RRF 融合
def hybrid_rank(q):
# BM25 排名
bm_scores = bm25.get_scores(list(jieba.cut(q)))
bm_rank = sorted(all_ids, key=lambda i: -bm_scores[i])
# 向量排名（全量召回）
q_emb = embedder.encode([q], normalize_embeddings=True)
all_res = col.query(query_embeddings=q_emb.tolist(), n_results=len(chunks))
v_ids = all_res["ids"][0]
v_rank = [int(i.strip("c")) for i in v_ids]
return rrf_fusion([bm_rank, v_rank])
print("=" * 50)
evaluate("纯向量检索", vector_rank)
evaluate("混合检索(BM25+Vec+RRF)", hybrid_rank)
# 逐条对比哪些查询混合更好
print("\n逐条分析（Top-3 排名）：")
for q, gold in questions:
v_top3 = vector_rank(q)[:3]
h_top3 = hybrid_rank(q)[:3]
def pos(lst):
return lst.index(gold) + 1 if gold in lst else "-"
print(f" {q[:20]:<20s} 向量排名:{pos(v_top3):>3} 混合排名:{pos(h_top3):>3}")运行后你会看到混合检索通常在含数字、专有名词（如「7 天」「VIP」「运费」）的查询上优于纯向量 -- 因为这些是字面匹配的强项，向量检索可能被近义但不相关的「工作日处理时间」「会员权益」干扰。双路的好处是互补：BM25 保证精确匹配不丢，向量保证语义扩展有回旋余地。
评估
RAG 要分别评估检索（召回率、MRR、NDCG）与生成（忠实度、答案相关度）。可用 RAGAS、TruLens 等框架自动打分，再人工抽检。上手第一步不需要框架：构造 20 至 50 条「问题 → 应命中的块」标注对，先算检索侧的 Hit Rate 与 MRR：
# 检索评估：Hit Rate（命中率）与 MRR（平均倒数排名）
# eval_set 每条含 question 与 gold_id（应命中的 chunk id）
eval_set = [
{"question": "退货期限是多久?", "gold_id": "chunk-12"},
{"question": "运费谁承担?", "gold_id": "chunk-15"},
]
hit, rr_sum = 0, 0.0
for item in eval_set:
q = embedder.encode([item["question"]], normalize_embeddings=True)
ids = col.query(query_embeddings=q.tolist(), n_results=5)["ids"][0]
if item["gold_id"] in ids:
hit += 1
rr_sum += 1 / (ids.index(item["gold_id"]) + 1) # 名次的倒数
n = len(eval_set)
print(f"Hit@5 = {hit / n:.2%}, MRR = {rr_sum / n:.3f}")Hit@5 低说明切分或召回有问题，先别动生成端。检索达标后再用 RAGAS 评生成侧的忠实度（faithfulness，回答是否被资料支撑）与答案相关度。检索背后的向量库选型，详见 向量数据库 模块；如何量化整体表现，见 评估与评测。
动手练习
三个练习由浅入深，建议全部在本地完成，总耗时约 2 至 3 小时。
练习 1：切分策略对照实验（约 30 分钟）。取一篇 3000 字以上的中文文章（如产品手册或长篇博客），分别用固定长度切分（500 字符 / overlap 100）与语义切分（threshold 0.55）处理，打印两种结果的块数、平均块长，并肉眼检查各 5 个块的完整性。
验收标准：能指出至少 2 处「固定切分把完整语义切断、语义切分保住了」的具体例子，并写一句话结论说明该文章适合哪种策略。
练习 2：对一个 PDF 建迷你 RAG 并问答（约 60 分钟）。选一份 10 页以上的 PDF（论文、合同模板、员工手册均可），用 pypdf 抽取文本后接入本章的 mini_rag.py 管线，建库后提出 5 个问题：3 个答案确实在文档里，2 个文档里没有。
# PDF 抽取（pip install pypdf）, 后续接 mini_rag.py 的 chunk_text
from pypdf import PdfReader
reader = PdfReader("paper.pdf")
text = "\n".join(page.extract_text() for page in reader.pages)
print(f"共 {len(reader.pages)} 页, {len(text)} 字符")验收标准：3 个在库问题全部回答正确且标注了资料编号；2 个不在库问题模型明确回答「资料中未提及」而不是编造；把 6 步管线各步的输入输出打印出来能对上（切了几块、召回了哪几块、最终提示词长什么样）。
练习 3：量化混合召回的收益（约 60 分钟）。基于练习 2 的 PDF，手工标注 15 条「问题 → 应命中块」评测对（其中至少 3 条含专有名词或数字，如型号、日期、金额），分别测「纯向量召回」与「BM25 + 向量 RRF 混合召回」的 Hit@5 与 MRR。
验收标准：产出一张两行四列的对比表（两种方案 × Hit@5 / MRR）；能解释混合召回在哪几条查询上赢了、为什么（通常是字面匹配型查询）；若混合没有提升，给出你的原因分析（如文档太短、分词质量）。
练习 4：搭建带引用追踪的 RAG（约 90 分钟）。在前面 mini_rag.py 的基础上增加引用追踪能力：每条回答除了给出答案，还要标注答案中的每条事实来自哪块文档。具体做法是将每条查询结果的资料编号信息注入生成环节，并要求模型以 [资料N] 格式标注引用。
# citation_rag.py：在 mini_rag.py 基础上增加引用追踪
from sentence_transformers import SentenceTransformer
import chromadb
from openai import OpenAI
from collections import defaultdict
# ---------- 前 4 步与 mini_rag.py 相同 ----------
def chunk_text(text, size=500, overlap=100):
chunks, start = [], 0
while start < len(text):
chunks.append(text[start:start + size])
start += size - overlap
return chunks
# （省略 文档读取 / embedder 初始化 / Chroma 写库, 同 mini_rag.py）
# ---------- 步骤 5：拼装上下文（增强版：显式给每个块加资料编号） ----------
def build_citation_prompt(question, ctx_chunks):
"""
构造带引用标注要求的提示词。
关键：明确要求模型 (1) 按资料编号引用 (2) 回答每条信息标注来源
"""
numbered_context = "\n\n".join(
f"[资料{i+1}] {c[:500]}" for i, c in enumerate(ctx_chunks)
)
return f"""你是一个严格基于资料的问答助手。
请仅根据以下资料回答问题。每条答案中的事实都必须标注其来源的资料编号，格式为 [资料N]。
资料：
{numbered_context}
问题：{question}
请用以下格式回答：
**答案**：(你的回答)
**引用**：
- [资料N]：(资料中的原句或段落)
# ---------- 步骤 6：解析引用 ----------
def parse_citations(answer, ctx_chunks):
"""
提取答案中引用的资料编号, 映射至原文块内容。
返回: {资料编号: 匹配的原文块内容}
"""
import re
refs = re.findall(r"\[资料(\d+)\]", answer)
cited = {}
for ref in set(refs):
idx = int(ref) - 1
if 0 <= idx < len(ctx_chunks):
cited[idx + 1] = ctx_chunks[idx][:200]
return cited
# ---------- 步骤 7：引用验证 ----------
def validate_citations(question, answer, ctx_chunks, gold_citations):
"""
验证回答中的引用是否与预期（gold_citations）一致。
gold_citations: 应引用的资料编号集合, 如 {1, 3}
返回: (正确引用数, 遗漏引用数, 错误引用数)
"""
found = set(parse_citations(answer, ctx_chunks).keys())
correct = found & gold_citations
missed = gold_citations - found
fp = found - gold_citations
return correct, missed, fp
# ---------- 使用示例 ----------
llm = OpenAI()
questions = [
("退货期限是多久?", {1}), # 应引用资料 1
("运费由谁承担?", {2}), # 应引用资料 2
("退款几天到账? 保修几年?", {3, 5}), # 应引用资料 3 和 5
("VIP 有什么权益?", {7}), # 应引用资料 7
("可以积分转让吗?", {6}), # 应引用资料 6
]
total_q = len(questions)
passed = 0
for q, gold_set in questions:
ctx_chunks = retrieve(q) # 复用 mini_rag.py 的检索函数
prompt = build_citation_prompt(q, ctx_chunks)
resp = llm.chat.completions.create(
model="gpt-4o-mini", temperature=0,
messages=[{"role": "user", "content": prompt}]
)
answer = resp.choices[0].message.content
correct, missed, fp = validate_citations(q, answer, ctx_chunks, gold_set)
ok = len(missed) == 0 and len(fp) == 0
if ok:
passed += 1
print(f"Q: {q[:40]}")
print(f" 正确引用: {correct}, 遗漏: {missed}, 错误: {fp} {'PASS' if ok else 'FAIL'}")
print()
print(f"\n引用正确率: {passed}/{total_q} ({passed/total_q:.0%})")
if passed >= 4:
print("验收通过: 5 题中至少 4 题引用正确")验收标准：5 个问题中至少 4 个引用全部正确（无遗漏、无错误引用）；能定位到具体的文档块；用 gold_citations 自动化验证而非肉眼检查。如果某题引用错误，检查那条查询召回的块是否确实包含该信息 -- 问题很可能出在检索端而非生成端。