企业级RAG系统架构与实战完全教程

教程简介

检索增强生成（Retrieval-Augmented Generation, RAG）已成为企业级AI应用的核心架构模式。本教程从RAG的基础原理出发，系统讲解企业级RAG系统的架构设计、核心组件、实战部署与运维优化，帮助开发者构建生产级别的智能检索系统。

第一章：RAG架构演进与设计原则

1.1 从朴素RAG到高级RAG

RAG的核心思想是将外部知识库与大语言模型（LLM）结合，通过检索相关文档来增强模型的生成能力。

朴素RAG（Naive RAG） 的基本流程：

用户查询 → 文档检索 → 拼接上下文 → LLM生成 → 返回结果

高级RAG（Advanced RAG） 在此基础上增加了查询优化、多路检索、重排序、幻觉检测等环节：

用户查询 → 查询理解/改写 → 多路检索 → 重排序 → 上下文压缩 → LLM生成 → 事实性验证 → 返回结果

模块化RAG（Modular RAG） 将系统拆分为可组合的独立模块：

class ModularRAG:
    """模块化RAG架构"""
    def __init__(self):
        self.query_processor = QueryProcessor()    # 查询处理
        self.retriever = HybridRetriever()         # 混合检索
        self.reranker = CrossEncoderReranker()     # 重排序
        self.context_manager = ContextManager()     # 上下文管理
        self.generator = LLMGenerator()            # 生成
        self.validator = HallucinationValidator()   # 幻觉验证
        self.memory = ConversationMemory()         # 对话记忆

1.2 企业级RAG的特殊挑战

企业场景与Demo级RAG有本质区别：

第二章：多模态文档解析

2.1 PDF文档解析

企业文档大量以PDF格式存在，解析质量直接影响RAG效果。

# 使用 PyMuPDF4LLM 进行高质量PDF解析
import pymupdf4llm

def parse_pdf_with_pymupdf(pdf_path: str) -> str:
    """将PDF转换为Markdown格式，保留结构信息"""
    md_text = pymupdf4llm.to_markdown(
        pdf_path,
        page_chunks=True,       # 按页分块
        write_images=False,     # 不提取图片
        image_path="images/",
        dpi=150
    )
    return md_text


# 使用 Unstructured 进行复杂PDF解析
from unstructured.partition.pdf import partition_pdf

def parse_complex_pdf(pdf_path: str) -> list:
    """解析包含表格、图片的复杂PDF"""
    elements = partition_pdf(
        filename=pdf_path,
        strategy="hi_res",           # 高精度模式
        model_name="yolox",          # 版面分析模型
        extract_images_in_pdf=True,
        infer_table_structure=True,
        chunking_strategy="by_title",
        max_characters=1000,
        new_after_n_chars=800,
        overlap=200
    )
    return elements

2.2 表格数据处理

表格是企业文档中最常见也最难处理的结构。

import pandas as pd
from io import StringIO

class TableProcessor:
    """表格处理器：将表格转换为LLM可理解的文本"""
    
    def table_to_natural_language(self, table: pd.DataFrame, context: str = "") -> str:
        """将表格转换为自然语言描述"""
        descriptions = []
        
        # 表头信息
        headers = ", ".join(table.columns.tolist())
        descriptions.append(f"该表格包含以下列：{headers}")
        
        # 逐行描述
        for idx, row in table.iterrows():
            row_desc = []
            for col in table.columns:
                row_desc.append(f"{col}为{row[col]}")
            descriptions.append(f"第{idx+1}行：{'，'.join(row_desc)}")
        
        return "\n".join(descriptions)
    
    def table_to_markdown(self, table: pd.DataFrame) -> str:
        """将DataFrame转为Markdown表格"""
        return table.to_markdown(index=False)
    
    def smart_chunk_table(self, table: pd.DataFrame, max_rows: int = 20) -> list:
        """大表格智能分块"""
        chunks = []
        for i in range(0, len(table), max_rows):
            chunk = table.iloc[i:i+max_rows]
            chunks.append(self.table_to_markdown(chunk))
        return chunks

2.3 图片与多模态内容

import base64

class ImageProcessor:
    """图片处理器：提取图片中的文字和语义信息"""
    
    def __init__(self, vlm_client):
        self.vlm = vlm_client  # 视觉语言模型
    
    async def extract_from_image(self, image_path: str) -> dict:
        """使用VLM提取图片中的文字和语义"""
        with open(image_path, "rb") as f:
            img_base64 = base64.b64encode(f.read()).decode()
        
        prompt = """请分析这张图片，提取以下信息：
1. 图片中的所有文字内容（OCR）
2. 图片的类型（图表/流程图/截图/照片等）
3. 图片所表达的核心信息
4. 如果是图表，请描述数据趋势和关键数值
        
请以JSON格式返回结果。"""
        
        result = await self.vlm.chat(
            messages=[{
                "role": "user",
                "content": [
                    {"type": "text", "text": prompt},
                    {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{img_base64}"}}
                ]
            }]
        )
        return result

第三章：Chunking策略与优化

3.1 常见分块策略

Chunking是RAG系统中影响检索质量的关键环节。

from typing import List

class TextChunker:
    """文本分块器"""
    
    def fixed_size_chunk(self, text: str, chunk_size: int = 512, overlap: int = 50) -> List[str]:
        """固定大小分块"""
        chunks = []
        start = 0
        while start < len(text):
            end = start + chunk_size
            chunks.append(text[start:end])
            start = end - overlap
        return chunks
    
    def recursive_split(self, text: str, chunk_size: int = 1000, 
                        separators: List[str] = None) -> List[str]:
        """递归字符分割：按语义边界分块"""
        if separators is None:
            separators = ["\n\n", "\n", "。", "！", "？", ".", "!", "?", " "]
        
        chunks = []
        current_chunks = [text]
        
        for sep in separators:
            new_chunks = []
            for chunk in current_chunks:
                if len(chunk) <= chunk_size:
                    new_chunks.append(chunk)
                else:
                    splits = chunk.split(sep)
                    current = ""
                    for split in splits:
                        if len(current) + len(split) + len(sep) <= chunk_size:
                            current = current + sep + split if current else split
                        else:
                            if current:
                                new_chunks.append(current)
                            current = split
                    if current:
                        new_chunks.append(current)
            current_chunks = new_chunks
        
        return [c.strip() for c in current_chunks if c.strip()]
    
    def semantic_chunk(self, text: str, embedding_model, threshold: float = 0.5) -> List[str]:
        """语义分块：基于嵌入相似度的智能分块"""
        import numpy as np
        
        # 按句子分割
        sentences = self._split_sentences(text)
        
        # 计算每个句子的嵌入
        embeddings = embedding_model.encode(sentences)
        
        # 计算相邻句子的相似度
        similarities = []
        for i in range(len(embeddings) - 1):
            sim = np.dot(embeddings[i], embeddings[i+1]) / (
                np.linalg.norm(embeddings[i]) * np.linalg.norm(embeddings[i+1])
            )
            similarities.append(sim)
        
        # 在相似度低于阈值处切分
        chunks = []
        current_chunk = [sentences[0]]
        for i, sim in enumerate(similarities):
            if sim < threshold:
                chunks.append(" ".join(current_chunk))
                current_chunk = [sentences[i+1]]
            else:
                current_chunk.append(sentences[i+1])
        
        if current_chunk:
            chunks.append(" ".join(current_chunk))
        
        return chunks

3.2 高级分块：Parent-Child策略

class ParentChildChunker:
    """父子分块策略：检索小块，返回大块"""
    
    def __init__(self, child_size=256, parent_size=1024, overlap=64):
        self.child_size = child_size
        self.parent_size = parent_size
        self.overlap = overlap
    
    def chunk(self, text: str) -> dict:
        """生成父子块"""
        # 生成大块（父块）
        parent_chunks = self._split(text, self.parent_size, self.overlap)
        
        result = {"parents": [], "children": []}
        
        for parent_idx, parent in enumerate(parent_chunks):
            result["parents"].append({
                "id": f"parent_{parent_idx}",
                "text": parent,
                "metadata": {"parent_id": f"parent_{parent_idx}"}
            })
            
            # 将大块拆分为小块（子块）
            child_chunks = self._split(parent, self.child_size, self.overlap // 2)
            for child_idx, child in enumerate(child_chunks):
                result["children"].append({
                    "id": f"child_{parent_idx}_{child_idx}",
                    "text": child,
                    "parent_id": f"parent_{parent_idx}",
                    "metadata": {"parent_id": f"parent_{parent_idx}"}
                })
        
        return result
    
    def _split(self, text, size, overlap):
        """按句子边界分块"""
        sentences = text.replace("。", "。\n").replace("！", "！\n").replace("？", "？\n").split("\n")
        chunks = []
        current = ""
        for sent in sentences:
            if len(current) + len(sent) <= size:
                current += sent
            else:
                if current:
                    chunks.append(current.strip())
                current = sent
        if current:
            chunks.append(current.strip())
        return chunks

第四章：向量数据库选型与实战

4.1 主流向量数据库对比

4.2 Milvus实战

from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType, utility

class MilvusVectorStore:
    """Milvus向量数据库操作封装"""
    
    def __init__(self, host="localhost", port="19530"):
        connections.connect(host=host, port=port)
        self.collection = None
    
    def create_collection(self, name: str, dim: int = 1536):
        """创建集合"""
        fields = [
            FieldSchema(name="id", dtype=DataType.VARCHAR, is_primary=True, max_length=64),
            FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=65535),
            FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=dim),
            FieldSchema(name="metadata", dtype=DataType.JSON),
        ]
        schema = CollectionSchema(fields, description="RAG document chunks")
        self.collection = Collection(name, schema)
        
        # 创建向量索引
        index_params = {
            "metric_type": "COSINE",
            "index_type": "HNSW",
            "params": {"M": 16, "efConstruction": 256}
        }
        self.collection.create_index("embedding", index_params)
        return self.collection
    
    def insert(self, ids: list, texts: list, embeddings: list, metadata: list):
        """批量插入数据"""
        data = [ids, texts, embeddings, metadata]
        self.collection.insert(data)
        self.collection.flush()
    
    def search(self, query_embedding: list, top_k: int = 10, 
               filter_expr: str = None) -> list:
        """向量检索"""
        self.collection.load()
        
        search_params = {"metric_type": "COSINE", "params": {"ef": 128}}
        
        results = self.collection.search(
            data=[query_embedding],
            anns_field="embedding",
            param=search_params,
            limit=top_k,
            expr=filter_expr,
            output_fields=["text", "metadata"]
        )
        
        return [
            {
                "id": hit.id,
                "text": hit.entity.get("text"),
                "score": hit.score,
                "metadata": hit.entity.get("metadata")
            }
            for hit in results[0]
        ]

4.3 Qdrant实战

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct, Filter, FieldCondition

class QdrantVectorStore:
    """Qdrant向量数据库操作封装"""
    
    def __init__(self, host="localhost", port=6333):
        self.client = QdrantClient(host=host, port=port)
    
    def create_collection(self, name: str, dim: int = 1536):
        """创建集合"""
        self.client.create_collection(
            collection_name=name,
            vectors_config=VectorParams(size=dim, distance=Distance.COSINE)
        )
    
    def upsert(self, collection: str, points: list):
        """插入或更新数据"""
        self.client.upsert(
            collection_name=collection,
            points=[
                PointStruct(
                    id=p["id"],
                    vector=p["embedding"],
                    payload={"text": p["text"], "metadata": p.get("metadata", {})}
                )
                for p in points
            ]
        )
    
    def search(self, collection: str, query_vector: list, 
               top_k: int = 10, filter_conditions: dict = None) -> list:
        """向量检索"""
        query_filter = None
        if filter_conditions:
            query_filter = Filter(
                must=[
                    FieldCondition(key=k, match={"value": v})
                    for k, v in filter_conditions.items()
                ]
            )
        
        results = self.client.search(
            collection_name=collection,
            query_vector=query_vector,
            limit=top_k,
            query_filter=query_filter
        )
        
        return [
            {
                "id": r.id,
                "text": r.payload.get("text"),
                "score": r.score,
                "metadata": r.payload.get("metadata", {})
            }
            for r in results
        ]

第五章：混合检索策略

5.1 稠密检索 + 稀疏检索

单一检索方式难以覆盖所有场景，混合检索能显著提升召回率。

from sentence_transformers import SentenceTransformer
from rank_bm25 import BM25Okapi
import numpy as np

class HybridRetriever:
    """混合检索器：稠密向量 + BM25稀疏检索"""
    
    def __init__(self, embedding_model_name: str = "BAAI/bge-large-zh-v1.5"):
        self.dense_model = SentenceTransformer(embedding_model_name)
        self.bm25 = None
        self.documents = []
        self.dense_index = None
    
    def build_index(self, documents: list):
        """构建索引"""
        self.documents = documents
        texts = [doc["text"] for doc in documents]
        
        # BM25索引
        tokenized = [list(text) for text in texts]  # 中文按字符分词
        self.bm25 = BM25Okapi(tokenized)
        
        # 稠密向量索引
        self.dense_index = self.dense_model.encode(texts, show_progress_bar=True)
    
    def search(self, query: str, top_k: int = 10, 
               dense_weight: float = 0.7, sparse_weight: float = 0.3) -> list:
        """混合检索"""
        # 稠密检索
        query_embedding = self.dense_model.encode([query])[0]
        dense_scores = np.dot(self.dense_index, query_embedding)
        
        # 稀疏检索
        query_tokens = list(query)
        sparse_scores = self.bm25.get_scores(query_tokens)
        
        # 归一化
        dense_scores = (dense_scores - dense_scores.min()) / (dense_scores.max() - dense_scores.min() + 1e-8)
        sparse_scores = (sparse_scores - sparse_scores.min()) / (sparse_scores.max() - sparse_scores.min() + 1e-8)
        
        # 加权融合
        combined_scores = dense_weight * dense_scores + sparse_weight * sparse_scores
        
        # 排序
        top_indices = np.argsort(combined_scores)[::-1][:top_k]
        
        results = []
        for idx in top_indices:
            results.append({
                "text": self.documents[idx]["text"],
                "score": float(combined_scores[idx]),
                "dense_score": float(dense_scores[idx]),
                "sparse_score": float(sparse_scores[idx]),
                "metadata": self.documents[idx].get("metadata", {})
            })
        
        return results

5.2 ColBERT晚期交互检索

# 使用 ColBERT 进行晚期交互检索
# pip install ragatouille

from ragatouille import RAGPretrainedModel

class ColBERTRetriever:
    """ColBERT晚期交互检索器"""
    
    def __init__(self, model_path: str = "colbert-ir/colbertv2.0"):
        self.model = RAGPretrainedModel.from_pretrained(model_path)
    
    def index(self, documents: list, collection_name: str = "my_index"):
        """构建ColBERT索引"""
        self.model.index(
            collection=[doc["text"] for doc in documents],
            index_name=collection_name,
            max_document_length=256,
            split_documents=True
        )
    
    def search(self, query: str, top_k: int = 10) -> list:
        """ColBERT检索"""
        results = self.model.search(query, k=top_k)
        return results

第六章：重排序（Reranking）

6.1 Cross-Encoder重排序

检索返回的候选文档需要经过精排来提升相关性。

from sentence_transformers import CrossEncoder

class Reranker:
    """Cross-Encoder重排序器"""
    
    def __init__(self, model_name: str = "BAAI/bge-reranker-v2-m3"):
        self.model = CrossEncoder(model_name, max_length=512)
    
    def rerank(self, query: str, documents: list, top_k: int = 5) -> list:
        """对检索结果重排序"""
        pairs = [[query, doc["text"]] for doc in documents]
        scores = self.model.predict(pairs)
        
        for i, doc in enumerate(documents):
            doc["rerank_score"] = float(scores[i])
        
        # 按重排序分数排序
        reranked = sorted(documents, key=lambda x: x["rerank_score"], reverse=True)
        return reranked[:top_k]


class LLMReranker:
    """使用LLM进行重排序（适合高质量场景）"""
    
    def __init__(self, llm_client):
        self.llm = llm_client
    
    async def rerank(self, query: str, documents: list, top_k: int = 5) -> list:
        """LLM重排序"""
        doc_list = "\n".join([
            f"[{i+1}] {doc['text'][:200]}..."
            for i, doc in enumerate(documents)
        ])
        
        prompt = f"""请根据与查询的相关性，对以下文档进行排序。
查询：{query}

文档列表：
{doc_list}

请返回排序后的文档编号（从最相关到最不相关），用逗号分隔。
例如：3,1,5,2,4

排序结果："""
        
        result = await self.llm.chat(messages=[{"role": "user", "content": prompt}])
        
        # 解析排序结果
        try:
            order = [int(x.strip()) - 1 for x in result.strip().split(",")]
            reranked = [documents[i] for i in order if i < len(documents)]
            return reranked[:top_k]
        except:
            return documents[:top_k]

第七章：查询理解与改写

7.1 查询意图识别

class QueryProcessor:
    """查询处理器"""
    
    def __init__(self, llm_client):
        self.llm = llm_client
    
    async def classify_intent(self, query: str) -> str:
        """查询意图分类"""
        prompt = f"""请对以下查询进行意图分类：
查询：{query}

可选分类：
- factual：事实性问答
- comparison：对比分析
- how-to：操作指导
- summary：总结归纳
- analysis：深入分析

请只返回分类名称。"""
        
        result = await self.llm.chat(messages=[{"role": "user", "content": prompt}])
        return result.strip().lower()
    
    async def rewrite_query(self, query: str, history: list = None) -> str:
        """查询改写：优化检索效果"""
        history_text = ""
        if history:
            history_text = "\n对话历史：\n" + "\n".join([
                f"用户：{h['user']}\n助手：{h['assistant']}"
                for h in history[-3:]
            ])
        
        prompt = f"""请将以下查询改写为更适合检索的形式，保持核心意图不变。
{history_text}

原始查询：{query}

改写后的查询（只返回改写结果）："""
        
        result = await self.llm.chat(messages=[{"role": "user", "content": prompt}])
        return result.strip()
    
    async def generate_sub_queries(self, query: str) -> list:
        """查询分解：将复杂查询拆分为子查询"""
        prompt = f"""请将以下复杂查询分解为2-4个更具体的子查询：
原始查询：{query}

请以JSON数组格式返回子查询列表。
示例：["子查询1", "子查询2", "子查询3"]"""
        
        result = await self.llm.chat(messages=[{"role": "user", "content": prompt}])
        import json
        try:
            return json.loads(result.strip())
        except:
            return [query]

第八章：幻觉检测与事实性验证

8.1 基于检索的验证

class HallucinationValidator:
    """幻觉检测器"""
    
    def __init__(self, llm_client, embedding_model):
        self.llm = llm_client
        self.embedding_model = embedding_model
    
    async def validate(self, query: str, answer: str, 
                       source_documents: list) -> dict:
        """验证生成内容的事实性"""
        
        # 提取答案中的关键声明
        claims = await self._extract_claims(answer)
        
        # 对每个声明进行验证
        validations = []
        for claim in claims:
            # 在源文档中查找支持证据
            evidence = await self._find_evidence(claim, source_documents)
            
            # 判断是否有充分支持
            is_supported = await self._check_support(claim, evidence)
            
            validations.append({
                "claim": claim,
                "is_supported": is_supported,
                "evidence": evidence
            })
        
        # 计算事实性得分
        supported_count = sum(1 for v in validations if v["is_supported"])
        factuality_score = supported_count / len(validations) if validations else 1.0
        
        return {
            "factuality_score": factuality_score,
            "validations": validations,
            "is_faithful": factuality_score >= 0.8
        }
    
    async def _extract_claims(self, text: str) -> list:
        """提取文本中的关键声明"""
        prompt = f"""请从以下文本中提取所有事实性声明（可验证的陈述）：
{text}

请以JSON数组格式返回，每个元素为一个声明。"""
        
        result = await self.llm.chat(messages=[{"role": "user", "content": prompt}])
        import json
        try:
            return json.loads(result.strip())
        except:
            return [text]
    
    async def _find_evidence(self, claim: str, documents: list) -> list:
        """在源文档中查找支持证据"""
        claim_embedding = self.embedding_model.encode([claim])[0]
        
        evidence = []
        for doc in documents:
            doc_embedding = self.embedding_model.encode([doc["text"]])[0]
            similarity = np.dot(claim_embedding, doc_embedding) / (
                np.linalg.norm(claim_embedding) * np.linalg.norm(doc_embedding)
            )
            if similarity > 0.5:
                evidence.append({"text": doc["text"], "score": float(similarity)})
        
        return sorted(evidence, key=lambda x: x["score"], reverse=True)[:3]
    
    async def _check_support(self, claim: str, evidence: list) -> bool:
        """判断声明是否有充分证据支持"""
        if not evidence:
            return False
        
        evidence_text = "\n".join([e["text"][:300] for e in evidence])
        
        prompt = f"""根据以下证据，判断声明是否得到支持。

声明：{claim}

证据：
{evidence_text}

请回答：SUPPORTED（有支持）或 NOT_SUPPORTED（无支持）
只返回判断结果。"""
        
        result = await self.llm.chat(messages=[{"role": "user", "content": prompt}])
        return "SUPPORTED" in result.upper()

第九章：RAG评估框架

9.1 RAGAS评估

# pip install ragas

from ragas import evaluate
from ragas.metrics import (
    faithfulness,       # 忠实度
    answer_relevancy,   # 答案相关性
    context_precision,  # 上下文精确度
    context_recall      # 上下文召回率
)
from datasets import Dataset

class RAGEvaluator:
    """RAG系统评估器"""
    
    def evaluate_with_ragas(self, eval_data: list) -> dict:
        """使用RAGAS框架评估"""
        dataset = Dataset.from_list(eval_data)
        
        result = evaluate(
            dataset=dataset,
            metrics=[
                faithfulness,
                answer_relevancy,
                context_precision,
                context_recall
            ]
        )
        
        return {
            "faithfulness": result["faithfulness"],
            "answer_relevancy": result["answer_relevancy"],
            "context_precision": result["context_precision"],
            "context_recall": result["context_recall"],
            "overall": result["faithfulness"] * 0.3 + result["answer_relevancy"] * 0.3 + 
                       result["context_precision"] * 0.2 + result["context_recall"] * 0.2
        }
    
    def evaluate_retrieval(self, queries: list, retrieved_docs: list, 
                           ground_truth: list) -> dict:
        """评估检索质量"""
        hits = 0
        mrr_sum = 0
        
        for i, (retrieved, relevant) in enumerate(zip(retrieved_docs, ground_truth)):
            retrieved_ids = [d["id"] for d in retrieved]
            
            # Hit@K
            if any(doc_id in relevant for doc_id in retrieved_ids[:5]):
                hits += 1
            
            # MRR
            for rank, doc_id in enumerate(retrieved_ids[:10]):
                if doc_id in relevant:
                    mrr_sum += 1 / (rank + 1)
                    break
        
        n = len(queries)
        return {
            "hit_at_5": hits / n,
            "mrr_at_10": mrr_sum / n
        }

第十章：完整RAG Pipeline实战

10.1 端到端RAG系统

class EnterpriseRAG:
    """企业级RAG系统完整实现"""
    
    def __init__(self, config: dict):
        self.config = config
        
        # 初始化各组件
        self.document_parser = DocumentParser()
        self.chunker = ParentChildChunker(
            child_size=config.get("child_size", 256),
            parent_size=config.get("parent_size", 1024)
        )
        self.embedding_model = SentenceTransformer(config.get("embedding_model", "BAAI/bge-large-zh-v1.5"))
        self.vector_store = MilvusVectorStore()
        self.reranker = Reranker()
        self.query_processor = QueryProcessor(llm_client=None)
        self.validator = HallucinationValidator(llm_client=None, embedding_model=self.embedding_model)
        self.memory = ConversationMemory(max_turns=5)
    
    async def ingest(self, file_path: str):
        """文档摄入流程"""
        # 1. 解析文档
        content = self.document_parser.parse(file_path)
        
        # 2. 分块
        chunks = self.chunker.chunk(content)
        
        # 3. 生成嵌入
        child_texts = [c["text"] for c in chunks["children"]]
        embeddings = self.embedding_model.encode(child_texts)
        
        # 4. 存入向量数据库
        self.vector_store.insert(
            ids=[c["id"] for c in chunks["children"]],
            texts=child_texts,
            embeddings=embeddings.tolist(),
            metadata=[c["metadata"] for c in chunks["children"]]
        )
        
        # 5. 存储父块映射
        self._store_parent_mapping(chunks["parents"])
        
        return len(chunks["children"])
    
    async def query(self, question: str, session_id: str = None) -> dict:
        """查询流程"""
        # 1. 获取对话历史
        history = self.memory.get_history(session_id)
        
        # 2. 查询处理
        rewritten = await self.query_processor.rewrite_query(question, history)
        
        # 3. 混合检索
        query_embedding = self.embedding_model.encode([rewritten])[0]
        raw_results = self.vector_store.search(
            query_embedding.tolist(), top_k=20
        )
        
        # 4. 重排序
        reranked = self.reranker.rerank(rewritten, raw_results, top_k=5)
        
        # 5. 获取父块上下文
        parent_ids = list(set(r["metadata"].get("parent_id") for r in reranked))
        context = self._get_parent_context(parent_ids)
        
        # 6. 生成回答
        answer = await self._generate(question, context, history)
        
        # 7. 幻觉验证
        validation = await self.validator.validate(question, answer, reranked)
        
        # 8. 更新对话记忆
        self.memory.add(session_id, question, answer)
        
        return {
            "answer": answer,
            "sources": [{"text": r["text"][:200], "score": r.get("rerank_score", 0)} for r in reranked],
            "factuality_score": validation["factuality_score"],
            "is_faithful": validation["is_faithful"]
        }
    
    async def _generate(self, question: str, context: str, history: list) -> str:
        """生成回答"""
        history_text = ""
        if history:
            history_text = "\n".join([
                f"用户：{h['user']}\n助手：{h['assistant']}"
                for h in history[-3:]
            ])
        
        prompt = f"""基于以下上下文信息回答用户的问题。如果上下文中没有相关信息，请明确说明无法回答。

上下文：
{context}

对话历史：
{history_text}

用户问题：{question}

请给出准确、详细的回答，并在回答中标注信息来源。"""
        
        return await self.llm.chat(messages=[{"role": "user", "content": prompt}])

第十一章：大规模RAG系统运维

11.1 索引更新策略

class IndexManager:
    """索引管理器"""
    
    def __init__(self, vector_store, embedding_model):
        self.vector_store = vector_store
        self.embedding_model = embedding_model
    
    async def incremental_update(self, new_documents: list):
        """增量更新索引"""
        for doc in new_documents:
            # 检查文档是否已存在
            existing = await self._check_exists(doc["id"])
            
            if existing:
                # 文档已更新，删除旧数据后重新索引
                await self._delete_document(doc["id"])
            
            # 索引新文档
            await self._index_document(doc)
    
    async def rebuild_index(self, all_documents: list):
        """全量重建索引"""
        # 创建临时集合
        temp_name = f"{self.collection_name}_temp"
        self.vector_store.create_collection(temp_name)
        
        # 批量索引
        batch_size = 1000
        for i in range(0, len(all_documents), batch_size):
            batch = all_documents[i:i+batch_size]
            embeddings = self.embedding_model.encode([d["text"] for d in batch])
            self.vector_store.insert(
                ids=[d["id"] for d in batch],
                texts=[d["text"] for d in batch],
                embeddings=embeddings.tolist(),
                metadata=[d.get("metadata", {}) for d in batch]
            )
        
        # 原子切换
        self._swap_collection(self.collection_name, temp_name)

11.2 性能优化建议

1. 缓存策略：对高频查询结果进行缓存
2. 异步处理：全文档摄入采用异步流水线
3. 批量嵌入：批量调用embedding模型减少API调用次数
4. 预加载：向量数据库索引预加载到内存
5. 连接池：数据库和API连接复用

最佳实践总结

1. 文档预处理是关键：投入足够精力优化文档解析和分块策略
2. 混合检索优于单一检索：结合稠密和稀疏检索提升召回率
3. 重排序必不可少：Cross-Encoder重排序能显著提升精确度
4. 查询改写提升体验：LLM驱动的查询改写能理解用户真实意图
5. 幻觉检测保障质量：生产环境必须有事实性验证机制
6. 持续评估迭代：建立评估体系，定期评估和优化系统
7. 权限控制不可忽视：企业场景必须考虑文档级权限控制

总结

本教程系统讲解了企业级RAG系统的完整技术栈，从文档解析、分块策略、向量数据库、混合检索、重排序、查询改写到幻觉检测和评估框架。RAG系统的核心在于将正确的上下文以正确的方式传递给LLM，每个环节都直接影响最终效果。

随着技术的发展，Agentic RAG、GraphRAG、多模态RAG等新范式不断涌现，RAG正在从简单的检索增强向智能化、自适应的方向演进。掌握本教程的核心技术，将为构建生产级AI应用打下坚实基础。