大模型 API 编排与 RAG 架构深度实践

一、从理论到生产：RAG 的工程化挑战

检索增强生成（Retrieval-Augmented Generation, RAG）已经成为大模型应用的主流架构。理论很简单：让模型在回答问题时先从知识库检索相关内容，再基于检索结果生成答案。但当开发者真正把 RAG 部署到生产环境时，才发现这个"简单"的架构充满了工程挑战。

Chunk 大小如何选择？向量数据库选哪个？如何提升召回率？如何处理检索结果与生成质量的关系？这些问题没有标准答案，需要根据具体场景反复调优。

本文从工程实践角度深入探讨 RAG 架构的各个环节，从数据处理到检索优化，从生成策略到效果评估，提供可落地的解决方案。

二、RAG 架构总览

2.1 经典 RAG 流程

flowchart TD
    A[用户问题] --> B[问题向量化]
    B --> C[向量数据库检索]
    C --> D[Top-K 相关文档]
    D --> E[上下文组装]
    E --> F[Prompt 构建]
    F --> G[LLM 生成]
    G --> H[答案输出]
    
    subgraph 索引构建
        I[原始文档] --> J[文档分块]
        J --> K[块向量化]
        K --> L[向量索引]
    end
    
    L -.->|离线| C

2.2 RAG 变体架构

graph LR
    A[基础 RAG] --> B[检索-生成]
    
    C[Self-RAG] --> D[检索+自评估+生成]
    
    E[ReAct RAG] --> F[推理+行动+检索]
    
    G[HyDE] --> H[假设文档检索]
    
    I[RRR] --> J[检索-相关性-生成]
    
    style A fill:#ffcccc
    style C fill:#ffcc99
    style E fill:#ffffcc
    style G fill:#ccffcc

三、数据处理与索引构建

3.1 文档分块策略

from typing import Callable, Iterator
from dataclasses import dataclass

@dataclass
class Chunk:
    """文档块"""
    content: str
    metadata: dict
    chunk_id: str

class TextSplitter:
    """
    文档分块器
    策略：固定大小、重叠、语义分句
    """
    
    def __init__(
        self,
        chunk_size: int = 500,
        chunk_overlap: int = 50,
        separators: list[str] = ["\n\n", "\n", "。", "！", "？", " "],
    ):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        self.separators = separators
    
    def split_text(self, text: str, metadata: dict = None) -> list[Chunk]:
        """
        文本分块
        
        chunk_size 选择原则：
        - 太小：丢失上下文
        - 太大：引入噪声，降低召回精度
        - 经验值：300-800 tokens
        """
        chunks = []
        
        # 按段落初步分割
        paragraphs = self._split_by_separators(text)
        
        # 合并成 chunk
        current_chunk = ""
        current_size = 0
        
        for para in paragraphs:
            para_size = len(para)
            
            if current_size + para_size <= self.chunk_size:
                current_chunk += para + "\n"
                current_size += para_size
            else:
                # 保存当前 chunk
                if current_chunk.strip():
                    chunks.append(Chunk(
                        content=current_chunk.strip(),
                        metadata=metadata or {},
                        chunk_id=self._generate_id(current_chunk),
                    ))
                
                # 开始新 chunk（带 overlap）
                overlap_text = current_chunk[-self.chunk_overlap:] if current_chunk else ""
                current_chunk = overlap_text + para + "\n"
                current_size = len(current_chunk)
        
        # 最后一个 chunk
        if current_chunk.strip():
            chunks.append(Chunk(
                content=current_chunk.strip(),
                metadata=metadata or {},
                chunk_id=self._generate_id(current_chunk),
            ))
        
        return chunks
    
    def _split_by_separators(self, text: str) -> list[str]:
        """按分隔符分割"""
        # 简化实现
        return text.split("\n")

3.2 元数据增强

class MetadataEnricher:
    """
    元数据增强器
    为每个 chunk 添加丰富元数据
    """
    
    def enrich(self, chunks: list[Chunk], source_doc: dict) -> list[Chunk]:
        """为 chunk 添加元数据"""
        enriched = []
        
        for i, chunk in enumerate(chunks):
            chunk.metadata.update({
                "source": source_doc.get("title", "unknown"),
                "source_type": source_doc.get("type", "text"),
                "chunk_index": i,
                "total_chunks": len(chunks),
                "created_at": source_doc.get("created_at"),
            })
            enriched.append(chunk)
        
        return enriched

3.3 向量化与索引

from openai import OpenAI
import chromadb

class VectorIndexer:
    """
    向量化索引器
    """
    
    def __init__(self, embed_model: str = "text-embedding-3-small"):
        self.embed_client = OpenAI()
        self.embed_model = embed_model
        self.vector_db = chromadb.PersistentClient(path="./chroma_db")
    
    def create_collection(self, name: str):
        """创建 collection"""
        return self.vector_db.get_or_create_collection(
            name=name,
            metadata={"hnsw:space": "cosine"},  # 余弦相似度
        )
    
    def embed_chunks(self, chunks: list[Chunk]) -> list[list[float]]:
        """
        批量向量化 chunk
        """
        texts = [chunk.content for chunk in chunks]
        
        response = self.embed_client.embeddings.create(
            model=self.embed_model,
            input=texts,
        )
        
        return [item.embedding for item in response.data]
    
    def index_chunks(
        self,
        collection_name: str,
        chunks: list[Chunk],
        batch_size: int = 100,
    ):
        """
        批量索引 chunk
        """
        collection = self.create_collection(collection_name)
        
        for i in range(0, len(chunks), batch_size):
            batch = chunks[i:i + batch_size]
            embeddings = self.embed_chunks(batch)
            
            ids = [chunk.chunk_id for chunk in batch]
            documents = [chunk.content for chunk in batch]
            metadatas = [chunk.metadata for chunk in batch]
            
            collection.add(
                ids=ids,
                embeddings=embeddings,
                documents=documents,
                metadatas=metadatas,
            )
            
            print(f"Indexed {min(i + batch_size, len(chunks))}/{len(chunks)}")

四、检索优化实践

4.1 混合检索

class HybridRetriever:
    """
    混合检索器：向量检索 + 关键词检索
    """
    
    def __init__(self, vector_store, keyword_store):
        self.vector_store = vector_store
        self.keyword_store = keyword_store
        self.vector_weight = 0.7
        self.keyword_weight = 0.3
    
    def retrieve(
        self,
        query: str,
        top_k: int = 5,
        filter_criteria: dict = None,
    ) -> list[dict]:
        """
        混合检索
        """
        # 1. 向量检索
        vector_results = self.vector_store.query(
            query_texts=[query],
            n_results=top_k * 2,  # 多取一些，后面融合
            where=filter_criteria,
        )
        
        # 2. BM25 关键词检索
        keyword_results = self.keyword_store.query(
            query=query,
            top_k=top_k * 2,
            filter=filter_criteria,
        )
        
        # 3. RRF 融合
        fused = self._reciprocal_rank_fusion(
            vector_results,
            keyword_results,
            k=60,  # RRF 参数
        )
        
        # 4. 重排序
        reranked = self._rerank(fused[:top_k])
        
        return reranked
    
    def _reciprocal_rank_fusion(
        self,
        results1: list,
        results2: list,
        k: int = 60,
    ) -> list:
        """
        倒数排序融合（RRF）
        
        公式：RRF(doc) = Σ 1/(k + rank(doc))
        """
        scores = {}
        
        # 第一个结果集的分数
        for rank, doc in enumerate(results1):
            doc_id = doc["id"]
            scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank + 1)
        
        # 第二个结果集的分数
        for rank, doc in enumerate(results2):
            doc_id = doc["id"]
            scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank + 1)
        
        # 按分数排序
        sorted_docs = sorted(scores.items(), key=lambda x: x[1], reverse=True)
        
        return [doc_id for doc_id, _ in sorted_docs]

4.2 查询扩展与改写

class QueryExpander:
    """
    查询扩展器
    """
    
    def __init__(self, llm_client):
        self.llm = llm_client
    
    def expand_query(self, query: str) -> list[str]:
        """
        将单一查询扩展为多个相关查询
        
        目的：解决表达方式差异导致的召回不全
        """
        prompt = f"""给定用户查询，生成3个不同的搜索查询，用于检索相关文档。

原始查询：{query}

要求：
1. 生成语义相近但表达不同的查询
2. 覆盖查询的不同角度
3. 简短直接，每个不超过20字

输出格式：
查询1
查询2
查询3
"""
        
        response = self.llm.generate(prompt)
        expanded = [query] + response.strip().split("\n")
        
        return [q.strip() for q in expanded if q.strip()]

五、生成优化与评估

5.1 Prompt 模板设计

RAG_PROMPT_TEMPLATE = """你是一个知识问答助手。请根据以下检索到的文档回答用户问题。

检索到的文档：
---
{context}
---

用户问题：{question}

回答要求：
1. 仅基于检索文档回答，不要编造信息
2. 如果检索文档中没有相关信息，诚实地说明"我没有找到相关内容"
3. 回答要准确、完整、简洁
4. 适当引用文档来源

回答："""

5.2 生成质量评估

class RAGEvaluator:
    """
    RAG 系统评估
    """
    
    METRICS = {
        "context_precision": "检索结果的相关性",
        "context_recall": "检索结果覆盖问题的程度",
        "faithfulness": "生成答案与检索内容的一致性",
        "answer_relevancy": "生成答案与问题的相关性",
    }
    
    def evaluate(
        self,
        question: str,
        retrieved_docs: list[dict],
        generated_answer: str,
    ) -> dict:
        """
        综合评估
        """
        return {
            "context_precision": self._compute_precision(
                question, retrieved_docs
            ),
            "faithfulness": self._compute_faithfulness(
                retrieved_docs, generated_answer
            ),
            "answer_relevancy": self._compute_relevancy(
                question, generated_answer
            ),
        }
    
    def _compute_faithfulness(
        self,
        docs: list[dict],
        answer: str,
    ) -> float:
        """
        忠诚度：答案中是否有检索文档未支持的内容
        
        简化实现：实际需要更复杂的 NLP 分析
        """
        # 启发式：检查答案中是否有明显"编造"迹象
        fabricated_indicators = [
            "根据文档显示" in answer,
            "文档提到" in answer,
            len(answer) < 20,  # 太短可能没有充分利用检索内容
        ]
        
        if any(fabricated_indicators):
            return 0.5
        
        return 0.9  # 占位

六、边界分析与最佳实践

6.1 RAG 优化决策树

graph TD
    A[RAG 系统问题] --> B{检索问题}
    A --> C{生成问题}
    
    B --> B1[召回率低]
    B1 --> B1a[扩展查询]
    B1a --> B1b[混合检索]
    B1b --> B1c[增加索引]
    
    B --> B2[精确度低]
    B2 --> B2a[重排序]
    B2b[增加元数据过滤]
    
    C --> C1[幻觉]
    C1 --> C1a[Few-shot 示例]
    C1b[Prompt 强调只依据文档]
    
    C --> C2[答案不完整]
    C2 --> C2a[多次检索迭代]
    C2b[查询扩展]

6.2 Chunk 大小选择

Chunk 大小	适用场景	优缺点
100-200 tokens	简单问答	上下文少，精度高
300-500 tokens	通用场景	平衡
800-1000 tokens	复杂分析	上下文丰富，可能有噪声

七、总结

RAG 的工程化是一个持续优化的过程，没有银弹解决方案。

关键优化建议：

1. 先评估再优化：用量化指标定位瓶颈
2. 检索优先：生成质量的上限由检索质量决定
3. 迭代调优：chunk 大小、top_k、权重系数都需要实验
4. 监控生产：持续收集用户反馈，发现问题及时修复

生产环境 checklist：

• 数据 pipeline 自动化
• 定期重新索引（知识更新）
• 检索质量监控
• 答案质量人工抽检