RAG 系统深度实战：从朴素检索到 Agentic RAG 的完整演进

文档 → 解析 → 分块 → Embedding → 存储（离线阶段）
                                        ↓
问题 → Query处理 → 检索 → 重排序 → 生成（在线阶段）

固定分块（512 tokens）：
"...模型的核心在于注意力机制的"   ← 这里断了
"多头设计使得模型能够同时关注..."   ← 后半句在下一个 Chunk

语义分块：
"...模型的核心在于注意力机制的多头设计，使得模型能够同时关注不同位置的语义信息。"

import re
from typing import List

class SemanticChunker:
    """基于语义边界的中文文本分块器。"""

    def __init__(self, min_chunk_size: int = 200, max_chunk_size: int = 1000):
        self.min_chunk = min_chunk_size
        self.max_chunk = max_chunk_size

    def chunk(self, text: str) -> List[dict]:
        """
        语义分块策略：
        1. 先按自然段落分割
        2. 判断相邻段落是否语义连贯（用标题、转折词、主题词判断）
        3. 连贯的合并，独立的断开
        4. 超长段落按句子边界二次切割
        """
        paragraphs = self._split_by_paragraph(text)
        chunks = []
        current_chunk = []
        current_size = 0

        for para in paragraphs:
            para_size = len(para)

            # 碰到标题 → 新 Chunk 开始
            if self._is_heading(para) and current_chunk:
                chunks.append(self._finalize(current_chunk))
                current_chunk = [para]
                current_size = para_size
                continue

            # 语义转折词 → 新 Chunk 开始
            if self._has_transition(para) and current_chunk:
                chunks.append(self._finalize(current_chunk))
                current_chunk = [para]
                current_size = para_size
                continue

            # 超长段落 → 按句子边界切割
            if para_size > self.max_chunk:
                if current_chunk:
                    chunks.append(self._finalize(current_chunk))
                    current_chunk = []
                    current_size = 0
                sub_chunks = self._split_long_paragraph(para)
                chunks.extend(sub_chunks)
                continue

            # 常规合并
            if current_size + para_size  List[str]:
        """按自然段落分割。"""
        return [p.strip() for p in re.split(r'\n\s*\n', text) if p.strip()]

    def _is_heading(self, text: str) -> bool:
        """判断是否为标题。"""
        heading_patterns = [
            r'^第[一二三四五六七八九十\d]+[章节篇]',
            r'^[一二三四五六七八九十]、',
            r'^\d+[\.\、]',
            r'^##+\s',
        ]
        return any(re.match(p, text) for p in heading_patterns)

    def _has_transition(self, text: str) -> bool:
        """判断是否包含语义转折标记。"""
        transition_words = [
            '但是', '然而', '另一方面', '与此相反', '相比之下',
            '更重要的是', '需要强调的是', '总的来看', '综上所述',
            '不过', '尽管如此', '无论如何',
        ]
        return any(w in text[:30] for w in transition_words)

    def _split_long_paragraph(self, para: str) -> List[dict]:
        """超长段落的句子级切割。"""
        sentences = re.split(r'[。！？；\n]', para)
        chunks = []
        current = []
        current_size = 0

        for sent in sentences:
            sent = sent.strip()
            if not sent:
                continue
            sent_size = len(sent)
            if current_size + sent_size > self.max_chunk and current:
                chunks.append(self._finalize(current))
                current = [sent]
                current_size = sent_size
            else:
                current.append(sent)
                current_size += sent_size

        if current:
            chunks.append(self._finalize(current))
        return chunks

    def _finalize(self, paragraphs: List[str]) -> dict:
        text = '\n\n'.join(paragraphs)
        return {
            'content': text,
            'size': len(text),
            'paragraph_count': len(paragraphs),
        }

场景一：个人项目 / 离线环境 → stella-base-zh-v3-1792d（本地免费，性能最佳）
场景二：企业级 / 大规模索引 → BGE-M3（成熟稳定，社区活跃，多语言支持）
场景三：极致效果 / 预算充足 → text-embedding-3-large（维度高，长文本表现好）
场景四：长文档（>8K tokens）→ Jina Embeddings v3（原生支持 8192 token 输入）

class QueryTransformer:
    """查询改写器——把用户自然语言转换为更适合检索的查询。"""

    def __init__(self, llm_client):
        self.llm = llm_client

    def rewrite(self, user_query: str, conversation_history: list = None) -> list:
        """生成多个检索查询变体。"""
        prompt = f"""将以下用户问题改写为 3 个不同角度的检索查询。每个查询应该使用文档中可能出现的术语：

用户问题：{user_query}

要求：
1. 查询 1：精确匹配——用关键词和术语
2. 查询 2：语义扩展——用同义词和相关概念
3. 查询 3：问题分解——如果问题包含多个子问题，拆开

仅输出 3 个查询，每行一个，不要编号。"""

        response = self.llm.chat.completions.create(
            model='deepseek-chat',
            messages=[{'role': 'user', 'content': prompt}],
            temperature=0.1,
        )
        queries = response.choices[0].message.content.strip().split('\n')
        return [q.strip() for q in queries if q.strip()]

class HybridSearcher:
    """混合检索：向量检索 + BM25 关键词检索。"""

    def __init__(self, vector_db, bm25_index, alpha: float = 0.7):
        """
        alpha: 向量检索的权重（0-1）
        0.7 意味着向量检索占 70%，BM25 占 30%
        这个值需要通过你的实际数据做 Grid Search 来确定最优值
        """
        self.vector_db = vector_db
        self.bm25 = bm25_index
        self.alpha = alpha

    def search(self, query: str, top_k: int = 10) -> list:
        # 向量检索
        vector_results = self.vector_db.search(query, limit=top_k * 2)
        # BM25 检索
        bm25_results = self.bm25.search(query, limit=top_k * 2)

        # 分数归一化 + 加权融合
        vector_scores = self._normalize({r['id']: r['score'] for r in vector_results})
        bm25_scores = self._normalize({r['id']: r['score'] for r in bm25_results})

        # 合并分数
        all_ids = set(vector_scores.keys()) | set(bm25_scores.keys())
        combined = {}
        for doc_id in all_ids:
            vs = vector_scores.get(doc_id, 0)
            bs = bm25_scores.get(doc_id, 0)
            combined[doc_id] = self.alpha * vs + (1 - self.alpha) * bs

        # 排序返回
        sorted_ids = sorted(combined, key=combined.get, reverse=True)[:top_k]
        return [{'id': i, 'score': combined[i]} for i in sorted_ids]

    def _normalize(self, scores: dict) -> dict:
        """Min-Max 归一化，确保向量和 BM25 分数可比。"""
        if not scores:
            return {}
        values = list(scores.values())
        min_v, max_v = min(values), max(values)
        if max_v == min_v:
            return {k: 0.5 for k in scores}
        return {k: (v - min_v) / (max_v - min_v) for k, v in scores.items()}

class Reranker:
    """基于 Cross-Encoder 的重排序器。"""

    def __init__(self, model_name: str = 'BAAI/bge-reranker-v2-m3'):
        from sentence_transformers import CrossEncoder
        self.model = CrossEncoder(model_name)

    def rerank(self, query: str, documents: list, top_k: int = 3) -> list:
        """从候选文档中精选最相关的 top_k 条。"""
        pairs = [(query, doc['content']) for doc in documents]
        scores = self.model.predict(pairs)

        # 分数 + 原始信息
        for i, doc in enumerate(documents):
            doc['rerank_score'] = float(scores[i])

        documents.sort(key=lambda d: d['rerank_score'], reverse=True)
        return documents[:top_k]

class SelfRAG:
    """Self-RAG：LLM 自主决定是否检索、检索什么、检索结果是否可用。"""

    REFLECTION_TOKENS = {
        'Retrieve': '需要检索外部知识',
        'NoRetrieve': '不需要检索，凭已有知识回答',
        'Relevant': '检索结果相关，可以使用',
        'Irrelevant': '检索结果不相关，需要重新检索或放弃',
    }

    def generate(self, query: str, retriever, generator) -> str:
        # Step 1: 判断是否需要检索
        decision = generator.predict(f'判断是否需要检索来回答：{query}')
        if 'NoRetrieve' in decision:
            return generator.generate(query)

        # Step 2: 检索
        docs = retriever.search(query)

        # Step 3: 逐条判断相关性
        relevant_docs = []
        for doc in docs:
            relevance = generator.predict(
                f'问题：{query}\n文档：{doc["content"]}\n判断：Relevant 或 Irrelevant'
            )
            if 'Relevant' in relevance:
                relevant_docs.append(doc)

        # Step 4: 如果全不相关，重新检索
        if not relevant_docs:
            rewritten_query = generator.rewrite(query)
            docs = retriever.search(rewritten_query)
            relevant_docs = [d for d in docs if 'Relevant' in generator.predict(
                f'问题：{query}\n文档：{d["content"]}\n判断：'
            )]

        # Step 5: 基于精选文档生成
        return generator.generate(query, context=relevant_docs)

class AgenticRAG:
    """Agentic RAG：多步检索 + 信息缺口分析。"""

    def research(self, question: str, max_steps: int = 5) -> str:
        findings = []
        sub_questions = [question]

        for step in range(max_steps):
            # 检索当前子问题
            new_findings = []
            for sq in sub_questions:
                docs = self.retriever.search(sq)
                new_findings.extend(docs)

            findings.extend(new_findings)

            # 分析信息缺口
            gap_analysis = self.llm.chat.completions.create(
                model='deepseek-chat',
                messages=[{
                    'role': 'user',
                    'content': f"""基于已收集的信息，分析还缺少什么：

原始问题：{question}
已收集信息：{findings[-5:]}  # 最近 5 条

列出 1-3 个还需要搜索的子问题（用列表格式）。如果信息已足够，回复"SUFFICIENT"."""
                }],
            )

            response = gap_analysis.choices[0].message.content
            if 'SUFFICIENT' in response:
                break

            sub_questions = [q.strip('- ') for q in response.split('\n') if q.strip('- ')]

        # 综合所有发现生成最终回答
        return self.llm.chat.completions.create(
            model='deepseek-chat',
            messages=[{
                'role': 'user',
                'content': f'基于以下研究发现回答用户问题：\n问题：{question}\n发现：{findings}'
            }],
        ).choices[0].message.content