跳到主要内容

RAG 落地实践

目标:在本地小文档库上跑通 ingest → 检索 → 生成(生成可用模板拼接,不强制真模型)。本篇 deliberately 不用框架,先建立流水线直觉;跑通后再对照 [LangChain RAG 实践](/docs/ai/RAG/LangChain RAG 实践)。概念见 RAG 与检索增强。

目标与验收​

验收项标准
ingest2~3 个 .txt 切成 chunks 并建索引
检索提问能返回相关 chunk
生成Prompt 中可见引用片段;库中无依据时拒答
调参能说明改 chunk/top-k 后的现象

建议目录​

mini_rag/
data/ # 放几篇短文
ingest.py
retrieve.py
answer.py
eval_set.jsonl # 问题 + 期望含某关键字

切分与索引(无向量库版)​

PoC 可用「词袋重叠打分」代替向量库,逻辑同构:

# ingest.py
from pathlib import Path

def chunk_text(text: str, size: int = 80, overlap: int = 20):
chunks = []
i = 0
while i < len(text):
chunks.append(text[i : i + size])
i += size - overlap
return chunks

def build_index(data_dir: str):
index = []
for path in Path(data_dir).glob("*.txt"):
text = path.read_text(encoding="utf-8")
for c in chunk_text(text):
index.append({"source": path.name, "text": c})
return index

检索​

# retrieve.py
def score(query: str, chunk: str) -> int:
q = set(query.lower().split())
t = set(chunk.lower().split())
return len(q & t)

def retrieve(index, query: str, k: int = 3):
ranked = sorted(index, key=lambda c: score(query, c["text"]), reverse=True)
return [c for c in ranked[:k] if score(query, c["text"]) > 0]

生成(模板 + 拒答)​

# answer.py
def answer(query: str, hits: list[dict]) -> str:
if not hits:
return "知识库中未找到依据,无法回答。"
context = "\n---\n".join(f"[{h['source']}] {h['text']}" for h in hits)
return f"依据如下片段作答:\n{context}\n\n(此处可替换为 LLM 调用)\n问题:{query}"

小评测集​

eval_set.jsonl 示例:

{"q": "退货几天内可以?", "must_include": "7"}
{"q": "公司地址在哪?", "must_include": "找不到"}

跑批:对每条调用 retrieve + answer,检查输出是否包含 must_include。先保证可复现,再换 embedding / 向量库。

调参直觉(动手)​

改动观察
size 变大单段更全,但噪声与定位变差
k 变大更易召回,也更易掺无关段
数据加专有名词文词袋法对同义词弱 → 引出「为何要 embedding」

实践要点​

  1. 先 Naive 基线(可无真向量库),再换 embedding。
  2. 无命中必须拒答,避免「空检索还生成」。
  3. 用 5~20 条 jsonl 做回归,比凭感觉调参靠谱。
  4. 与 [LangChain RAG 实践](/docs/ai/RAG/LangChain RAG 实践) 做组件 1:1 对照后,再换真实 embedding / 向量库。
  5. 面试挂钩:能讲清流水线与失效模式即可;框架是同一流水线的另一种写法。