RAG 落地实践
目标:在本地小文档库上跑通 ingest → 检索 → 生成(生成可用模板拼接,不强制真模型)。本篇 deliberately 不用框架,先建立流水线直觉;跑通后再对照 [LangChain RAG 实践](/docs/ai/RAG/LangChain RAG 实践)。概念见 RAG 与检索增强。
目标与验收
| 验收项 | 标准 |
|---|---|
| ingest | 2~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」 |
实践要点
- 先 Naive 基线(可无真向量库),再换 embedding。
- 无命中必须拒答,避免「空检索还生成」。
- 用 5~20 条 jsonl 做回归,比凭感觉调参靠谱。
- 与 [LangChain RAG 实践](/docs/ai/RAG/LangChain RAG 实践) 做组件 1:1 对照后,再换真实 embedding / 向量库。
- 面试挂钩:能讲清流水线与失效模式即可;框架是同一流水线的另一种写法。