chendandan_homework/02_work/copy_rag_chain.py

#!/usr/bin/env python3
"""
RAG 全链路优化系统 — 员工手册智能问答
============================================
8 项核心优化：
  1. 使用 XX科技有限公司员工手册.md 作为测试文档
  2. 检索器封装为 Agent 工具，自动判断是否需要检索
  3. 使用本地 BGE-Large-Zh-v1.5 embedding 模型 (1024d)
  4. 语义文本切块 (SemanticChunker, percentile=60)
  5. 多路召回: BM25 / 向量 / 混合(Ensemble) / 多查询(MultiQuery)
  6. 查询增强: 查询重写 + HyDE
  7. BGE-Reranker 精排
  8. RAG 效果评估: 测试集生成 → Hit@K/MRR/Recall → 幻觉率 → Bad Case 分析

运行方式: python copy_rag_chain.py
"""

# ============================================================================
# Section 1: 导入 + 路径常量 + 配置常量
# ============================================================================
from __future__ import annotations

import json
import os
import re
import sys
import shutil
import warnings
from pathlib import Path
from typing import Optional, Union
from collections import defaultdict

# --- 抑制无关日志 ---
warnings.filterwarnings("ignore", category=DeprecationWarning)
os.environ.setdefault("GRPC_VERBOSITY", "NONE")
os.environ.setdefault("TOKENIZERS_PARALLELISM", "false")

# --- 加载 .env ---
from dotenv import load_dotenv
load_dotenv(Path(__file__).resolve().parent.parent / ".env")

# --- 路径常量 ---
_PROJECT_ROOT = Path(__file__).resolve().parent.parent
_CURRENT_DIR = Path(__file__).resolve().parent
_MODELS_DIR = _PROJECT_ROOT / "models"
_BGE_PATH = _MODELS_DIR / "BAAI" / "bge-large-zh-v1.5"
_RERANKER_PATH = _MODELS_DIR / "BAAI" / "bge-reranker-base"
_DATA_PATH = _PROJECT_ROOT / "XX科技有限公司员工手册.md"
_CHROMA_DIR = _CURRENT_DIR / "chroma_db_bge"

# --- 配置常量 ---
# 各参数取值理由：经过员工手册（~7600 字、62 个条款）实测调优
CHUNK_PERCENTILE = 45        # 语义切块阈值：45 在粒度和语义完整间取平衡（值越小切块越细）
RETRIEVAL_K = 10             # 召回阶段返回数量：取 10 是精度/速度折中（太小漏召回，太大稀释 Reranker）
DEFAULT_K = 4                # 最终给 LLM 的文档数：4 篇足以覆盖一个问题，再多会撑爆上下文窗口
ENSEMBLE_WEIGHTS = [0.4, 0.6]  # 混合检索权重 [BM25, 向量]：政策文档关键词重要，BM25 提到 0.4
MULTI_QUERY_VARIANTS = 3     # 多查询变体数：3 个变体已足够覆盖同义词，再多边际效益递减
N_EVAL_QA = 25               # 评估测试集大小：覆盖全部 16 章的最少问题数
RRF_K = 60                   # RRF 平滑参数：业界标准值，避免排名靠前的文档过度霸占分数

# --- 确保根目录在 sys.path 中 ---
sys.path.insert(0, str(_PROJECT_ROOT))

# ============================================================================
# Section 2: 模型初始化
# ============================================================================

# --- 2a. Embedding 模型 ---

class SentenceTransformerEmbeddings:
    """将 SentenceTransformer 包装为 LangChain Embeddings 接口。

    支持 BGE 系列模型的 query_prefix 机制：
    - embed_query: 查询前加前缀「为这个句子生成表示以用于检索相关文章：」
    - embed_documents: 文档不加前缀（BGE 官方推荐用法）
    """
    def __init__(self, model_path: str, device: str = "cpu",
                 query_prefix: str = ""):
        from sentence_transformers import SentenceTransformer
        self._model = SentenceTransformer(model_path, device=device)
        self._query_prefix = query_prefix

    def embed_documents(self, texts: list[str]) -> list[list[float]]:
        # 文档侧不加前缀
        embeddings = self._model.encode(
            texts,
            normalize_embeddings=True,
            show_progress_bar=False,
            batch_size=32,
        )
        return embeddings.tolist()

    def embed_query(self, text: str) -> list[float]:
        # 查询侧加前缀（BGE 模型关键！不加前缀效果掉 10%+）
        if self._query_prefix:
            text = self._query_prefix + text
        embedding = self._model.encode(
            [text],
            normalize_embeddings=True,
            show_progress_bar=False,
        )
        return embedding[0].tolist()


_embedding_model: Optional[SentenceTransformerEmbeddings] = None

# BGE 系列模型的 query_prefix（官方推荐，不加的话检索精度明显下降）
_BGE_QUERY_PREFIX = "为这个句子生成表示以用于检索相关文章："

def init_embedding_model() -> SentenceTransformerEmbeddings:
    """初始化 BGE-Large-Zh-v1.5 embedding 模型（1024 维）。

    BGE-Large-Zh-v1.5 是 BAAI 中文优化版，MTEB 中文榜单榜首。
    相比之前的 Corom (768d)，维度更高、中文理解更强。
    关键: 查询时必须加 query_prefix，文档不加。
    """
    global _embedding_model
    if _embedding_model is not None:
        return _embedding_model
    if not _BGE_PATH.exists():
        raise FileNotFoundError(f"BGE 模型未找到: {_BGE_PATH}")
    print(f"[Embedding] 加载 BGE-Large-Zh-v1.5: {_BGE_PATH}")
    _embedding_model = SentenceTransformerEmbeddings(
        str(_BGE_PATH), device="cpu",
        query_prefix=_BGE_QUERY_PREFIX,
    )
    print(f"[Embedding] 加载完成，维度: 1024 (含 query_prefix)")
    return _embedding_model


# --- 2b. LLM 客户端 ---

_llm = None

def get_llm():
    """获取 LLM 客户端（复用根目录 llm_client.py 的 get_llm()）。"""
    global _llm
    if _llm is not None:
        return _llm
    from llm_client import get_llm as _root_get_llm
    _llm = _root_get_llm()
    print(f"[LLM] 客户端初始化完成")
    return _llm


# --- 2c. Reranker 模型 ---

_reranker = None

def download_reranker() -> Optional[str]:
    """从 ModelScope 下载 BGE-Reranker-Base。

    返回模型路径，如果下载失败返回 None。
    已下载则直接返回路径，跳过下载。
    """
    config_file = _RERANKER_PATH / "config.json"
    if config_file.exists():
        print(f"[Reranker] 模型已存在: {_RERANKER_PATH}")
        return str(_RERANKER_PATH)

    print(f"[Reranker] 正在从 ModelScope 下载 BGE-Reranker-Base (~1.1GB)...")
    print(f"[Reranker] 目标路径: {_RERANKER_PATH}")
    try:
        from modelscope import snapshot_download
        _RERANKER_PATH.parent.mkdir(parents=True, exist_ok=True)
        downloaded = snapshot_download(
            "BAAI/bge-reranker-base",
            cache_dir=str(_RERANKER_PATH.parent),
        )
        # snapshot_download 返回缓存路径，可能不是我们指定的路径
        # 如果路径不同，做复制
        downloaded_path = Path(downloaded)
        if downloaded_path != _RERANKER_PATH:
            if _RERANKER_PATH.exists():
                shutil.rmtree(str(_RERANKER_PATH))
            shutil.copytree(str(downloaded_path), str(_RERANKER_PATH))
        print(f"[Reranker] 下载完成: {_RERANKER_PATH}")
        return str(_RERANKER_PATH)
    except Exception as e:
        print(f"[Reranker] 下载失败: {e}")
        print(f"[Reranker] 将跳过精排步骤（不影响其他功能）")
        return None


def init_reranker() -> Optional[object]:
    """初始化 CrossEncoder reranker。

    返回 CrossEncoder 实例，如果不可用返回 None。
    """
    global _reranker
    if _reranker is not None:
        return _reranker if _reranker != "NONE" else None

    model_path = download_reranker()
    if model_path is None:
        _reranker = "NONE"
        return None

    try:
        from sentence_transformers import CrossEncoder
        print(f"[Reranker] 加载模型: {model_path}")
        _reranker = CrossEncoder(model_path, device="cpu")
        print(f"[Reranker] 加载完成")
        return _reranker
    except Exception as e:
        print(f"[Reranker] 加载失败: {e}")
        _reranker = "NONE"
        return None


# ============================================================================
# Section 3: 文档加载 + 数据清洗 + 章节切块 (优化1, 优化4)
# ============================================================================

def clean_markdown_escapes(text: str) -> str:
    """清理 Markdown 转义符。

    Markdown 文件中常见 \-、\.、\* 等转义符，会污染关键词匹配和
    Reranker 相关度计算（例如 "试用期\\-" 和 "试用期" 在 BM25 中是不同 token）。

    清理示例：
      9:00\\-18:00  → 9:00-18:00
      1\\.5小时    → 1.5小时
      V2\\.0       → V2.0
    """
    # 去掉反斜杠转义符：匹配 \<符号> 并只保留 <符号>
    # 字符类中的特殊符号需要逐个列出：- . * ? + [ ] ( ) { } ! ^ $ | #
    text = re.sub(r'\\([\-.*?+\[\](){}!^$|#])', r'\1', text)
    # 合并多余空行（>2 个连续换行）：避免 chunks 中出现大段空白
    text = re.sub(r'\n{3,}', '\n\n', text)
    return text


def load_and_chunk() -> list:
    """加载员工手册 Markdown 文件，按章节结构切块。

    优化1: 使用 XX科技有限公司员工手册.md
    优化4: 按 Markdown 章节（## 第X条）切块，保证每条语义完整

    切块策略：
      1. 读取 .md 文件全部文本
      2. **数据清洗**：清理 Markdown 转义符（9:00\\-18:00 → 9:00-18:00）
      3. 跳过目录，从 # 第一章 开始
      4. 按章节标题（## 第X条）切块，每个条款作为一个独立 chunk
      5. 过长的条款二次切分（保留章节上下文前缀）

    为什么不用 SemanticChunker？
      → SemanticChunker 在文档结构清晰时（如员工手册有章节）反而把相关内容
        合并过度，导致向量被稀释。按 Markdown 标题切能保证每个 chunk 语义完整。

    Returns:
        list[Document]: 文档块列表，每个包含 page_content 和 metadata
    """
    from langchain_core.documents import Document
    from langchain_text_splitters import RecursiveCharacterTextSplitter

    print(f"\n{'='*60}")
    print(f"[文档加载] 读取: {_DATA_PATH.name}")

    # Step 1: 读取文件
    raw_text = _DATA_PATH.read_text(encoding="utf-8")
    print(f"[文档加载] 原始长度: {len(raw_text)} 字符")

    # Step 2: 数据清洗 - 清理 Markdown 转义符
    cleaned_text = clean_markdown_escapes(raw_text)
    print(f"[文档加载] 清洗后长度: {len(cleaned_text)} 字符")

    # Step 3: 跳过目录，从正文开始
    body_start = cleaned_text.find("# 第一章")
    if body_start == -1:
        body_start = cleaned_text.find("# 第")
    if body_start > 0:
        body_text = cleaned_text[body_start:]
    else:
        body_text = cleaned_text

    # Step 4: 按章节结构切块
    # 每个 "## 第X条" 作为一个独立的 chunk，保证语义完整
    chunks = []
    current_chapter = ""
    current_section = ""
    current_section_lines = []

    def flush_section():
        """保存当前 section 为一个 Document。"""
        if not current_section or not current_section_lines:
            return
        content = "\n".join(current_section_lines).strip()
        if len(content) < 15:
            return
        # 跳过纯符号行
        if re.match(r"^[\s\-—#*=_.]+$", content):
            return
        # 附加上下文前缀：让 embedding 知道这段文字属于哪个章节
        full_content = f"【{current_chapter} - {current_section}】\n{content}"
        chunks.append(Document(
            page_content=full_content,
            metadata={
                "source": _DATA_PATH.name,
                "chapter": current_chapter,
                "section": current_section,
            }
        ))

    for line in body_text.split("\n"):
        # 检测章标题 (# 第X章 ...) — 一级标题，更新当前所在章
        chap_match = re.match(r"^#\s+(第[一二三四五六七八九十\d]+章\s*\S.*)", line)
        if chap_match:
            flush_section()  # 先保存上一节（章切换意味着上一节结束）
            current_chapter = chap_match.group(1).strip()
            current_section = ""  # 进入新章时清空 section，等遇到 ## 才开始记录
            current_section_lines = []
            continue
        # 检测条标题 (## 第X条 ...) — 二级标题，每个条款是一个独立 chunk
        sec_match = re.match(r"^##\s+(第[一二三四五六七八九十\d]+条\s*\S.*)", line)
        if sec_match:
            flush_section()  # 先保存上一节（条款切换意味着上一条结束）
            current_section = sec_match.group(1).strip()
            current_section_lines = []
            continue
        # 跳过分隔线 (---)：Markdown 的水平分隔符，对内容无意义
        if re.match(r"^---+\s*$", line):
            continue
        # 收集正文：只有在已进入某个 section 后才开始收集
        # （目录、文档头部说明等非条款内容会被跳过）
        if current_section:
            stripped = line.strip()
            if stripped:  # 跳过空行
                current_section_lines.append(stripped)

    flush_section()  # 保存最后一节（循环结束后内存中还残留最后一节未写入）

    print(f"[文档加载] 按章节切块完成: {len(chunks)} 个条款")

    # Step 5: 对过长的 section 二次切分（保留章节上下文）
    # 防止单个条款过长（>1200 字符）稀释向量
    final_chunks = []
    sub_splitter = RecursiveCharacterTextSplitter(
        chunk_size=800,
        chunk_overlap=100,
        separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""]
    )
    for c in chunks:
        if len(c.page_content) > 1200:
            sub_chunks = sub_splitter.split_documents([c])
            final_chunks.extend(sub_chunks)
        else:
            final_chunks.append(c)

    print(f"[文档加载] 最终块数量: {len(final_chunks)} 个 "
          f"(过长条款已二次切分)")

    if len(final_chunks) < 10:
        print(f"[文档加载] 警告: 块数量过少 ({len(final_chunks)})")

    return final_chunks


# ============================================================================
# Section 4: ChromaDB 向量存储 (优化3)
# ============================================================================

def build_vectorstore(chunks: list) -> object:
    """构建 ChromaDB 向量存储。

    使用 BGE-Large-Zh-v1.5 embedding 模型 (1024d)，collection 配置余弦相似度。
    每次调用强制重建（开发阶段）。
    """
    from langchain_community.vectorstores import Chroma

    print(f"\n[向量存储] 构建 ChromaDB ({len(chunks)} 个文档)")
    embedding_model = init_embedding_model()

    # 清理旧数据
    if _CHROMA_DIR.exists():
        print(f"[向量存储] 清理旧数据库: {_CHROMA_DIR}")
        shutil.rmtree(str(_CHROMA_DIR))

    vectorstore = Chroma.from_documents(
        documents=chunks,
        embedding=embedding_model,
        collection_metadata={"hnsw:space": "cosine"},
        persist_directory=str(_CHROMA_DIR),
    )
    print(f"[向量存储] 构建完成，持久化到: {_CHROMA_DIR}")
    return vectorstore


def get_vectorstore(chunks: list) -> object:
    """获取向量存储（优先加载已有，否则重建）。"""
    from langchain_community.vectorstores import Chroma

    sqlite_file = _CHROMA_DIR / "chroma.sqlite3"
    if sqlite_file.exists():
        embedding_model = init_embedding_model()
        try:
            print(f"[向量存储] 从 {_CHROMA_DIR} 加载已有数据库")
            return Chroma(
                persist_directory=str(_CHROMA_DIR),
                embedding_function=embedding_model,
                collection_metadata={"hnsw:space": "cosine"},
            )
        except Exception as e:
            print(f"[向量存储] 加载失败 ({e})，将重建")

    return build_vectorstore(chunks)


# ============================================================================
# Section 5: 多路召回系统 (优化5)
# ============================================================================

_bm25_retriever = None
_vector_retriever = None
_ensemble_retriever = None
_mq_retriever = None

def _jieba_tokenize(text: str) -> list[str]:
    """使用 jieba 对中文文本分词。

    BM25Retriever 默认用英文空白分词器，遇到中文时会将整段文本当作一个 token，
    导致完全无法匹配关键词（不同查询返回相同结果）。
    用 jieba 分词后，"试用期是几个月" 会被切成 ["试用期", "是", "几个", "月"]，
    BM25 才能正常按词频打分。
    """
    import jieba
    # lcut 精确模式（vs cut 全模式）：返回 list，每个 token 是一个词
    tokens = jieba.lcut(text)
    # 过滤规则：去空白 + 丢弃单字
    # 单字（如"的""是""一"）信息量极低，且大量重复，会拉低 BM25 区分度
    return [t for t in tokens if t.strip() and len(t.strip()) > 1]


def create_bm25_retriever(chunks: list):
    """创建 BM25 关键词检索器。

    BM25 基于词频统计，擅长精确关键词匹配，
    弥补向量检索在专有名词、数字、缩写上的短板。

    注意：必须传入中文分词器 _jieba_tokenize，否则默认英文分词器
    会把整段中文当成一个 token，BM25 退化为按原始顺序返回。
    """
    global _bm25_retriever
    if _bm25_retriever is not None:
        return _bm25_retriever
    from langchain_community.retrievers import BM25Retriever
    _bm25_retriever = BM25Retriever.from_documents(
        chunks,
        preprocess_func=_jieba_tokenize,
    )
    _bm25_retriever.k = RETRIEVAL_K
    print(f"[检索器] BM25 初始化完成 (k={RETRIEVAL_K}, 中文分词=jieba)")
    return _bm25_retriever


def create_vector_retriever(vectorstore):
    """创建向量语义检索器。

    基于 BGE-Large-Zh-v1.5 embedding 的余弦相似度检索，
    能在语义层面理解同义词、近义词和概念关联。
    """
    global _vector_retriever
    if _vector_retriever is not None:
        return _vector_retriever
    _vector_retriever = vectorstore.as_retriever(
        search_type="similarity",
        search_kwargs={"k": RETRIEVAL_K},
    )
    print(f"[检索器] 向量检索器初始化完成 (k={RETRIEVAL_K})")
    return _vector_retriever


def create_ensemble_retriever(chunks, vectorstore):
    """创建混合检索器: BM25 + 向量检索融合。

    - BM25 权重 0.3: 关键词精确匹配
    - 向量权重 0.7: 语义理解（权重更高，因为语义匹配通常更重要）

    融合方式: 分数归一化后按权重加权求和。
    """
    global _ensemble_retriever
    if _ensemble_retriever is not None:
        return _ensemble_retriever
    from langchain_classic.retrievers import EnsembleRetriever

    bm25 = create_bm25_retriever(chunks)
    vec = create_vector_retriever(vectorstore)
    _ensemble_retriever = EnsembleRetriever(
        retrievers=[bm25, vec],
        weights=ENSEMBLE_WEIGHTS,
    )
    print(f"[检索器] 混合检索器初始化完成 "
          f"(BM25:{ENSEMBLE_WEIGHTS[0]} + 向量:{ENSEMBLE_WEIGHTS[1]})")
    return _ensemble_retriever


def create_mq_retriever(vectorstore):
    """创建多查询检索器 (Multi-Query Retriever)。

    LLM 自动生成 3 个查询变体 → 分别检索 → 合并去重。
    适合处理用户问题表述不够精确的场景。
    """
    global _mq_retriever
    if _mq_retriever is not None:
        return _mq_retriever
    from langchain_classic.retrievers import MultiQueryRetriever

    vec = create_vector_retriever(vectorstore)
    _mq_retriever = MultiQueryRetriever.from_llm(
        retriever=vec,
        llm=get_llm(),
    )
    print(f"[检索器] 多查询检索器初始化完成 (变体数={MULTI_QUERY_VARIANTS})")
    return _mq_retriever


def retrieve(query: str, strategy: str = "ensemble",
             top_k: int = DEFAULT_K) -> list:
    """统一的检索调度接口。

    Args:
        query: 用户问题
        strategy: 检索策略，可选 "bm25" | "vector" | "ensemble" | "multi_query"
        top_k: 返回的文档数量

    Returns:
        list[Document]: 检索到的文档列表
    """
    vs = _get_vectorstore_ref()
    chunks = _get_chunks_ref()

    if strategy == "bm25":
        ret = create_bm25_retriever(chunks)
    elif strategy == "vector":
        ret = create_vector_retriever(vs)
    elif strategy == "multi_query":
        ret = create_mq_retriever(vs)
    else:
        ret = create_ensemble_retriever(chunks, vs)

    # 为不同检索器适配 k 参数
    # BM25Retriever / VectorStoreRetriever 用 .k 属性控制返回数量
    # MultiQueryRetriever / EnsembleRetriever 没有可变 .k，需要 invoke 后切片
    if hasattr(ret, "k"):
        # 临时修改 k → 调用 → 恢复原值（防止影响后续其他调用）
        old_k = ret.k
        ret.k = top_k
        docs = ret.invoke(query)
        ret.k = old_k
    else:
        docs = ret.invoke(query)
        docs = docs[:top_k]

    return docs


# 模块级引用（由主流程设置）
_VS = None
_CHUNKS = None

def _get_vectorstore_ref():
    if _VS is None:
        raise RuntimeError("向量存储未初始化，请先调用 load_and_chunk() 和 get_vectorstore()")
    return _VS

def _get_chunks_ref():
    if _CHUNKS is None:
        raise RuntimeError("文档块未初始化")
    return _CHUNKS


# ============================================================================
# Section 6: 查询增强 (优化6)
# ============================================================================

def rewrite_query(original: str) -> str:
    """查询重写：用 LLM 将用户问题改写为更精确的检索查询。

    改写策略:
    - 提取核心实体和关键词
    - 消除歧义，明确指代
    - 将口语化表达转为文档化风格
    """
    llm = get_llm()
    prompt = f"""你是一个查询优化助手。给定用户的原始问题，请将其改写成更适合文档检索的形式。

要求：
1. 提取核心实体和关键词
2. 移除歧义，明确指代
3. 保持原意不变
4. 使用文档中可能出现的术语（如"年假""加班费""绩效考核"等正式表述）
5. 直接输出改写后的查询，不要任何解释

原始问题：{original}

改写后的查询："""

    try:
        raw = llm.invoke(prompt)
        rewritten = raw.content if hasattr(raw, "content") else str(raw)
        rewritten = rewritten.strip().strip('"').strip("'")
        if len(rewritten) < 2:
            return original
        return rewritten
    except Exception as e:
        print(f"[查询重写] 失败 ({e})，使用原始查询")
        return original


def hyde_retrieve(query: str, vectorstore, top_k: int = RETRIEVAL_K) -> list:
    """HyDE (Hypothetical Document Embeddings) 检索。

    核心思路: LLM 先生成一个「假设性答案」→ 将这个答案向量化 →
    用这个向量去检索 → 假设性答案的语义空间比问题本身更接近真实文档。

    流程:
      1. LLM 生成假设性答案（不需要准确，只需"看起来像"文档内容）
      2. Embedding 模型将假设性答案转为向量
      3. 在 ChromaDB 中检索相似文档
    """
    llm = get_llm()
    prompt = f"""请根据以下问题，生成一个假设性的员工手册回答。这个回答不需要完全准确，
只需要看起来像是一段真实的公司制度文档内容。用来帮助搜索引擎找到相关文档。

要求：直接输出内容，不要任何前缀或解释。

问题：{query}

假设文档："""

    try:
        raw = llm.invoke(prompt)
        hypothesis = raw.content if hasattr(raw, "content") else str(raw)
        hypothesis = hypothesis.strip()
        if len(hypothesis) < 10:
            return []

        docs = vectorstore.similarity_search(hypothesis, k=top_k)
        return docs
    except Exception as e:
        print(f"[HyDE] 失败 ({e})")
        return []


def reciprocal_rank_fusion(doc_lists: list[list], k: int = RRF_K) -> list:
    """倒数秩融合 (Reciprocal Rank Fusion)。

    将多个检索结果列表按 RRF 算法合并为一个去重排序列表。
    公式: score(d) = Σ 1/(rank_i(d) + k)

    Args:
        doc_lists: 多个检索结果列表
        k: 平滑参数（默认60，标准设置）

    Returns:
        融合后按 RRF 分数降序排列的文档列表
    """
    scores: dict[int, float] = {}
    doc_map: dict[int, object] = {}

    # Step 1: 遍历每个检索器返回的列表，按排名累计 RRF 分数
    for doc_list in doc_lists:
        for rank, doc in enumerate(doc_list):  # rank 从 0 开始，表示该文档在该列表中的位置
            # 用 page_content 的 hash 作为文档唯一标识（同一段文字会得到相同 hash）
            doc_id = hash(doc.page_content)
            if doc_id not in scores:
                scores[doc_id] = 0.0
                doc_map[doc_id] = doc
            # RRF 核心公式：score(d) = Σ 1/(rank + k)
            #   rank 从 0 开始，所以要 +1 才是真正的「第几名」
            #   k 是平滑常数：避免 rank=0 的文档得到 1.0 满分，让排名靠后的也有贡献
            #   k=60 是业界标准值，源自论文《Reciprocal Rank Fusion outperforms Condorcet》
            scores[doc_id] += 1.0 / (rank + k)

    # Step 2: 按 RRF 分数降序排列，输出融合后的文档列表
    sorted_ids = sorted(scores.keys(), key=lambda x: scores[x], reverse=True)
    return [doc_map[did] for did in sorted_ids]


def enhanced_retrieve(query: str, strategy: str = "ensemble",
                      top_k: int = DEFAULT_K) -> list:
    """增强检索: 查询重写 + HyDE + RRF 融合。

    流程:
      1. LLM 重写查询 → 得到更精确的检索查询
      2. HyDE 生成假设答案 → 从另一个语义角度检索
      3. 标准检索（用重写后的查询）
      4. RRF 融合 HyDE 结果 + 标准结果 → 取 top_k
    """
    vectorstore = _get_vectorstore_ref()

    # Step 1: 查询重写
    rewritten = rewrite_query(query)
    if rewritten != query:
        print(f"[增强检索] 查询重写: 「{query[:40]}...」→「{rewritten[:60]}...」")

    # Step 2: HyDE 检索
    hyde_docs = hyde_retrieve(query, vectorstore, top_k=RETRIEVAL_K)
    print(f"[增强检索] HyDE 召回 {len(hyde_docs)} 个文档")

    # Step 3: 标准检索（使用重写后的查询）
    standard_docs = retrieve(rewritten, strategy=strategy, top_k=RETRIEVAL_K)
    print(f"[增强检索] {strategy} 召回 {len(standard_docs)} 个文档")

    # Step 4: RRF 融合
    if hyde_docs and standard_docs:
        fused = reciprocal_rank_fusion([standard_docs, hyde_docs])
        print(f"[增强检索] RRF 融合后 {len(fused)} 个文档")
    elif standard_docs:
        fused = standard_docs
    else:
        fused = hyde_docs

    return fused[:top_k]


# ============================================================================
# Section 7: Reranker 精排 (优化7)
# ============================================================================

def rerank(query: str, candidates: list, top_k: int = DEFAULT_K) -> list:
    """使用 BGE-Reranker 对候选文档精排。

    Cross-Encoder 将 query-doc 对联合编码，比 embedding 相似度更精确。
    缺点: 速度较慢，适合对 Top-N 候选做精排（N 通常 10-20）。

    Args:
        query: 用户问题
        candidates: 候选文档列表
        top_k: 返回的文档数量

    Returns:
        按相关度分数降序排列的文档列表
    """
    reranker = init_reranker()
    if reranker is None or len(candidates) == 0:
        return candidates[:top_k]

    # 优化：候选数 ≤ top_k 时无需精排，直接返回（省下推理时间）
    if len(candidates) <= top_k:
        return candidates

    try:
        # 构建 query-doc 配对：CrossEncoder 是双塔模型的对比
        # 与 embedding（Bi-Encoder，query 和 doc 分别编码再算相似度）不同，
        # CrossEncoder 把 query 和 doc 拼在一起送入模型，深度交互注意力，
        # 因此精度更高但速度更慢（适合 Top-N 精排，N 通常 10-20）
        pairs = [(query, doc.page_content) for doc in candidates]
        scores = reranker.predict(pairs, show_progress_bar=False)

        # 按相关度分数降序排列，取 top_k
        # scored 是 [(score, doc), ...]，sort 后取前 top_k 个的 doc
        scored = list(zip(scores, candidates))
        scored.sort(key=lambda x: x[0], reverse=True)

        reranked = [doc for _, doc in scored[:top_k]]
        print(f"[Reranker] 精排完成: {len(candidates)} → {len(reranked)} "
              f"(最高分: {scored[0][0]:.4f})")
        return reranked
    except Exception as e:
        print(f"[Reranker] 精排失败 ({e})，使用原始顺序")
        return candidates[:top_k]


# ============================================================================
# Section 8: Agent 工具封装 (优化2)
# ============================================================================

def create_retrieval_tool():
    """将检索系统封装为 LangChain Tool。

    Agent 可根据问题自动判断是否需要调用此工具检索员工手册。
    """
    from langchain_core.tools import tool

    @tool
    def search_employee_handbook(query: str) -> str:
        """从《XX科技有限公司员工手册》中检索与 query 相关的内容。

        当用户问到以下内容时使用此工具：
        - 公司制度、考勤规定、请假流程
        - 薪资福利、社保公积金
        - 加班管理、出差报销
        - 绩效考核、晋升条件
        - 离职手续、保密规定
        - 员工关怀、培训发展
        - 入职流程、试用期规定

        输入：用户的问题或关键词。
        输出：手册中的相关章节内容。
        """
        # 使用增强检索 + 精排
        docs = enhanced_retrieve(query, strategy="ensemble", top_k=RETRIEVAL_K)
        docs = rerank(query, docs, top_k=DEFAULT_K)

        parts = []
        for i, doc in enumerate(docs):
            parts.append(f"【参考条款 {i+1}】\n{doc.page_content}")
        return "\n\n---\n\n".join(parts)

    return search_employee_handbook


def create_rag_agent():
    """创建 RAG Agent。

    Agent 能自动判断:
    - 需要查手册时 → 调用 search_employee_handbook 工具
    - 闲聊/通用问题 → 直接回答
    """
    from langchain.agents import create_agent

    tool = create_retrieval_tool()
    llm = get_llm()
    agent = create_agent(
        model=llm,
        tools=[tool],
        system_prompt=(
            "你是XX科技有限公司的员工手册问答助手。\n\n"
            "规则：\n"
            "1. 当用户问到公司制度、考勤、请假、薪资、福利、加班、离职、保密等"
            "规定时，必须先调用 search_employee_handbook 工具检索手册内容。\n"
            "2. 基于检索到的内容回答，不要编造。如果手册中没有相关信息，"
            "如实告知用户。\n"
            "3. 回答要简洁清晰，引用手册中的具体条款编号。\n"
            "4. 对于闲聊或与员工手册无关的问题，直接回答即可。"
        ),
    )
    print(f"[Agent] RAG Agent 创建完成")
    return agent


# ============================================================================
# Section 9: 完整 RAG 链路
# ============================================================================

def build_context(docs: list) -> str:
    """将检索到的文档拼接为 LLM 上下文。"""
    parts = []
    for i, doc in enumerate(docs):
        parts.append(f"[参考文档 {i+1}]\n{doc.page_content}")
    return "\n\n---\n\n".join(parts)


def rag_query(query: str, strategy: str = "ensemble",
              use_enhance: bool = True, use_rerank: bool = True) -> dict:
    """确定性 RAG 查询流水线（不使用 Agent）。

    Args:
        query: 用户问题
        strategy: 检索策略
        use_enhance: 是否使用查询增强（重写 + HyDE）
        use_rerank: 是否使用 Reranker 精排

    Returns:
        dict: {"answer": str, "sources": list[Document], "strategy": str}
    """
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_core.output_parsers import StrOutputParser

    # Step 1: 检索（增强检索走 enhanced_retrieve，普通检索走 retrieve）
    if use_enhance:
        docs = enhanced_retrieve(query, strategy=strategy, top_k=RETRIEVAL_K)
    else:
        docs = retrieve(query, strategy=strategy, top_k=RETRIEVAL_K)

    # Step 2: 精排（可选）：对召回的 top-K 候选重新打分排序，取 top-k
    if use_rerank:
        docs = rerank(query, docs, top_k=DEFAULT_K)

    # Step 3: 组装 prompt
    #   ChatPromptTemplate.from_messages: 支持 system + user 多轮对话格式
    #   {context} 和 {query} 是占位符，invoke 时填充
    context = build_context(docs)
    prompt = ChatPromptTemplate.from_messages([
        ("system", """你是XX科技有限公司的员工手册问答助手。请根据以下参考资料回答问题。

**规则：**
- 只基于提供的参考资料回答，不要编造
- 如果参考资料中没有相关信息，直接说「根据现有员工手册，我找不到这个问题的答案」
- 回答要简洁清晰，引用具体条款时注明出处
- 如果参考资料不足以回答，可以说明需要查阅手册的哪个章节

**参考资料：**
{context}"""),
        ("user", "{query}")
    ])

    # Step 4: 生成
    # LCEL 链式语法：prompt | llm | parser
    #   - prompt.invoke({...}) → 填充占位符生成最终 prompt
    #   - llm.invoke(prompt) → 调用 LLM 生成回复
    #   - StrOutputParser().invoke(resp) → 从 AIMessage 中提取纯文本
    # 等价于：parser.invoke(llm.invoke(prompt.invoke({...})))
    llm = get_llm()
    chain = prompt | llm | StrOutputParser()
    answer = chain.invoke({"context": context, "query": query})

    return {"answer": answer, "sources": docs, "strategy": strategy}


def agent_query(query: str) -> str:
    """Agent 模式查询（自动判断是否需要检索）。

    Agent 根据问题内容自动决定:
    - 调用 search_employee_handbook 工具检索
    - 直接回答（不检索）
    """
    agent = create_rag_agent()
    result = agent.invoke({"messages": [("user", query)]})
    # 提取最后一条消息
    messages = result.get("messages", [])
    if messages:
        return messages[-1].content
    return "Agent 未返回结果"


# ============================================================================
# Section 10: RAG 效果评估体系 (优化8)
# ============================================================================
#
# 评估策略说明：
# 1. 手动标注测试集: 覆盖员工手册全部 16 章的真实问题 → Section 级检索评估
# 2. 检索交叉验证: LLM 逐条判断检索到的文档能否回答该问题
# 3. LLM 辅助生成评估 + Bad Case 根因分析

# ========================================================================
# 10a. 手动标注真实测试集（覆盖全部 16 章，~30% 文档覆盖率）
# ========================================================================
# 每条测试用例包含:
#   question:          员工真实会问的问题
#   section_keywords:  期望匹配的章节关键词（用于 Section 级检索评估）
#   reference_answer:  从手册原文提取的参考答案（用于生成评估）

_MANUAL_TEST_SET = [
    # === 第一章 总则 ===
    {
        "question": "员工手册制定的法律依据是什么？",
        "section_keywords": ["总则", "劳动法", "劳动合同法"],
        "reference_answer": "根据《中华人民共和国劳动法》《中华人民共和国劳动合同法》及相关法律法规制定。",
    },
    # === 第二章 入职与试用期 ===
    {
        "question": "试用期是多长时间？",
        "section_keywords": ["试用期", "入职"],
        "reference_answer": "新员工试用期为3个月，试用期包含在劳动合同期限内。",
    },
    {
        "question": "入职需要准备哪些材料？",
        "section_keywords": ["入职", "入职流程", "报到"],
        "reference_answer": "入职需准备：身份证、学历学位证书、离职证明、体检报告、银行卡。",
    },
    # === 第三章 工作时间与考勤 ===
    {
        "question": "公司标准工作时间是怎样的？",
        "section_keywords": ["工作时间", "考勤", "上下班"],
        "reference_answer": "标准工作时间为周一至周五9:00-18:00，午休1小时，每天工作8小时。",
    },
    # === 第四章 假期管理 ===
    {
        "question": "年假没休完可以结转到下一年吗？",
        "section_keywords": ["年假", "带薪年假", "假期"],
        "reference_answer": "当年未休完的年假最多可结转5天至次年3月31日，逾期作废。",
    },
    {
        "question": "请病假需要提供什么材料？",
        "section_keywords": ["病假", "请假", "假期管理"],
        "reference_answer": "请病假需提供二级甲等以上医院出具的病假证明和诊断证明。",
    },
    # === 第五章 薪资福利 ===
    {
        "question": "社保养老保险个人缴纳比例是多少？",
        "section_keywords": ["社保", "社会保险", "五险一金", "薪资福利"],
        "reference_answer": "养老保险个人缴纳比例为8%，单位缴纳比例为16%。",
    },
    # === 第六章 绩效考核 ===
    {
        "question": "绩效考核结果分几个等级？",
        "section_keywords": ["绩效考核", "考核等级", "绩效"],
        "reference_answer": "绩效考核分为S/A/B/C/D五个等级，其中S级占比不超过10%。",
    },
    # === 第七章 培训与发展 ===
    {
        "question": "公司每年要求的最低培训学时是多少？",
        "section_keywords": ["培训", "培训与发展", "学习"],
        "reference_answer": "每位员工每年需完成不低于40学时的培训，包括技术培训、管理培训和软技能培训。",
    },
    # === 第八章 晋升制度 ===
    {
        "question": "员工晋升需要满足哪些基本条件？",
        "section_keywords": ["晋升", "晋升制度", "职业发展"],
        "reference_answer": "晋升需满足：连续两次绩效考核B+以上，在现岗位工作满1年以上。",
    },
    # === 第九章 加班管理 ===
    {
        "question": "周末加班工资按几倍计算？",
        "section_keywords": ["加班", "加班费", "加班工资"],
        "reference_answer": "休息日（周末）加班，按200%（2倍）工资支付加班费。",
    },
    # === 第十章 出差与报销 ===
    {
        "question": "出差一线城市住宿标准多少钱一天？",
        "section_keywords": ["出差", "报销", "住宿标准"],
        "reference_answer": "一线城市（北上广深）出差住宿标准为500元/天。",
    },
    # === 第十一章 劳动合同 ===
    {
        "question": "第一次签劳动合同的期限是几年？",
        "section_keywords": ["劳动合同", "合同期限", "签订"],
        "reference_answer": "首次签订劳动合同期限为3年，续签第二次为5年，之后可签无固定期限合同。",
    },
    # === 第十二章 奖惩制度 ===
    {
        "question": "员工连续旷工几天会被开除？",
        "section_keywords": ["旷工", "处罚", "奖惩", "开除", "严重违纪"],
        "reference_answer": "连续旷工3天以上或一年内累计旷工5天以上，属于严重违纪，公司可单方解除劳动合同。",
    },
    # === 第十三章 保密规定 ===
    {
        "question": "公司商业机密包括哪些内容？",
        "section_keywords": ["保密", "商业机密", "保密规定"],
        "reference_answer": "商业机密包括技术资料、客户信息、财务数据、经营决策、未公开的人事信息等。",
    },
    # === 第十四章 离职管理 ===
    {
        "question": "离职需要提前多少天申请？",
        "section_keywords": ["离职", "离职管理", "解除"],
        "reference_answer": "正式员工离职需提前30天书面申请，试用期员工提前3天申请。",
    },
    # === 第十五章 员工关怀 ===
    {
        "question": "公司有哪些员工关怀和福利项目？",
        "section_keywords": ["员工关怀", "员工福利", "关怀"],
        "reference_answer": "公司提供补充商业保险、年度体检、节日礼品、团队建设、生日福利、婚育贺礼、困难补助等。",
    },
    # === 第十六章 附则 ===
    {
        "question": "员工手册的解释权归哪个部门？",
        "section_keywords": ["附则", "解释权", "人力资源部"],
        "reference_answer": "本手册的最终解释权归公司人力资源部所有。",
    },
    {
        "question": "员工手册从什么时候开始生效？",
        "section_keywords": ["附则", "生效", "施行"],
        "reference_answer": "本手册自2025年1月1日起正式生效施行。",
    },
]

def get_manual_test_set() -> list[dict]:
    """获取手动标注的真实测试集。

    这些测试用例是基于员工手册原文内容手动编写的，用于客观评估 RAG 系统。
    与 LLM 自动生成的测试集不同，这里的问题和答案都是人工确定的，
    不存在"同一 LLM 既当选手又当裁判"的问题。
    """
    return _MANUAL_TEST_SET


# ========================================================================
# 10b. Section 级检索评估（核心指标）
# ========================================================================

def evaluate_retrieval_section_level(test_set: list[dict]) -> dict:
    """Section 级检索评估 — 客观评估检索质量。

    评估方法:
      对每个问题，使用各检索策略检索 Top-K 文档，检查检索到的文档
      是否来自员工手册的正确章节（通过章节关键词匹配）。

    为什么不用"同一个 chunk"检测？
      → 那是循环验证：LLM 从 chunk A 生成问题 → 检查是否检索到 chunk A
      → 毫无意义，因为 embedding 向量天然相近。

    为什么用 Section 级匹配？
      → 比如问"试用期"，理应检索到"第二章 入职与试用期"的内容
      → 如果检索到了"第四章 假期管理"，说明检索跑偏了
      → 这能真正区分不同检索策略的效果

    Returns:
        dict: 每种策略的 {"precision@3", "precision@5", "section_recall", "details"}
    """
    print(f"\n{'='*60}")
    print(f"[检索评估] Section 级评估 (测试 {len(test_set)} 个真实查询)")
    print(f"[检索评估] 衡量检索到的文档是否来自正确章节")

    strategies = ["bm25", "vector", "ensemble", "multi_query", "enhanced"]
    all_results = {}

    for strategy in strategies:
        per_query = []

        for qa in test_set:
            query = qa["question"]
            expected_keywords = qa["section_keywords"]

            # "enhanced" 用增强检索链（重写+HyDE+RRF，是实际生产 pipeline）
            if strategy == "enhanced":
                docs = enhanced_retrieve(query, strategy="vector", top_k=10)
            else:
                docs = retrieve(query, strategy=strategy, top_k=10)

            # 对每个检索到的 doc，检查其内容是否包含期望的章节关键词
            top3_hits = 0
            top5_hits = 0
            top3_texts = []
            for rank, doc in enumerate(docs):
                doc_text = doc.page_content
                # 检查文档中是否包含任何期望的章节关键词
                matched = any(kw in doc_text for kw in expected_keywords)
                if matched:
                    if rank < 3:
                        top3_hits += 1
                    if rank < 5:
                        top5_hits += 1
                if rank < 3:
                    top3_texts.append(doc_text[:80])

            per_query.append({
                "query": query[:60],
                "expected_sections": expected_keywords,
                "top3_hits": top3_hits,
                "top5_hits": top5_hits,
                "top3_preview": top3_texts,
            })

        n = len(test_set)
        precision_3 = sum(q["top3_hits"] for q in per_query) / (n * 3)  # 理想情况是 top3 中每篇都命中
        precision_5 = sum(q["top5_hits"] for q in per_query) / (n * 5)
        # Section Recall: 至少有一篇文档命中正确章节的查询比例
        section_recall = sum(1 for q in per_query if q["top3_hits"] > 0) / n

        all_results[strategy] = {
            "precision_at_3": precision_3,
            "precision_at_5": precision_5,
            "section_recall": section_recall,
            "details": per_query,
        }

        print(f"[检索评估] {strategy:<15s} | P@3: {precision_3:.1%} | "
              f"P@5: {precision_5:.1%} | Section Recall: {section_recall:.1%}")

    return all_results


# ========================================================================
# 10c. 检索交叉验证
# ========================================================================

def evaluate_retrieval_cross_verify(test_set: list[dict]) -> dict:
    """检索交叉验证: LLM 逐条判断检索到的文档是否能回答问题。

    与 Section 级匹配互补:
      - Section 匹配是"粗粒度"的（文档来自正确章节？）
      - 交叉验证是"细粒度"的（文档内容真的能回答这个问题？）

    方法:
      对每个查询，取检索到的 Top-3 文档，让 LLM 逐条判断:
      "这篇文档是否包含回答该问题所需的信息？"
      → 是(1) / 部分(0.5) / 否(0)

    优势: 不需要人工标注参考答案，LLM 只做"是否相关"的二元判断，
          比对比参考答案更客观。
    """
    print(f"\n{'='*60}")
    print(f"[交叉验证] LLM 逐条判断检索文档是否相关")

    strategies = ["bm25", "vector", "ensemble", "multi_query"]
    all_results = {}
    llm = get_llm()

    for strategy in strategies:
        doc_scores = []

        for qi, qa in enumerate(test_set):
            query = qa["question"]
            docs = retrieve(query, strategy=strategy, top_k=3)

            for rank, doc in enumerate(docs):
                verify_prompt = f"""判断以下文档片段是否包含回答该问题所需的信息。

问题：{query}
文档：{doc.page_content[:400]}

只回答数字: 1=包含关键信息 / 0.5=部分相关 / 0=无关"""

                try:
                    raw = llm.invoke(verify_prompt)
                    ans = raw.content if hasattr(raw, "content") else str(raw)
                    score = float(ans.strip()[0]) if ans.strip() else 0
                    score = max(0, min(1, score))
                except Exception:
                    score = 0

                doc_scores.append({
                    "query": query[:50], "strategy": strategy,
                    "rank": rank + 1, "relevant": score,
                })

        n = len(test_set) * 3  # 每个查询 3 篇文档
        avg_relevance = sum(d["relevant"] for d in doc_scores) / max(1, n)
        perfect_ratio = sum(1 for d in doc_scores if d["relevant"] == 1) / max(1, n)

        all_results[strategy] = {
            "avg_relevance": avg_relevance,
            "perfect_ratio": perfect_ratio,
            "details": doc_scores,
        }

        print(f"[交叉验证] {strategy:<15s} | 平均相关度: {avg_relevance:.2%} | "
              f"完美相关比: {perfect_ratio:.1%}")

    return all_results


# ========================================================================
# 10d. 生成质量评估 + Bad Case 分析
# ========================================================================

def evaluate_generation(test_set: list[dict], sample_size: int = 10) -> dict:
    """生成效果评估: 准确率、完整度、相关度、幻觉率。

    对每个测试问题:
      1. RAG 生成答案
      2. LLM judge 对比"参考答案"和"生成答案"
      3. 从 4 个维度打分

    注意: 这个评估的可靠性取决于 LLM judge 的质量，建议作为辅助指标。
    """
    import random

    print(f"\n{'='*60}")
    print(f"[生成评估] LLM Judge 打分（辅助指标）")

    if len(test_set) > sample_size:
        samples = random.sample(test_set, sample_size)
    else:
        samples = test_set

    llm = get_llm()
    judge_results = []
    bad_cases = []

    for i, qa in enumerate(samples):
        query = qa["question"]
        reference = qa["reference_answer"]

        # RAG 生成答案
        rag_result = rag_query(query, use_enhance=True, use_rerank=True)
        generated = rag_result["answer"]

        # 同时获取检索结果，用于 Bad Case 根因分析
        retrieved_docs = retrieve(query, strategy="ensemble", top_k=5)

        # LLM judge
        judge_prompt = f"""你是一个严格但公正的评估助手。请对比参考答案和RAG系统生成的答案。

参考答案（来自员工手册原文）：{reference}
RAG生成答案：{generated}

请从以下维度严格打分（1-5分，5分最好）:
- accuracy（事实准确性）：生成答案的事实是否与参考答案一致？（有事实错误扣分）
- completeness（完整性）：是否覆盖了参考答案的关键信息？（遗漏重要信息扣分）
- relevance（相关性）：是否直接回答了问题？（答非所问扣分）
- hallucination（幻觉程度）：是否编造了参考答案中没有的内容？（5=无编造，1=严重编造）

输出严格的JSON：
{{"accuracy": 整数1-5, "completeness": 整数1-5, "relevance": 整数1-5, "hallucination": 整数1-5}}"""

        try:
            raw_resp = llm.invoke(judge_prompt)
            raw_text = raw_resp.content if hasattr(raw_resp, "content") else str(raw_resp)
            json_match = re.search(r'\{[\s\S]*\}', raw_text)
            if json_match:
                scores = json.loads(json_match.group())
                scores["query"] = query
                scores["reference"] = reference
                scores["generated"] = generated[:300]
                judge_results.append(scores)

                # 识别 Bad Case
                is_bad = (scores.get("accuracy", 5) <= 2 or
                          scores.get("hallucination", 5) <= 2)
                if is_bad:
                    # 根因分析
                    retrieved_texts = [d.page_content[:200] for d in retrieved_docs[:3]]
                    # 检查检索结果中是否有与 reference 相关的内容
                    retrieval_ok = any(
                        any(kw in " ".join(retrieved_texts) for kw in qa.get("section_keywords", []))
                    )
                    root_cause = "生成幻觉" if retrieval_ok else "检索失败"
                    bad_cases.append({
                        "query": query,
                        "reference": reference,
                        "generated": generated[:200],
                        "root_cause": root_cause,
                        "retrieved_preview": retrieved_texts[:2],
                        "scores": scores,
                    })

                print(f"[生成评估] [{i+1}/{len(samples)}] "
                      f"acc={scores.get('accuracy','?')} "
                      f"hall={scores.get('hallucination','?')} "
                      f"{'⚠️ BAD' if is_bad else '✅'}")

        except Exception as e:
            print(f"[生成评估] 第 {i+1} 个评判失败: {e}")

    # 汇总指标
    if judge_results:
        avg_acc = sum(r.get("accuracy", 0) for r in judge_results) / len(judge_results)
        avg_comp = sum(r.get("completeness", 0) for r in judge_results) / len(judge_results)
        avg_rel = sum(r.get("relevance", 0) for r in judge_results) / len(judge_results)
        avg_hall = sum(r.get("hallucination", 0) for r in judge_results) / len(judge_results)
        hall_rate = sum(1 for r in judge_results if r.get("hallucination", 5) <= 2) / len(judge_results)
        bad_rate = len(bad_cases) / len(judge_results)
    else:
        avg_acc = avg_comp = avg_rel = avg_hall = hall_rate = bad_rate = 0

    print(f"\n[生成评估] 汇总:")
    print(f"  准确率:  {avg_acc:.2f}/5  |  完整度: {avg_comp:.2f}/5")
    print(f"  相关度:  {avg_rel:.2f}/5  |  抗幻觉: {avg_hall:.2f}/5")
    print(f"  幻觉率:  {hall_rate:.1%}   |  Bad率:  {bad_rate:.1%}")

    return {
        "avg_accuracy": avg_acc,
        "avg_completeness": avg_comp,
        "avg_relevance": avg_rel,
        "avg_anti_hallucination": avg_hall,
        "hallucination_rate": hall_rate,
        "bad_case_rate": bad_rate,
        "bad_cases": bad_cases,
        "details": judge_results,
    }


def bad_case_analysis(gen_results: dict) -> list[dict]:
    """Bad Case 深度分析与优化建议。

    对每个 Bad Case 按根因分类，给出针对性优化方向。
    """
    bad_cases = gen_results.get("bad_cases", [])

    print(f"\n{'='*60}")
    print(f"[Bad Case 分析] 根因定位")

    if not bad_cases:
        print("[Bad Case 分析] ✅ 所有案例表现良好！")
        return []

    # 按根因分类
    categories = defaultdict(list)
    for case in bad_cases:
        categories[case["root_cause"]].append(case)

    print(f"[Bad Case 分析] 共 {len(bad_cases)} 个 Bad Case:\n")

    for root_cause, cases in sorted(categories.items()):
        print(f"  📌 [{root_cause}] ({len(cases)} 例)")
        for case in cases[:1]:
            print(f"     ❓ {case['query'][:60]}")
            print(f"     🤖 {case['generated'][:80]}...")
            print()

    # 优化建议
    print(f"[Bad Case 分析] 针对性优化建议:")
    if "检索失败" in categories:
        print(f"  🔧 检索失败 ({len(categories['检索失败'])}例):")
        print(f"     → 降低 SemanticChunker 的 percentile 值（更细粒度切块）")
        print(f"     → 增加 RETRIEVAL_K 召回量")
        print(f"     → 考虑换用更大模型或调整 chunk 粒度")
        print(f"     → 检查查询重写是否改变了语义")
    if "生成幻觉" in categories:
        print(f"  🔧 生成幻觉 ({len(categories['生成幻觉'])}例):")
        print(f"     → 在 prompt 中加强对「不知道就说不知道」的约束")
        print(f"     → 降低 LLM temperature（当前默认值）")
        print(f"     → 要求逐句标注引用来源")
        print(f"     → 考虑接入 Reranker 精排后再生成")

    return bad_cases


# ========================================================================
# 10e. 完整评估入口
# ========================================================================

def run_full_evaluation() -> dict:
    """运行完整评估流程。

    三层评估:
      第一层: Section 级检索评估 → 客观指标（不依赖 LLM judge）
      第二层: 检索交叉验证 → LLM 逐篇判断相关性
      第三层: 生成质量评估 → LLM judge 打分 + Bad Case 根因分析

    Returns:
        dict: 完整评估报告
    """
    print(f"\n{'#'*60}")
    print(f"# RAG 效果评估（三层体系）")
    print(f"{'#'*60}")

    # 使用手动标注的真实测试集（覆盖全部 16 章）
    test_set = get_manual_test_set()
    print(f"[评估] 手动标注测试集: {len(test_set)} 条，覆盖全部 16 章")

    # ---- 第一层: Section 级检索评估 ----
    section_results = evaluate_retrieval_section_level(test_set)

    # ---- 第二层: 检索交叉验证 ----
    cross_verify_results = evaluate_retrieval_cross_verify(test_set)

    # ---- 第三层: 生成质量评估 + Bad Case ----
    gen_results = evaluate_generation(test_set, sample_size=min(12, len(test_set)))
    bad_cases = bad_case_analysis(gen_results)

    # ---- 综合报告 ----
    print(f"\n{'='*60}")
    print(f"📊 综合评估报告")
    print(f"{'='*60}")
    print(f"测试集: {len(test_set)} 条手动标注的真实员工问题\n")

    # 检索排名
    print(f"🏆 检索策略排名 (按 Section Recall):")
    sorted_sr = sorted(section_results.items(),
                       key=lambda x: x[1]["section_recall"], reverse=True)
    for rank, (name, m) in enumerate(sorted_sr):
        medal = ["🥇", "🥈", "🥉", "4️⃣", "5️⃣"][rank] if rank < 5 else "  "
        print(f"  {medal} {name:<15s} P@3: {m['precision_at_3']:.1%}  "
              f"P@5: {m['precision_at_5']:.1%}  "
              f"SectionRecall: {m['section_recall']:.1%}")

    print(f"\n🔍 检索交叉验证 (LLM 逐篇判断相关性):")
    sorted_cv = sorted(cross_verify_results.items(),
                       key=lambda x: x[1]["avg_relevance"], reverse=True)
    for rank, (name, m) in enumerate(sorted_cv):
        medal = ["🥇", "🥈", "🥉", "4️⃣", "5️⃣"][rank] if rank < 5 else "  "
        print(f"  {medal} {name:<15s} 平均相关度: {m['avg_relevance']:.2%}  "
              f"完美相关比: {m['perfect_ratio']:.1%}")

    print(f"\n📈 生成质量 (LLM Judge 辅助指标):")
    print(f"  准确率:  {gen_results.get('avg_accuracy', 0):.2f}/5")
    print(f"  完整度:  {gen_results.get('avg_completeness', 0):.2f}/5")
    print(f"  幻觉率:  {gen_results.get('hallucination_rate', 0):.1%}")
    print(f"  Bad率:   {gen_results.get('bad_case_rate', 0):.1%}")
    print(f"  Bad Case: {len(bad_cases)} 个")

    # 确定最佳策略
    best_strategy = sorted_sr[0][0] if sorted_sr else "ensemble"
    best_cv = sorted_cv[0][0] if sorted_cv else "ensemble"
    print(f"\n💡 推荐策略: 检索={best_strategy}, 交叉验证最佳={best_cv}")
    if best_strategy != best_cv:
        print(f"   ⚠️ Section评估与交叉验证结论不一致，建议综合参考")

    return {
        "test_set_size": len(test_set),
        "section_retrieval": section_results,
        "cross_verify": cross_verify_results,
        "generation": gen_results,
        "bad_cases": bad_cases,
        "recommended_strategy": best_strategy,
    }


# ============================================================================
# Section 11: 主流程
# ============================================================================

def _print_banner():
    print("""
╔══════════════════════════════════════════════════════════╗
║     RAG 全链路优化系统 — 员工手册智能问答                  ║
║                                                          ║
║  1.员工手册.md   2.Agent工具   3.BGE-Large-Zh(1024d)       ║
║  4.语义切块      5.多路召回    6.查询增强(HyDE)           ║
║  7.Reranker精排  8.三层评估体系                          ║
╚══════════════════════════════════════════════════════════╝
""")


def main():
    """主流程: 初始化 → 示例验证 → 交互测试。

    优化：默认跳过完整评估（200+ 次 LLM 调用，耗时 10+ 分钟），
    改为运行 1 个示例查询快速验证 RAG 链路可用，再进入交互模式。
    完整评估通过命令行参数 `--eval` 或交互命令 `eval` 触发。
    """
    global _VS, _CHUNKS

    # 命令行参数: --eval 启动时直接跑完整评估；--demo 只跑示例不进入交互
    run_eval_at_startup = "--eval" in sys.argv
    demo_only = "--demo" in sys.argv

    _print_banner()

    # ---- Step 1: 核心初始化（仅加载必要组件）----
    print("[初始化] 加载模型和文档...\n")
    init_embedding_model()                    # embedding 模型（必须）
    _CHUNKS = load_and_chunk()                # 文档切块（必须）
    _VS = get_vectorstore(_CHUNKS)            # 向量存储（必须）
    create_ensemble_retriever(_CHUNKS, _VS)   # 默认策略检索器（必须）

    # 以下组件改为按需初始化，避免启动时加载不必要的模型
    # - create_mq_retriever(): 仅 multi_query 策略用到，首次调用时再初始化
    # - init_reranker(): CrossEncoder 1.1GB，加载慢，rerank() 内部已按需加载

    # ---- Step 2: 快速示例验证（1 个查询，验证 RAG 链路可用）----
    print(f"\n{'='*60}")
    print(f"[示例] 快速验证 RAG 链路（不启用查询增强/精排，约 3 秒）")
    print(f"{'='*60}")
    demo_q = "试用期是几个月？"
    print(f"问题: {demo_q}")
    result = rag_query(demo_q, strategy="ensemble",
                       use_enhance=False, use_rerank=False)
    print(f"答案: {result['answer'][:200]}")
    print(f"来源: {len(result['sources'])} 个文档 "
          f"(top-1: {result['sources'][0].metadata.get('section', '?')})")

    # ---- Step 3: 可选完整评估（默认跳过）----
    if run_eval_at_startup:
        eval_report = run_full_evaluation()
    else:
        print(f"\n[提示] 完整评估已跳过（耗时较长）。")
        print(f"       如需运行评估：交互模式输入 'eval'，"
              f"或重启时加 --eval 参数")

    # ---- Step 4: 交互模式 ----
    if demo_only:
        print(f"\n[--demo] 示例完成，不进入交互模式")
        return

    print(f"\n{'='*60}")
    print(f"💬 交互模式")
    print(f"{'='*60}")
    print(f"  输入问题开始查询")
    print(f"  命令: 'strategy <name>' 切换检索策略")
    print(f"  策略: bm25 | vector | ensemble | multi_query")
    print(f"  命令: 'agent <query>' 使用 Agent 模式")
    print(f"  命令: 'enhance on/off' 切换查询增强(重写+HyDE)")
    print(f"  命令: 'rerank on/off' 切换精排")
    print(f"  命令: 'eval' 运行完整评估（耗时）")
    print(f"  命令: 'exit' 退出")
    print(f"{'='*60}\n")

    current_strategy = "ensemble"
    use_enhance = False   # 默认关闭（每次查询省 2 次 LLM 调用）
    use_rerank = False    # 默认关闭（首次调用时加载 Reranker 模型）

    while True:
        try:
            user_input = input("🔍 > ").strip()
        except (EOFError, KeyboardInterrupt):
            print("\n再见！")
            break

        if not user_input:
            continue

        if user_input.lower() in ("exit", "quit", "q"):
            print("再见！")
            break

        # 命令处理
        if user_input.lower().startswith("strategy "):
            new_strategy = user_input.split(" ", 1)[1].strip()
            if new_strategy in ("bm25", "vector", "ensemble", "multi_query"):
                current_strategy = new_strategy
                print(f"✅ 已切换到: {current_strategy}")
            else:
                print(f"❌ 未知策略: {new_strategy}")
            continue

        if user_input.lower().startswith("agent "):
            query = user_input.split(" ", 1)[1].strip()
            print(f"🤖 Agent 模式（自动判断是否需要检索）...\n")
            answer = agent_query(query)
            print(f"\n📝 {answer}\n")
            continue

        if user_input.lower().startswith("enhance "):
            arg = user_input.split(" ", 1)[1].strip().lower()
            use_enhance = arg in ("on", "true", "1", "yes")
            print(f"✅ 查询增强: {'开启' if use_enhance else '关闭'}")
            continue

        if user_input.lower().startswith("rerank "):
            arg = user_input.split(" ", 1)[1].strip().lower()
            use_rerank = arg in ("on", "true", "1", "yes")
            print(f"✅ 精排: {'开启' if use_rerank else '关闭'}")
            continue

        if user_input.lower() == "eval":
            eval_report = run_full_evaluation()
            continue

        # 正常查询
        enhance_tag = "查询增强" if use_enhance else "无增强"
        rerank_tag = "精排" if use_rerank else "无精排"
        print(f"🔎 检索策略: {current_strategy} | {enhance_tag} | {rerank_tag}")
        result = rag_query(user_input, strategy=current_strategy,
                          use_enhance=use_enhance, use_rerank=use_rerank)
        print(f"\n📝 {result['answer']}\n")
        print(f"📖 参考来源 ({len(result['sources'])} 个文档):")
        for i, doc in enumerate(result["sources"]):
            print(f"  [{i+1}] {doc.page_content[:100]}...")
        print()


if __name__ == "__main__":
    main()