RAG 的完整流程是什么？

#RAG#Agent#面试八股

RAG 的完整流程是什么？

结论：RAG 由离线索引与在线查询两条流水线组成，二者只在向量库（Vector Store）处汇合。 离线把"文档"变成"可检索的向量 + 原文 + 元数据"，在线把"问题"变成"带证据的提示词（Prompt）"。 面试时按"离线四步、在线五步"讲，再点出每一步的失败模式，比背名词更有说服力。

1. 全景图

离线索引（一次性 / 增量）
  采集 → 解析 → 清洗 → 分块 → 向量化 → 入库
  raw    text    text   chunk   vector   index + metadata

                     └──────┬──────┘
                        向量库（唯一交汇点）
                     ┌──────┴──────┐
在线查询（每次请求）
  问题 → 改写 → 向量化 → 召回 → 重排 → 编排 → 生成 → 引用
 query  rewrite  vector  top-k  top-n  prompt  answer  cite

两条线唯一的强耦合是向量化必须用同一个 Embedding 模型，模型一换，全库重算。

2. 离线阶段：把文档变成可检索的数据

步骤输入输出关键决策常见故障
采集PDF/Word/网页/库表原始字节全量还是增量、去重策略版本混杂，旧制度和新制度同时入库
解析原始字节带结构的文本版式还原、表格转 Markdown多栏 PDF 读成乱序，页眉页脚混进正文
分块长文本300~800 token 的片段切分单位、重叠比例一句话被切成两半，主语和谓语分家
向量化片段文本定长浮点向量模型、维度、是否归一化中英混排、缩写术语召回差
入库向量 + 原文索引记录元数据字段设计缺 source/page，导致无法给出处

解析这一步最容易被低估：一份扫描件 OCR 出错，后面所有环节都在为它擦屁股。

3. 在线阶段：把问题变成带证据的回答

步骤作用典型耗时不做会怎样
改写补全指代、拆子问题100~300 ms"它多少钱" 检索不到任何东西
向量化问题转向量30~80 ms—
召回top-k 粗排5~30 ms没有证据，模型只能编
重排top-k → top-n50~200 ms噪声片段挤掉正确片段
编排拼提示词、截断、排序<5 ms长上下文里关键证据被忽略
生成基于证据作答0.5~3 s—

4. 一个能跑通的主干实现

# rag.py —— 只保留主干，异常与重试从略
import hashlib
from openai import OpenAI
import chromadb

llm = OpenAI()
store = chromadb.PersistentClient(path="./chroma").get_or_create_collection(
    name="kb", metadata={"hnsw:space": "cosine"}
)

def embed(texts: list[str]) -> list[list[float]]:
    # 一次批量送，比逐条调用省一个数量级的往返开销
    resp = llm.embeddings.create(model="text-embedding-3-small", input=texts)
    return [d.embedding for d in resp.data]

def index(chunks: list[str], source: str) -> None:
    # 用内容哈希做主键：同一片段重复入库自然覆盖，增量更新无需删表
    ids = [hashlib.md5(c.encode()).hexdigest() for c in chunks]
    store.upsert(
        ids=ids, documents=chunks, embeddings=embed(chunks),
        metadatas=[{"source": source, "seq": i} for i in range(len(chunks))],
    )

def answer(question: str, k: int = 5) -> str:
    hits = store.query(query_embeddings=embed([question]), n_results=k)
    ctx = "\n\n".join(f"[{i + 1}] {d}" for i, d in enumerate(hits["documents"][0]))
    resp = llm.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system",
             "content": "只依据资料回答；资料不足就直说不知道；引用资料时用 [n] 标注编号。"},
            {"role": "user", "content": f"资料：\n{ctx}\n\n问题：{question}"},
        ],
    )
    return resp.choices[0].message.content

5. 常见追问

Q：能边查边切块吗？ 可以把切片放到查询时做，但每次请求都要重新解析和向量化，延迟翻倍；只有文档极少时才划算。

Q：k 取多少？ 召回阶段宁多勿少，k 取 2050，交给重排压到 38 条。直接 k=3 是把召回和精排的职责混在一起。

Q：引用怎么落地？ 靠入库时写进元数据的 source、page、section，生成阶段让模型输出编号，前端再把编号映射回原文，别指望模型自己复述文件路径。

Q：文档更新了怎么办？ 按内容哈希做幂等写入，改过的片段覆盖，删掉的文档按 source 批量清理，避免旧版本残留污染召回。

小结

RAG 的流程本身不复杂，复杂的是每个环节的数据质量与可观测性。先把解析和分块做扎实，再谈容器与索引选型。

笔记加载中…