RAG系统设计面试题

难度：⭐⭐ | 高频指数：🔥🔥🔥 | 应用岗相关度：★★★

面试回答

常见问法

• “设计一个企业知识库 RAG 系统”
• “你们的 RAG 系统架构是怎样的？”
• “如果让你从零搭一个文档问答系统，你怎么做？“

回答

这是一道系统设计开放题，面试时建议先画架构、再逐步展开每一层的选型和 trade-off。

整体数据流：文档 → 切分 → Embedding → 向量库 → 检索 → Rerank → Prompt 组装 → LLM → 回答

# RAG系统整体架构
class EnterpriseRAGSystem:
    """
    企业知识库 RAG 系统
    
    数据流：
    文档摄入 → 预处理 → Chunk切分 → Embedding → 向量库写入
    用户查询 → 查询改写 → 混合检索 → Rerank → Prompt组装 → LLM生成 → 回答
    """
    
    def __init__(self, config):
        # 离线链路组件
        self.doc_parser = DocumentParser(config['parser'])
        self.chunker = SmartChunker(config['chunk'])
        self.embedder = EmbeddingModel(config['embedding'])
        self.vector_store = VectorStore(config['vector_db'])
        
        # 在线链路组件
        self.query_rewriter = QueryRewriter(config['rewriter'])
        self.retriever = HybridRetriever(config['retriever'])
        self.reranker = Reranker(config['reranker'])
        self.prompt_builder = PromptBuilder(config['prompt'])
        self.llm = LLMClient(config['llm'])
        
        # 观测组件
        self.tracer = Tracer(config['tracing'])
    
    # ========== 离线链路：文档摄入 ==========
    def ingest_document(self, document_path, metadata=None):
        """
        文档摄入流程
        
        1. 解析文档（支持 PDF/Word/HTML/Markdown）
        2. 预处理（清洗、去重、提取结构）
        3. 切分 Chunk
        4. 生成 Embedding
        5. 写入向量库
        """
        with self.tracer.span("document_ingestion"):
            # 1. 解析
            raw_text = self.doc_parser.parse(document_path)
            
            # 2. 预处理
            cleaned_text = self.preprocess(raw_text)
            
            # 3. 切分
            chunks = self.chunker.chunk(
                cleaned_text,
                metadata=metadata
            )
            
            # 4. Embedding
            embeddings = self.embedder.encode_batch(
                [chunk['content'] for chunk in chunks]
            )
            
            # 5. 写入向量库
            for chunk, embedding in zip(chunks, embeddings):
                self.vector_store.upsert(
                    id=chunk['id'],
                    vector=embedding,
                    metadata={
                        'content': chunk['content'],
                        'source': document_path,
                        'chunk_index': chunk['index'],
                        **chunk.get('metadata', {}),
                        **(metadata or {})
                    }
                )
            
            return {
                'status': 'success',
                'chunks_count': len(chunks),
                'document': document_path
            }
    
    # ========== 在线链路：查询回答 ==========
    def query(self, user_question, user_context=None):
        """
        在线查询流程
        
        1. 查询改写
        2. 混合检索
        3. Rerank
        4. Prompt 组装
        5. LLM 生成
        6. 后处理
        """
        with self.tracer.span("rag_query") as span:
            span.set_attribute("user_question", user_question)
            
            # 1. 查询改写（处理模糊查询、多轮对话）
            rewritten_queries = self.query_rewriter.rewrite(
                user_question, 
                context=user_context
            )
            
            # 2. 混合检索
            candidates = self.retriever.retrieve(
                queries=rewritten_queries,
                top_k=20
            )
            
            # 3. Rerank
            reranked = self.reranker.rerank(
                query=user_question,
                documents=candidates,
                top_n=5
            )
            
            # 4. Prompt 组装
            prompt = self.prompt_builder.build(
                question=user_question,
                context_docs=reranked,
                system_instructions="基于提供的资料回答，无法确定时说明"
            )
            
            # 5. LLM 生成
            response = self.llm.generate(
                prompt=prompt,
                temperature=0.1,  # 知识问答用低温度
                max_tokens=1024
            )
            
            # 6. 后处理（添加引用、格式化）
            final_response = self.postprocess(
                response, reranked
            )
            
            span.set_attribute("response_length", len(final_response))
            return final_response
    
    def preprocess(self, raw_text):
        """文档预处理"""
        import re
        # 去除多余空白
        text = re.sub(r'\s+', ' ', raw_text)
        # 去除特殊字符
        text = re.sub(r'[^\w\s\u4e00-\u9fff.,;:!?()（）。，；：！？]', '', text)
        return text.strip()
    
    def postprocess(self, response, source_docs):
        """后处理：添加引用"""
        citations = []
        for i, doc in enumerate(source_docs):
            citations.append(f"[{i+1}] {doc['metadata'].get('source', '未知来源')}")
        
        return {
            'answer': response,
            'citations': citations,
            'confidence': self.calculate_confidence(response, source_docs)
        }
    
    def calculate_confidence(self, response, source_docs):
        """计算回答置信度"""
        if not source_docs:
            return 0.0
        avg_score = sum(d.get('score', 0) for d in source_docs) / len(source_docs)
        return min(avg_score, 1.0)

面试时每一步都可以展开讲选型：

追问

多语言怎么处理？

class MultilingualRAG:
    """多语言处理策略"""
    
    def handle_multilingual(self, query, target_lang='zh'):
        # 策略1：统一语言embedding（推荐）
        # 使用多语言embedding模型（如 multilingual-e5）
        
        # 策略2：翻译后检索
        # 先把query翻译成文档语言，再检索
        
        # 策略3：多语言索引
        # 每种语言建独立索引，查询时并行检索
        
        # 实际推荐：多语言embedding + 语言标签过滤
        results = self.retriever.retrieve(
            query=query,
            filters={'language': target_lang}  # 可选过滤
        )
        return results

文档更新怎么增量？

class IncrementalIngestion:
    """增量更新策略"""
    
    def incremental_update(self, document_path, version):
        # 1. 计算文档hash，判断是否有变化
        doc_hash = self.compute_hash(document_path)
        
        if self.is_unchanged(document_path, doc_hash):
            return {'status': 'skipped', 'reason': 'no_change'}
        
        # 2. 删除旧版本的chunks
        self.vector_store.delete_by_filter(
            filter={'source': document_path, 'version': {'$lt': version}}
        )
        
        # 3. 重新摄入
        result = self.ingest_document(
            document_path, 
            metadata={'version': version, 'hash': doc_hash}
        )
        
        return result

权限怎么控制？

class PermissionAwareRAG:
    """权限控制策略"""
    
    def query_with_permission(self, user_question, user_id):
        # 1. 获取用户权限
        user_permissions = self.get_user_permissions(user_id)
        
        # 2. 检索时加权限过滤
        results = self.retriever.retrieve(
            query=user_question,
            filters={
                'access_level': {'$lte': user_permissions['level']},
                'department': {'$in': user_permissions['departments']}
            }
        )
        
        # 3. 生成回答（只基于有权限的文档）
        return self.generate_answer(user_question, results)

原理展开

架构设计思路

面试时画架构图的顺序：

1. 先画数据流：文档从哪来 → 怎么处理 → 存到哪
2. 再画查询流：用户问题 → 怎么检索 → 怎么生成回答
3. 最后补观测：哪里加 tracing、哪里加指标

# 效果评估体系
class RAGEvaluator:
    """RAG系统效果评估"""
    
    def evaluate(self, test_dataset):
        """
        评估维度：
        1. 召回率（Recall）：相关文档是否被检索到
        2. 准确率（Precision）：检索到的文档是否相关
        3. 回答质量：生成的回答是否正确、完整
        4. 用户满意度：实际使用反馈
        """
        results = {
            'retrieval_metrics': {},
            'generation_metrics': {},
            'overall_metrics': {}
        }
        
        for sample in test_dataset:
            # 检索评估
            retrieved = self.system.retrieve(sample['question'])
            retrieval_score = self.compute_retrieval_metrics(
                retrieved, sample['relevant_docs']
            )
            
            # 生成评估
            answer = self.system.query(sample['question'])
            generation_score = self.compute_generation_metrics(
                answer, sample['expected_answer']
            )
            
            results['retrieval_metrics'][sample['id']] = retrieval_score
            results['generation_metrics'][sample['id']] = generation_score
        
        # 汇总
        results['overall_metrics'] = {
            'avg_recall': self.average(results['retrieval_metrics'], 'recall'),
            'avg_precision': self.average(results['retrieval_metrics'], 'precision'),
            'avg_answer_quality': self.average(results['generation_metrics'], 'quality'),
            'hallucination_rate': self.compute_hallucination_rate(results)
        }
        
        return results
    
    def compute_retrieval_metrics(self, retrieved, relevant):
        """计算检索指标"""
        retrieved_ids = set(d['id'] for d in retrieved)
        relevant_ids = set(relevant)
        
        recall = len(retrieved_ids & relevant_ids) / len(relevant_ids) if relevant_ids else 0
        precision = len(retrieved_ids & relevant_ids) / len(retrieved_ids) if retrieved_ids else 0
        
        return {'recall': recall, 'precision': precision}

面试时怎么讲

推荐表达顺序：

1. 先说业务场景：“这是一个企业内部知识库问答系统，用户是内部员工”
2. 画架构图：离线摄入 + 在线查询两条线
3. 逐步展开选型：每一步说选了什么、为什么
4. 主动提优化点：缓存、增量更新、权限、多语言
5. 说评估方法：怎么知道系统效果好不好

# 面试时可以提的优化点
OPTIMIZATION_POINTS = {
    "性能优化": [
        "热门问题缓存（语义缓存）",
        "Embedding 预计算和批量处理",
        "向量库分片和负载均衡",
    ],
    "效果优化": [
        "查询改写提升召回",
        "Rerank 提升精度",
        "Prompt 模板 A/B 测试",
    ],
    "工程优化": [
        "增量更新避免全量重建",
        "权限控制保证数据安全",
        "Tracing 和监控保证可观测",
    ],
    "成本优化": [
        "小模型做初筛，大模型做精排",
        "缓存减少重复调用",
        "按需选择 embedding 维度",
    ]
}

易错点

• 只说”用了 LangChain”但讲不清每一步在做什么
• 架构图画得太简单，只有”用户→模型→回答”
• 不提评估方法，面试官会追问”你怎么知道效果好”
• 忽略工程细节（增量更新、权限、缓存）
• 把所有问题都归结为”换个更好的模型”

记忆技巧

记住 RAG 系统设计的”两条线 + 三个追问”：

两条线：

1. 离线线：文档 → Chunk → Embedding → 向量库
2. 在线线：Query → 检索 → Rerank → Prompt → LLM → 回答

三个必答追问：

1. 怎么评估效果？（召回率 + 准确率 + 用户满意度）
2. 文档更新怎么办？（增量摄入 + 版本管理）
3. 怎么保证安全？（权限过滤 + 内容审核）

面试速答版

• RAG 系统设计题先画两条线：离线摄入（文档→Chunk→Embedding→向量库）和在线查询（Query→检索→Rerank→Prompt→LLM→回答）
• 每一步都要能说出选型和 trade-off，不能只说”用了 LangChain”
• 关键评估维度：召回率、准确率、回答质量、用户满意度
• 必须主动提的优化点：缓存、增量更新、权限控制、多语言
• 常见坑：架构太简单、不提评估方法、把问题全归结为换模型
• 面试表达顺序：业务场景 → 架构图 → 逐步选型 → 主动提优化 → 说评估
• 一句话：这是系统设计题不是知识题，展示工程判断力比背技术名词重要