你刷到过多少篇"十分钟上手LangChain RAG"的教程?GitHub上跑不完的demo,文档厚得像砖头,跑起来还是一堆依赖报错。
我们被工具绑架了。
RAG的本质就三步:切片、向量化、相似度检索。这就是全部真相。剩下的都是包装。
我花了个周末,用纯Python写了一个能跑的RAG,代码量不到500行。不依赖LangChain,不依赖LlamaIndex。
下面是结果。
RAG到底在干什么
你问LLM一个问题,它没见过你的私有文档,所以答不上来。
RAG的思路很简单:先去你的文档库里找相关段落,把找到的段落塞给LLM,让它基于这些内容回答。
就这么简单。
文档长了不能全塞,切成小段。每段变成向量——机器"理解"文本的方式。来了新问题,把问题也变成向量,用余弦相似度算一下,找出最相近的段落。拼成prompt发给LLM。
这三步就是RAG的全部。中间的抽象层、工具链、"最佳实践",本质都是包装。
第一步:文档切片
切太碎上下文丢失,切太长向量计算慢。递归切片器是最务实的方案:先按段落切,太长再按句子切,句子太长按字符切。
class RecursiveTextSplitter: def __init__(self, chunk_size=500, chunk_overlap=50, separators=None): self.chunk_size = chunk_size self.chunk_overlap = chunk_overlap self.separators = separators or ["\n\n", "\n", "。", "!", "?", ".", "!", "?", " ", ""] def split_text(self, text): if len(text) <= self.chunk_size: return [text] for sep in self.separators: if sep not in text: continue parts = text.split(sep) if len(parts) > 1: chunks, current = [], "" for part in parts: if len(current) + len(part) + len(sep) <= self.chunk_size: current += part + sep else: if current: chunks.append(current.rstrip()) current = part + sep if len(current) > self.chunk_size: chunks.extend(self.split_text(current)) current = "" if current: chunks.append(current.rstrip()) return chunks return [text[i:i+self.chunk_size] for i in range(0, len(text), self.chunk_size - self.chunk_overlap)]
段落优先,句子次之,字符兜底。重叠50字符防止上下文断裂。实测效果不输LangChain的RecursiveCharacterTextSplitter。
第二步:Embedding生成
直接调API就完了。
class EmbeddingClient: def __init__(self, api_key, base_url="https://api.openai.com/v1"): self.api_key = api_key self.base_url = base_url def embed_texts(self, texts): resp = requests.post(f"{self.base_url}/embeddings", json={"model": "text-embedding-3-small", "input": texts}, headers={"Authorization": f"Bearer {self.api_key}"}) resp.raise_for_status() return [item["embedding"] for item in resp.json()["data"]]
想换模型?改一行参数。想本地跑?换成ollama的API调用。
第三步:向量存储+检索
文档库在几千段以内,存json文件,用numpy做余弦相似度就够了。
class SimpleVectorStore: def __init__(self): self.embeddings, self.texts, self.metadata = [], [], [] def add_texts(self, texts, embeddings, metadata=None): for i, text in enumerate(texts): self.embeddings.append(np.array(embeddings[i])) self.texts.append(text) self.metadata.append(metadata[i] if metadata else {}) def search(self, query_embedding, top_k=3): query_vec = np.array(query_embedding) results = [] for i, doc_vec in enumerate(self.embeddings): sim = np.dot(query_vec, doc_vec) / (np.linalg.norm(query_vec) * np.linalg.norm(doc_vec)) results.append((self.texts[i], float(sim), self.metadata[i])) results.sort(key=lambda x: x[1], reverse=True) return results[:top_k]
3000个文档,top-3检索,10ms以内。够用了。
第四步:组装Prompt调用LLM
class RAGPipeline: def __init__(self, embedding_client, llm_api_key, vector_store): self.embedding_client = embedding_client self.llm_api_key = llm_api_key self.vector_store = vector_store def query(self, question, context_count=3): query_emb = self.embedding_client.embed_texts([question])[0] results = self.vector_store.search(query_emb, top_k=context_count) context = "\n\n".join([f"[片段{i+1}] {t}" for i, (t, _, _) in enumerate(results)]) prompt = f"基于以下信息回答问题,不要编造:\n\n{context}\n\n问题:{question}" resp = requests.post("https://api.openai.com/v1/chat/completions", json={"model": "gpt-4o-mini", "messages": [{"role": "user", "content": prompt}], "temperature": 0.7}, headers={"Authorization": f"Bearer {self.llm_api_key}"}) return resp.json()["choices"][0]["message"]["content"]
整个RAG管道就这些。加个命令行交互循环,总计不到500行。
效果对比
我用同一份文档测试了LangChain RAG和手写版。
测试问题:"RAG的核心步骤是什么?"
两个版本回答质量差不多。但手写版启动更快(1秒 vs 3秒),依赖更少(requests + numpy vs 20+个包)。
文档库几千段以内,手写版完全够用。
什么时候该用框架
用框架: 文档量10万段以上需要专业向量数据库、需要多轮检索/重排序等复杂链路、团队协作需要标准化。
手写: 学习原理、小项目追求可控、需要调试中间结果、不想背依赖包袱。
⚠️ 踩坑提醒:
你有没有过这种经历:跟着教程搭建项目,跑起来后一头雾水。为什么这样做?那个抽象层干什么的?
这就是我抛弃框架的原因。透明比便利更重要。
当你能从零写出能跑的RAG,再去用LangChain,那些抽象层就没那么可怕了。
💡 一句话带走
RAG的本质就是三步:切片、向量化、检索,剩下的都是包装。
灵魂拷问: 你有没有在某个技术栈里越陷越深,最后忘了它要解决的原始问题是什么?