实现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而非配置ElasticsearchLLM用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(fPDF解析暂未实现跳过: {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, encodingencoding) 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( # modelself.config.embedding_model, # inputtexts, # ) # 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, dtypenp.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, dtypenp.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_kself.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调用LLMMock实现 # 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( modelself.config.llm_model, messages[{role: user, content: prompt}], temperatureself.config.llm_temperature, max_tokensself.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, sourcedoc_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( fEmbedding进度: {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(fRAG原型构建完成: {json.dumps(build_info, ensure_asciiFalse)}) return pipeline, build_info # 使用示例 if __name__ __main__: logging.basicConfig(levellogging.INFO) # 创建测试数据目录 test_dir Path(./rag_test_data) test_dir.mkdir(exist_okTrue) # 创建示例知识文档 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天年假。 年假可分次使用最小单位为半天。 , encodingutf-8) # 构建RAG原型 pipeline, build_info build_rag_prototype(str(test_dir)) print(f\n构建信息: {json.dumps(build_info, ensure_asciiFalse, indent2)}) # 执行查询 print(\n RAG查询测试 \n) questions [ 年假有多少天, 怎么请年假, 加班有什么补贴, ] for q in questions: print(fQ: {q}) result pipeline.query(q) print(fA: {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天, 年假], categorypolicy, ), TestCase( question年假可以一天一天地请吗, expected_keywords[半天, 分次], categorypolicy, ), ] 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个月后的人能快速理解当时的选择和后续的工作量这是极速开发最重要的工程习惯。