当前位置：首页 > news >正文

从零搭建你的第一个 RAG 应用：百行 Python 就够了

news 2026/7/17 6:52:01

RAG 就六个字：先查资料，再回答。系统分三步走——建库、检索、生成。

今天，我们把这三步变成能跑的代码。读完并跟着敲一遍，你会得到一个基于你自己文档的问答机器人：它只回答资料里有的内容，会标注出处，遇到不知道的问题会老实承认，而不是胡编。

全部代码百行左右，不依赖任何 RAG 框架。

一、开工前：技术选型

市面上大多数教程一上来就是 LangChain、LlamaIndex。框架当然好用，但对初学者有两个问题：一是封装太深，跑通了也不知道里面干了啥；二是框架 API 更新极快，教程隔三个月就过期。

所以这篇我们纯手写，让你看清每一个环节。等你理解了原理，再回头上框架，会快得多。

三个组件的选型如下：

向量模型（Embedding）：Qwen/Qwen3-Embedding-0.6B。通义千问开源的轻量向量模型，中文效果好，通过硅基流动的 API 调用它，登录有免费试用额度。

向量数据库：Chroma。pip 装完就能用，不用部署服务，数据直接存在本地文件夹里，对新手比较友好。

大模型：DeepSeek-V4-Pro，同样通过硅基流动调用。接口与 OpenAI 完全兼容——以后想换成 Qwen、智谱、Kimi 或本地 Ollama，只改两行配置就行（最后会给出改法）。

为什么选硅基流动（SiliconFlow）？两个理由：一是向量模型和大模型共用同一个平台、同一个 API Key、同一个 base_url，环境配置直接减半；二是它的模型广场有近百个模型可选，从免费小模型到 DeepSeek 旗舰版，改一行模型名就能切换。

对应上一篇的三步走：

建库= 读取文档 + 切分 + 向量化存入 Chroma
检索= 把问题向量化，去 Chroma 里找最相近的段落
生成= 把"问题 + 段落"打包发给大模型

二、环境准备（3 分钟）

第 1 步：安装依赖（需要 Python 3.9 以上）

pip install chromadb openai

国内网络下载慢的话，加个清华镜像源：

pip install -i https://pypi.tuna.tsinghua.edu.cn/simple chromadb openai

第 2 步：申请 API Key

到硅基流动官网（cloud.siliconflow.cn）注册，在控制台的「API 密钥」页面创建一个 Key，然后把它设置成环境变量——注意，永远不要把 Key 直接写进代码里，一旦代码外传或提交到git，Key 就泄露了。

macOS / Linux：

export SILICONFLOW_API_KEY="sk-你的key"

Windows PowerShell：

$env:SILICONFLOW_API_KEY="sk-你的key"

三、准备一份"私有资料"

新建一个项目文件夹，在里面再建一个docs文件夹，用来放你的知识库文档（支持.txt和.md）。

还记得上一篇开头的吐槽吗——问 AI"我们公司的报销流程"，它给你现编一套。今天我们就用一份报销制度文档，治好它这个毛病。

在docs里新建差旅报销制度.md，贴入下面的示例内容（你也可以换成自己的任何文档）：

# 差旅报销制度（示例） ## 出差申请员工出差需提前 3 个工作日在 OA 系统提交申请，写明目的地、事由和预计天数，经直属主管审批通过后方可预订行程。 ## 交通标准市内交通实报实销；城际出行默认高铁二等座或飞机经济舱，总监及以上级别可乘坐高铁一等座。 ## 住宿标准一线城市（北京、上海、广州、深圳）每晚不超过 500 元，其他城市每晚不超过 350 元。 ## 餐费补贴出差期间按每天 100 元发放餐补，无需发票；客户招待餐费另行走招待费流程，需事前审批。 ## 报销流程与时限出差结束后 30 天内，在 OA 系统提交报销单并粘贴发票原件，经主管与财务审核后，款项在 10 个工作日内打入工资卡。逾期提交需部门负责人特批。 ## 发票要求所有报销票据须为增值税发票，抬头为公司全称，个人抬头或抬头错误的发票不予报销。

四、代码分步讲解

在项目文件夹里新建rag_demo.py，我们一段一段往里写。

第 1 段：配置

import os import chromadb from openai import OpenAI API_KEY = os.getenv("SILICONFLOW_API_KEY") if not API_KEY: raise SystemExit("未检测到环境变量 SILICONFLOW_API_KEY，请先配置。") # 硅基流动的接口与 OpenAI 兼容：向量模型和大模型共用一个客户端、一个 Key client = OpenAI(api_key=API_KEY, base_url="https://api.siliconflow.cn/v1") LLM_MODEL = "deepseek-ai/DeepSeek-V4-Pro" # 大模型（按量付费，效果稳） EMBED_MODEL = "Qwen/Qwen3-Embedding-0.6B" # 向量模型 # 向量数据库：数据会持久化到本地 chroma_db 文件夹 chroma = chromadb.PersistentClient(path="./chroma_db") collection = chroma.get_or_create_collection( name="my_knowledge_base", metadata={"hnsw:space": "cosine"}, # 用余弦相似度衡量语义远近 )

这里就能体现了"OpenAI 兼容接口"的好处：调用硅基流动用的就是 openai 这个库，只是把base_url指向了它的服务器；而且向量模型和大模型共用同一个client。以后想换模型，改LLM_MODEL、EMBED_MODEL两个字符串即可。

另外，创建集合时多传了一个metadata，告诉 Chroma 用余弦相似度来比较向量——这是文本语义检索最常用的度量方式，照抄即可。

第 2 段：读取文档

def load_documents(folder: str = "docs") -> list[dict]: if not os.path.isdir(folder): raise SystemExit(f"未找到 {folder}/ 文件夹，请先创建。") docs = [] for name in os.listdir(folder): if name.endswith((".txt", ".md")): with open(os.path.join(folder, name), encoding="utf-8") as f: docs.append({"name": name, "text": f.read()}) if not docs: raise SystemExit(f"{folder}/ 文件夹里还没有任何 .txt 或 .md 文件。") print(f"读取到 {len(docs)} 份文档") return docs

没什么玄机：遍历docs文件夹，把每份文档的文件名和全文读进来。文件名后面会作为"出处"展示。

第 3 段：切分文本（Chunking）

def split_text(text: str, chunk_size: int = 300, overlap: int = 50) -> list[str]: chunks, start = [], 0 while start < len(text): chunk = text[start : start + chunk_size] if chunk.strip(): chunks.append(chunk) start += chunk_size - overlap return chunks

为什么要切？两个原因：一是向量模型能处理的文本长度有限；二是检索的粒度越合适，找到的内容越精准——拿整本手册去匹配一个具体问题，反而找不准。

两个参数值得注意：

chunk_size=300：每块 300 个字符；
overlap=50：相邻两块重叠 50 个字符，避免一句话正好被"拦腰斩断"后语义丢失。

这是最简单粗暴的"固定长度切分"。它够用，但远谈不上好——比如可能把一条完整的制度切成两半。怎么切才科学，是 RAG 效果好坏的关键之一，下一篇专门讲。

第 4 段：向量化，建立索引

def embed(texts: list[str]) -> list[list[float]]: """调用硅基流动的向量接口，把一批文本变成向量（按 32 条分批，兼容各模型的单次上限）""" vectors = [] for i in range(0, len(texts), 32): resp = client.embeddings.create(model=EMBED_MODEL, input=texts[i : i + 32]) vectors.extend(item.embedding for item in resp.data) return vectors def build_index(docs: list[dict]) -> None: all_chunks, ids, metas = [], [], [] for doc in docs: for i, chunk in enumerate(split_text(doc["text"])): all_chunks.append(chunk) ids.append(f"{doc['name']}-{i}") # 每个块一个唯一 ID metas.append({"source": doc["name"]}) # 记录出处 collection.add( ids=ids, documents=all_chunks, embeddings=embed(all_chunks), metadatas=metas, ) print(f"索引完成，共写入 {collection.count()} 个文本块")

这一段做的就是上一篇说的"给每段文字生成语义坐标"：embed函数调用硅基流动的 embeddings 接口，把每个文本块变成一串数字（一个 1024 维的向量），然后连同原文、出处一起存进 Chroma。

一个小细节：embeddings 接口对单次能接收的文本条数有上限（不同模型的上限不一样），所以embed里按 32 条一批做了分批，换什么模型都稳妥。示例文档只有 2 个文本块、一批就够，但等你往 docs 里塞几十份文档时，这个循环就派上用场了。

第 5 段：检索

def retrieve(query: str, top_k: int = 3): res = collection.query(query_embeddings=embed([query]), n_results=top_k) return res["documents"][0], res["metadatas"][0]

用户的问题也走同一个向量模型，变成同一空间里的"坐标"，然后让 Chroma 找出坐标最接近的 3 个文本块（top_k=3，可自行调整）。坐标近，就是语义近。

第 6 段：生成——Prompt 是防幻觉的关键

PROMPT_TEMPLATE = """你是一个严谨的知识库问答助手。请只根据下面提供的资料回答问题： 1. 如果资料里有答案，用简洁的中文回答，并在结尾注明出处文件名； 2. 如果资料里没有相关信息，直接回答"根据现有资料，我无法回答这个问题"，禁止编造。【资料】 {context} 【问题】 {question}""" def answer(question: str) -> str: chunks, metas = retrieve(question) context = "\n\n".join( f"（出处：{m['source']}）\n{c}" for c, m in zip(chunks, metas) ) prompt = PROMPT_TEMPLATE.format(context=context, question=question) resp = client.chat.completions.create( model=LLM_MODEL, messages=[{"role": "user", "content": prompt}], temperature=0.3, # 温度调低，回答更稳定、少发挥 ) return resp.choices[0].message.content

这段 Prompt 有三个设计点，正是 RAG 对抗幻觉的核心手法：

限定信息来源："只根据下面提供的资料回答"，把模型从"凭记忆答题"锁死在"看资料答题"；
要求注明出处：答案可追溯、可核对；
允许说"不知道"：明确给模型一个"承认无知"的出口，它才不会硬编。

第 7 段：主程序

if __name__ == "__main__": if collection.count() == 0: print("首次运行，正在建立索引……") build_index(load_documents("docs")) while True: q = input("\n请提问（输入 q 退出）：").strip() if q.lower() in {"q", "quit", "exit"}: print("再见！") break if q: print("\n" + answer(q))

首次运行建索引，之后直接进入命令行问答循环。索引是持久化的，第二次运行不用重建；如果你更新了文档，删掉chroma_db文件夹再跑一次即可。

五、跑起来看效果

$ python rag_demo.py 首次运行，正在建立索引…… 读取到 1 份文档索引完成，共写入 2 个文本块请提问（输入 q 退出）：出差住宿一晚最多能报销多少？一线城市（北京、上海、广州、深圳）每晚不超过 500 元，其他城市每晚不超过 350 元。（出处：差旅报销制度.md）请提问（输入 q 退出）：公司年会一般在哪里举办？根据现有资料，我无法回答这个问题。

（输出仅为示意，文本块数量取决于文档长度，回答措辞会因模型而略有差异。）

注意第二个问题——资料里没有年会的信息，它没有编，而是老实承认。这就是 RAG 与"裸问大模型"最直观的区别。

现在，把你自己的笔记、手册、周报丢进docs文件夹，删掉chroma_db重建索引，你就拥有了一个真正的私人知识库助手。

六、想换别的模型？改一行就行

想更省钱？改一行：硅基流动的模型广场上有近百个模型，DeepSeek、Qwen、GLM、Kimi 全都在。本文用的旗舰版效果最稳；想降低成本，换成更快更便宜的 Flash 版即可：

LLM_MODEL = "deepseek-ai/DeepSeek-V4-Flash"

模型广场里还有一批免费档小模型，预算为零也能跑通全流程——不过免费档有限流、输出质量偶有波动，正式使用建议按量付费档。

向量模型同理，改EMBED_MODEL就行（注意：不同向量模型生成的向量互不通用，换完记得删掉chroma_db重建索引）。

换成其他厂商的 API（DeepSeek 官方、通义千问、智谱等大多兼容 OpenAI 接口），替换base_url、Key 和模型名即可，例如 DeepSeek 官方：

client = OpenAI( api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com", ) LLM_MODEL = "deepseek-chat"

不过注意：多数大模型厂商不提供向量接口，这样换过去之后，向量部分要么继续走硅基流动，要么改用下面的本地方案。

完全本地、零 API 依赖：大模型用 Ollama（安装后ollama pull qwen2.5:7b），向量模型用sentence-transformers在本地跑 BGE：

client = OpenAI(api_key="ollama", base_url="http://localhost:11434/v1") LLM_MODEL = "qwen2.5:7b" # 需先 pip install sentence-transformers，然后把 embed 函数整体替换为： from sentence_transformers import SentenceTransformer st_model = SentenceTransformer("BAAI/bge-small-zh-v1.5") def embed(texts: list[str]) -> list[list[float]]: return st_model.encode(texts, normalize_embeddings=True).tolist()

这样一改，整套 RAG 就 100% 离线了——涉密资料也能放心用。