你公司那几百页产品手册、员工制度、历史答疑，全躺在硬盘里吃灰。新人问一句「差旅报销上限是多少」，没人答得上来，或者要翻半小时群聊记录。

大模型很聪明，但它没读过你家文档。RAG（Retrieval-Augmented Generation，检索增强生成）干的事很简单：把「你的资料」在提问的瞬间喂给大模型，让它基于事实开口。

相比训练或微调一个专属模型，RAG 是中小团队落地 AI 最快的那条路——不需要 GPU 集群，不需要标注上万条样本，一份 Python 脚本加一个向量库就能跑起来。这也是为什么 2026 年了，RAG 仍是落地最快、最容易被读者「真用起来」的 AI 应用。

读者现在不缺概念，缺的是能跑的代码。下面直接给骨架。

三件事讲清架构

RAG 本质就是三件小事，串起来就行：

检索（Retrieval）：把用户问题变成向量，去向量库里捞最相关的文档片段。

增强（Augmentation）：把捞回来的片段拼进提示词，作为「参考资料」塞给模型。

生成（Generation）：大模型只读这些资料来作答，并标注出处。

一句话总结：先找资料，再带着资料去问模型，最后只输出有依据的回答。

我们的脚本用 LangChain 串流程、Chroma 存向量、Streamlit 做界面，三者都是开源、轻量、本地能跑。

依赖与运行 

把依赖写进 requirements.txt，Python 用 3.10 及以上：

langchain>=0.2
langchain-community>=0.2
langchain-openai>=0.1
langchain-chroma>=0.1
chromadb>=0.5
streamlit>=1.36
openai>=1.4
pypdf>=4.0

装好后三步走：

设置密钥：export OPENAI_API_KEY=sk-xxx

在 ./docs 放入你的 PDF / TXT / MD 文档

streamlit run rag_kb.py，首次会自动建库并持久化到 ./chroma_db

想用命令行直接问，跑 python rag_kb.py --ask "退款政策是怎样的？"；加 --debug 看检索到的片段，加 --rebuild 强制重建索引。

160 行可运行骨架 

下面这约 160 行就是全部骨架，存成 rag_kb.py 即可运行，已经把「建库 → 检索 → 问答 → 前端」串成一条线。

# rag_kb.py — 约 160 行搭一个能问答自家文档的 RAG 机器人
# 运行：pip install -r requirements.txt && streamlit run rag_kb.py
# 环境：Python 3.10+ / langchain>=0.2 / chromadb>=0.5 / streamlit>=1.36 / openai>=1.4
import os
import shutil
import logging
from pathlib import Path

import streamlit as st
from langchain_community.document_loaders import PyPDFLoader, TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_chroma import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

logging.basicConfig(level=logging.INFO)

# ------------------------- 配置 -------------------------
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "")
ifnot OPENAI_API_KEY:
    raise RuntimeError("请先 export OPENAI_API_KEY=xxx")

DOCS_DIR = os.getenv("DOCS_DIR", "./docs")
CHROMA_DIR = os.getenv("CHROMA_DIR", "./chroma_db")
EMBED_MODEL = "text-embedding-3-small"
CHAT_MODEL = "gpt-4o-mini"
TOP_K = int(os.getenv("TOP_K", "5"))
SCORE_THRESHOLD = float(os.getenv("SCORE_THRESHOLD", "0.3"))
CHUNK_SIZE = int(os.getenv("CHUNK_SIZE", "800"))
CHUNK_OVERLAP = int(os.getenv("CHUNK_OVERLAP", "120"))

SYSTEM_PROMPT = """你是企业文档助手，严格遵守：
1. 仅依据【参考资料】作答，禁止编造或引入外部知识。
2. 资料中无相关信息时，明确回答"根据现有资料无法回答"。
3. 每条结论后用 [来源:文件名#片段号] 标注出处。
4. 回答控制在 200 字以内，先结论后依据。"""

# ------------------------- 1. 建库 -------------------------
def load_docs(docs_dir: str):
    """按扩展名选加载器；空文本层 PDF 触发 OCR 预警。"""
    docs = []
    for p in Path(docs_dir).rglob("*"):
        suf = p.suffix.lower()
        if suf == ".pdf":
            docs += PyPDFLoader(str(p)).load()
        elif suf in (".txt", ".md"):
            docs += TextLoader(str(p), encoding="utf-8").load()
        elif suf == ".docx":
            try:
                from langchain_community.document_loaders import Docx2txtLoader
                docs += Docx2txtLoader(str(p)).load()
            except ImportError:
                logging.warning("未安装 docx2txt，已跳过 %s", p)
    empty = [d.metadata["source"] for d in docs ifnot d.page_content.strip()]
    if empty:
        logging.warning("以下文件提取到空文本层，请先做 OCR：%s", empty)
    return docs

def split_docs(docs):
    """递归切分：分隔符优先级 + chunk/overlap 调参。"""
    splitter = RecursiveCharacterTextSplitter(
        separators=["\n\n", "\n", "。", "；", " ", ""],  # 段落 > 句 > 字
        chunk_size=CHUNK_SIZE,
        chunk_overlap=CHUNK_OVERLAP,
        length_function=len,
    )
    chunks = splitter.split_documents(docs)
    for i, c in enumerate(chunks):
        c.metadata["chunk_id"] = i          # 编号便于引用来源
    return chunks

@st.cache_resource(show_spinner="正在构建向量库…")
def build_index():
    embeddings = OpenAIEmbeddings(model=EMBED_MODEL)
    if Path(CHROMA_DIR).exists():
        return Chroma(persist_directory=CHROMA_DIR, embedding_function=embeddings)
    chunks = split_docs(load_docs(DOCS_DIR))
    logging.info("切片数：%d，开始写入向量库", len(chunks))
    return Chroma.from_documents(chunks, embeddings, persist_directory=CHROMA_DIR)

# ------------------------- 2. 检索 -------------------------
def retrieve(query: str) -> str:
    """相似度阈值过滤 + 同源去重裁剪，降低噪声。"""
    vs = build_index()
    retriever = vs.as_retriever(
        search_type="similarity_score_threshold",
        search_kwargs={"k": TOP_K, "score_threshold": SCORE_THRESHOLD},
    )
    hits = retriever.invoke(query)
    seen, ctx = set(), []
    for i, d in enumerate(hits, 1):
        src = d.metadata.get("source", "未知")
        if src in seen:                     # 同源去重，省出上下文空间
            continue
        seen.add(src)
        name = Path(src).name
        ctx.append(f"[片段{i}|{name}#{d.metadata.get('chunk_id')}]\n{d.page_content}")
    return"\n\n".join(ctx[:TOP_K])

# ------------------------- 3. 问答 -------------------------
def answer(query: str) -> str:
    """检索增强 + 边界约束的系统提示词。"""
    context = retrieve(query)
    ifnot context:
        return"根据现有资料无法回答。"
    prompt = ChatPromptTemplate.from_messages([
        ("system", SYSTEM_PROMPT),
        ("human", "【参考资料】\n{context}\n\n【问题】{question}"),
    ])
    chain = prompt | ChatOpenAI(model=CHAT_MODEL, temperature=0) | StrOutputParser()
    return chain.invoke({"context": context, "question": query})

# ------------------------- 4. 前端 -------------------------
def main():
    st.set_page_config(page_title="文档问答机器人")
    st.title("自家文档问答机器人")
    st.caption("LangChain + Chroma + Streamlit 的 RAG 最小骨架")
    build_index()                            # 首次启动即建库（已缓存）
    q = st.text_input("问点什么？", placeholder="例如：退款政策是怎样的？")
    if q.strip():
        try:
            with st.spinner("检索中…"):
                st.markdown(answer(q.strip()))
        except Exception as e:               # 网络/密钥异常时不白屏
            st.error(f"出错了：{e}")

# ------------------------- 命令行模式 -------------------------
if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser(description="RAG 文档问答")
    parser.add_argument("--ask", help="直接提问，不走前端")
    parser.add_argument("--rebuild", action="store_true", help="强制重建向量库")
    parser.add_argument("--debug", action="store_true", help="打印检索到的片段")
    args = parser.parse_args()
    if args.rebuild and Path(CHROMA_DIR).exists():
        shutil.rmtree(CHROMA_DIR)
    if args.ask:
        if args.debug:
            print("=== 检索到的片段 ===")
            print(retrieve(args.ask))
        print(answer(args.ask))
    else:
        main()

跑起来后，把文档丢进 ./docs，在网页输入框问问题，模型就会带着检索到的片段作答，并标出每段话的出处。

切片三个坑 

代码能跑，不代表答得准。第一步「切片」就藏着三个最常见的坑。

坑一：空 PDF 文本层。 扫描件、图片型 PDF 没有文本层，PyPDFLoader 提取到的 page_content 是空字符串。这种文档永远检索不到，却没人发现。上面 load_docs 里我们检测空内容并打日志告警，正解是先用 OCR（如 PaddleOCR、腾讯云 OCR）把图片转成文本再入库。

坑二：递归切分分隔符顺序。RecursiveCharacterTextSplitter 按 separators 顺序尝试切分——能按段落切就不动句子，能按句子切就不动字。顺序写反（比如把 "" 放最前面）会直接按字符硬切，把一句话拆得七零八落。我们按 \n\n > \n > 。 >  > "" 排，保证语义尽量完整。

坑三：chunk_size / overlap 调参。 块太大，检索不精准、喂给模型的噪声多；块太小，一句话被切断、语义丢失。split_docs 里给了经验起点：中文文档 500–1000 字/块，overlap 取块的 10%–15%（我们 800/120）。调完必须用评测集验证命中率，别凭感觉。

检索增强两招 

检索回来的片段经常「看起来相关、其实没用」，直接拼进提示词只会让模型更糊涂。两招压噪声：

第一招：相关性阈值过滤。retriever 用 similarity_score_threshold，只保留相关性 ≥ 阈值的片段，把「搜出来但不相关」的挡在门外。SCORE_THRESHOLD 按你的嵌入模型实测：text-embedding-3-small 下 0.3 是个起点——调到模型开始「说不知道」说明太严，调到开始胡说说明太松。

第二招：上下文裁剪。TOP_K 取前 5 个，再做同源去重——同一篇文档命中多个相邻片段时只留一个，把宝贵的上下文窗口让给不同来源。否则 5 个槽位被同一篇文档占满，模型根本看不到别的资料，反而答偏。

提示词三道边界 

系统提示词给模型划了三条红线（见代码 SYSTEM_PROMPT），这是压制「幻觉」的关键：

只基于材料答：禁止引入训练知识或现场编造。大部分胡说八道都来自这一条没守住。

无信息说不知：检索为空或资料里没有相关内容时，明确回「根据现有资料无法回答」，而不是硬编一个看似合理的答案。

标注来源：每条结论带 [来源:文件名#片段号]，既方便人工核对，也倒逼模型真的用了检索内容而不是自说自话。

配合 temperature=0，进一步压住随机发挥。

用评测守住质量 

代码能跑、界面能开，不代表答得对。最后一步是用黄金测试集（golden set）在 CI 里把质量钉死。每个用例是一句问题 + 期望在回答中出现的关键词（或人工标注的标准答案）。

我们盯两个指标：

命中率（Hit Rate）：回答是否包含期望关键词，检验「检索 + 生成」整条链路通不通。

回答质量：相关性、来源准确性，可用 LLM-as-judge 或人工抽检。

最小可运行评测脚本如下，CI 里命中率低于 80% 直接失败，提醒你回去调切片或阈值：

# eval.py — 用黄金测试集在 CI 里验证命中率与回答质量
from rag_kb import answer   # 直接复用上面的 answer()

# 黄金测试集：问题 + 期望在回答中出现的关键词
GOLDEN = [
    {"q": "退款政策是怎样的？", "must_include": ["7天", "原路退回"]},
    {"q": "支持哪些支付方式？", "must_include": ["微信", "支付宝"]},
    {"q": "账号如何注销？",     "must_include": ["设置", "客服"]},
]

def hit_rate(golden):
    hits = 0
    for item in golden:
        reply = answer(item["q"])
        ok = all(k in reply for k in item["must_include"])
        hits += ok
        print(f"[{'PASS' if ok else 'FAIL'}] {item['q']}")
    return hits / len(golden)

if __name__ == "__main__":
    rate = hit_rate(GOLDEN)
    print(f"命中率: {rate:.0%}")
    # CI 门禁：低于 80% 直接失败，提醒回去调切片/阈值
    assert rate >= 0.8, "命中率低于 80%，请检查切片与检索参数"

把 eval.py 接进 GitHub Actions 或任意 CI，每次改了切片参数或提示词，自动跑一遍黄金集，质量就有了底线

结尾与获取 

RAG 不性感，但它今天就能让你的文档「开口说话」。上面这约 160 行骨架，已经覆盖了建库、切片、检索增强、提示词边界和评测门禁——复制粘贴就能跑。

更多相关技术内容咨询欢迎前往并持续关注好学星城论坛了解详情。

想高效系统的学习Python编程语言，推荐大家关注一个微信公众号：Python编程学习圈。每天分享行业资讯、技术干货供大家阅读，关注即可免费领取整套Python入门到进阶的学习资料以及教程，感兴趣的小伙伴赶紧行动起来吧。