RAG驱动的智能答疑系统让AI读懂你的问题再回答一、为什么我的代码超时了——这个问题的答案不在模型参数里用户提交了一段代码输出是 Time Limit Exceeded他问系统为什么超时怎么优化这是一个典型的需要领域知识的问题。GPT-4 可以回答通用的检查你的时间复杂度但无法回答在你这个具体问题LeetCode 239. 滑动窗口最大值中你用 List.remove(0) 的时间复杂度是 O(k)导致总复杂度 O(n*k)你应该用双端队列优化到 O(n)。通用模型不知道你的题库细节、不知道这道题的官方解法、不知道常见错误模式。RAGRetrieval-Augmented Generation检索增强生成的解决方案是在回答前先从知识库中检索相关信息再让模型基于检索结果生成答案。RAG 不改变模型本身——它改变的是输入给模型的上下文。好的检索结果让模型看到相关信息基于它生成精准答案差劲的检索结果让模型猜输出就变得空泛或错误。二、知识库构建Embedding 向量检索RAG 的精度取决于检索质量。检索质量的瓶颈不在向量数据库Milvus/Qdrant 都很快而在于知识库的内容质量和Embedding 模型的选择。import numpy as np import hashlib from typing import List, Dict, Tuple, Optional from dataclasses import dataclass, field from collections import OrderedDict dataclass class Document: 知识库文档 doc_id: str title: str content: str source: str # 来源official_solution | user_explanation | faq problem_id: Optional[int] None # 关联的题目ID tags: List[str] field(default_factorylist) embedding: Optional[np.ndarray] None # 文档向量 class SimpleEmbedder: 简化的文本向量化器 生产环境应使用 - OpenAI text-embedding-3-small1536维效果好 - BGE-M31024维开源支持中英文 - Jina embeddings8192维支持长文本 这里使用 TF-IDF 作为简化演示展示向量化的核心思想。 def __init__(self, dim: int 128): self.dim dim self._vocab: Dict[str, int] {} self._idf: Dict[str, float] {} def _tokenize(self, text: str) - List[str]: 简易中文 英文分词 实际项目请使用 jieba中文或 tiktoken英文。 import re # 提取中文词2-4个字和英文单词 text text.lower() tokens [] # 英文单词 tokens.extend(re.findall(r[a-z_][a-z0-9_]*, text)) # 中文字符按2-3字切割 chinese re.findall(r[\u4e00-\u9fff], text) for chunk in chinese: for i in range(len(chunk)): if i 2 len(chunk): tokens.append(chunk[i:i2]) if i 3 len(chunk): tokens.append(chunk[i:i3]) return tokens def fit(self, documents: List[str]): 构建词汇表和IDF doc_count len(documents) doc_freq: Dict[str, int] {} for doc in documents: tokens set(self._tokenize(doc)) for token in tokens: doc_freq[token] doc_freq.get(token, 0) 1 # 构建词汇表取前 dim 个最常见的词 sorted_tokens sorted( doc_freq.items(), keylambda x: x[1], reverseTrue ) for i, (token, _) in enumerate(sorted_tokens[:self.dim]): self._vocab[token] i # 计算 IDF for token, idx in self._vocab.items(): self._idf[token] np.log( (doc_count 1) / (doc_freq.get(token, 0) 1) ) 1 def encode(self, text: str) - np.ndarray: 将文本编码为向量 使用 TF-IDF 加权词袋模型 vec[i] TF(token_i, text) * IDF(token_i) 然后做 L2 归一化使余弦相似度退化为内积。 tokens self._tokenize(text) vec np.zeros(self.dim) # 计算 TF tf: Dict[str, int] {} for token in tokens: if token in self._vocab: tf[token] tf.get(token, 0) 1 # TF-IDF 加权 for token, count in tf.items(): if token in self._vocab: idx self._vocab[token] vec[idx] count * self._idf.get(token, 0.0) # L2 归一化 norm np.linalg.norm(vec) if norm 0: vec / norm return vec class VectorStore: 向量存储与检索 核心操作 1. 添加文档embedding 元数据 2. 相似度搜索余弦相似度 Top-K 3. 删除/更新文档 def __init__(self, embedder: SimpleEmbedder): self.embedder embedder self._documents: Dict[str, Document] {} self._embeddings: Dict[str, np.ndarray] {} def add(self, doc: Document): 添加或更新文档 if doc.embedding is None: doc.embedding self.embedder.encode( f{doc.title}\n{doc.content} ) self._documents[doc.doc_id] doc self._embeddings[doc.doc_id] doc.embedding def add_batch(self, documents: List[Document]): 批量添加文档 for doc in documents: self.add(doc) def _cosine_similarity( self, query_vec: np.ndarray, doc_vec: np.ndarray ) - float: 计算余弦相似度 因为向量已经 L2 归一化所以直接内积 余弦相似度。 范围 [-1, 1]值越大越相似。 return float(np.dot(query_vec, doc_vec)) def search( self, query: str, top_k: int 5, min_score: float 0.3 ) - List[Tuple[Document, float]]: 检索与查询最相关的 Top-K 文档 Args: query: 用户查询 top_k: 返回文档数 min_score: 最低相似度阈值过滤不相关文档 Returns: [(文档, 相似度得分), ...] 按得分降序排列 if not self._documents: return [] query_vec self.embedder.encode(query) # 计算所有文档的相似度 scores [] for doc_id, emb in self._embeddings.items(): score self._cosine_similarity(query_vec, emb) if score min_score: scores.append((doc_id, score)) # Top-K 排序 scores.sort(keylambda x: x[1], reverseTrue) results [] for doc_id, score in scores[:top_k]: results.append((self._documents[doc_id], score)) return results def search_with_filter( self, query: str, problem_id: int, top_k: int 5, min_score: float 0.3 ) - List[Tuple[Document, float]]: 带过滤条件的检索 先检索 Top-K*2 个候选再过滤出不匹配的文档。 这样可以保证即使 Top-K 个中有些被过滤掉 仍有足够的候选项。 candidates self.search(query, top_k * 2, min_score) filtered [] for doc, score in candidates: if doc.problem_id problem_id: filtered.append((doc, score)) if len(filtered) top_k: break return filtered class RAGAnswerGenerator: RAG 回答生成器 流程 1. 接收用户问题 上下文题目ID、用户代码 2. 构建检索查询整合问题 代码片段中的错误信息 3. 检索相关知识库文档 4. 拼接上下文并调用 LLM 生成回答 SYSTEM_PROMPT 你是一个算法刷题助手。请基于以下检索到的知识库文档回答用户问题。 要求 1. 回答要具体到用户正在做的这道题不要给出泛泛的建议 2. 如果检索到的文档中有对应题的官方题解必须引用 3. 如果用户提供了代码指出具体哪一行有问题以及为什么 4. 给出优化建议时说明优化前和优化后的时间复杂度 5. 如果没有找到相关知识诚实告知不要编造 def __init__(self, vector_store: VectorStore): self.vector_store vector_store def _build_query( self, question: str, problem_title: str , user_code: str , error_msg: str ) - str: 构建增强的检索查询 将用户的问题、题目信息、错误信息拼接成一个检索查询。 包含更多上下文能让检索更精准。 parts [question] if problem_title: parts.append(f题目{problem_title}) if error_msg: parts.append(f错误信息{error_msg}) if user_code: # 对代码做轻量处理只取函数定义和关键逻辑 lines user_code.strip().split(\n) key_lines [ l for l in lines if any(kw in l for kw in [def , class , return]) ][:5] if key_lines: parts.append(f代码片段{; .join(key_lines)}) return \n.join(parts) def _build_context( self, retrieved: List[Tuple[Document, float]] ) - str: 将检索结果拼接为 LLM 的上下文 每个文档格式化 [来源官方题解 | 用户题解 | FAQ] 标题xxx 内容xxx --- 限制总长度在 3000 token 内粗略估计 超出部分截断最新的文档。 context_parts [] total_chars 0 max_chars 6000 # 约 3000 token for doc, score in retrieved: source_label { official_solution: 官方题解, user_explanation: 用户题解, faq: FAQ, error_pattern: 常见错误 }.get(doc.source, doc.source) block ( f[来源{source_label} | 相似度{score:.2f}]\n f标题{doc.title}\n f内容{doc.content}\n f--- ) if total_chars len(block) max_chars: break context_parts.append(block) total_chars len(block) return \n.join(context_parts) def generate( self, question: str, problem_id: Optional[int] None, problem_title: str , user_code: str , error_msg: str ) - dict: 生成回答 Args: question: 用户问题必须 problem_id: 题目ID用于过滤检索范围 problem_title: 题目标题 user_code: 用户代码 error_msg: 编译/运行错误信息 Returns: { answer: str, # 生成的回答 sources: list, # 引用的来源文档 retrieval_time_ms: float } import time as _time # Step 1: 构建检索查询 query self._build_query( question, problem_title, user_code, error_msg ) # Step 2: 检索相关文档 start _time.perf_counter() if problem_id is not None: retrieved self.vector_store.search_with_filter( query, problem_id, top_k5 ) # 如果题目相关的文档不够补充全局检索 if len(retrieved) 3: global_results self.vector_store.search( query, top_k5 ) retrieved.extend(global_results) else: retrieved self.vector_store.search(query, top_k5) retrieval_time (_time.perf_counter() - start) * 1000 # Step 3: 构建上下文 context self._build_context(retrieved) # Step 4: 构建 prompt这里只构建实际调用 LLM API 需添加 prompt ( f{self.SYSTEM_PROMPT}\n\n f## 参考文档\n{context}\n\n f## 用户问题\n题目{problem_title}\n f问题{question}\n ) if error_msg: prompt f错误信息{error_msg}\n if user_code: prompt f用户代码\n\n{user_code}\n\n # 实际调用 LLM这里返回模拟结果 answer ( f基于检索到的 {len(retrieved)} 篇参考文档 f以下是针对你的问题的回答\n\n f[此处应由 LLM 基于上下文生成具体回答]\n\n f检索到的参考文档以 [来源xxx] 标注。 ) return { answer: answer, sources: [ { doc_id: doc.doc_id, title: doc.title, source: doc.source, score: round(score, 4) } for doc, score in retrieved ], retrieval_time_ms: round(retrieval_time, 2), context_prompt: prompt[:500] ... # 截断展示 } def demo_rag(): 演示完整的 RAG 问答流程 # 构建知识库 documents [ Document( doc_idsol_239, title滑动窗口最大值 - 官方题解, content使用双端队列维护窗口内的最大值候选。 队首始终是当前窗口的最大值。 时间复杂度 O(n)空间复杂度 O(k)。 关键点入队时从队尾移除所有小于当前值的元素 出队时如果队首元素滑出窗口则移除。, sourceofficial_solution, problem_id239, tags[sliding_window, deque, monotonic_queue] ), Document( doc_iderr_239_list_remove, title滑动窗口最大值 - 常见错误使用 List.remove(0), content很多初学者用 List.remove(0) 来移除窗口最左元素。 但 List.remove(0) 的时间复杂度是 O(k) 导致总复杂度变为 O(n*k)在 k 较大时会超时。 应使用 collections.dequepopleft() 是 O(1)。, sourceerror_pattern, problem_id239, tags[timeout, python, performance] ), Document( doc_idfaq_timeout, titleFAQ代码超时怎么办, content代码超时的常见原因\n 1. 算法复杂度过高O(n²) 而数据范围 n10^5\n 2. Python 中用了 O(n) 的列表操作而应使用 O(1) 的数据结构\n 3. 未做剪枝/记忆化\n 4. 递归深度过高\n 建议先用小数据量测试通过率再分析复杂度瓶颈。, sourcefaq, tags[timeout, optimization] ), Document( doc_idfaq_deque, titleFAQ什么时候用双端队列, content双端队列collections.deque适用场景\n 1. 滑动窗口最大值/最小值问题\n 2. BFS 中的队列操作\n 3. 需要 O(1) 的头部和尾部插入/删除\n 在 Python 中deque 的 popleft() 和 appendleft() 都是 O(1)而 list.pop(0) 和 list.insert(0, x) 是 O(n)。, sourcefaq, tags[data_structure, python, deque] ), ] # 初始化 Embedder 和 VectorStore embedder SimpleEmbedder(dim128) all_texts [f{d.title}\n{d.content} for d in documents] embedder.fit(all_texts) vector_store VectorStore(embedder) vector_store.add_batch(documents) # 创建 RAG 生成器 rag RAGAnswerGenerator(vector_store) # 测试问题 question 为什么我的代码在滑动窗口最大值这题超时了 user_code def maxSlidingWindow(nums, k): result [] window [] for i, num in enumerate(nums): window.append(num) if i k: window.pop(0) # ← 这里很慢 if i k - 1: result.append(max(window)) return result result rag.generate( questionquestion, problem_id239, problem_title滑动窗口最大值, user_codeuser_code, error_msgTime Limit Exceeded on test case 45 ) print( * 50) print(RAG 回答生成结果) print( * 50) print(f\n检索耗时: {result[retrieval_time_ms]:.1f}ms) print(f\n检索到的文档) for src in result[sources]: print(f [{src[source]}] {src[title]} f(相似度: {src[score]:.2f})) print(f\n生成的回答\n{result[answer]}) if __name__ __main__: demo_rag()三、RAG 的查询优化技巧直接拿用户问题去检索效果往往不好。以下是两个提升检索质量的技巧技巧 1HyDE假设性文档嵌入让 LLM 先生成一个假设的完美回答然后以这个回答来检索。因为回答和知识库文档的语言风格更接近检索精度更高。技巧 2Query 分解将复杂问题拆成子问题分别检索。例如 为什么超时怎么优化 拆成Query 1: 滑动窗口最大值 时间复杂度分析 为什么超时Query 2: 滑动窗口最大值 Python 优化方法 deque分别检索后合并结果去重。四、RAG 的边界和局限局限 1知识库不包含答案时模型可能编造答案幻觉。解决方案设置最低相似度阈值min_score0.5如果检索结果为空直接回答抱歉我目前的知识库中没有找到相关信息。局限 2长文本检索失焦长文档2000 token的关键信息可能分布在文档不同位置。解决方案将长文档切片为多个 chunk每个 chunk 独立 embedding 和检索。局限 3时效性问题知识库是静态的新题上线后知识库更新有延迟。解决方案新题上线时自动生成向量、加入知识库。五、总结RAG 检索 推理RAG 不强求模型记住一切——它让模型遇到不懂的先查资料。这个思路在刷题系统中的价值尤为明显题库是有限的、确定的领域知识大部分问题有明确的最优解常见错误模式可以被系统性地收录当用户问为什么超时时系统从知识库中精准检索到你用了 List.pop(0)时间复杂度是 O(n)应改用 deque.popleft()然后基于这个检索结果生成回答——这是 RAG 的典型胜利。而如果用户问为什么我学了三个月算法还是没进步这个问题不在知识库的任何文档中检索结果为空——RAG 应该诚实告知而不是编造一篇学习建议。这是 RAG 的边界也是它的可靠性所在。