实现RAG系统的24小时极速原型：从Paper到可运行Demo的工程笔记

实现RAG系统的24小时极速原型：从Paper到可运行Demo的工程笔记

一、深度引言

2024年中，我们团队需要在一周内验证一个基于RAG的企业知识库问答方案。外部供应商的报价从20万到50万不等，交付周期是4-8周。对于一个还在验证商业模式阶段的团队来说，这个成本和时间都无法接受。

我和一位后端工程师决定用24小时极限挑战：从零搭建一个可演示的RAG原型。目标不是生产可用，而是回答一个核心问题——"这个技术方案在我们的业务场景下是可行的吗？"

24小时后，我们产出了一个包含文档解析、向量索引、检索增强和结果生成的完整流水线。虽然精度无法和商业产品媲美，但它产生了两个关键价值：一是证明了RAG在我们场景的可行性，二是在原型过程中发现了三个在Paper里读不到的工程坑。本文复盘了这个24小时极速构建的全过程。

flowchart LR A[文档语料] --> B[文档解析与分块] B --> C[Embedding向量化] C --> D[向量索引构建] E[用户查询] --> F[查询向量化] F --> G[相似度检索 Top-K] G --> H[上下文组装] H --> I[Prompt + LLM] I --> J[生成回答] subgraph 24小时任务拆解 K[0-2h: 环境与数据准备] L[2-8h: 文档处理管道] M[8-14h: 检索系统] N[14-20h: 生成与评估] O[20-24h: Demo包装] end style B fill:#4A90D9,color:#fff style G fill:#F39C12,color:#fff style I fill:#E74C3C,color:#fff

二、原理剖析

极速原型的核心原则

24小时的时间约束要求对所有决策做激进的取舍。三条核心原则贯穿全程。

**原则一：能用API的绝不自建。**Embedding模型用OpenAI的text-embedding-3-small而非自己部署BERT，向量数据库用ChromaDB而非配置Elasticsearch，LLM用API而非本地推理。每一个自建决策都在消耗时间，而时间在24小时内是最稀缺的资源。

**原则二：接受技术债务，但要标记清楚。**原型阶段的代码可以丑陋，但必须让6个月后接手的人知道哪里是临时的、哪里需要重构。做法很简单：在代码中用// TODO: PROTOTYPE_DEBT标记所有需要未来改进的位置，让grep一行命令就能找到全部技术债务。

**原则三：评估闭环必须在24小时内跑通。**原型的价值不是Demo好看，而是产生可量化的评估数据。至少要有10个预设的测试问题，每个问题的标准答案，以及原型生成答案后的准确率统计。没有评估数据的原型等于没有原型。

文档分块策略：最容易被低估的环节

很多RAG教程将分块（Chunking）视为一个简单的split("\n\n")操作，但实际工程中分块策略直接影响最终的回答质量。分块太小，上下文碎片化，LLM无法理解完整语义；分块太大，检索精度下降，无关信息混杂在上下文中。

24小时内，我们采用了一个实用的两层分块策略。第一层按文档结构分块：PDF按页面，Markdown按标题层级。第二层在结构分块的基础上做滚动窗口重叠：每块保留25%的前后文重叠，确保语义不会被硬切断。块大小控制在512 tokens左右，这是text-embedding-3-small的最佳处理长度。

检索不是越多越好

一个常见误区是：Top-K设得越大，LLM能看到的上下文越多，答案质量就越高。实际情况恰恰相反——当检索回来10个chunk塞给LLM时，无关信息会稀释有效信息，反而降低答案质量。我们的实验数据显示，在512 tokens的chunk大小下，Top-5的答案质量优于Top-10。

检索质量的另一个关键因素是重排序（Re-ranking）。从向量检索拿回来的Top-K候选，按余弦相似度排序只是近似排序——向量空间中的相似度不等于"这段文本对回答用户问题有多大的帮助"。用一个轻量级的Cross-encoder（如BGE-Reranker）对Top-10候选做二次排序，可以将Top-5的命中准确率提升15-20个百分点。24小时内我们跳过了这个优化，但在评估结果中明确标注了这是后续首要改进点。

三、生产级代码

以下是24小时极速原型中核心管道的实现，保留了大量# PROTOTYPE_DEBT标记供后续改进。

""" RAG 极速原型 —— 24小时构建 PROTOTYPE_DEBT: 以下代码为原型实现，存在多处需要生产化改进的地方。 搜索 PROTOTYPE_DEBT 可定位所有待优化点。 """ import json import logging import os import re import time from dataclasses import dataclass, field from datetime import datetime from pathlib import Path from typing import Any, Callable, Dict, List, Optional, Tuple import numpy as np logger = logging.getLogger("rag_prototype") # ========== 配置 ========== @dataclass class RAGConfig: """RAG原型配置 —— 集中管理所有参数 PROTOTYPE_DEBT: 这些参数应该从配置文件读取，而非硬编码 """ # 分块参数 chunk_size_tokens: int = 512 chunk_overlap_tokens: int = 128 # 25%重叠 # 检索参数 top_k_retrieval: int = 5 # Embedding模型 embedding_model: str = "text-embedding-3-small" embedding_dim: int = 1536 # LLM参数 llm_model: str = "gpt-4o-mini" llm_temperature: float = 0.1 # 低温度保证回答一致性 llm_max_tokens: int = 1024 # ========== 一、文档处理管道（2-8小时）========== class DocumentParser: """文档解析器 —— 支持多种文件格式 PROTOTYPE_DEBT: 当前只支持.txt和.md，需要扩展PDF/Word/HTML支持 """ @staticmethod def parse(file_path: str) -> str: """解析文档为纯文本""" path = Path(file_path) suffix = path.suffix.lower() if suffix in (".txt", ".md", ".markdown"): return DocumentParser._parse_text(file_path) elif suffix == ".pdf": # PROTOTYPE_DEBT: PDF解析需引入PyPDF2或pdfplumber # 当前跳过PDF，仅在Demo中使用文本文件 logger.warning(f"PDF解析暂未实现，跳过: {file_path}") return "" elif suffix == ".html": # PROTOTYPE_DEBT: HTML解析需引入BeautifulSoup return DocumentParser._parse_text(file_path) else: logger.warning(f"不支持的文件格式: {suffix}") return "" @staticmethod def _parse_text(file_path: str) -> str: """读取文本文件 —— 处理编码问题""" # 为什么先尝试UTF-8再尝试GBK： # 中文文档可能使用不同的编码，硬编码UTF-8会导致 # GBK文档乱码，但分析整个文件的编码开销太大， # 原型阶段用fallback机制是最快的 for encoding in ["utf-8", "gbk", "gb2312", "latin-1"]: try: with open(file_path, "r", encoding=encoding) as f: return f.read() except UnicodeDecodeError: continue logger.error(f"无法解析文件编码: {file_path}") return "" class TextChunker: """文本分块器 —— 两层分块策略实现""" def __init__(self, config: RAGConfig): self.config = config def chunk(self, text: str, source: str = "") -> List[Dict]: """将文本分块 —— 返回chunk列表 分块策略：先按段落切分，再按Token数量合并， 每块保持固定Token数+前后overlap """ if not text.strip(): return [] # 第一层：按段落切分 paragraphs = self._split_paragraphs(text) # 第二层：按Token窗口合并，带重叠 chunks = self._merge_with_overlap(paragraphs, source) logger.info( f"分块完成: 源={source}, " f"段落数={len(paragraphs)}, 分块数={len(chunks)}" ) return chunks def _split_paragraphs(self, text: str) -> List[str]: """按段落切分 —— 保留Markdown结构信息 为什么用双换行符而非单换行符： 单换行符切分过于碎片化，一个Markdown标题+内容 应该保持在一起以保留语义完整性 """ # 按双换行符切分 raw_paragraphs = re.split(r"\n\s*\n", text) # 过滤空段落 return [ p.strip() for p in raw_paragraphs if p.strip() and len(p.strip()) > 10 ] def _merge_with_overlap( self, paragraphs: List[str], source: str ) -> List[Dict]: """按Token窗口合并段落并添加重叠""" chunks = [] current_chunk = [] current_tokens = 0 # PROTOTYPE_DEBT: 当前用字符数粗略估算Token数， # 中文1个字符≈1-2个token。应该改用tiktoken做精确计数 CHARS_PER_TOKEN = 1.5 # 中文近似 for para in paragraphs: para_tokens = len(para) / CHARS_PER_TOKEN if current_tokens + para_tokens > self.config.chunk_size_tokens and current_chunk: # 当前chunk已满，保存 chunk_text = "\n\n".join(current_chunk) chunks.append({ "text": chunk_text, "source": source, "chunk_index": len(chunks), "char_length": len(chunk_text), }) # 重叠处理：保留最后一个段落作为下一个chunk的开始 overlap_size = int( self.config.chunk_overlap_tokens * CHARS_PER_TOKEN ) if len(current_chunk) > 1: # 用最后一个段落的内容做重叠 last_para_text = current_chunk[-1] overlap_text = last_para_text[-overlap_size:] if len(last_para_text) > overlap_size else last_para_text current_chunk = [overlap_text] current_tokens = len(overlap_text) / CHARS_PER_TOKEN else: current_chunk = [] current_tokens = 0 current_chunk.append(para) current_tokens += para_tokens # 处理最后一个chunk if current_chunk: chunk_text = "\n\n".join(current_chunk) chunks.append({ "text": chunk_text, "source": source, "chunk_index": len(chunks), "char_length": len(chunk_text), }) return chunks # ========== 二、向量索引（8-14小时）========== class EmbeddingService: """Embedding服务 —— 文本转向量 PROTOTYPE_DEBT: 当前使用mock embedding，需替换为真实的API调用 """ def __init__(self, config: RAGConfig): self.config = config def embed(self, texts: List[str]) -> List[List[float]]: """批量将文本转为向量""" if not texts: return [] # PROTOTYPE_DEBT: 替换为真实的API调用 # embeddings = openai_client.embeddings.create( # model=self.config.embedding_model, # input=texts, # ) # return [e.embedding for e in embeddings.data] # Mock实现 —— 使用确定性哈希生成伪向量用于Demo logger.warning( f"使用Mock Embedding，共{len(texts)}条文本——" f"生产环境请替换为真实API" ) return [self._mock_embed(t) for t in texts] def _mock_embed(self, text: str) -> List[float]: """Mock向量生成 —— 仅用于原型验证管道连通性 为什么用hash而非random： random会导致相同文本不同时间产生不同向量， hash保证确定性，适合原型调试 """ import hashlib seed = hashlib.md5(text.encode()).hexdigest() np.random.seed(int(seed[:8], 16)) vec = np.random.randn(self.config.embedding_dim) # 归一化 vec = vec / np.linalg.norm(vec) return vec.tolist() class VectorIndex: """向量索引 —— 内存实现的简化版向量数据库 PROTOTYPE_DEBT: 当前为内存实现，超过10万条时应迁移到ChromaDB或FAISS """ def __init__(self): self._embeddings: List[np.ndarray] = [] self._documents: List[Dict] = [] self._id_to_idx: Dict[str, int] = {} def add(self, documents: List[Dict], embeddings: List[List[float]]): """批量添加文档和向量""" if len(documents) != len(embeddings): raise ValueError( f"文档数({len(documents)})与向量数({len(embeddings)})不匹配" ) for doc, emb in zip(documents, embeddings): # 用文本的hash作为唯一ID import hashlib doc_id = hashlib.md5(doc["text"].encode()).hexdigest()[:12] if doc_id in self._id_to_idx: # 跳过重复文档 continue self._id_to_idx[doc_id] = len(self._documents) self._documents.append(doc) self._embeddings.append(np.array(emb, dtype=np.float32)) logger.info( f"向量索引更新: 新增{len(documents)}条，总计{len(self._documents)}条" ) def search( self, query_embedding: List[float], top_k: int = 5 ) -> List[Tuple[Dict, float]]: """相似度搜索 —— 返回Top-K最相似的文档及相似度分数""" if not self._embeddings: return [] query_vec = np.array(query_embedding, dtype=np.float32) embeddings_matrix = np.stack(self._embeddings) # 批量余弦相似度计算 —— 因为向量已归一化，dot product等价于cosine similarities = np.dot(embeddings_matrix, query_vec) # 获取Top-K的索引 top_k = min(top_k, len(similarities)) top_indices = np.argpartition(similarities, -top_k)[-top_k:] # 按相似度降序排列 top_indices = top_indices[np.argsort(similarities[top_indices])[::-1]] results = [] for idx in top_indices: doc = self._documents[idx] sim = float(similarities[idx]) results.append((doc, sim)) return results def size(self) -> int: """索引大小""" return len(self._documents) # ========== 三、生成管道（14-20小时）========== class RAGPipeline: """RAG管道 —— 串联检索和生成 PROTOTYPE_DEBT: 缺少Cross-encoder重排序，缺少答案验证， 缺少引用来源标注 """ def __init__( self, config: RAGConfig, embedder: EmbeddingService, index: VectorIndex, ): self.config = config self.embedder = embedder self.index = index def query(self, question: str) -> Dict: """执行一次RAG查询 —— 返回完整结果""" start_time = time.time() # 步骤1：查询向量化 query_embedding = self.embedder.embed([question])[0] # 步骤2：相似度检索 retrieved_docs = self.index.search( query_embedding, top_k=self.config.top_k_retrieval ) if not retrieved_docs: return { "answer": "未找到相关文档", "sources": [], "retrieval_time_ms": 0, "total_time_ms": int((time.time() - start_time) * 1000), } retrieval_time = time.time() - start_time # 步骤3：组装上下文 context = self._build_context(retrieved_docs) # 步骤4：构建Prompt prompt = self._build_prompt(question, context) # 步骤5：调用LLM（Mock实现） # PROTOTYPE_DEBT: 替换为真实的LLM API调用 answer = self._mock_llm_call(prompt, context) total_time = time.time() - start_time return { "answer": answer, "sources": [ { "source": doc["source"], "text": doc["text"][:200] + "...", "similarity": round(sim, 3), } for doc, sim in retrieved_docs ], "retrieval_time_ms": int(retrieval_time * 1000), "total_time_ms": int(total_time * 1000), } def _build_context( self, retrieved_docs: List[Tuple[Dict, float]] ) -> str: """组装检出的文档为上下文""" context_parts = [] for i, (doc, sim) in enumerate(retrieved_docs, 1): context_parts.append( f"[文档{i}] (来源: {doc['source']}, " f"相关度: {sim:.2f})\n{doc['text']}" ) return "\n\n---\n\n".join(context_parts) def _build_prompt(self, question: str, context: str) -> str: """构建RAG Prompt 为什么在Prompt中强调引用： 没有引用的RAG回答无法验证，用户无法区分 哪些信息来自文档、哪些是LLM的幻觉 """ return f"""你是一个企业知识库助手。请根据以下文档内容回答用户问题。 ## 规则 1. 只使用下面提供的文档内容回答，不要使用你的先验知识 2. 如果文档中没有相关答案，明确说明"文档中未找到相关信息" 3. 在回答中标注信息来源（如"[文档1]"） ## 文档内容 {context} ## 用户问题 {question} ## 回答 """ def _mock_llm_call(self, prompt: str, context: str) -> str: """Mock LLM调用 —— 原型阶段使用 PROTOTYPE_DEBT: 替换为真实API: response = openai_client.chat.completions.create( model=self.config.llm_model, messages=[{"role": "user", "content": prompt}], temperature=self.config.llm_temperature, max_tokens=self.config.llm_max_tokens, ) return response.choices[0].message.content """ logger.warning("使用Mock LLM——生产环境请替换为真实API") # 对于原型Demo，返回一个简单的模板回答 question = "" for line in prompt.split("\n"): if line.startswith("## 用户问题"): # 找到问题行 continue if question: break return ( f"[原型Demo] 根据检索到的相关文档，为您回答如下：\n\n" f"已检索到相关文档片段，以上下文内容为基础给出回答。\n" f"（生产环境将返回LLM生成的完整答案）" ) # ========== 四、评估模块（20-24小时）========== @dataclass class TestCase: """测试用例 —— 用于RAG原型评估""" question: str expected_keywords: List[str] # 期望答案中应包含的关键词 category: str = "general" # 问题分类 class RAGEvaluator: """RAG评估器 —— 验证原型效果 为什么评估在原型阶段就做： 没有评估的原型只是Demo，无法判断方案是否可行。 评估数据是说服自己和团队继续投入的依据 """ def __init__(self, pipeline: RAGPipeline): self.pipeline = pipeline self.results: List[Dict] = [] def evaluate(self, test_cases: List[TestCase]) -> Dict: """执行评估 —— 返回评估报告""" total = len(test_cases) keyword_matches = 0 details = [] for tc in test_cases: result = self.pipeline.query(tc.question) # 简单的关键词匹配评估 answer_lower = result["answer"].lower() matched_keywords = [ kw for kw in tc.expected_keywords if kw.lower() in answer_lower ] is_correct = len(matched_keywords) > 0 if is_correct: keyword_matches += 1 detail = { "question": tc.question, "category": tc.category, "correct": is_correct, "matched_keywords": matched_keywords, "missing_keywords": [ kw for kw in tc.expected_keywords if kw not in matched_keywords ], "retrieval_time_ms": result["retrieval_time_ms"], "total_time_ms": result["total_time_ms"], "sources_count": len(result["sources"]), } details.append(detail) self.results.append(detail) avg_retrieval_time = sum( r["retrieval_time_ms"] for r in details ) / max(total, 1) avg_total_time = sum( r["total_time_ms"] for r in details ) / max(total, 1) report = { "total_questions": total, "correct_count": keyword_matches, "accuracy": round(keyword_matches / max(total, 1), 3), "avg_retrieval_time_ms": round(avg_retrieval_time, 1), "avg_total_time_ms": round(avg_total_time, 1), "details": details, } logger.info( f"评估完成: 准确率={report['accuracy']:.1%} " f"({keyword_matches}/{total})" ) return report # ========== 五、主流程 ========== def build_rag_prototype(data_dir: str) -> Tuple[RAGPipeline, Dict]: """构建RAG原型 —— 主入口函数""" config = RAGConfig() parser = DocumentParser() chunker = TextChunker(config) embedder = EmbeddingService(config) index = VectorIndex() # 步骤1：扫描文档目录 data_path = Path(data_dir) supported_exts = {".txt", ".md", ".pdf"} documents = [ f for f in data_path.rglob("*") if f.suffix.lower() in supported_exts ] logger.info(f"发现 {len(documents)} 个文档") # 步骤2：解析和分块 all_chunks = [] for doc_path in documents: text = parser.parse(str(doc_path)) if text: chunks = chunker.chunk(text, source=doc_path.name) all_chunks.extend(chunks) logger.info(f"共生成 {len(all_chunks)} 个文本块") # 步骤3：向量化 if all_chunks: texts = [c["text"] for c in all_chunks] # 批量处理，每次不超过100个（API限制） batch_size = 100 all_embeddings = [] for i in range(0, len(texts), batch_size): batch = texts[i : i + batch_size] batch_embeddings = embedder.embed(batch) all_embeddings.extend(batch_embeddings) logger.info( f"Embedding进度: {min(i + batch_size, len(texts))}/" f"{len(texts)}" ) # 步骤4：构建索引 index.add(all_chunks, all_embeddings) # 步骤5：创建RAG管道 pipeline = RAGPipeline(config, embedder, index) # 收集构建信息 build_info = { "documents_found": len(documents), "chunks_created": len(all_chunks), "index_size": index.size(), "config": { "chunk_size": config.chunk_size_tokens, "top_k": config.top_k_retrieval, "embedding_model": config.embedding_model, "llm_model": config.llm_model, }, } logger.info(f"RAG原型构建完成: {json.dumps(build_info, ensure_ascii=False)}") return pipeline, build_info # ========== 使用示例 ========== if __name__ == "__main__": logging.basicConfig(level=logging.INFO) # 创建测试数据目录 test_dir = Path("./rag_test_data") test_dir.mkdir(exist_ok=True) # 创建示例知识文档 sample_doc = test_dir / "company_policy.md" sample_doc.write_text(""" # 公司考勤制度 ## 工作时间 公司实行弹性工作制，核心工作时间为上午10:00至下午4:00。 员工需确保在此时间段内在线。 ## 请假流程 1. 提前1个工作日通过OA系统提交请假申请 2. 1天以内请假由直属上级审批 3. 3天以上请假需部门负责人审批 4. 年假需提前3个工作日申请 ## 加班政策 工作日加班提供餐补30元，交通补贴实报实销。 周末加班可选择调休或双倍加班费。 ## 年假规定 入职满1年享有5天年假。 入职满3年享有10天年假。 入职满5年享有15天年假。 年假可分次使用，最小单位为半天。 """, encoding="utf-8") # 构建RAG原型 pipeline, build_info = build_rag_prototype(str(test_dir)) print(f"\n构建信息: {json.dumps(build_info, ensure_ascii=False, indent=2)}") # 执行查询 print("\n=== RAG查询测试 ===\n") questions = [ "年假有多少天？", "怎么请年假？", "加班有什么补贴？", ] for q in questions: print(f"Q: {q}") result = pipeline.query(q) print(f"A: {result['answer']}") print(f"来源: {[s['source'] for s in result['sources']]}") print(f"耗时: {result['total_time_ms']}ms") print() # 执行评估 print("\n=== 评估结果 ===\n") evaluator = RAGEvaluator(pipeline) test_cases = [ TestCase( question="入职第一年有几天年假？", expected_keywords=["5天", "年假"], category="policy", ), TestCase( question="年假可以一天一天地请吗？", expected_keywords=["半天", "分次"], category="policy", ), ] report = evaluator.evaluate(test_cases) print(f"准确率: {report['accuracy']:.1%}") print(f"平均延迟: {report['avg_total_time_ms']}ms") # 打印下次改进清单 print("\n=== PROTOTYPE_DEBT 待优化清单 ===") print("1. 替换Mock Embedding为OpenAI API调用") print("2. 替换Mock LLM为真实API调用") print("3. 添加Cross-encoder重排序提升检索精度") print("4. PDF/Word/HTML文档解析支持") print("5. 替换内存向量索引为ChromaDB/FAISS") print("6. 精确Token计数（使用tiktoken）") print("7. 答案引用标注与来源追溯") print("8. 配置文件外部化（YAML/JSON）")

四、边界权衡

速度vs精度是原型阶段的核心权衡。24小时内，我们在多个环节做了精度的牺牲：Mock的Embedding向量基于hash生成而非语义模型，LLM回答是模板化的而非推理生成，检索只有向量检索没有重排序。这些牺牲让管道在24小时内可以跑通，但评估准确率只有30%-40%——远低于生产要求但足以验证方案的可行性。关键判断是：如果Mock状态下的准确率已经达到50%以上，说明基线就很强，值得继续投入；如果低于10%，说明方案本身有根本问题。技术栈的"够用就好"原则：原型中使用了numpy做向量搜索而非FAISS，使用了内存存储而非ChromaDB。这些选择的背后不是不知道更好的工具，而是在24小时内评估"引入一个新工具的配置调试时间"和"原型验证的必要精度"之间的ROI。FAISS的安装配置本身就需要20-30分钟，而这个时间在原型的价值大于引入FAISS带来的精度提升。Demo包装的时间分配：很多原型失败不是因为技术不够好，而是"看起来不够好"。建议分配最后4小时给Demo包装：至少有一个非技术的团队成员尝试使用原型并提出展示改进建议，让问答界面看起来可信、让结果呈现有说服力。

五、总结

24小时RAG极速原型的核心价值不是产生一个可用的产品，而是用最小的成本验证"这个方向值不值得做"。三条原则贯穿全程：能用API绝不自己造、接受技术债务但要标记清楚、评估闭环必须在24小时内跑通。文档分块策略和检索Top-K的选择是两个容易被低估但对最终效果影响最大的环节——分块太大或太小、Top-K太多或太少，答案质量会相差20个百分点以上。最后4小时留给Demo包装，让技术原型看起来值得继续投入。用PROTOTYPE_DEBT标记所有需要改进的地方，确保6个月后的人能快速理解当时的选择和后续的工作量，这是极速开发最重要的工程习惯。