3.1 RAG系统架构设计

3.1 RAG系统架构设计

导读：RAG（Retrieval-Augmented Generation，检索增强生成）是大模型应用中最核心的技术架构之一。本节将从系统架构的角度，全面介绍RAG系统的设计思路、核心组件及其协作方式。

3.1.1 RAG系统概述

什么是RAG

RAG是将信息检索与大语言模型生成能力结合的一种架构模式。它通过在生成回答之前先从知识库中检索相关信息，有效解决大模型的幻觉问题（Hallucination），并让模型能够基于特定领域的知识进行回答。

# RAG系统核心流程示意
class RAGPipeline:
    """RAG核心流程"""
    
    def query(self, user_question: str) -> str:
        """完整的RAG查询流程"""
        # Step 1: 用户问题预处理
        processed_query = self.preprocess(user_question)
        
        # Step 2: 向量化查询
        query_vector = self.embed(processed_query)
        
        # Step 3: 从知识库检索相关文档
        relevant_docs = self.retrieve(query_vector, top_k=5)
        
        # Step 4: 构建增强Prompt
        augmented_prompt = self.build_prompt(user_question, relevant_docs)
        
        # Step 5: LLM生成回答
        answer = self.generate(augmented_prompt)
        
        # Step 6: 后处理（可选）
        final_answer = self.postprocess(answer, relevant_docs)
        
        return final_answer
    
    def build_prompt(self, question: str, docs: list) -> str:
        """构建增强Prompt"""
        context = "\n".join(f"[{i+1}] {d}" for i, d in enumerate(docs))
        return f"""基于以下参考资料回答问题。

参考资料：
{context}

问题：{question}

请基于参考资料回答，如果参考资料中没有相关信息，请明确说明。"""
​

为什么需要RAG

大模型虽然拥有丰富的知识，但存在以下局限性，RAG正是为了解决这些问题：

1. 知识时效性：模型的知识截止于训练数据的时间点
2. 领域知识不足：通用模型在特定专业领域的知识有限
3. 幻觉问题：模型可能"编造"不存在的事实
4. 隐私合规：企业数据不能直接用于模型训练

3.1.2 RAG系统架构设计

整体架构

一个完整的RAG系统由以下核心层组成：

用户查询
    │
    ▼
┌─────────────────────────────────────────┐
│              应用层                       │
│  ┌─────────┐  ┌──────────┐  ┌────────┐ │
│  │ API网关  │→ │ 认证授权  │→ │ 路由    │ │
│  └─────────┘  └──────────┘  └────────┘ │
├─────────────────────────────────────────┤
│              处理层                       │
│  ┌──────────┐  ┌──────────┐  ┌────────┐ │
│  │ 查询预处理│→ │ 检索引擎  │→ │重排序   │ │
│  └──────────┘  └──────────┘  └────────┘ │
├─────────────────────────────────────────┤
│              增强层                       │
│  ┌──────────┐  ┌──────────┐  ┌────────┐ │
│  │ 上下文构建│→ │ Prompt组装 │→ │ LLM调用│ │
│  └──────────┘  └──────────┘  └────────┘ │
├─────────────────────────────────────────┤
│              数据层                       │
│  ┌──────────┐  ┌──────────┐  ┌────────┐ │
│  │ 向量数据库│  │ 文档存储  │→ │ 缓存层 │ │
│  └──────────┘  └──────────┘  └────────┘ │
└─────────────────────────────────────────┘
​

数据摄入管道

数据摄入是RAG系统的第一个关键环节，将原始文档转化为可检索的向量数据。

# 数据摄入管道
from typing import List, Dict
from pathlib import Path
import re

class DataIngestionPipeline:
    """文档数据摄入管道"""
    
    def __init__(self, embedder, vector_store, chunker):
        self.embedder = embedder
        self.vector_store = vector_store
        self.chunker = chunker
    
    def ingest_file(self, file_path: str) -> List[str]:
        """摄入单个文件"""
        content = self._read_file(file_path)
        cleaned = self._clean_text(content)
        chunks = self.chunker.split(cleaned)
        vectors = self.embedder.embed_batch(chunks)
        metadata = [{"source": file_path, "chunk_index": i} for i in range(len(chunks))]
        self.vector_store.upsert(vectors, chunks, metadata)
        return chunks
    
    def _read_file(self, file_path: str) -> str:
        """读取文件内容"""
        suffix = Path(file_path).suffix.lower()
        if suffix in [".txt", ".md"]:
            return Path(file_path).read_text(encoding="utf-8")
        elif suffix == ".pdf":
            import fitz
            doc = fitz.open(file_path)
            return "\n".join(page.get_text() for page in doc)
        else:
            raise ValueError(f"不支持的文件格式: {suffix}")
    
    def _clean_text(self, text: str) -> str:
        """文本清洗"""
        text = re.sub(r'\n{3,}', '\n\n', text)
        text = re.sub(r' {2,}', ' ', text)
        return text.strip()
​

文档分块策略

分块（Chunking）是RAG系统中极其重要的一步，分块质量直接影响检索效果。

# 多种分块策略实现
from typing import List

class DocumentChunker:
    """文档分块器，支持多种分块策略"""
    
    def __init__(self, chunk_size: int = 500, chunk_overlap: int = 50):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
    
    def split_by_length(self, text: str) -> List[Dict]:
        """按固定长度分块（带重叠）"""
        chunks = []
        start = 0
        while start < len(text):
            end = start + self.chunk_size
            chunk = text[start:end]
            chunks.append({"content": chunk, "start": start, "end": end})
            start = end - self.chunk_overlap
        return chunks
    
    def split_by_paragraph(self, text: str) -> List[Dict]:
        """按段落分块，合并过短的段落"""
        paragraphs = [p.strip() for p in text.split('\n\n') if p.strip()]
        chunks = []
        current_chunk = ""
        for para in paragraphs:
            if len(current_chunk) + len(para) <= self.chunk_size:
                current_chunk += f"\n\n{para}" if current_chunk else para
            else:
                if current_chunk:
                    chunks.append({"content": current_chunk})
                current_chunk = para
        if current_chunk:
            chunks.append({"content": current_chunk})
        return chunks
    
    def split_by_markdown(self, text: str) -> List[Dict]:
        """按Markdown标题分块"""
        sections = re.split(r'^(#{1,4}\s+.+)$', text, flags=re.MULTILINE)
        chunks = []
        for i in range(1, len(sections) - 1, 2):
            title = sections[i].strip()
            content = sections[i + 1].strip()
            chunks.append({"content": f"{title}\n{content}", "title": title})
        return chunks
    
    def split(self, text: str, strategy: str = "paragraph") -> List[str]:
        """统一分块接口"""
        strategies = {
            "length": self.split_by_length,
            "paragraph": self.split_by_paragraph,
            "markdown": self.split_by_markdown,
        }
        if strategy not in strategies:
            raise ValueError(f"未知的分块策略: {strategy}")
        result = strategies[strategy](text)
        return [chunk["content"] for chunk in result]
​

3.1.3 向量化与嵌入模型

嵌入模型选择

嵌入模型负责将文本转化为高维向量，是RAG系统的核心组件之一。

# 嵌入模型使用示例
from openai import OpenAI
from typing import List

class TextEmbedder:
    """文本向量化工具"""
    
    def __init__(self, api_key: str = None, model: str = "text-embedding-3-small"):
        self.client = OpenAI(api_key=api_key)
        self.model = model
    
    def embed(self, text: str) -> List[float]:
        """单条文本向量化"""
        response = self.client.embeddings.create(model=self.model, input=text)
        return response.data[0].embedding
    
    def embed_batch(self, texts: List[str], batch_size: int = 100) -> List[List[float]]:
        """批量文本向量化"""
        all_embeddings = []
        for i in range(0, len(texts), batch_size):
            batch = texts[i:i + batch_size]
            response = self.client.embeddings.create(model=self.model, input=batch)
            batch_embeddings = [item.embedding for item in response.data]
            all_embeddings.extend(batch_embeddings)
        return all_embeddings
​

3.1.4 检索引擎设计

混合检索策略

# 混合检索引擎
class HybridRetriever:
    """混合检索引擎，结合向量检索和关键词检索"""
    
    def __init__(self, vector_store, keyword_store, alpha: float = 0.7):
        self.vector_store = vector_store
        self.keyword_store = keyword_store
        self.alpha = alpha  # 向量检索权重
    
    def retrieve(self, query: str, query_vector: list, top_k: int = 10) -> List[Dict]:
        """混合检索"""
        vector_results = self.vector_store.search(query_vector, top_k=top_k * 2)
        keyword_results = self.keyword_store.search(query, top_k=top_k * 2)
        merged = self._reciprocal_rank_fusion(vector_results, keyword_results, self.alpha)
        return merged[:top_k]
    
    def _reciprocal_rank_fusion(self, results_a, results_b, alpha):
        """倒数排名融合"""
        scores = {}
        k = 60
        for rank, item in enumerate(results_a):
            scores[item["id"]] = scores.get(item["id"], 0) + alpha / (k + rank + 1)
        for rank, item in enumerate(results_b):
            scores[item["id"]] = scores.get(item["id"], 0) + (1 - alpha) / (k + rank + 1)
        return sorted(scores.items(), key=lambda x: x[1], reverse=True)
​

重排序机制

检索出候选文档后，使用重排序模型对结果进行精排。

# 重排序器
class Reranker:
    """检索结果重排序"""
    
    def __init__(self, rerank_model, top_n: int = 5):
        self.rerank_model = rerank_model
        self.top_n = top_n
    
    def rerank(self, query: str, documents: List[Dict]) -> List[Dict]:
        """对检索结果重排序"""
        scored_docs = []
        for doc in documents:
            score = self.rerank_model.compute_score(query, doc["content"])
            scored_docs.append({**doc, "rerank_score": score})
        scored_docs.sort(key=lambda x: x["rerank_score"], reverse=True)
        return scored_docs[:self.top_n]
​

3.1.5 完整RAG系统实现

# 完整的RAG系统
class RAGSystem:
    """完整的RAG系统"""
    
    def __init__(self, config):
        self.embedder = TextEmbedder(model=config["embedding_model"])
        self.chunker = DocumentChunker(
            chunk_size=config.get("chunk_size", 500),
            chunk_overlap=config.get("chunk_overlap", 50)
        )
        self.llm = OpenAI()
    
    def query(self, question: str) -> Dict:
        """完整的RAG查询"""
        import time
        start_time = time.time()
        
        # Step 1: 查询预处理（查询扩展、改写）
        expanded_query = self._expand_query(question)
        
        # Step 2: 向量化
        query_vector = self.embedder.embed(expanded_query)
        
        # Step 3: 检索
        candidates = self.retrieve(query_vector, top_k=10)
        
        # Step 4: 重排序
        reranked = self.rerank(expanded_query, candidates, top_n=5)
        
        # Step 5: 构建上下文
        context = "\n\n".join(
            f"参考资料{i+1}:\n{doc['content']}" 
            for i, doc in enumerate(reranked)
        )
        
        # Step 6: 生成回答
        prompt = f"""基于以下参考资料回答问题。

{context}

问题：{question}

请基于参考资料回答，并在回答末尾标注信息来源。"""
        
        response = self.llm.chat.completions.create(
            model="gpt-4-turbo-preview",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.3
        )
        
        answer = response.choices[0].message.content
        elapsed = time.time() - start_time
        
        return {
            "question": question,
            "answer": answer,
            "sources": [{"content": doc["content"][:100], "score": doc.get("rerank_score", 0)} for doc in reranked],
            "elapsed_seconds": round(elapsed, 2),
            "tokens_used": response.usage.total_tokens if response.usage else 0
        }
    
    def _expand_query(self, query: str) -> str:
        """查询扩展（简化版）"""
        return query
​

本章小结

通过本节的学习，你将：

✅ 理解RAG系统的核心概念和设计动机
✅ 掌握RAG系统的四层架构设计
✅ 学会数据摄入管道和文档分块策略
✅ 了解向量化、嵌入模型和检索引擎
✅ 能够构建完整的RAG查询系统

在3.2节中，我们将深入向量数据库的具体应用，在3.3节中探讨检索策略的优化方法。

[FAQ] 常见问题

Q: RAG系统和传统的搜索引擎有什么区别？
A: 传统搜索引擎返回文档列表让用户自己阅读，而RAG系统直接生成基于检索结果的精准回答。RAG结合了大模型的理解和生成能力，能够提供更自然、更有针对性的回答。

Q: 分块大小应该怎么选择？
A: 没有万能的最优值。一般建议在300-1000字符之间，取决于文档类型和应用场景。较短的块检索精度高但上下文少，较长的块上下文丰富但可能包含无关信息。建议通过实验找到最佳值。

Q: 如何评估RAG系统的效果？
A: 可以从多个维度评估：检索召回率、回答准确性、回答相关性、来源可追溯性等。常用的评估框架包括RAGAS、TruLens等。

Q: 向量数据库选型应该考虑什么？
A: 主要考虑：数据规模、查询延迟要求、是否需要过滤、部署方式（云/自托管）、成本等。小规模应用可以用ChromaDB或FAISS，大规模生产环境推荐Milvus或Qdrant。

Q: RAG系统能处理多语言吗？
A: 可以，但需要选择支持多语言的嵌入模型。OpenAI的text-embedding-3系列对中文支持较好，也可以考虑多语言专用模型如multilingual-e5。