RAG（Retrieval-Augmented Generation，检索增强生成）是当前最热门的AI应用架构之一，它通过结合外部知识库检索和大语言模型生成，解决了LLM的知识时效性和幻觉问题。本文将深入解析RAG的架构原理、工作流程和实现细节。

什么是RAG？

RAG的定义与核心思想

RAG是一种将信息检索与文本生成相结合的技术架构，其核心思想是：

• 不依赖模型内部的参数化知识
• 从外部知识库动态检索相关信息
• 将检索到的信息作为上下文提供给LLM
• 基于检索内容生成准确、可靠的答案

class RAGConcept:
    """
    RAG核心概念
    """
    
    def __init__(self):
        self.definition = "检索增强生成（RAG）= 信息检索 + 大语言模型生成"
        
        self.core_components = {
            '知识库': '存储领域知识的文档集合',
            '检索器': '根据查询找到相关文档',
            '生成器': '基于检索内容生成答案的LLM'
        }
        
        self.advantages = [
            '知识可更新：无需重新训练模型',
            '减少幻觉：答案基于真实文档',
            '可追溯性：可以引用信息来源',
            '成本更低：比微调模型成本低',
            '领域适配：快速适配特定领域'
        ]
    
    def show_comparison(self):
        """展示RAG与传统方法的对比"""
        print("RAG vs 传统方法对比")
        print("=" * 80)
        
        comparison = {
            '纯LLM': {
                '知识来源': '模型参数（训练时学到的知识）',
                '知识更新': '需要重新训练，成本高',
                '准确性': '可能产生幻觉，编造信息',
                '可追溯': '无法提供信息来源',
                '成本': '推理成本较低',
                '适用场景': '通用问答、创意生成'
            },
            '微调模型': {
                '知识来源': '微调数据注入的知识',
                '知识更新': '需要重新微调',
                '准确性': '在特定领域准确',
                '可追溯': '无法提供信息来源',
                '成本': '训练和存储成本高',
                '适用场景': '特定任务优化'
            },
            'RAG系统': {
                '知识来源': '外部知识库（可实时更新）',
                '知识更新': '更新文档即可，成本低',
                '准确性': '基于真实文档，准确性高',
                '可追溯': '可以引用原始文档',
                '成本': '检索+推理成本适中',
                '适用场景': '企业知识库、客服、文档问答'
            }
        }
        
        for method, features in comparison.items():
            print(f"
【{method}】")
            for key, value in features.items():
                print(f"  {key}: {value}")

concept = RAGConcept()
concept.show_comparison()

RAG解决的核心问题

class RAGProblems:
    """
    RAG解决的核心问题
    """
    
    def __init__(self):
        self.problems = [
            {
                'problem': '知识时效性问题',
                'description': 'LLM的知识截止于训练时间，无法获取最新信息',
                'example': '询问"2024年最新的AI政策"，模型无法回答',
                'rag_solution': 'RAG从实时更新的知识库检索最新文档'
            },
            {
                'problem': '幻觉问题',
                'description': 'LLM可能生成看似合理但实际错误的信息',
                'example': '编造不存在的论文引用、虚构的统计数据',
                'rag_solution': 'RAG基于真实文档生成，减少虚构内容'
            },
            {
                'problem': '领域知识不足',
                'description': 'LLM对特定领域的深度知识有限',
                'example': '企业内部流程、专业技术文档',
                'rag_solution': 'RAG可以接入领域专属知识库'
            },
            {
                'problem': '可追溯性缺失',
                'description': '无法验证信息来源和准确性',
                'example': '用户无法确认答案的可信度',
                'rag_solution': 'RAG可以返回引用的原始文档'
            },
            {
                'problem': '上下文长度限制',
                'description': 'LLM的上下文窗口有限，无法处理大量文档',
                'example': '需要分析数百页的技术文档',
                'rag_solution': 'RAG先检索相关片段，只传递必要信息'
            }
        ]
    
    def show_problems(self):
        """展示问题和解决方案"""
        print("
RAG解决的核心问题")
        print("=" * 80)
        
        for i, item in enumerate(self.problems, 1):
            print(f"
问题{i}：{item['problem']}")
            print(f"  描述：{item['description']}")
            print(f"  示例：{item['example']}")
            print(f"  ✅ RAG方案：{item['rag_solution']}")

problems = RAGProblems()
problems.show_problems()

RAG架构详解

基础架构

class RAGArchitecture:
    """
    RAG基础架构
    """
    
    def __init__(self):
        self.components = {
            '离线索引阶段': {
                '文档加载': '从各种来源加载文档（PDF、Word、网页等）',
                '文档切分': '将长文档切分为小块（Chunks）',
                '向量化': '使用Embedding模型将文本转为向量',
                '存储': '将向量存入向量数据库'
            },
            '在线检索阶段': {
                '查询处理': '理解和优化用户查询',
                '查询向量化': '将查询转为向量',
                '相似度检索': '在向量数据库中检索相似文档',
                '重排序': '对检索结果进行精排',
                '上下文构建': '将检索内容整合到Prompt中'
            },
            '生成阶段': {
                'Prompt构建': '组合查询和检索内容',
                'LLM调用': '调用大语言模型生成答案',
                '后处理': '格式化、引用标注等'
            }
        }
    
    def show_architecture(self):
        """展示架构组件"""
        print("
RAG架构组件")
        print("=" * 80)
        
        for stage, components in self.components.items():
            print(f"
【{stage}】")
            for comp, desc in components.items():
                print(f"  • {comp}: {desc}")

arch = RAGArchitecture()
arch.show_architecture()

核心组件详解

1. 文档处理器

class DocumentProcessor:
    """
    文档处理器：负责加载和预处理文档
    """
    
    def __init__(self):
        self.supported_formats = ['PDF', 'DOCX', 'TXT', 'HTML', 'Markdown']
        
    def load_document(self, file_path, file_type):
        """
        加载文档
        
        Args:
            file_path: 文件路径
            file_type: 文件类型
        
        Returns:
            文档内容
        """
        print(f"加载文档: {file_path} (类型: {file_type})")
        
        # 根据文件类型选择加载器
        loaders = {
            'PDF': 'PyPDFLoader',
            'DOCX': 'Docx2txtLoader',
            'TXT': 'TextLoader',
            'HTML': 'UnstructuredHTMLLoader',
            'Markdown': 'UnstructuredMarkdownLoader'
        }
        
        loader = loaders.get(file_type, 'TextLoader')
        print(f"  使用加载器: {loader}")
        
        # 模拟加载
        return f"文档内容来自 {file_path}"
    
    def clean_text(self, text):
        """
        清洗文本
        
        Args:
            text: 原始文本
        
        Returns:
            清洗后的文本
        """
        steps = [
            '移除多余空白字符',
            '统一换行符',
            '移除特殊字符',
            '修正编码问题'
        ]
        
        print("
文本清洗步骤:")
        for step in steps:
            print(f"  ✓ {step}")
        
        return text
    
    def extract_metadata(self, document):
        """
        提取元数据
        
        Args:
            document: 文档对象
        
        Returns:
            元数据字典
        """
        metadata = {
            'title': '文档标题',
            'author': '作者',
            'created_date': '创建日期',
            'source': '来源',
            'page_count': '页数',
            'language': '语言'
        }
        
        print("
提取的元数据:")
        for key, value in metadata.items():
            print(f"  {key}: {value}")
        
        return metadata

# 演示
processor = DocumentProcessor()
doc = processor.load_document('/path/to/doc.pdf', 'PDF')
cleaned = processor.clean_text(doc)
metadata = processor.extract_metadata(doc)

2. 文档切分器

class TextSplitter:
    """
    文档切分器：将长文档切分为小块
    """
    
    def __init__(self, chunk_size=500, chunk_overlap=50):
        """
        Args:
            chunk_size: 每个块的大小（字符数或token数）
            chunk_overlap: 块之间的重叠大小
        """
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
    
    def split_by_characters(self, text):
        """
        按字符数切分
        
        Args:
            text: 待切分文本
        
        Returns:
            文本块列表
        """
        chunks = []
        start = 0
        
        while start < len(text):
            end = start + self.chunk_size
            chunk = text[start:end]
            chunks.append(chunk)
            start = end - self.chunk_overlap
        
        print(f"
按字符切分:")
        print(f"  原文长度: {len(text)} 字符")
        print(f"  块大小: {self.chunk_size} 字符")
        print(f"  重叠: {self.chunk_overlap} 字符")
        print(f"  生成块数: {len(chunks)}")
        
        return chunks
    
    def split_by_sentences(self, text):
        """
        按句子切分（保持语义完整性）
        
        Args:
            text: 待切分文本
        
        Returns:
            文本块列表
        """
        # 简化示例：按句号切分
        sentences = text.split('。')
        chunks = []
        current_chunk = ""
        
        for sentence in sentences:
            if len(current_chunk) + len(sentence) < self.chunk_size:
                current_chunk += sentence + "。"
            else:
                if current_chunk:
                    chunks.append(current_chunk)
                current_chunk = sentence + "。"
        
        if current_chunk:
            chunks.append(current_chunk)
        
        print(f"
按句子切分:")
        print(f"  句子数: {len(sentences)}")
        print(f"  生成块数: {len(chunks)}")
        
        return chunks
    
    def split_by_semantic(self, text):
        """
        按语义切分（使用语义相似度）
        
        Args:
            text: 待切分文本
        
        Returns:
            文本块列表
        """
        print(f"
按语义切分:")
        print(f"  方法: 使用Embedding计算句子相似度")
        print(f"  策略: 相似度低的地方作为切分点")
        print(f"  优势: 保持语义连贯性")
        
        # 实际实现需要使用Embedding模型
        return ["语义块1", "语义块2", "语义块3"]
    
    def show_splitting_strategies(self):
        """展示不同切分策略"""
        print("
文档切分策略对比")
        print("=" * 80)
        
        strategies = {
            '固定长度切分': {
                '方法': '按固定字符数或token数切分',
                '优点': '简单快速，实现容易',
                '缺点': '可能破坏语义完整性',
                '适用': '对语义要求不高的场景'
            },
            '句子边界切分': {
                '方法': '在句子边界处切分',
                '优点': '保持句子完整性',
                '缺点': '块大小不均匀',
                '适用': '需要保持语义的场景'
            },
            '段落切分': {
                '方法': '按段落切分',
                '优点': '保持段落主题完整',
                '缺点': '段落长度差异大',
                '适用': '结构化文档'
            },
            '语义切分': {
                '方法': '基于语义相似度切分',
                '优点': '最佳语义连贯性',
                '缺点': '计算成本高',
                '适用': '高质量要求的场景'
            },
            '递归切分': {
                '方法': '先按段落，再按句子，最后按字符',
                '优点': '平衡语义和大小',
                '缺点': '实现复杂',
                '适用': '通用场景'
            }
        }
        
        for strategy, info in strategies.items():
            print(f"
【{strategy}】")
            for key, value in info.items():
                print(f"  {key}: {value}")

# 演示
splitter = TextSplitter(chunk_size=500, chunk_overlap=50)
splitter.show_splitting_strategies()

# 示例文本
sample_text = "这是一个示例文本。" * 100
chunks = splitter.split_by_characters(sample_text)

3. 向量化器（Embedder）

class Embedder:
    """
    向量化器：将文本转换为向量
    """
    
    def __init__(self, model_name='text-embedding-ada-002'):
        """
        Args:
            model_name: Embedding模型名称
        """
        self.model_name = model_name
        self.dimension = self._get_dimension()
    
    def _get_dimension(self):
        """获取向量维度"""
        dimensions = {
            'text-embedding-ada-002': 1536,  # OpenAI
            'text-embedding-3-small': 1536,
            'text-embedding-3-large': 3072,
            'bge-large-zh': 1024,  # 智源
            'm3e-base': 768,  # M3E
            'sentence-transformers': 768
        }
        return dimensions.get(self.model_name, 768)
    
    def embed_text(self, text):
        """
        将文本转为向量
        
        Args:
            text: 输入文本
        
        Returns:
            向量（列表）
        """
        print(f"
向量化文本:")
        print(f"  模型: {self.model_name}")
        print(f"  文本长度: {len(text)} 字符")
        print(f"  向量维度: {self.dimension}")
        
        # 实际实现会调用Embedding API或模型
        # 这里返回模拟向量
        import random
        vector = [random.random() for _ in range(self.dimension)]
        
        return vector
    
    def embed_batch(self, texts):
        """
        批量向量化
        
        Args:
            texts: 文本列表
        
        Returns:
            向量列表
        """
        print(f"
批量向量化:")
        print(f"  文本数量: {len(texts)}")
        print(f"  批次大小: 建议32-128")
        
        vectors = [self.embed_text(text) for text in texts]
        return vectors
    
    def show_embedding_models(self):
        """展示常用Embedding模型"""
        print("
常用Embedding模型对比")
        print("=" * 80)
        
        models = {
            'OpenAI text-embedding-ada-002': {
                '维度': 1536,
                '语言': '多语言',
                '性能': '优秀',
                '成本': '$0.0001/1K tokens',
                '优点': '质量高，多语言支持好',
                '缺点': '需要API调用，有成本'
            },
            'OpenAI text-embedding-3-large': {
                '维度': 3072,
                '语言': '多语言',
                '性能': '最优',
                '成本': '$0.00013/1K tokens',
                '优点': '最新模型，性能最佳',
                '缺点': '维度大，存储成本高'
            },
            'BGE-large-zh（智源）': {
                '维度': 1024,
                '语言': '中文优化',
                '性能': '优秀',
                '成本': '开源免费',
                '优点': '中文效果好，可本地部署',
                '缺点': '需要GPU资源'
            },
            'M3E': {
                '维度': 768,
                '语言': '中文',
                '性能': '良好',
                '成本': '开源免费',
                '优点': '轻量级，中文效果好',
                '缺点': '多语言支持一般'
            },
            'Sentence-Transformers': {
                '维度': 768,
                '语言': '多语言',
                '性能': '良好',
                '成本': '开源免费',
                '优点': '生态丰富，模型多',
                '缺点': '需要选择合适的模型'
            }
        }
        
        for model, info in models.items():
            print(f"
【{model}】")
            for key, value in info.items():
                print(f"  {key}: {value}")

# 演示
embedder = Embedder('text-embedding-ada-002')
embedder.show_embedding_models()

text = "RAG是一种检索增强生成技术"
vector = embedder.embed_text(text)
print(f"
向量示例（前10维）: {vector[:10]}")

4. 检索器

class Retriever:
    """
    检索器：从向量数据库检索相关文档
    """
    
    def __init__(self, top_k=5, similarity_threshold=0.7):
        """
        Args:
            top_k: 返回最相关的K个文档
            similarity_threshold: 相似度阈值
        """
        self.top_k = top_k
        self.similarity_threshold = similarity_threshold
    
    def retrieve(self, query_vector, vector_db):
        """
        检索相关文档
        
        Args:
            query_vector: 查询向量
            vector_db: 向量数据库
        
        Returns:
            检索结果列表
        """
        print(f"
执行检索:")
        print(f"  Top-K: {self.top_k}")
        print(f"  相似度阈值: {self.similarity_threshold}")
        
        # 模拟检索结果
        results = [
            {
                'id': 'doc_1',
                'text': 'RAG技术结合了检索和生成...',
                'score': 0.95,
                'metadata': {'source': 'doc1.pdf', 'page': 1}
            },
            {
                'id': 'doc_2',
                'text': '向量数据库用于存储文档向量...',
                'score': 0.88,
                'metadata': {'source': 'doc2.pdf', 'page': 3}
            },
            {
                'id': 'doc_3',
                'text': 'Embedding模型将文本转为向量...',
                'score': 0.82,
                'metadata': {'source': 'doc1.pdf', 'page': 5}
            }
        ]
        
        # 过滤低于阈值的结果
        filtered_results = [
            r for r in results 
            if r['score'] >= self.similarity_threshold
        ]
        
        print(f"  检索到 {len(filtered_results)} 个相关文档")
        
        return filtered_results[:self.top_k]
    
    def show_retrieval_strategies(self):
        """展示检索策略"""
        print("
检索策略")
        print("=" * 80)
        
        strategies = {
            '密集检索（Dense Retrieval）': {
                '方法': '使用向量相似度检索',
                '优点': '语义理解好，召回率高',
                '缺点': '计算成本高',
                '实现': 'FAISS, Milvus, Pinecone'
            },
            '稀疏检索（Sparse Retrieval）': {
                '方法': '基于关键词匹配（BM25）',
                '优点': '精确匹配，速度快',
                '缺点': '语义理解弱',
                '实现': 'Elasticsearch, BM25'
            },
            '混合检索（Hybrid Retrieval）': {
                '方法': '结合密集和稀疏检索',
                '优点': '兼顾语义和精确匹配',
                '缺点': '实现复杂',
                '实现': '自定义融合策略'
            },
            '重排序（Reranking）': {
                '方法': '对初步检索结果精排',
                '优点': '提高Top-K准确性',
                '缺点': '增加延迟',
                '实现': 'Cross-Encoder模型'
            }
        }
        
        for strategy, info in strategies.items():
            print(f"
【{strategy}】")
            for key, value in info.items():
                print(f"  {key}: {value}")

# 演示
retriever = Retriever(top_k=5, similarity_threshold=0.7)
retriever.show_retrieval_strategies()

5. 生成器

class Generator:
    """
    生成器：基于检索内容生成答案
    """
    
    def __init__(self, model='gpt-3.5-turbo'):
        """
        Args:
            model: LLM模型名称
        """
        self.model = model
    
    def build_prompt(self, query, retrieved_docs):
        """
        构建Prompt
        
        Args:
            query: 用户查询
            retrieved_docs: 检索到的文档
        
        Returns:
            完整的Prompt
        """
        # 整合检索内容
        context = "

".join([
            f"文档{i+1}（相似度：{doc['score']:.2f}）:
{doc['text']}"
            for i, doc in enumerate(retrieved_docs)
        ])
        
        # 构建Prompt
        prompt = f"""请基于以下参考文档回答用户问题。

参考文档：
{context}

用户问题：{query}

要求：
1. 答案必须基于参考文档
2. 如果文档中没有相关信息，明确说明
3. 引用具体的文档来源
4. 保持客观准确

答案："""
        
        print("
构建的Prompt:")
        print("=" * 80)
        print(prompt)
        print("=" * 80)
        
        return prompt
    
    def generate(self, prompt):
        """
        生成答案
        
        Args:
            prompt: 输入Prompt
        
        Returns:
            生成的答案
        """
        print(f"
调用LLM生成答案:")
        print(f"  模型: {self.model}")
        print(f"  Prompt长度: {len(prompt)} 字符")
        
        # 实际实现会调用LLM API
        answer = """根据参考文档，RAG（检索增强生成）是一种结合了信息检索和文本生成的技术架构。
它通过从外部知识库检索相关文档，然后将这些文档作为上下文提供给大语言模型来生成答案。

主要优势包括：
1. 知识可实时更新，无需重新训练模型
2. 减少模型幻觉，答案基于真实文档
3. 可以追溯信息来源

（参考：文档1，相似度0.95）"""
        
        return answer
    
    def show_prompt_templates(self):
        """展示Prompt模板"""
        print("
Prompt模板示例")
        print("=" * 80)
        
        templates = {
            '基础模板': """参考文档：
{context}

问题：{query}

请基于参考文档回答问题。""",
            
            '带引用模板': """参考文档：
{context}

问题：{query}

要求：
1. 基于参考文档回答
2. 标注信息来源
3. 如果文档中没有答案，明确说明

答案：""",
            
            '结构化输出模板': """参考文档：
{context}

问题：{query}

请按以下格式回答：
## 答案
[你的答案]

## 依据
[引用的文档片段]

## 来源
[文档来源]

## 置信度
[高/中/低]""",
            
            '对话式模板': """你是一个专业的AI助手，请基于以下知识库内容回答用户问题。

知识库内容：
{context}

对话历史：
{chat_history}

用户问题：{query}

请提供准确、有帮助的回答。"""
        }
        
        for name, template in templates.items():
            print(f"
【{name}】")
            print(template)
            print("-" * 60)

# 演示
generator = Generator('gpt-3.5-turbo')
generator.show_prompt_templates()

RAG工作流程

完整流程示例

class RAGPipeline:
    """
    完整的RAG流程
    """
    
    def __init__(self):
        self.document_processor = DocumentProcessor()
        self.text_splitter = TextSplitter(chunk_size=500, chunk_overlap=50)
        self.embedder = Embedder('text-embedding-ada-002')
        self.retriever = Retriever(top_k=5, similarity_threshold=0.7)
        self.generator = Generator('gpt-3.5-turbo')
    
    def index_documents(self, documents):
        """
        索引文档（离线阶段）
        
        Args:
            documents: 文档列表
        
        Returns:
            索引结果
        """
        print("
" + "="*80)
        print("阶段1：文档索引（离线）")
        print("="*80)
        
        all_chunks = []
        
        for doc in documents:
            print(f"
处理文档: {doc}")
            
            # 1. 加载文档
            content = self.document_processor.load_document(doc, 'PDF')
            
            # 2. 清洗文本
            cleaned = self.document_processor.clean_text(content)
            
            # 3. 切分文档
            chunks = self.text_splitter.split_by_characters(cleaned)
            
            # 4. 向量化
            print(f"
向量化 {len(chunks)} 个文本块...")
            vectors = self.embedder.embed_batch(chunks)
            
            # 5. 存储到向量数据库
            print(f"存储到向量数据库...")
            
            all_chunks.extend(chunks)
        
        print(f"
✅ 索引完成！共处理 {len(all_chunks)} 个文本块")
        return all_chunks
    
    def query(self, question):
        """
        查询（在线阶段）
        
        Args:
            question: 用户问题
        
        Returns:
            答案
        """
        print("
" + "="*80)
        print("阶段2：查询处理（在线）")
        print("="*80)
        
        print(f"
用户问题: {question}")
        
        # 1. 查询向量化
        print("
步骤1: 查询向量化")
        query_vector = self.embedder.embed_text(question)
        
        # 2. 检索相关文档
        print("
步骤2: 检索相关文档")
        retrieved_docs = self.retriever.retrieve(query_vector, None)
        
        print(f"
检索结果:")
        for i, doc in enumerate(retrieved_docs, 1):
            print(f"  {i}. [{doc['score']:.2f}] {doc['text'][:50]}...")
        
        # 3. 构建Prompt
        print("
步骤3: 构建Prompt")
        prompt = self.generator.build_prompt(question, retrieved_docs)
        
        # 4. 生成答案
        print("
步骤4: 生成答案")
        answer = self.generator.generate(prompt)
        
        print("
" + "="*80)
        print("最终答案:")
        print("="*80)
        print(answer)
        
        return answer
    
    def show_complete_workflow(self):
        """展示完整工作流程"""
        print("
RAG完整工作流程")
        print("=" * 80)
        
        workflow = """
【离线阶段：文档索引】
1. 文档加载
   ├─ 支持多种格式（PDF、Word、HTML等）
   ├─ 提取文本内容
   └─ 提取元数据

2. 文档预处理
   ├─ 文本清洗
   ├─ 格式统一
   └─ 去除噪声

3. 文档切分
   ├─ 选择切分策略
   ├─ 设置块大小和重叠
   └─ 生成文本块

4. 向量化
   ├─ 选择Embedding模型
   ├─ 批量向量化
   └─ 生成向量表示

5. 存储
   ├─ 选择向量数据库
   ├─ 创建索引
   └─ 存储向量和元数据

【在线阶段：查询处理】
1. 查询理解
   ├─ 查询预处理
   ├─ 查询扩展（可选）
   └─ 查询向量化

2. 检索
   ├─ 向量相似度检索
   ├─ 过滤和排序
   └─ 返回Top-K结果

3. 重排序（可选）
   ├─ 使用重排序模型
   ├─ 精确评分
   └─ 优化Top-K

4. Prompt构建
   ├─ 整合检索内容
   ├─ 添加指令
   └─ 格式化Prompt

5. 生成答案
   ├─ 调用LLM
   ├─ 生成答案
   └─ 后处理

6. 返回结果
   ├─ 答案
   ├─ 引用来源
   └─ 置信度（可选）
"""
        print(workflow)

# 演示完整流程
pipeline = RAGPipeline()
pipeline.show_complete_workflow()

# 模拟索引文档
documents = ['doc1.pdf', 'doc2.pdf', 'doc3.pdf']
pipeline.index_documents(documents)

# 模拟查询
question = "什么是RAG技术？"
answer = pipeline.query(question)

RAG的变体与优化

class RAGVariants:
    """
    RAG的变体和优化方法
    """
    
    def show_variants(self):
        """展示RAG变体"""
        print("
RAG变体")
        print("=" * 80)
        
        variants = {
            'Naive RAG（基础RAG）': {
                '描述': '最基本的RAG实现',
                '流程': '检索 → 生成',
                '优点': '简单直接，易于实现',
                '缺点': '检索质量依赖查询质量',
                '适用': '简单问答场景'
            },
            'Advanced RAG（高级RAG）': {
                '描述': '增加预处理和后处理',
                '流程': '查询优化 → 检索 → 重排序 → 生成',
                '优点': '检索质量更高',
                '缺点': '复杂度增加',
                '适用': '对准确性要求高的场景'
            },
            'Modular RAG（模块化RAG）': {
                '描述': '可自由组合的模块化架构',
                '流程': '灵活组合各种模块',
                '优点': '高度可定制',
                '缺点': '需要专业知识',
                '适用': '复杂业务场景'
            },
            'Self-RAG': {
                '描述': 'LLM自我反思和纠正',
                '流程': '检索 → 生成 → 自我评估 → 重新检索/生成',
                '优点': '答案质量更高',
                '缺点': '多次调用LLM，成本高',
                '适用': '高质量要求场景'
            },
            'Corrective RAG（CRAG）': {
                '描述': '纠正性RAG，评估检索质量',
                '流程': '检索 → 评估 → 纠正 → 生成',
                '优点': '减少错误检索的影响',
                '缺点': '需要额外的评估模型',
                '适用': '检索质量不稳定的场景'
            }
        }
        
        for variant, info in variants.items():
            print(f"
【{variant}】")
            for key, value in info.items():
                print(f"  {key}: {value}")
    
    def show_optimization_techniques(self):
        """展示优化技术"""
        print("

RAG优化技术")
        print("=" * 80)
        
        techniques = {
            '查询优化': [
                'Query Rewriting: 重写查询以提高检索效果',
                'Query Expansion: 扩展查询词，增加召回',
                'Query Decomposition: 将复杂查询分解为子查询',
                'HyDE: 生成假设性文档，用于检索'
            ],
            '检索优化': [
                'Hybrid Search: 结合密集和稀疏检索',
                'Reranking: 使用重排序模型精排',
                'Metadata Filtering: 基于元数据过滤',
                'Recursive Retrieval: 递归检索相关内容'
            ],
            '上下文优化': [
                'Context Compression: 压缩检索内容',
                'Context Selection: 智能选择最相关片段',
                'Context Reordering: 重新排序上下文',
                'Sliding Window: 滑动窗口处理长文档'
            ],
            '生成优化': [
                'Prompt Engineering: 优化Prompt模板',
                'Few-shot Examples: 添加示例',
                'Chain-of-Thought: 引导推理过程',
                'Self-Consistency: 多次生成取最佳'
            ]
        }
        
        for category, items in techniques.items():
            print(f"
【{category}】")
            for item in items:
                print(f"  • {item}")

# 演示
variants = RAGVariants()
variants.show_variants()
variants.show_optimization_techniques()

总结

本文深入解析了RAG的架构和工作流程：

核心要点

1. RAG定义：检索增强生成 = 信息检索 + LLM生成
2. 核心优势：知识可更新、减少幻觉、可追溯、成本低
3. 关键组件：文档处理器、切分器、向量化器、检索器、生成器
4. 工作流程：离线索引 + 在线检索生成

架构组件

• 文档处理：加载、清洗、元数据提取
• 文档切分：固定长度、句子边界、语义切分
• 向量化：Embedding模型选择和使用
• 检索：密集检索、稀疏检索、混合检索
• 生成：Prompt构建和LLM调用

RAG变体

• Naive RAG：基础实现
• Advanced RAG：增加优化步骤
• Modular RAG：模块化架构
• Self-RAG：自我反思
• Corrective RAG：纠正性检索

下一步

在下一篇文章中，我们将详细对比各种向量数据库（Milvus、Pinecone、Chroma等），帮助你选择最适合的方案。

---

相关文章：

• 10.1 RAG架构与工作流程（本文）
• 10.2 向量数据库选择（Milvus、Pinecone、Chroma等）
• 10.3 文档切分、向量化与检索优化
• 10.4 完整实战：构建企业级知识库问答系统