1. 为什么我建议你先手写一遍 RAG,而不是直接上框架
RAG 这个词听起来唬人,拆开看就六个字:先查资料,再回答。你问它一个私有文档里的问题,它不会凭记忆瞎编,而是先去你的资料库里翻出最相关的几段,再把「问题 + 这几段」一起交给大模型,让它照着资料回答。整个过程分三步走——建库、检索、生成。适合谁?适合手上有几份 PDF、Markdown 笔记、公司制度文档,想让 AI 只基于这些内容作答的零基础读者。
市面上大多数教程一上来就是 LangChain、LlamaIndex。框架当然好用,但对初学者有两个问题:一是封装太深,跑通了也不知道里面干了啥;二是框架 API 更新极快,教程隔三个月就过期。所以这篇我们纯手写,用大约一百行 Python,让你看清每一个环节。等你理解了原理,再回头上框架,会快得多。
三个组件的选型我这样定:向量模型用 Qwen/Qwen3-Embedding-0.6B,中文效果好、体积轻;向量数据库用 Chroma,pip 装完就能用,不用部署服务,数据直接存在本地文件夹里;大模型用 DeepSeek 系列,接口与 OpenAI 完全兼容。关键点在于,这三者我都通过 TaoToken 的统一 Key 和 API 通道来调用——向量模型和大模型共用一个 base_url、一个 Key,环境配置直接减半,以后想换模型只改一行字符串。
对应到三步走:建库 = 读取文档 + 切分 + 向量化存入 Chroma;检索 = 把问题向量化,去 Chroma 里找最相近的段落;生成 = 把「问题 + 段落」打包发给大模型。下面我把每一步都拆成能直接复制的代码,跑完你会得到一个基于自己文档的问答机器人,它会标注出处,遇到不知道的问题会老实承认。
2. TaoToken 前置准备:一个 Key 打通 Embedding 和对话
在动手写代码前,先把「通道」铺好。传统做法是向量模型找一个平台、大模型找另一个平台,两套 Key、两个 base_url,环境变量配到怀疑人生。TaoToken 的思路是把这些统一到一个入口:你只需要一个 API Key,就能同时调用 Embedding 接口和 Chat 接口,base_url 也只有一个。
第一步,打开 TaoToken 官网 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 注册账号。注册完成后进入控制台,在「API Keys」页面创建一个新的 Key。这里有个习惯要养成:永远不要把 Key 直接写进代码里,一旦代码外传或提交到 git,Key 就泄露了。正确做法是设置成环境变量。
macOS / Linux 下这样设置:
export TAOTOKEN_API_KEY="sk-你的key"Windows PowerShell 下这样设置:
$env:TAOTOKEN_API_KEY="sk-你的key"设置完记得在同一个终端窗口里运行脚本,或者重开终端重新设置,否则环境变量不生效,后面会报 401。
第二步,确认你要用的模型 ID。TaoToken 的模型列表里,Embedding 我选Qwen/Qwen3-Embedding-0.6B,对话模型我选 DeepSeek 系列。你可以在控制台的模型广场里看到当前可用的模型名,复制准确的 Model ID 备用。这一步别偷懒,模型名写错是最常见的报错来源之一。
第三步,记住两个地址。API 根地址是https://taotoken.net/api,注意这个地址后面不加任何 UTM 参数,直接用于代码里的 base_url。而官网链接带 UTM 是为了统计来源,两者用途不同,别混用。如果你后面想用 Claude Code 这类编码工具接入,可以在文档里找到对应的接入说明;想长期跑编码或 Agent 任务,可以了解 Coding Plan;想直接在网页里验证模型效果,用模型对话就行。
到这里前置就齐了:一个 Key、一个 base_url、两个模型 ID。接下来所有代码都围绕这几个变量展开,换模型时你只需要改字符串,不用动逻辑。
3. 可复制配置:目录结构、依赖清单与完整脚本
先把项目骨架搭起来。新建一个文件夹,比如rag_demo,在里面建一个docs子文件夹放你的知识库文档,支持.txt和.md。目录结构长这样:
rag_demo/ ├── docs/ │ └── 差旅报销制度.md ├── chroma_db/ # 运行后自动生成,存向量 └── rag_demo.py依赖只有两个,Python 3.9 以上即可:
pip install chromadb openai国内网络下载慢的话加个镜像源:
pip install -i https://pypi.tuna.tsinghua.edu.cn/simple chromadb openai准备一份私有资料。在docs里新建差旅报销制度.md,贴入示例内容(你也可以换成自己的任何文档):
# 差旅报销制度(示例) ## 出差申请 员工出差需提前 3 个工作日在 OA 系统提交申请,写明目的地、 事由和预计天数,经直属主管审批通过后方可预订行程。 ## 交通标准 市内交通实报实销;城际出行默认高铁二等座或飞机经济舱, 总监及以上级别可乘坐高铁一等座。 ## 住宿标准 一线城市(北京、上海、广州、深圳)每晚不超过 500 元, 其他城市每晚不超过 350 元。 ## 餐费补贴 出差期间按每天 100 元发放餐补,无需发票;客户招待餐费 另行走招待费流程,需事前审批。 ## 报销流程与时限 出差结束后 30 天内,在 OA 系统提交报销单并粘贴发票原件, 经主管与财务审核后,款项在 10 个工作日内打入工资卡。 逾期提交需部门负责人特批。 ## 发票要求 所有报销票据须为增值税发票,抬头为公司全称, 个人抬头或抬头错误的发票不予报销。现在写主脚本rag_demo.py。第一段是配置,注意 base_url 指向 TaoToken:
import os import chromadb from openai import OpenAI API_KEY = os.getenv("TAOTOKEN_API_KEY") if not API_KEY: raise SystemExit("未检测到环境变量 TAOTOKEN_API_KEY,请先配置。") # TaoToken 的接口与 OpenAI 兼容:向量模型和大模型共用一个客户端、一个 Key client = OpenAI(api_key=API_KEY, base_url="https://taotoken.net/api") LLM_MODEL = "deepseek-ai/DeepSeek-V3" # 对话模型,按控制台实际 Model ID 填写 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 兼容接口」的好处:调用 TaoToken 用的就是openai这个库,只是把 base_url 指向了它的服务器,而且向量模型和大模型共用同一个 client。以后想换模型,改LLM_MODEL、EMBED_MODEL两个字符串即可。创建集合时多传了一个 metadata,告诉 Chroma 用余弦相似度来比较向量,这是文本语义检索最常用的度量方式,照抄即可。
第二段读取文档:
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 文件夹,把每份文档的文件名和全文读进来,文件名后面会作为「出处」展示。
第三段切分文本:
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 效果好坏的关键之一。
第四段向量化并建索引:
def embed(texts: list[str]) -> list[list[float]]: """调用 TaoToken 的向量接口,把一批文本变成向量(按 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}") metas.append({"source": doc["name"]}) collection.add( ids=ids, documents=all_chunks, embeddings=embed(all_chunks), metadatas=metas, ) print(f"索引完成,共写入 {collection.count()} 个文本块")embed函数调用 TaoToken 的 embeddings 接口,把每个文本块变成一串数字(一个向量),然后连同原文、出处一起存进 Chroma。注意 embeddings 接口对单次能接收的文本条数有上限,所以按 32 条一批做了分批,换什么模型都稳妥。
第五段检索:
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 个文本块。坐标近,就是语义近。
第六段生成,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 有三个设计点:限定信息来源,把模型从「凭记忆答题」锁死在「看资料答题」;要求注明出处,答案可追溯;允许说「不知道」,明确给模型一个承认无知的出口,它才不会硬编。
第七段主程序:
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文件夹再跑一次即可。
4. 验证请求:跑通一次问答并打印命中的 chunk 与最终 Prompt
代码写完了,先别急着问问题,我们加一段调试输出,把命中的 chunk 和最终拼好的 Prompt 都打印出来。这是理解 RAG 最直观的方式——你能亲眼看到模型到底「看」到了什么。把answer函数改成下面这样:
def answer(question: str, debug: bool = True) -> 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) if debug: print("=" * 50) print(f"命中 {len(chunks)} 个文本块:") for i, (c, m) in enumerate(zip(chunks, metas), 1): print(f"\n[chunk {i}] 出处={m['source']}") print(c[:120] + ("..." if len(c) > 120 else "")) print("\n最终 Prompt:") print(prompt) print("=" * 50) resp = client.chat.completions.create( model=LLM_MODEL, messages=[{"role": "user", "content": prompt}], temperature=0.3, ) return resp.choices[0].message.content现在运行脚本:
python rag_demo.py首次运行会看到建索引的过程,然后进入问答循环。输入第一个问题「出差住宿一晚最多能报销多少?」,你会看到类似下面的输出:
首次运行,正在建立索引…… 读取到 1 份文档 索引完成,共写入 2 个文本块 请提问(输入 q 退出):出差住宿一晚最多能报销多少? ================================================== 命中 2 个文本块: [chunk 1] 出处=差旅报销制度.md ## 住宿标准 一线城市(北京、上海、广州、深圳)每晚不超过 500 元, 其他城市每晚不超过 350 元。 [chunk 2] 出处=差旅报销制度.md ## 餐费补贴 出差期间按每天 100 元发放餐补,无需发票;客户招待餐费 另行走招待费流程,需事前审批。 最终 Prompt: 你是一个严谨的知识库问答助手。请只根据下面提供的资料回答问题: ... ================================================== 一线城市(北京、上海、广州、深圳)每晚不超过 500 元, 其他城市每晚不超过 350 元。 (出处:差旅报销制度.md)重点看两处:一是命中的 chunk,你能确认检索确实找到了「住宿标准」这一段;二是最终 Prompt,你能看到模型收到的完整上下文,包括资料和问题。这就是 RAG 的「黑盒」被打开的样子。
再问一个资料里没有的问题,比如「公司年会一般在哪里举办?」,输出会是:
根据现有资料,我无法回答这个问题。它没有编,而是老实承认。这就是 RAG 与「裸问大模型」最直观的区别。到这里,验证动作就完成了:跑通一次问答、打印命中的 chunk、打印最终 Prompt,三件事都做到了。
5. 本篇常见错排查:401、local proxy failed、reading choices 与 OAuth
跑不通是常态,我把这一路踩过的坑按报错原文列出来,你对照着查。
报 401 / API Key 错误。最常见的原因是环境变量没生效。设置完环境变量后要在同一个终端窗口里运行脚本,或者重开终端重新设置。另一个原因是 Key 复制时带了空格或换行,重新复制一遍。还有一种情况是 base_url 写错了,注意 TaoToken 的 API 根地址是https://taotoken.net/api,不要多加/v1也不要带 UTM 参数。
报 local proxy failed 或连接超时。这类报错通常是本地网络环境或代理设置导致的。检查你的终端有没有设置HTTP_PROXY、HTTPS_PROXY环境变量,如果有,先清掉再试。如果你在公司内网,确认防火墙没有拦截对taotoken.net的访问。这个报错和 Key 无关,别急着去重新生成 Key。
报 reading choices 或'NoneType' object has no attribute 'choices'。这说明接口返回的结构和你预期的不一样。先打印resp看看原始返回,常见原因是模型 ID 写错了,接口返回了错误信息而不是正常的 choices 结构。回到控制台核对LLM_MODEL的准确 Model ID,注意大小写和斜杠。
报 OAuth 相关错误。如果你用的是 Claude Code 这类工具接入,可能会遇到 OAuth 认证流程的问题。这类工具通常需要配置三件套:Base URL、API Key、Model ID。Base URL 填https://taotoken.net/api,Key 填你创建的 Key,Model ID 填控制台里的准确名称。三件套缺一不可,只填 Key 不填 Base URL 是最常见的错误。如果你用的是 Cline 或 CC Switch 这类工具,同样检查这三项配置是否完整。
报维度错误或换了向量模型后检索异常。不同向量模型生成的向量互不通用,换完EMBED_MODEL后必须删掉chroma_db文件夹重建索引,否则新旧向量维度对不上,查询会直接报错。
中文乱码。确保你的文档以 UTF-8 编码保存,Windows 记事本另存为时可以选择编码。
偶发 429(Too Many Requests)。这是请求频率限制,等几秒重试即可;频繁触发的话,换一个限流更宽松的模型。
排查的顺序建议是:先看报错原文,再查环境变量和 base_url,然后核对模型 ID,最后才怀疑代码逻辑。大部分问题都出在前三步。
6. 想换模型?改一行就行,以及接下来该学什么
跑通之后,你会发现换模型这件事比想象中简单。想更省钱,改一行:
LLM_MODEL = "deepseek-ai/DeepSeek-V3"换成控制台里更轻量的模型即可。向量模型同理,改EMBED_MODEL就行,但记住换完要删掉chroma_db重建索引。如果你想换成其他厂商的 API,只要对方兼容 OpenAI 接口,替换 base_url、Key 和模型名即可。不过要注意,多数大模型厂商不提供向量接口,这样换过去之后,向量部分要么继续走 TaoToken,要么改用本地方案。
完全本地、零 API 依赖的方案也有:大模型用 Ollama,向量模型用 sentence-transformers 在本地跑 BGE。这样整套 RAG 就 100% 离线了,涉密资料也能放心用。
最后说句实在的:跑通只是起点,调优才是 RAG 的主战场。固定长度切分太粗暴,可能把完整语义切碎;向量模型选了最轻量的,中文场景怎么选型、要不要上更大的模型;纯向量检索会漏掉一些「关键词明明对上了」的内容,需要混合检索与重排序。这些都不是一百行代码能解决的,但你现在有了一个能跑、能改、能观察的最小系统,往上加任何东西都不会迷路。
如果你想把模型调用通道统一管理,可以去 TaoToken 的 API Keys 页面创建一个 Key,再对照接入文档把 base_url 和模型 ID 填进代码;想先验证模型效果,用模型对话试几句;打算长期跑编码或 Agent 任务,了解一下 Coding Plan 会更省心。