【Agent开发】第八阶段：RAG 自动化评估体系构建 (RAG Evaluation Framework) —— 给你的 RAG 系统发一张“体检报告” – pd的AI Agent开发笔记

文章目录


前置环境：

  1. 当前环境是基于WSL2 + Ubuntu 24.04 + Docker Desktop构建的云原生开发平台，所有服务（MySQL、Redis、Qwen）均以独立容器形式运行并通过Docker Compose统一编排。如何配置请参考我的博客 WSL2 + Ubuntu 24.04 + Docker Desktop 配置双内核环境
  2. 补充了milvus相关的配置，如何配置请参考我的博客 【Agent开发】第三阶段：RAG 实战 —— 赋予 Agent “外脑”。
  3. 引入了ES检索，并且配置了ES服务，ES部分的配置请查看我的博客 【Agent开发】第五阶段：RAG 深度优化实战 —— 从“可用”到“卓越”。
  4. 补充了Json形式的存储，引入了pgsql，pgsql的配置请查看我的博客 【Agent开发】第七阶段：RAG 自动化评估体系构建 (RAG Evaluation Framework) —— 从“凭感觉调优”到“数据驱动决策”。

第 3 讲：集成 Ragas 框架进行自动化评估

本讲目标：利用现有的 augmented 架构，无缝集成 Ragas 框架，将 PostgreSQL 中的测试数据转化为可视化的评估报告，精准量化 HyDE、Sentence-Child 等策略的真实收益。

1. 为什么选择 Ragas？

在 2026 年的今天，Ragas 依然是 RAG 评估的事实标准，原因有三：

  1. 无参考评估 (Reference-Free)：完美契合我们第 2 讲生成的“无标准答案”测试集（只需 query, context, answer）。
  2. LLM-as-a-Judge：内部封装了复杂的 Prompt 工程，让强 LLM 自动充当裁判，无需人工打分。
  3. 指标全面：原生支持 Faithfulness, Answer Relevance, Context Precision, Context Recall 等核心指标。

💡 版本提示：Ragas 在 0.2.x 版本后 API 有较大变化，本讲基于 最新稳定版 (v0.2+) 编写，确保代码未来可用。

2. 架构集成设计

既然 augmented 模块已经非常成熟（Milvus -> Strategy -> LLM -> PostgreSQL），我们不需要推翻重来，只需做一个“评估适配器”。

🏗️ 新增文件结构

在 ./augmented/ 下新增 evaluator.py，并在test目录创建 run_evaluation.py 脚本。

./augmented/
├── ... (原有文件)
├── evaluator.py        # [新增] Ragas 评估核心逻辑：读取 PG -> 运行 Pipeline -> 计算指标
└── ...

./test/
└──run_evaluation.py     # [新增] 评估入口脚本：配置模型 -> 加载数据 -> 输出报告

🔄 数据流向

  1. 输入：从 PostgreSQL 读取 golden_dataset (Query + Ground Truth Context)。
  2. 执行：调用你现有的 data_generator.py 中的策略逻辑（或复用 Pipeline），对每个 Query 运行检索 + 生成，得到 Retrieved Context 和 Generated Answer。
    • 注意：评估时需要的是系统实际检索到的 Context，而不是黄金测试集里的 Ground Truth Context（后者只用于算 Recall）。
  3. 计算：将 (query, contexts, answer, ground_truths) 喂给 Ragas。
  4. 输出：Pandas DataFrame + 可视化图表。

3. 实战代码实现

🛠️ 步骤 1: 安装依赖

pip install ragas datasets pandas matplotlib
# 确保你的 langchain 和 openai 库也是最新版

🛠️ 步骤 2: 修改之前的数据管道

因为评估时需要用到不同的数据源、不同的检索方式，所以需要修改数据管道，使其根据测试要求灵活变化。

随着配置项的增多，数据保存策略的多样化、检索策略的多样化，通过全局配置来设置，不太适合用于频繁的评估方式，通过数据管道的配置项对这一系列进行配置。

  1. 把数据库 client 改为支持“参数注入 + 默认兼容”
    • milvus_clinet.py
    • es_client.py
@dataclass
class MilvusClientConfig:
    host: str = settings.db.milvus_host
    port: str = settings.db.milvus_port
    collection_name: str = settings.db.milvus_collection
    metric_type: str = settings.db.milvus_metric_type
    index_type: str = settings.db.milvus_index_type
    index_m: int = settings.db.milvus_index_m
    index_ef_construction: int = settings.db.milvus_index_ef_construction
    search_ef: int = settings.db.milvus_search_ef
    embedding_model_name: str = settings.embedding.model_name

    @classmethod
    def from_any(cls, value: Optional["MilvusClientConfig | Dict[str, Any]"]) -> "MilvusClientConfig":
        if value is None:
            return cls()
        if isinstance(value, cls):
            return value
        if isinstance(value, dict):
            allowed = {f.name for f in cls.__dataclass_fields__.values()}  # type: ignore[attr-defined]
            payload = {k: v for k, v in value.items() if k in allowed}
            return cls(**payload)
        raise TypeError("Milvus config must be None, MilvusClientConfig or dict")

    # 原有逻辑 ... 

# 为了方便工具调用，可以提供一个全局单例实例
# 注意：在多线程/多进程环境下可能需要更复杂的单例管理
milvus_client_instance = None
milvus_client_instances: Dict[tuple, MilvusClient] = {}

def get_milvus_client(config: Optional[MilvusClientConfig | Dict[str, Any]] = None) -> MilvusClient:
    global milvus_client_instance
    if config is None:
        if milvus_client_instance is None:
            milvus_client_instance = MilvusClient()
        return milvus_client_instance

    cfg = MilvusClientConfig.from_any(config)
    cache_key = (
        cfg.host,
        cfg.port,
        cfg.collection_name,
        cfg.embedding_model_name,
        cfg.metric_type,
        cfg.index_type,
        cfg.index_m,
        cfg.index_ef_construction,
        cfg.search_ef,
    )
    if cache_key not in milvus_client_instances:
        milvus_client_instances[cache_key] = MilvusClient(cfg)
    return milvus_client_instances[cache_key]
  1. 将检索器也设置成注入的形式
@dataclass
class ComposerConfig:
    """RetrieverComposer 运行时配置，可由实验代码动态注入。"""
    enable_hybrid_search: bool = settings.search.enable_hybrid_search
    plugin_rewritten_query: bool = settings.search.plugin_rewritten_query
    plugin_rewritten_hyde: bool = settings.search.plugin_rewritten_hyde
    plugin_es_questions: bool = settings.search.plugin_es_questions
    plugin_es_summaries: bool = settings.search.plugin_es_summaries
    rrf_k: int = settings.search.rrf_k
    es_host: Optional[str] = settings.db.es_host
    milvus_config: Optional[Dict[str, Any]] = None
    es_config: Optional[Dict[str, Any]] = None

    @classmethod
    def from_any(cls, value: Optional["ComposerConfig | Dict[str, Any]"]) -> "ComposerConfig":
        if value is None:
            return cls()
        if isinstance(value, cls):
            return value
        if isinstance(value, dict):
            allowed = {f.name for f in cls.__dataclass_fields__.values()}  # type: ignore[attr-defined]
            payload = {k: v for k, v in value.items() if k in allowed}
            return cls(**payload)
        raise TypeError("composer config must be None, ComposerConfig or dict")


class RetrieverComposer:
    """
    检索器组装器
    职责：根据配置动态加载多个检索插件，并行执行，并使用 RRF 融合结果。
    """
    def __init__(self, config: Optional[ComposerConfig | Dict[str, Any]] = None):
        self.config = ComposerConfig.from_any(config)
        self.retrievers: List[BaseRetrievalStrategy] = []
        self.rrf_engine = RRFFusionEngine(k=self.config.rrf_k)
        self.milvus_client = get_milvus_client(self.config.milvus_config)
        self.es_client = get_es_client(self.config.es_config)
        
        self._load_plugins()

    def _load_plugins(self):
        """根据配置动态加载插件"""
        # 1. 主路：永远加载
        self.retrievers.append(VectorTextRetriever(milvus_client=self.milvus_client))
        logger.info("✅ [Composer] 已加载主路：VectorText")
        
        # 2. 变体路：如果开启混合检索
        if self.config.enable_hybrid_search:
            # 2. 改写路：如果开启改写
            if self.config.plugin_rewritten_query:
                self.retrievers.append(VectorRewrittenRetriever('standard', milvus_client=self.milvus_client))
                logger.info("✅ [Composer] 已加载变体路：VectorRewritten-standard")
            
            if self.config.plugin_rewritten_hyde:
                self.retrievers.append(VectorRewrittenRetriever('hyde', milvus_client=self.milvus_client))
                logger.info("✅ [Composer] 已加载变体路：VectorRewritten-hyde")

            # 3. ES 路：如果配置了 ES
            if self.config.es_host:
                # 3. ES - Questions 路
                if self.config.plugin_es_questions:
                    es_retriever = ESQuestionsRetriever(es_client=self.es_client)
                    if es_retriever.es.is_available(): # 只有连接成功才加入
                        self.retrievers.append(es_retriever)
                        logger.info("✅ [Composer] 已加载 ES - Questions 路：ESQuestions")
                # 4. ES - Summaries 路
                if self.config.plugin_es_summaries:
                    es_retriever = ESSummariesRetriever(es_client=self.es_client)
                    if es_retriever.es.is_available(): # 只有连接成功才加入
                        self.retrievers.append(es_retriever)
                        logger.info("✅ [Composer] 已加载 ES - Summaries 路：ESSummaries")
  1. 修改数据管道
class RetrievalPipeline:
    def __init__(self, composer_config: Optional[ComposerConfig | Dict[str, Any]] = None):
        # 导入元数据过滤组件
        self.filter_builder = MetadataFilterBuilder()
        self.default_filter_category = settings.search.default_filter_category
        # 导入多路召回组件（支持运行时注入配置，便于实验）
        self.composer = RetrieverComposer(config=composer_config)
        # 👇 修改点：使用工厂获取重排器
        self.reranker = RerankerFactory.get_reranker()
        logger.info(f"⚙️ Pipeline 初始化完成 (重排器：{'已加载' if self.reranker else '未加载'})")
        
        # 读取配置
        self.rough_top_k = settings.rag_online.rough_top_k
        self.final_top_k = settings.rag_online.final_top_k
        self.dynamic_threshold = settings.rag_online.score_threshold
        
        logger.info(f"⚙️ Pipeline 初始化：粗排Top{self.rough_top_k}, 动态阈值={self.dynamic_threshold}")

🛠️ 步骤 2: 编写评估器 (augmented/evaluator.py)

这个模块负责连接现有系统和 Ragas。

"""
RAG evaluator (sync).
- Read generated eval samples from PostgreSQL.
- Reuse existing retrieval pipeline to produce answer + contexts.
- Run RAGAS metrics and return a pandas DataFrame.
"""

from __future__ import annotations

import asyncio
import logging
from typing import Any, Dict, List, Optional

import pandas as pd
from datasets import Dataset as HFDataset
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from openai import OpenAI
from ragas import evaluate
from ragas.llms import llm_factory
from ragas.metrics import Faithfulness, AnswerRelevancy, ContextPrecision, ContextRecall

from src.core.config import settings
from src.core.embedding_client import get_ragas_shared_embedding
from src.core.models import RagEvalSample
from src.core.postgres_client import get_postgres_client
from src.rag.pipeline import RetrievalPipeline
from src.utils.xml_parser import remove_think_and_n
from sqlalchemy import func

logger = logging.getLogger(__name__)


class RAGEvaluator:
    """Evaluator aligned with current architecture."""

    def __init__(self, top_k: int = 3, composer_config: Optional[Dict[str, Any]] = None):
        self.top_k = top_k
        self.pg_client = get_postgres_client()

        # Independent retrieval pipeline for evaluation experiments.
        self.pipeline = RetrievalPipeline(composer_config=composer_config)

        # 关键节点1：初始化两套 LLM
        # - answer_llm: 生成最终回答（RAG 回答阶段）
        # - ragas_llm: 供 RAGAS 指标打分使用（评估阶段）
        self.answer_llm = self._build_llm_from_global_config()
        self.ragas_llm = self._build_ragas_llm()
        # 关键节点2：复用全局 embedding 单例，避免重复加载模型。
        self.ragas_embeddings = get_ragas_shared_embedding(
            self.pipeline.composer.milvus_client.model_name
        )

        self.answer_prompt = ChatPromptTemplate.from_template(
            """
你是一个严谨的问答助手。请只依据给定上下文回答问题。
如果上下文无法支持答案，请明确回答：根据提供的上下文无法回答。
问题：{question}

上下文：
{contexts}
"""
        )

    @staticmethod
    def _run_async(coro):
        """Run async pipeline in sync context."""
        try:
            asyncio.get_running_loop()
            new_loop = asyncio.new_event_loop()
            try:
                return new_loop.run_until_complete(coro)
            finally:
                new_loop.close()
        except RuntimeError:
            return asyncio.run(coro)

    def _build_llm_from_global_config(self) -> ChatOpenAI:
        logger.info("Evaluator uses global LLM: %s", settings.llm.model_name)
        return ChatOpenAI(
            base_url=settings.llm.base_url,
            model=settings.llm.model_name,
            api_key=settings.llm.api_key,
            temperature=float(settings.llm.temperature),
        )

    def _build_ragas_llm(self):
        """
        Ragas collections metrics require InstructorLLM.
        Build it from the same global model endpoint.
        """
        client = OpenAI(
            base_url=settings.llm.base_url,
            api_key=settings.llm.api_key,
        )
        return llm_factory(
            model=settings.llm.model_name,
            provider="openai",
            client=client,
        )

    def load_samples_from_postgres(self, limit: int = 100) -> List[Dict[str, Any]]:
        """Load eval samples from rag_eval_samples table."""
        with self.pg_client.get_session() as session:
            rows: List[RagEvalSample] = (
                session.query(RagEvalSample)
                .order_by(func.random())
                .limit(limit)
                .all()
            )

        samples: List[Dict[str, Any]] = []
        for r in rows:
            samples.append(
                {
                    "id": r.id,
                    "query": r.query,
                    "ground_truth_answer": r.ground_truth_answer,
                    "ground_truth_context": r.ground_truth_context or [],
                }
            )
        logger.info("Loaded %s eval samples from PostgreSQL", len(samples))
        return samples

    def run_pipeline_for_query(self, query: str) -> Dict[str, Any]:
        """Retrieve contexts then generate answer."""
        results = self._run_async(self.pipeline.run(query=query, top_k=self.top_k))
        contexts = [r.text for r in results] if results else []

        chain = self.answer_prompt | self.answer_llm
        resp = chain.invoke({"question": query, "contexts": "\n\n".join(contexts)})
        # 清理部分模型返回的 <think> 推理片段，避免污染评估与报告展示。
        answer = remove_think_and_n(getattr(resp, "content", "") or "")

        return {"contexts": contexts, "answer": answer}

    def evaluate_dataset(self, test_samples: List[Dict[str, Any]]) -> pd.DataFrame:
        """Run pipeline and compute RAGAS metrics."""
        if not test_samples:
            return pd.DataFrame()

        logger.info("Start evaluation, sample count=%s", len(test_samples))
        rows_for_ragas: List[Dict[str, Any]] = []

        # 关键节点3：逐条样本执行“检索+回答”，并转换到 ragas 所需字段。
        for s in test_samples:
            out = self.run_pipeline_for_query(s["query"])
            rows_for_ragas.append(
                {
                    # New ragas schema keys.
                    "user_input": s["query"],
                    "response": out["answer"],
                    "retrieved_contexts": out["contexts"],
                    "reference": s["ground_truth_answer"],
                    "reference_contexts": s.get("ground_truth_context", []),
                    # Keep compatibility aliases for older tooling/exports.
                    "question": s["query"],
                    "answer": out["answer"],
                    "contexts": out["contexts"],
                    "ground_truth": s["ground_truth_answer"],
                }
            )

        hf_dataset = HFDataset.from_list(rows_for_ragas)

        # 关键节点4：实例化评估指标对象（不是函数），并注入 llm/embedding 依赖。
        metrics = [
            Faithfulness(llm=self.ragas_llm),
            AnswerRelevancy(llm=self.ragas_llm, embeddings=self.ragas_embeddings),
            ContextPrecision(llm=self.ragas_llm),
            ContextRecall(llm=self.ragas_llm),
        ]

        # 关键节点5：执行 ragas evaluate，返回结构化 DataFrame 结果。
        result = evaluate(
            dataset=hf_dataset,
            metrics=metrics,
            show_progress=True,
        )
        return result.to_pandas()

    def evaluate_from_postgres(self, limit: int = 100) -> pd.DataFrame:
        samples = self.load_samples_from_postgres(limit=limit)
        return self.evaluate_dataset(samples)


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
    evaluator = RAGEvaluator(top_k=3)
    df = evaluator.evaluate_from_postgres(limit=10)
    print(df.head())

🛠️ 步骤 3: 编写评估脚本 (src\test\run_evaluation.py)

"""
运行评估脚本（固定配置版）
直接修改下方 CONFIG 字典后执行：
python -m src.test.run_evaluation
"""

import logging
import sys
from pathlib import Path


# 评估运行配置（按需直接修改）
CONFIG = {
    "limit": 15,
    "top_k": 3,
    "output_csv": "C:\\Users\\pdnbplus\\Documents\\python全系列\\AIAgent开发\\data\\eval\\recursive_result.csv",
    "composer_config": {
        "enable_hybrid_search": True,
        "plugin_rewritten_query": True,
        "plugin_rewritten_hyde": True,
        "plugin_es_questions": False,
        "plugin_es_summaries": False,
        "rrf_k": 60,
        "milvus_config": {"collection_name": "recursive"},
        "es_config": {"index_questions": "recursive_questions", "index_summaries": "recursive_summaries"},
    },
}

# 在 CONFIG 后做路径检查：若输出目录不存在则直接退出
_output_path = Path(CONFIG["output_csv"])
_output_parent = _output_path.parent
if not _output_parent.exists():
    print(f"配置错误：输出目录不存在 -> {_output_parent}")
    sys.exit(1)


def main():
    from src.augmented.evaluator import RAGEvaluator
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")

    evaluator = RAGEvaluator(
        top_k=CONFIG["top_k"],
        composer_config=CONFIG["composer_config"],
    )
    df = evaluator.evaluate_from_postgres(limit=CONFIG["limit"])

    if df.empty:
        print("评估结果为空（可能没有可用样本）。")
        return

    print(df.head())
    print("\n=== 指标均值 ===")
    numeric_cols = [c for c in df.columns if df[c].dtype.kind in ("i", "f")]
    if numeric_cols:
        print(df[numeric_cols].mean(numeric_only=True))

    if CONFIG["output_csv"]:
        df.to_csv(CONFIG["output_csv"], index=False, encoding="utf-8-sig")
        print(f"\n已输出 CSV: {CONFIG['output_csv']}")


if __name__ == "__main__":
    main()

4. 解读评估报告：如何看懂分数？

运行完脚本后，你会得到类似下面的输出：

在这里插入图片描述

🧐 深度诊断指南

1. Faithfulness (忠实度) < 0.7
  • 现象：模型在瞎编，答案里有上下文没提到的信息。
  • 原因：
    • Prompt 约束不够强（没强调“仅依据上下文”）。
    • 检索到的 contexts 太杂，干扰了模型。
    • 模型本身能力不足（温度太高）。
  • 对策：修改 System Prompt，增加“如果上下文中没有答案，请直接说不知道”；降低 Temperature。
2. Answer Relevancy (相关性) < 0.7
  • 现象：答非所问，或者啰嗦一大堆才说到重点。
  • 原因：
    • 生成 Prompt 没有要求“直接回答”。
    • 检索到的内容太多，模型迷失了重点。
  • 对策：优化 Prompt：“请用一句话直接回答问题，不要包含无关背景”。
3. Context Precision (精度) < 0.6
  • 现象：相关信息找到了，但排在第 4、5 位，模型没注意到（Lost in the Middle）。
  • 原因：
    • 向量检索排序不准。
    • 重排序 (Rerank) 没生效或模型太弱。
  • 对策：检查 RRF 融合权重；升级 Rerank 模型；尝试 HyDE 改变查询向量分布。
4. Context Recall (召回率) < 0.6
  • 现象：根本就没找到相关的文档片段。
  • 原因：
    • 分块策略不合理（关键信息被切断了）。
    • 查询词与文档词汇不匹配。
  • 对策：启用 HyDE；调整 Chunk Size；使用混合检索 (Keyword + Vector)。

5. 📝 本讲总结与行动清单

核心知识点

  1. Ragas 工作流：Load Data -> Run Pipeline -> LLM answer-> Evaluate Metrics -> Analyze。
  2. 指标含义：Faithfulness (防幻觉), Relevance (不啰嗦), Precision (排得对), Recall (找得到)。
  3. 集成模式：在不破坏现有 augmented 架构的前提下，通过 evaluator.py 进行旁路评估。

✅ 实际行动

  1. 环境准备：安装 ragas 和相关依赖。
  2. 小范围测试：修改 run_evaluation.py 中的 limit=5，先跑通 5 条数据，确保流程无误。
  3. 全量运行：对第 2 讲生成的 50 条数据进行评估。
  4. 基线记录：将当前的平均分记录下来（例如：Faithfulness=0.75），这是你后续优化的基准线 (Baseline)。

第 4 讲：深度诊断与瓶颈分析 (Debugging RAG) —— 从“看平均分”到“揪出坏孩子”

本讲目标：不仅知道系统“得了多少分”，更要通过自动化脚本挖掘 Bad Cases，利用 LLM 裁判的推理过程 (Reasoning) 进行根因分析，并输出可视化的诊断报告。

1. 为什么只看平均分不够？

假设你的评估报告如下：

  • Faithfulness: 0.85 (看起来不错！)
  • Answer Relevance: 0.82 (也挺好！)

陷阱：这可能是因为 80% 的问题很简单（如“公司叫什么？”），模型都答对了；而 20% 的关键复杂问题（如“跨部门报销流程中的特殊例外情况？”）模型完全在胡说八道，但被平均分掩盖了。

我们的策略：

  1. 聚焦尾部：只看得分最低的 Top 10 案例。
  2. 挖掘理由：Ragas 不仅能打分，还能让 LLM 裁判写出**“扣分理由”**（Reasoning）。
  3. 归因分类：将错误自动归类为 Retrieval_Failure (没找到), Ranking_Failure (排错了), Hallucination (瞎编), Irrelevant (啰嗦)。

2. 实战：构建 Bad Case 挖掘器

我们将扩展上一讲的 evaluator.py，增加一个 analyze_bad_cases 方法。

🛠️ 代码实现 (augmented/analyst.py)

"""RAG 评估结果分析器。通过 LLMRouter 进行诊断生成。"""
# src/augmented/analyst.py
from __future__ import annotations

import asyncio
import ast
import logging
import re
from typing import Dict, List, Optional

import pandas as pd
from langchain_core.prompts import ChatPromptTemplate

from src.augmented.config import GeneratorConfig, build_default_config
from src.augmented.llm_router import LLMRouter

logger = logging.getLogger(__name__)


class RAGAnalyst:
    """分析低分样本并给出诊断建议。"""

    def __init__(self, generator_config: Optional[GeneratorConfig] = None):
        # 使用 llm_router 管理多模型降级调用。
        self.generator_config = generator_config or build_default_config()
        self.router = LLMRouter(self.generator_config, llm_group="analyst_llms")

        self._prompt = ChatPromptTemplate.from_template(
            """
你是一个资深的 RAG 系统调试专家。
以下是一条低分样本，请输出结构化诊断：

【用户问题】{question}
【检索上下文】{contexts}
【模型回答】{answer}
【标准答案】{ground_truth}
【标准上下文】{ground_truth_contexts}

【指标】
- Faithfulness: {faithfulness}
- Answer Relevancy: {answer_relevancy}
- Context Precision: {context_precision}
- Context Recall: {context_recall}

请输出：
1) 根因判断（检索失败 / 排序失败 / 生成失败 / 混合问题）
2) 证据（引用上下文或回答中的具体片段）
3) 1条可执行优化建议（具体到参数或策略）

直接输出分析结论，不要客套。
"""
        )

    @staticmethod
    def _run_async(coro):
        """在同步代码里安全运行协程。"""
        try:
            asyncio.get_running_loop()
            loop = asyncio.new_event_loop()
            try:
                return loop.run_until_complete(coro)
            finally:
                loop.close()
        except RuntimeError:
            return asyncio.run(coro)

    @staticmethod
    def _pick(row: pd.Series, *keys: str, default=None):
        for k in keys:
            if k in row and pd.notna(row[k]):
                return row[k]
        return default

    @staticmethod
    def _to_text_list(value) -> List[str]:
        """将上下文字段统一为字符串列表，兼容 CSV 中的字符串化 list。"""
        if value is None:
            return []
        if isinstance(value, list):
            return [str(x) for x in value]
        if isinstance(value, str):
            s = value.strip()
            if not s:
                return []
            # 兼容 "['a', 'b']" 这类字符串化列表。
            if s.startswith("[") and s.endswith("]"):
                try:
                    parsed = ast.literal_eval(s)
                    if isinstance(parsed, list):
                        return [str(x) for x in parsed]
                except Exception:
                    pass
            return [s]
        return [str(value)]

    @staticmethod
    def _normalize_text(text: str) -> str:
        """归一化文本以便做规则匹配：去掉 <think>、空白和中英文常见标点。"""
        s = str(text or "")
        s = re.sub(r"<think>.*?</think>", "", s, flags=re.S | re.I)
        s = s.strip().lower()
        s = re.sub(r"\s+", "", s)
        s = re.sub(r"[，。！？；：、,.!?;:\"'`~\-_\(\)\[\]{}<>]", "", s)
        return s

    def _is_unanswerable_text(self, text: str) -> bool:
        """判断文本是否表达“依据上下文无法回答”。"""
        s = self._normalize_text(text)
        cues = [
            "根据提供的上下文无法回答",
            "根据上下文无法回答",
            "无法根据提供的上下文回答",
            "上下文无法支持答案",
            "未明确提及",
            "信息不足",
            "无法依据现有信息判断",
            "无法确定",
            "无法回答",
        ]
        return any(cue in s for cue in cues)

    def _is_expected_unanswerable_and_answered_correctly(self, row: pd.Series) -> bool:
        """
        识别“正确拒答”样本：
        - 标准答案本身是不可回答
        - 模型回答也明确不可回答
        这类样本不应作为坏案例。
        """
        gt = self._pick(row, "ground_truth", "reference", "ground_truth_answer", default="")
        ans = self._pick(row, "answer", "response", default="")
        return self._is_unanswerable_text(gt) and self._is_unanswerable_text(ans)

    def _normalize_case(self, row: pd.Series) -> Dict:
        """兼容 evaluator 输出的多种字段名。"""
        contexts = self._to_text_list(self._pick(row, "contexts", "retrieved_contexts", default=[]))
        gt_contexts = self._to_text_list(
            self._pick(row, "ground_truth_contexts", "reference_contexts", default=[])
        )

        return {
            "question": self._pick(row, "question", "user_input", default=""),
            "answer": self._pick(row, "answer", "response", default=""),
            "contexts": contexts,
            "ground_truth": self._pick(row, "ground_truth", "reference", default=""),
            "ground_truth_contexts": gt_contexts,
            "faithfulness": float(self._pick(row, "faithfulness", default=0.0) or 0.0),
            "answer_relevancy": float(self._pick(row, "answer_relevancy", default=0.0) or 0.0),
            "context_precision": float(self._pick(row, "context_precision", default=0.0) or 0.0),
            "context_recall": float(self._pick(row, "context_recall", default=0.0) or 0.0),
        }

    def _score_and_select(self, df_results: pd.DataFrame, top_k: int) -> pd.DataFrame:
        required = ["faithfulness", "answer_relevancy", "context_precision", "context_recall"]
        missing = [c for c in required if c not in df_results.columns]
        if missing:
            raise ValueError(f"分析失败：缺少指标列 {missing}")

        work = df_results.copy()
        for c in required:
            work[c] = pd.to_numeric(work[c], errors="coerce").fillna(0.0)

        # 先按评估分数构建候选集。
        work["avg_score"] = work[required].mean(axis=1)

        # 关键改进：排除“正确拒答”样本，避免将其误报为坏案例。
        # 典型场景：标准答案就是“无法回答”，模型也确实返回“无法回答”。
        work["is_correct_abstention"] = work.apply(
            self._is_expected_unanswerable_and_answered_correctly, axis=1
        )
        candidates = work[~work["is_correct_abstention"]].copy()

        if candidates.empty:
            logger.info("坏案例筛选后为空：所有低分样本均为‘正确拒答’或无需诊断。")
            return candidates

        return candidates.nsmallest(top_k, "avg_score")

    async def _diagnose_one(self, payload: Dict) -> Dict[str, str]:
        """在线程池中执行同步 router.invoke，避免阻塞事件循环。"""
        text, model = await asyncio.to_thread(self.router.invoke, self._prompt, payload)
        return {"diagnosis": text, "model": model or self.generator_config.default_model_name}

    async def analyze_bad_cases(self, df_results: pd.DataFrame, top_k: int = 5) -> List[Dict]:
        """异步分析低分 Top-K 样本。"""
        if df_results is None or df_results.empty:
            return []

        worst_cases = self._score_and_select(df_results, top_k)
        logger.info("开始分析低分样本：top_k=%s", top_k)

        reports: List[Dict] = []
        for idx, row in worst_cases.iterrows():
            item = self._normalize_case(row)
            payload = {
                "question": item["question"],
                "contexts": "\n".join(item["contexts"]),
                "answer": item["answer"],
                "ground_truth": item["ground_truth"],
                "ground_truth_contexts": "\n".join(item["ground_truth_contexts"]),
                "faithfulness": item["faithfulness"],
                "answer_relevancy": item["answer_relevancy"],
                "context_precision": item["context_precision"],
                "context_recall": item["context_recall"],
            }

            try:
                diagnose_out = await self._diagnose_one(payload)
                diagnosis = diagnose_out["diagnosis"]
                diagnosis_model = diagnose_out["model"]
            except Exception as e:
                logger.exception("低分样本诊断失败 idx=%s", idx)
                diagnosis = f"诊断失败：{e}"
                diagnosis_model = self.generator_config.default_model_name

            reports.append(
                {
                    "id": idx,
                    "question": item["question"],
                    "answer": item["answer"],
                    "contexts": item["contexts"],
                    "ground_truth": item["ground_truth"],
                    "ground_truth_contexts": item["ground_truth_contexts"],
                    "scores": {
                        "faithfulness": item["faithfulness"],
                        "answer_relevancy": item["answer_relevancy"],
                        "context_precision": item["context_precision"],
                        "context_recall": item["context_recall"],
                        "avg_score": float(worst_cases.loc[idx, "avg_score"]),
                    },
                    "predicted_category": self._auto_categorize_error(item),
                    "diagnosis": diagnosis,
                    "diagnosis_model": diagnosis_model,
                }
            )

        return reports

    def analyze_bad_cases_sync(self, df_results: pd.DataFrame, top_k: int = 5) -> List[Dict]:
        """同步入口，便于在脚本中直接调用。"""
        return self._run_async(self.analyze_bad_cases(df_results=df_results, top_k=top_k))

    def _auto_categorize_error(self, row: Dict) -> str:
        """基于指标 heuristic 自动归类。"""
        if row["context_recall"] < 0.5:
            return "检索失败 (Retrieval Failure)"
        if row["context_precision"] < 0.5:
            return "排序失败 (Ranking Failure)"
        if row["faithfulness"] < 0.6:
            return "幻觉问题 (Hallucination)"
        if row["answer_relevancy"] < 0.6:
            return "答非所问 (Irrelevance)"
        return "混合问题 / 需人工复核"

🛠️ 步骤 2: 新增分析脚本 (run_evaluation.py)

"""
运行评估脚本（固定配置版）
直接修改下方 CONFIG 字典后执行：
python -m src.test.run_evaluation
"""

import logging
import ast
import re
import sys
from pathlib import Path
from datetime import datetime


# 评估运行配置（按需直接修改）
CONFIG = {
    "limit": 15,
    "top_k": 3,
    "bad_case_top_k": 3,
    "output_csv": "C:\\Users\\pdnbplus\\Documents\\python全系列\\AIAgent开发\\data\\eval\\recursive_result.csv",
    "composer_config": {
        "enable_hybrid_search": True,
        "plugin_rewritten_query": True,
        "plugin_rewritten_hyde": True,
        "plugin_es_questions": False,
        "plugin_es_summaries": False,
        "rrf_k": 60,
        "milvus_config": {"collection_name": "recursive"},
        "es_config": {"index_questions": "recursive_questions", "index_summaries": "recursive_summaries"},
    },
}

# 在 CONFIG 后做路径检查：若输出目录不存在则直接退出
_output_path = Path(CONFIG["output_csv"])
_output_parent = _output_path.parent
if not _output_parent.exists():
    print(f"配置错误：输出目录不存在 -> {_output_parent}")
    sys.exit(1)


def _format_diagnosis_text(text: str) -> str:
    """将诊断文本整理为更易读的段落/列表结构。"""
    if not isinstance(text, str):
        return ""

    formatted = text.strip().replace("\r\n", "\n").replace("\r", "\n")
    if not formatted:
        return ""

    # 先修复可能被错误拆开的加粗片段：**xxx\n** -> **xxx**
    formatted = re.sub(r"\*\*([^\n*]+)\n\*\*", r"**\1**", formatted)

    # 兼容被压成一行的 markdown 结构（标题、序号、列表）。
    formatted = re.sub(r"\s*(#{2,6}\s*)", r"\n\1", formatted)
    formatted = re.sub(r"\s*(\d+\)\s*)", r"\n\1", formatted)
    formatted = re.sub(r"\s*(-\s+)", r"\n- ", formatted)
    # 只对常见小节名加粗做断行，避免把普通加粗内容拆坏。
    formatted = re.sub(
        r"\s*(\*\*(?:根因判断|证据|可执行优化建议|优化建议)\*\*)",
        r"\n\1",
        formatted,
    )

    formatted = re.sub(r"[ \t]+\n", "\n", formatted)
    formatted = re.sub(r"\n{3,}", "\n\n", formatted)
    return formatted.strip()


def _extract_suggestion(diagnosis: str) -> str:
    """优先提取“优化建议”段落，提取失败则回退整段。"""
    text = _format_diagnosis_text(diagnosis)
    if not text:
        return ""

    patterns = [
        r"(?:^|\n)#{2,6}\s*3\)\s*.*?建议[：:]?\s*(.*)$",
        r"(?:^|\n)\*\*[^*\n]*优化建议[^*\n]*\*\*[：:]?\s*(.*)$",
        r"优化建议[：:]\s*(.*)$",
    ]

    for pattern in patterns:
        m = re.search(pattern, text, flags=re.S)
        if m and m.group(1).strip():
            return re.sub(r"\s+", " ", m.group(1).strip())

    return re.sub(r"\s+", " ", text)


def _clip_text(text: str, max_len: int = 500) -> str:
    if not isinstance(text, str):
        return ""
    s = text.strip()
    if len(s) <= max_len:
        return s
    return s[:max_len] + "..."


def _strip_think(text: str) -> str:
    """移除模型返回中的 <think> 思考片段，保留最终可展示答案。"""
    if not isinstance(text, str):
        return ""
    cleaned = re.sub(r"<think>.*?</think>", "", text, flags=re.S | re.I)
    return cleaned.strip()


def _to_text_list(value):
    """兼容 list、字符串化 list、单字符串三种上下文字段形态。"""
    if value is None:
        return []
    if isinstance(value, list):
        return [str(x) for x in value]
    if isinstance(value, str):
        s = value.strip()
        if not s:
            return []
        if s.startswith("[") and s.endswith("]"):
            try:
                parsed = ast.literal_eval(s)
                if isinstance(parsed, list):
                    return [str(x) for x in parsed]
            except Exception:
                pass
        return [s]
    return [str(value)]


def _to_blockquote(text: str) -> str:
    """将文本转换为 markdown 引用块格式。"""
    lines = (text or "").splitlines() or [""]
    return "\n".join([f"> {line}" if line.strip() else ">" for line in lines])


def _format_context_blocks(contexts, max_items: int = 3, each_max_len: int = 300) -> str:
    items = _to_text_list(contexts)
    if not items:
        return "> （无）"

    blocks = []
    for i, raw_ctx in enumerate(items[:max_items], start=1):
        ctx = _strip_think(str(raw_ctx))
        ctx = _clip_text(ctx, max_len=each_max_len)
        blocks.append(f"#### 引用 {i}\n{_to_blockquote(ctx)}")
    return "\n\n".join(blocks)


def analyze_bad_results(df, top_k: int = 5):
    """调用 analyst 对低分样本进行诊断。"""
    from src.augmented.analyst import RAGAnalyst

    analyst = RAGAnalyst()
    reports = analyst.analyze_bad_cases_sync(df_results=df, top_k=top_k)
    if not reports:
        print("\n=== 差结果分析 ===")
        print("无可分析样本。")
        return []

    print("\n" + "=" * 40)
    print("🚨 坏案例深度诊断报告")
    print("=" * 40)

    report_file = f"bad_cases_report_{datetime.now().strftime('%Y%m%d_%H%M')}.md"
    report_path = _output_parent / report_file

    with open(report_path, "w", encoding="utf-8") as f:
        f.write("# 🚨 RAG 坏案例深度诊断报告\n\n")

        for i, report in enumerate(reports, start=1):
            q = report.get("question", "")
            scores = report.get("scores", {})
            diagnosis = _format_diagnosis_text(report.get("diagnosis", ""))
            category = report.get("predicted_category", "")
            model = report.get("diagnosis_model", "")
            answer = _strip_think(report.get("answer", ""))
            contexts = report.get("contexts", [])
            ground_truth = report.get("ground_truth", "")
            gt_contexts = report.get("ground_truth_contexts", [])

            f.write(f"## 案例 {i}: {category}\n")
            f.write(f"**问题**: {q}\n")
            f.write(
                f"**得分**: Faithfulness={float(scores.get('faithfulness', 0.0)):.2f}, "
                f"Recall={float(scores.get('context_recall', 0.0)):.2f}\n"
            )
            f.write(f"**诊断模型**: {model}\n\n")
            f.write(f"### 🤖 模型生成答案\n{_clip_text(answer, max_len=800)}\n\n")
            f.write(f"### 📚 检索引用信息（Top-3）\n{_format_context_blocks(contexts, max_items=3, each_max_len=380)}\n\n")
            f.write(f"### ✅ 标准答案（Ground Truth）\n{_clip_text(ground_truth, max_len=800)}\n\n")
            f.write(
                f"### 📖 标准参考上下文（Top-2）\n{_format_context_blocks(gt_contexts, max_items=2, each_max_len=280)}\n\n"
            )
            f.write(f"### 🩺 LLM 专家诊断\n{diagnosis}\n\n")
            f.write("---\n\n")

            print(f"\n[案例 {i}] {category}")
            short_q = q[:50] + "..." if len(q) > 50 else q
            print(f"Q: {short_q}")
            suggestion = _extract_suggestion(diagnosis)
            short_suggestion = suggestion[:100] + "..." if len(suggestion) > 100 else suggestion
            print(f"💡 建议: {short_suggestion}")

    print(f"\n💾 详细诊断报告已保存至 {report_path}")

    print("\n=== 差结果分析 ===")
    for i, report in enumerate(reports, start=1):
        formatted_diagnosis = _format_diagnosis_text(report.get("diagnosis", ""))
        answer = _clip_text(_strip_think(report.get("answer", "")), max_len=220)
        ctx_text = _format_context_blocks(report.get("contexts", []), max_items=2, each_max_len=120)
        print(f"\n[{i}] 问题: {report.get('question', '')}")
        print(f"类别: {report.get('predicted_category', '')}")
        print(f"诊断模型: {report.get('diagnosis_model', '')}")
        print(f"模型答案: {answer}")
        print(f"检索引用:\n{ctx_text}")
        print(f"诊断:\n{formatted_diagnosis}")
    return reports


def main():
    from src.augmented.evaluator import RAGEvaluator
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")

    evaluator = RAGEvaluator(
        top_k=CONFIG["top_k"],
        composer_config=CONFIG["composer_config"],
    )
    df = evaluator.evaluate_from_postgres(limit=CONFIG["limit"])

    if df.empty:
        print("评估结果为空（可能没有可用样本）。")
        return

    print(df.head())
    print("\n=== 指标均值 ===")
    numeric_cols = [c for c in df.columns if df[c].dtype.kind in ("i", "f")]
    if numeric_cols:
        print(df[numeric_cols].mean(numeric_only=True))

    if CONFIG["output_csv"]:
        df.to_csv(CONFIG["output_csv"], index=False, encoding="utf-8-sig")
        print(f"\n已输出 CSV: {CONFIG['output_csv']}")

    analyze_bad_results(df, top_k=CONFIG["bad_case_top_k"])


if __name__ == "__main__":
    # main()
    import pandas as pd
    df = pd.read_csv(CONFIG["output_csv"])
    # print("\n=== 指标均值 ===")
    # numeric_cols = [c for c in df.columns if df[c].dtype.kind in ("i", "f")]
    # if numeric_cols:
    #     print(df[numeric_cols].mean(numeric_only=True))
    analyze_bad_results(df, top_k=CONFIG["bad_case_top_k"])

3. 解读诊断报告：典型模式与对策

在这里插入图片描述

运行脚本后，你会得到一份 Markdown 报告。以下是三种典型的诊断模式及对策：

🔴 模式 A：检索失败 (Retrieval Failure)

  • 特征：Context Recall 极低 (<0.4)，Faithfulness 可能也低（因为没依据只能瞎编）。
  • LLM 诊断示例：“检索到的上下文中完全没有提及‘跨部门审批’的具体流程，导致模型无法回答。”
  • 根因：
    • 查询词与文档词汇不匹配（如用户问“怎么报销”，文档写“费用核销流程”）。
    • 分块切碎了关键信息（如“流程”在第一块，“金额”在第二块）。
  • ✅ 对策：
    • 启用 HyDE：让 LLM 先改写查询，对齐文档术语。
    • 调整分块：增大 Chunk Size，或使用父子索引 (Parent-Child)。
    • 混合检索：加入关键词检索 (BM25) 弥补向量检索的不足。

🟠 模式 B：排序失败 (Ranking Failure / Lost in the Middle)

  • 特征：Context Recall 高 (>0.8)，但 Context Precision 低 (<0.4)，Faithfulness 中等。
  • LLM 诊断示例：“正确答案存在于检索到的第 4 个片段中，但前 3 个片段都是无关噪音，导致模型忽略了关键信息。”
  • 根因：向量相似度计算不准，或者噪音太多干扰了模型注意力。
  • ✅ 对策：
    • 引入 Rerank 模型：在检索后加一步重排序（如 BGE-Reranker）。
    • 减少 Top-K：如果不需要那么多上下文，减少传入 LLM 的片段数量。

🔵 模式 C：幻觉/生成失败 (Hallucination / Generation Failure)

  • 特征：Context Recall 高，Context Precision 高，但 Faithfulness 低。
  • LLM 诊断示例：“模型声称‘报销额度为 500 元’，但提供的上下文中明确写的是‘300 元’。这是典型的幻觉。”
  • 根因：
    • Prompt 约束力不够。
    • 模型为了“讨好”用户强行作答。
  • ✅ 对策：
    • 强化 Prompt：加入“如果上下文中没有明确数字，严禁猜测”、“必须引用原文”等指令。
    • Self-RAG：让模型在生成前先自我反思“这段上下文够吗？”。
    • 降低 Temperature：设为 0 或 0.1。

打通测试集与诊断服务

将评估结果、诊断结果全部存入数据库ogsql，避免csv结果，方便后续调用。

  1. 在pgsql中新建两张表，分别为评估结果表和诊断结果表
  2. 在原本的rag_eval_samples增加一个字段 batch_id, 用于记录生成改测试集的批次
  3. 计算评估结果的样本基于List[批次id] 诊断结果基于评估结果，也是间接基于List[批次id]
  4. 诊断结果表中，有字段bad_case_rank用于记录在本次批次中，该样本bad_case的排名, 有diagnosis_markdown字段，即上面从问题到诊断的所有内容的markdown。

1. 🐘 数据库修改

🏷 新增表结构

-- PostgreSQL: 评估结果/诊断结果建表与样本批次字段补齐
-- 执行方式示例：
-- psql "postgresql://agent_user:***@127.0.0.1:5433/agent_dev" -f sql/postgres_eval_results_and_diagnosis.sql

BEGIN;

-- 1) 给原始评测样本表补 batch_id 字段（按批次管理测试集）
ALTER TABLE IF EXISTS rag_eval_samples
    ADD COLUMN IF NOT EXISTS batch_id INTEGER NOT NULL DEFAULT 1;

CREATE INDEX IF NOT EXISTS idx_rag_eval_samples_batch_id
    ON rag_eval_samples(batch_id);

-- 将当前历史样本统一归为 batch_id=1
UPDATE rag_eval_samples
SET batch_id = 1
WHERE batch_id IS DISTINCT FROM 1;


-- 2) 评估结果表：保存每次评估的逐样本指标与问答内容
CREATE TABLE IF NOT EXISTS rag_eval_results (
    id BIGSERIAL PRIMARY KEY,
    eval_run_id VARCHAR(64) NOT NULL,
    sample_id VARCHAR(64) NOT NULL,
    sample_batch_id INTEGER NOT NULL DEFAULT 1,
    question TEXT NOT NULL,
    answer TEXT NOT NULL,
    contexts JSONB NOT NULL DEFAULT '[]'::jsonb,
    ground_truth TEXT NOT NULL,
    ground_truth_contexts JSONB NOT NULL DEFAULT '[]'::jsonb,
    faithfulness DOUBLE PRECISION NOT NULL DEFAULT 0,
    answer_relevancy DOUBLE PRECISION NOT NULL DEFAULT 0,
    context_precision DOUBLE PRECISION NOT NULL DEFAULT 0,
    context_recall DOUBLE PRECISION NOT NULL DEFAULT 0,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE INDEX IF NOT EXISTS idx_rag_eval_results_eval_run_id
    ON rag_eval_results(eval_run_id);

CREATE INDEX IF NOT EXISTS idx_rag_eval_results_batch
    ON rag_eval_results(sample_batch_id);

CREATE INDEX IF NOT EXISTS idx_rag_eval_results_sample_id
    ON rag_eval_results(sample_id);


-- 3) 诊断结果表：保存坏案例排名与完整 markdown 诊断内容
CREATE TABLE IF NOT EXISTS rag_eval_diagnoses (
    id BIGSERIAL PRIMARY KEY,
    eval_run_id VARCHAR(64) NOT NULL,
    sample_id VARCHAR(64) NOT NULL,
    sample_batch_id INTEGER NOT NULL DEFAULT 1,
    bad_case_rank INTEGER NOT NULL,
    predicted_category VARCHAR(128) NOT NULL DEFAULT '',
    diagnosis_model VARCHAR(128) NOT NULL DEFAULT '',
    diagnosis TEXT NOT NULL,
    diagnosis_markdown TEXT NOT NULL,
    question TEXT NOT NULL,
    answer TEXT NOT NULL,
    contexts JSONB NOT NULL DEFAULT '[]'::jsonb,
    ground_truth TEXT NOT NULL,
    ground_truth_contexts JSONB NOT NULL DEFAULT '[]'::jsonb,
    faithfulness DOUBLE PRECISION NOT NULL DEFAULT 0,
    answer_relevancy DOUBLE PRECISION NOT NULL DEFAULT 0,
    context_precision DOUBLE PRECISION NOT NULL DEFAULT 0,
    context_recall DOUBLE PRECISION NOT NULL DEFAULT 0,
    avg_score DOUBLE PRECISION NOT NULL DEFAULT 0,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE INDEX IF NOT EXISTS idx_rag_eval_diagnoses_eval_run_id
    ON rag_eval_diagnoses(eval_run_id);

CREATE INDEX IF NOT EXISTS idx_rag_eval_diagnoses_batch
    ON rag_eval_diagnoses(sample_batch_id);

CREATE INDEX IF NOT EXISTS idx_rag_eval_diagnoses_bad_case_rank
    ON rag_eval_diagnoses(bad_case_rank);

CREATE INDEX IF NOT EXISTS idx_rag_eval_diagnoses_sample_id
    ON rag_eval_diagnoses(sample_id);

COMMIT;

📑 修改数据模型 rag_eval_samples

class RagEvalSample(Base):
    __tablename__ = "rag_eval_samples"

    id = Column(String(64), primary_key=True)
    category = Column(String(64), nullable=False, default="general")
    difficulty = Column(String(16), nullable=False)

    # Legacy columns.
    query = Column(Text, nullable=True)
    ground_truth_context = Column(JSON, nullable=True)
    ground_truth_answer = Column(Text, nullable=True)

    # Canonical columns.
    question = Column(Text, nullable=True)
    ground_truth_contexts = Column(JSON, nullable=True)
    ground_truth = Column(Text, nullable=True)

    source_document = Column(String(255), nullable=True)
    model_name = Column(String(128), nullable=True)
    meta = Column("metadata", JSON, nullable=False, default=dict)

    source_chunk_index = Column(Integer, nullable=False)
    source_backend = Column(String(32), nullable=False, default="milvus")
    created_at = Column(BigInteger, nullable=False)
    batch_id = Column(Integer, nullable=False, default=1)

    __table_args__ = (
        Index("idx_rag_eval_samples_created_at", "created_at"),
        Index("idx_rag_eval_samples_difficulty", "difficulty"),
        Index("idx_rag_eval_samples_category", "category"),
        Index("idx_rag_eval_samples_batch_id", "batch_id"),
    )


class RagEvalResult(Base):
    __tablename__ = "rag_eval_results"

    id = Column(BigInteger, primary_key=True, autoincrement=True)
    # 一次评估运行的唯一标识，便于串联结果与诊断。
    eval_run_id = Column(String(64), nullable=False)
    sample_id = Column(String(64), nullable=False, index=True)
    sample_batch_id = Column(Integer, nullable=False, default=1)
    question = Column(Text, nullable=False)
    answer = Column(Text, nullable=False)
    contexts = Column(JSON, nullable=False, default=list)
    ground_truth = Column(Text, nullable=False)
    ground_truth_contexts = Column(JSON, nullable=False, default=list)
    faithfulness = Column(Float, nullable=False, default=0.0)
    answer_relevancy = Column(Float, nullable=False, default=0.0)
    context_precision = Column(Float, nullable=False, default=0.0)
    context_recall = Column(Float, nullable=False, default=0.0)
    created_at = Column(DateTime, nullable=False, default=datetime.utcnow)

    __table_args__ = (
        Index("idx_rag_eval_results_eval_run_id", "eval_run_id"),
        Index("idx_rag_eval_results_batch", "sample_batch_id"),
    )


class RagEvalDiagnosis(Base):
    __tablename__ = "rag_eval_diagnoses"

    id = Column(BigInteger, primary_key=True, autoincrement=True)
    eval_run_id = Column(String(64), nullable=False, index=True)
    sample_id = Column(String(64), nullable=False, index=True)
    sample_batch_id = Column(Integer, nullable=False, default=1)
    bad_case_rank = Column(Integer, nullable=False)
    predicted_category = Column(String(128), nullable=False, default="")
    diagnosis_model = Column(String(128), nullable=False, default="")
    diagnosis = Column(Text, nullable=False)
    diagnosis_markdown = Column(Text, nullable=False)
    question = Column(Text, nullable=False)
    answer = Column(Text, nullable=False)
    contexts = Column(JSON, nullable=False, default=list)
    ground_truth = Column(Text, nullable=False)
    ground_truth_contexts = Column(JSON, nullable=False, default=list)
    faithfulness = Column(Float, nullable=False, default=0.0)
    answer_relevancy = Column(Float, nullable=False, default=0.0)
    context_precision = Column(Float, nullable=False, default=0.0)
    context_recall = Column(Float, nullable=False, default=0.0)
    avg_score = Column(Float, nullable=False, default=0.0)
    created_at = Column(DateTime, nullable=False, default=datetime.utcnow)

    __table_args__ = (
        Index("idx_rag_eval_diagnoses_eval_run_id", "eval_run_id"),
        Index("idx_rag_eval_diagnoses_batch", "sample_batch_id"),
        Index("idx_rag_eval_diagnoses_bad_case_rank", "bad_case_rank"),
    )

修改相应config.py文件

@dataclass
class GeneratorConfig:
    # ...[原本内容]
    # 评测样本批次号：写入 rag_eval_samples.batch_id，便于按批次抽样评估。
    sample_batch_id: int = 1

将原有评估CSV存入pgsql

"""
将评估 CSV 导入 PostgreSQL 的 rag_eval_results 表。

用法：
python -m src.test.import_eval_csv_to_pg
"""

from __future__ import annotations

import ast
from datetime import datetime
from pathlib import Path
from typing import Any, List

import pandas as pd

from src.core.models import RagEvalResult
from src.core.postgres_client import get_postgres_client


CONFIG = {
    # 评估 CSV 路径（run_evaluation 旧流程输出的结果文件）
    "csv_path": "C:\\Users\\pdnbplus\\Documents\\python全系列\\AIAgent开发\\data\\eval\\recursive_result.csv",
    # 本次导入写入到哪个 eval_run_id
    "eval_run_id": f"import_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}",
    # 写入 sample_batch_id（当 CSV 内没有该字段时使用）
    "sample_batch_id": 1,
    # 若该 eval_run_id 已存在数据，是否先删除再重导
    "replace_if_exists": True,
}


def _to_text_list(value: Any) -> List[str]:
    """兼容 list、字符串化 list、普通字符串。"""
    if value is None:
        return []
    if isinstance(value, list):
        return [str(x) for x in value]
    if isinstance(value, str):
        s = value.strip()
        if not s:
            return []
        if s.startswith("[") and s.endswith("]"):
            try:
                parsed = ast.literal_eval(s)
                if isinstance(parsed, list):
                    return [str(x) for x in parsed]
            except Exception:
                pass
        return [s]
    return [str(value)]


def _pick(row: pd.Series, *keys: str, default=None):
    for k in keys:
        if k in row and pd.notna(row[k]):
            return row[k]
    return default


def main():
    csv_path = Path(CONFIG["csv_path"])
    if not csv_path.exists():
        raise FileNotFoundError(f"CSV 不存在: {csv_path}")

    df = pd.read_csv(csv_path)
    if df.empty:
        print("CSV 为空，无需导入。")
        return

    eval_run_id = str(CONFIG["eval_run_id"])
    default_batch = int(CONFIG["sample_batch_id"])

    rows = []
    now = datetime.utcnow()
    for i, row in df.iterrows():
        sample_id = str(_pick(row, "sample_id", default=f"csv_{default_batch}_{i + 1}"))
        sample_batch_id = int(_pick(row, "sample_batch_id", "batch_id", default=default_batch) or default_batch)

        rows.append(
            {
                "eval_run_id": eval_run_id,
                "sample_id": sample_id,
                "sample_batch_id": sample_batch_id,
                "question": str(_pick(row, "question", "user_input", default="") or ""),
                "answer": str(_pick(row, "answer", "response", default="") or ""),
                "contexts": _to_text_list(_pick(row, "contexts", "retrieved_contexts", default=[])),
                "ground_truth": str(_pick(row, "ground_truth", "reference", default="") or ""),
                "ground_truth_contexts": _to_text_list(
                    _pick(row, "ground_truth_contexts", "reference_contexts", default=[])
                ),
                "faithfulness": float(_pick(row, "faithfulness", default=0.0) or 0.0),
                "answer_relevancy": float(_pick(row, "answer_relevancy", default=0.0) or 0.0),
                "context_precision": float(_pick(row, "context_precision", default=0.0) or 0.0),
                "context_recall": float(_pick(row, "context_recall", default=0.0) or 0.0),
                "created_at": now,
            }
        )

    pg = get_postgres_client()
    RagEvalResult.__table__.create(bind=pg.engine, checkfirst=True)

    with pg.get_session() as session:
        if CONFIG.get("replace_if_exists", True):
            session.query(RagEvalResult).filter(RagEvalResult.eval_run_id == eval_run_id).delete()
        session.bulk_insert_mappings(RagEvalResult, rows)
        session.commit()

    print(f"导入完成: {len(rows)} 条 -> rag_eval_results")
    print(f"eval_run_id: {eval_run_id}")
    print(f"csv: {csv_path}")


if __name__ == "__main__":
    main()

执行导入

python -m src.test.import_eval_csv_to_pg

2. 🖌 写入batch_id

  • 生成测试集：data_gerenator.py
# 主编排器（策略模式）：
# Milvus 取 chunk -> 按策略构建任务 -> LLM 生成评估样本 -> PostgreSQL 持久化。
from __future__ import annotations

import ast
import json
import logging
import time
from datetime import date
from typing import Any, Dict, List, Optional, Tuple

from langchain_core.prompts import ChatPromptTemplate
from pydantic import ValidationError

from src.augmented.config import GeneratorConfig, build_default_config
from src.augmented.llm_router import LLMRouter
from src.augmented.prompts import PromptRegistry
from src.augmented.sinks import PostgresSink
from src.augmented.sources import MilvusSource
from src.augmented.strategies import BaseGenerationStrategy, StrategyTask, build_strategies
from src.schema.augmented_schema import GeneratedSample

logger = logging.getLogger(__name__)


class DatasetGenerator:
    """评估集生成器：聚合数据源、策略、LLM 与落库。"""

    def __init__(self, config: Optional[GeneratorConfig] = None):
        self.config = config or build_default_config()
        self.prompt_registry = PromptRegistry()
        self.prompt_cache: Dict[str, ChatPromptTemplate] = {}

        self.strategy_params = self._safe_parse_strategy_params(self.config.strategy_params_json)
        self.strategy_params.setdefault("standard", {})
        self.strategy_params["standard"].setdefault("num_questions", self.config.num_questions_per_chunk)

        self.router = LLMRouter(config=self.config)
        self.source = MilvusSource()
        self.sink = PostgresSink()
        self.strategies = build_strategies(self.config.enabled_strategies, self.strategy_params)

    def _get_prompt(self, profile: str) -> ChatPromptTemplate:
        if profile not in self.prompt_cache:
            self.prompt_cache[profile] = ChatPromptTemplate.from_template(self.prompt_registry.get(profile))
        return self.prompt_cache[profile]

    @staticmethod
    def _safe_parse_strategy_params(raw: str) -> Dict[str, Dict[str, Any]]:
        try:
            parsed = json.loads(raw) if raw else {}
            return parsed if isinstance(parsed, dict) else {}
        except Exception:
            return {}

    @staticmethod
    def _extract_json_candidate(text: str) -> str:
        left_arr = text.find("[")
        right_arr = text.rfind("]")
        if left_arr != -1 and right_arr != -1 and right_arr > left_arr:
            return text[left_arr : right_arr + 1]

        left_obj = text.find("{")
        right_obj = text.rfind("}")
        if left_obj != -1 and right_obj != -1 and right_obj > left_obj:
            return text[left_obj : right_obj + 1]

        return text

    @staticmethod
    def _safe_parse_json(content: str) -> List[Dict[str, Any]]:
        text = content.strip()
        if text.startswith("```json"):
            text = text[7:]
        if text.startswith("```"):
            text = text[3:]
        if text.endswith("```"):
            text = text[:-3]
        text = text.strip()

        try:
            data = json.loads(text)
        except json.JSONDecodeError:
            candidate = DatasetGenerator._extract_json_candidate(text)
            try:
                data = json.loads(candidate)
            except json.JSONDecodeError:
                data = ast.literal_eval(candidate)

        if isinstance(data, dict):
            data = [data]
        return data if isinstance(data, list) else []

    @staticmethod
    def _validate_generated_samples(raw_samples: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        valid_samples: List[Dict[str, Any]] = []
        for raw in raw_samples:
            try:
                if isinstance(raw.get("ground_truth_contexts"), str):
                    raw["ground_truth_contexts"] = [raw["ground_truth_contexts"]]
                valid_samples.append(GeneratedSample(**raw).model_dump())
            except ValidationError:
                continue
        return valid_samples

    def _load_and_filter_chunks(self) -> List[Dict[str, Any]]:
        chunks = self.source.load_chunks(limit=self.config.chunks_limit)
        logger.info(
            "开始生成评估集: chunks_limit=%s, min_chunk_length=%s, standard_num_questions=%s",
            self.config.chunks_limit,
            self.config.min_chunk_length,
            self.config.num_questions_per_chunk,
        )
        logger.info("启用策略=%s", [s.name for s in self.strategies])
        logger.info("策略参数=%s", self.strategy_params)
        logger.info("Milvus加载到chunk数量=%s", len(chunks))
        if not chunks:
            logger.warning("⚠️ 未加载到任何 chunk，请检查 source 配置。")
            return []

        filtered: List[Dict[str, Any]] = []
        filtered_short = 0
        for item in chunks:
            text = item.get("text", "")
            if len(text.strip()) < self.config.min_chunk_length:
                filtered_short += 1
                continue
            filtered.append(item)
        logger.info("过滤过短chunk=%s, 可用chunk=%s", filtered_short, len(filtered))
        return filtered

    def _build_strategy_tasks(
        self, chunks: List[Dict[str, Any]]
    ) -> Tuple[List[Tuple[StrategyTask, str]], Dict[str, BaseGenerationStrategy]]:
        strategy_tasks: List[Tuple[StrategyTask, str]] = []
        strategy_map: Dict[str, BaseGenerationStrategy] = {}
        for strategy in self.strategies:
            strategy_map[strategy.name] = strategy
            tasks = strategy.build_tasks(chunks)
            for task in tasks:
                strategy_tasks.append((task, strategy.name))
        logger.info("策略任务总数=%s", len(strategy_tasks))
        return strategy_tasks, strategy_map

    def generate_from_task(self, task: StrategyTask) -> Tuple[List[Dict[str, Any]], Optional[str]]:
        prompt = self._get_prompt(task.prompt_profile)
        last_error_msg = ""
        for attempt in range(self.config.max_retries_per_chunk + 1):
            try:
                content, model_used = self.router.invoke(prompt=prompt, payload=task.payload)
                raw_samples = self._safe_parse_json(content)
                valid_samples = self._validate_generated_samples(raw_samples)
                if valid_samples:
                    return valid_samples, model_used
                last_error_msg = f"JSON解析成功但无有效样本(raw={len(raw_samples)})"
            except Exception as exc:
                last_error_msg = str(exc)
                logger.error(
                    "任务生成失败 (attempt=%s): %s | strategy=%s | content_preview=%s",
                    attempt + 1,
                    exc,
                    task.strategy_name,
                    content[:200] if "content" in locals() else "",
                )
            time.sleep(0.2)

        logger.warning(
            "任务生成失败，已放弃。strategy=%s reason=%s | task_preview=%s",
            task.strategy_name,
            last_error_msg,
            str(task.payload)[:160],
        )
        return [], None

    def _execute_strategy_tasks(
        self,
        strategy_tasks: List[Tuple[StrategyTask, str]],
        strategy_map: Dict[str, BaseGenerationStrategy],
    ) -> List[Tuple[StrategyTask, str, List[Dict[str, Any]], Optional[str]]]:
        raw_results: List[Tuple[StrategyTask, str, List[Dict[str, Any]], Optional[str]]] = []
        for task, strategy_name in strategy_tasks:
            samples, model_used = self.generate_from_task(task)
            samples = strategy_map[strategy_name].postprocess_samples(samples, task)
            raw_results.append((task, strategy_name, samples, model_used))

        success_tasks = sum(1 for _, _, samples, _ in raw_results if samples)
        failed_tasks = len(raw_results) - success_tasks
        total_generated = sum(len(samples) for _, _, samples, _ in raw_results)
        logger.info(
            "任务生成统计: success_tasks=%s, failed_tasks=%s, generated_samples=%s",
            success_tasks,
            failed_tasks,
            total_generated,
        )
        return raw_results

    def _assemble_rows(
        self, raw_results: List[Tuple[StrategyTask, str, List[Dict[str, Any]], Optional[str]]]
    ) -> List[Dict[str, Any]]:
        all_samples: List[Dict[str, Any]] = []
        now_ts = int(time.time())
        today = date.today().isoformat()

        for task, strategy_name, samples, model_used in raw_results:
            first_idx = task.source_chunk_indices[0] if task.source_chunk_indices else -1
            if strategy_name == "mixed_pair":
                a_src = task.source_metadata.get("chunk_a_metadata", {}).get("source")
                b_src = task.source_metadata.get("chunk_b_metadata", {}).get("source")
                source_document = f"{a_src}|{b_src}"
            else:
                source_document = task.source_metadata.get("source")

            for sample_idx, sample in enumerate(samples):
                sample_meta = sample.get("metadata", {}) or {}
                sample_meta.setdefault("generated_by", model_used or self.config.default_model_name)
                sample_meta.setdefault("generation_date", today)
                sample_meta.setdefault("strategy", strategy_name)
                sample_meta.setdefault("source_chunk_indices", task.source_chunk_indices)

                all_samples.append(
                    {
                        "id": f"gen_{now_ts}_{strategy_name}_{first_idx}_{sample_idx}",
                        "category": sample.get("category", task.source_metadata.get("category", "general")),
                        "difficulty": sample["difficulty"],
                        "question": sample["question"],
                        "ground_truth_contexts": sample["ground_truth_contexts"],
                        "ground_truth": sample["ground_truth"],
                        "source_document": sample.get("source_document") or source_document,
                        "model_name": model_used or self.config.default_model_name,
                        "metadata": sample_meta,
                        "source_chunk_index": first_idx,
                        "source_backend": "milvus",
                        "created_at": now_ts,
                        "batch_id": int(self.config.sample_batch_id),
                    }
                )
        return all_samples

    def generate(self) -> List[Dict[str, Any]]:
        filtered_chunks = self._load_and_filter_chunks()
        if not filtered_chunks:
            return []

        strategy_tasks, strategy_map = self._build_strategy_tasks(filtered_chunks)
        raw_results = self._execute_strategy_tasks(strategy_tasks, strategy_map)
        all_samples = self._assemble_rows(raw_results)

        self.sink.save(all_samples)
        logger.info("✅ 数据集生成完成，样本数=%s，已写入 PostgreSQL", len(all_samples))
        return all_samples


if __name__ == "__main__":
    # python -m src.augmented.data_generator
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
    generator = DatasetGenerator()
    generator.generate()
  • 对应修改入库代码sinks.py
# augmented/sinks.py
# 存储模块：将评估样本批量 upsert 到 PostgreSQL。
from typing import Any, Dict, List

from sqlalchemy.dialects.postgresql import insert

from src.core.models import RagEvalSample
from src.core.postgres_client import get_postgres_client


class PostgresSink:
    def __init__(self) -> None:
        self.client = get_postgres_client()

    def save(self, rows: List[Dict[str, Any]]) -> None:
        if not rows:
            return

        # 仅做字段归一化，数据库写入采用单次 upsert。
        # 插入载荷：统一按 canonical 字段入库。
        payload: List[Dict[str, Any]] = []
        for row in rows:
            payload.append(
                {
                    "id": row["id"],
                    "category": row.get("category", "general"),
                    "difficulty": row["difficulty"],
                    "question": row["question"],
                    "ground_truth_contexts": row["ground_truth_contexts"],
                    "ground_truth": row["ground_truth"],
                    "source_document": row.get("source_document"),
                    "model_name": row.get("model_name"),
                    "metadata": row.get("metadata", {}),
                    "source_chunk_index": row["source_chunk_index"],
                    "source_backend": row.get("source_backend", "milvus"),
                    "created_at": row["created_at"],
                    "batch_id": int(row.get("batch_id", 1)),
                }
            )

        upsert_stmt = insert(RagEvalSample.__table__).values(payload)
        # 冲突更新载荷：这是 PostgreSQL upsert 语义所需，不是别名映射。
        upsert_stmt = upsert_stmt.on_conflict_do_update(
            index_elements=[RagEvalSample.id],
            set_={
                "category": upsert_stmt.excluded.category,
                "difficulty": upsert_stmt.excluded.difficulty,
                "question": upsert_stmt.excluded.question,
                "ground_truth_contexts": upsert_stmt.excluded.ground_truth_contexts,
                "ground_truth": upsert_stmt.excluded.ground_truth,
                "source_document": upsert_stmt.excluded.source_document,
                "model_name": upsert_stmt.excluded.model_name,
                "metadata": upsert_stmt.excluded.metadata,
                "source_chunk_index": upsert_stmt.excluded.source_chunk_index,
                "source_backend": upsert_stmt.excluded.source_backend,
                "created_at": upsert_stmt.excluded.created_at,
                "batch_id": upsert_stmt.excluded.batch_id,
            },
        )

        with self.client.get_session() as session:
            session.execute(upsert_stmt)
            session.commit()

3. 🚁 修改分析者与分析脚本

# src/augmented/analyst.py

  • Analyst.analyze() 一键调用，查询评估结果，诊断坏案例并入库，自动生成完整报告
  • Analyst.to_markdown() 生成报告
"""RAG 评估结果分析器。通过 LLMRouter 进行诊断生成。"""
# src/augmented/analyst.py
from __future__ import annotations

import ast
import logging
import re
import asyncio
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional

import pandas as pd
from langchain_core.prompts import ChatPromptTemplate
from sqlalchemy import desc

from src.augmented.config import GeneratorConfig, build_default_config
from src.augmented.llm_router import LLMRouter
from src.augmented.utils import run_async
from src.core.models import RagEvalDiagnosis, RagEvalResult
from src.core.postgres_client import get_postgres_client
from src.schema.augmented_schema import EvalResultSample

logger = logging.getLogger(__name__)


class RAGAnalyst:
    """分析低分样本并给出诊断建议。"""

    def __init__(self, generator_config: Optional[GeneratorConfig] = None):
        # 使用 llm_router 管理多模型降级调用。
        self.generator_config = generator_config or build_default_config()
        self.router = LLMRouter(self.generator_config, llm_group="analyst_llms")
        # 保存最近一次分析生成的完整 markdown 报告。
        self.report: str = ""

        self._prompt = ChatPromptTemplate.from_template(
            """
你是一个资深的 RAG 系统调试专家。
以下是一条低分样本，请输出结构化诊断：

【用户问题】{question}
【检索上下文】{contexts}
【模型回答】{answer}
【标准答案】{ground_truth}
【标准上下文】{ground_truth_contexts}

【指标】
- Faithfulness: {faithfulness}
- Answer Relevancy: {answer_relevancy}
- Context Precision: {context_precision}
- Context Recall: {context_recall}

请输出：
1) 根因判断（检索失败 / 排序失败 / 生成失败 / 混合问题）
2) 证据（引用上下文或回答中的具体片段）
3) 1条可执行优化建议（具体到参数或策略）

直接输出分析结论，不要客套。
"""
        )

    @staticmethod
    def _to_text_list(value) -> List[str]:
        """将上下文字段统一为字符串列表，兼容 CSV 中的字符串化 list。"""
        if value is None:
            return []
        if isinstance(value, list):
            return [str(x) for x in value]
        if isinstance(value, str):
            s = value.strip()
            if not s:
                return []
            # 兼容 "['a', 'b']" 这类字符串化列表。
            if s.startswith("[") and s.endswith("]"):
                try:
                    parsed = ast.literal_eval(s)
                    if isinstance(parsed, list):
                        return [str(x) for x in parsed]
                except Exception:
                    pass
            return [s]
        return [str(value)]

    @staticmethod
    def _normalize_text(text: str) -> str:
        """归一化文本以便做规则匹配：去掉 <think>、空白和中英文常见标点。"""
        s = str(text or "")
        s = re.sub(r"<think>.*?</think>", "", s, flags=re.S | re.I)
        s = s.strip().lower()
        s = re.sub(r"\s+", "", s)
        s = re.sub(r"[，。！？；：、,.!?;:\"'`~\\-_\\(\\)\\[\\]{}<>]", "", s)
        return s

    def _is_unanswerable_text(self, text: str) -> bool:
        """判断文本是否表达“依据上下文无法回答”。"""
        s = self._normalize_text(text)
        cues = [
            "根据提供的上下文无法回答",
            "根据上下文无法回答",
            "无法根据提供的上下文回答",
            "上下文无法支持答案",
            "未明确提及",
            "信息不足",
            "无法依据现有信息判断",
            "无法确定",
            "无法回答",
        ]
        return any(cue in s for cue in cues)

    def _is_expected_unanswerable_and_answered_correctly(self, row: pd.Series) -> bool:
        """
        识别“正确拒答”样本：
        - 标准答案本身是不可回答
        - 模型回答也明确不可回答
        这类样本不应作为坏案例。
        """
        gt = str(row.get("ground_truth", ""))
        ans = str(row.get("answer", ""))
        return self._is_unanswerable_text(gt) and self._is_unanswerable_text(ans)

    def _score_and_select(self, df_results: pd.DataFrame, top_k: int) -> pd.DataFrame:
        required = ["faithfulness", "answer_relevancy", "context_precision", "context_recall"]
        missing = [c for c in required if c not in df_results.columns]
        if missing:
            raise ValueError(f"分析失败：缺少指标列 {missing}")

        work = df_results.copy()
        for c in required:
            work[c] = pd.to_numeric(work[c], errors="coerce").fillna(0.0)

        # 先按评估分数构建候选集。
        work["avg_score"] = work[required].mean(axis=1)
        # 关键改进：排除“正确拒答”样本，避免将其误报为坏案例。
        # 典型场景：标准答案就是“无法回答”，模型也确实返回“无法回答”。
        work["is_correct_abstention"] = work.apply(
            self._is_expected_unanswerable_and_answered_correctly, axis=1
        )
        candidates = work[~work["is_correct_abstention"]].copy()

        if candidates.empty:
            logger.info("坏案例筛选后为空：所有低分样本均为‘正确拒答’或无需诊断。")
            return candidates

        return candidates.nsmallest(top_k, "avg_score")

    async def _diagnose_one(self, payload: Dict) -> Dict[str, str]:
        """在线程池中执行同步 router.invoke，避免阻塞事件循环。"""
        text, model = await asyncio.to_thread(self.router.invoke, self._prompt, payload)
        return {"diagnosis": text, "model": model or self.generator_config.default_model_name}

    async def analyze_bad_cases(self, df_results: pd.DataFrame, top_k: int = 5) -> List[Dict]:
        """异步分析低分 Top-K 样本。"""
        if df_results is None or df_results.empty:
            return []

        worst_cases = self._score_and_select(df_results, top_k)
        logger.info("开始分析低分样本：top_k=%s", top_k)

        reports: List[Dict] = []
        for idx, row in worst_cases.iterrows():
            item = EvalResultSample(
                sample_id=str(row.get("sample_id", "")),
                sample_batch_id=int(row.get("sample_batch_id", 1) or 1),
                question=str(row.get("question", "")),
                answer=str(row.get("answer", "")),
                contexts=self._to_text_list(row.get("contexts", [])),
                ground_truth=str(row.get("ground_truth", "")),
                ground_truth_contexts=self._to_text_list(row.get("ground_truth_contexts", [])),
                faithfulness=float(row.get("faithfulness", 0.0) or 0.0),
                answer_relevancy=float(row.get("answer_relevancy", 0.0) or 0.0),
                context_precision=float(row.get("context_precision", 0.0) or 0.0),
                context_recall=float(row.get("context_recall", 0.0) or 0.0),
            )
            payload = {
                "question": item.question,
                "contexts": "\n".join(item.contexts),
                "answer": item.answer,
                "ground_truth": item.ground_truth,
                "ground_truth_contexts": "\n".join(item.ground_truth_contexts),
                "faithfulness": item.faithfulness,
                "answer_relevancy": item.answer_relevancy,
                "context_precision": item.context_precision,
                "context_recall": item.context_recall,
            }

            try:
                diagnose_out = await self._diagnose_one(payload)
                diagnosis = diagnose_out["diagnosis"]
                diagnosis_model = diagnose_out["model"]
            except Exception as e:
                logger.exception("低分样本诊断失败 idx=%s", idx)
                diagnosis = f"诊断失败：{e}"
                diagnosis_model = self.generator_config.default_model_name

            reports.append(
                {
                    "id": idx,
                    "sample_id": item.sample_id,
                    "sample_batch_id": item.sample_batch_id,
                    "question": item.question,
                    "answer": item.answer,
                    "contexts": item.contexts,
                    "ground_truth": item.ground_truth,
                    "ground_truth_contexts": item.ground_truth_contexts,
                    "scores": {
                        "faithfulness": item.faithfulness,
                        "answer_relevancy": item.answer_relevancy,
                        "context_precision": item.context_precision,
                        "context_recall": item.context_recall,
                        "avg_score": float(worst_cases.loc[idx, "avg_score"]),
                    },
                    "predicted_category": self._auto_categorize_error(item.model_dump()),
                    "diagnosis": diagnosis,
                    "diagnosis_model": diagnosis_model,
                }
            )

        return reports

    def analyze_bad_cases_sync(self, df_results: pd.DataFrame, top_k: int = 5) -> List[Dict]:
        """同步入口，便于在脚本中直接调用。"""
        return run_async(self.analyze_bad_cases(df_results=df_results, top_k=top_k))

    @staticmethod
    def _clip_text(text: str, max_len: int = 500) -> str:
        if not isinstance(text, str):
            return ""
        s = text.strip()
        return s if len(s) <= max_len else s[:max_len] + "..."

    @staticmethod
    def _strip_think(text: str) -> str:
        """移除模型返回中的 <think> 思考片段，保留最终可展示答案。"""
        if not isinstance(text, str):
            return ""
        return re.sub(r"<think>.*?</think>", "", text, flags=re.S | re.I).strip()

    @staticmethod
    def _format_diagnosis_text(text: str) -> str:
        """将诊断文本整理为更易读的段落/列表结构。"""
        if not isinstance(text, str):
            return ""

        formatted = text.strip().replace("\r\n", "\n").replace("\r", "\n")
        if not formatted:
            return ""
        formatted = re.sub(r"\*\*([^\n*]+)\n\*\*", r"**\1**", formatted)
        formatted = re.sub(r"\s*(#{2,6}\s*)", r"\n\1", formatted)
        formatted = re.sub(r"\s*(\d+\)\s*)", r"\n\1", formatted)
        formatted = re.sub(r"\s*(-\s+)", r"\n- ", formatted)
        formatted = re.sub(r"\s*(\*\*(?:根因判断|证据|可执行优化建议|优化建议)\*\*)", r"\n\1", formatted)
        formatted = re.sub(r"[ \t]+\n", "\n", formatted)
        formatted = re.sub(r"\n{3,}", "\n\n", formatted)
        return formatted.strip()

    @staticmethod
    def _to_blockquote(text: str) -> str:
        lines = (text or "").splitlines() or [""]
        return "\n".join([f"> {line}" if line.strip() else ">" for line in lines])

    def _format_context_blocks(self, contexts, max_items: int = 3, each_max_len: int = 300) -> str:
        items = self._to_text_list(contexts)
        if not items:
            return "> （无）"

        blocks = []
        for i, raw_ctx in enumerate(items[:max_items], start=1):
            ctx = self._clip_text(self._strip_think(str(raw_ctx)), max_len=each_max_len)
            blocks.append(f"#### 引用 {i}\n{self._to_blockquote(ctx)}")
        return "\n\n".join(blocks)

    def _build_case_markdown(self, report: dict, rank: int) -> str:
        """构建单个坏案例 markdown，同时用于数据库 diagnosis_markdown 字段。"""
        q = report.get("question", "")
        scores = report.get("scores", {})
        diagnosis = self._format_diagnosis_text(report.get("diagnosis", ""))
        category = report.get("predicted_category", "")
        model = report.get("diagnosis_model", "")
        answer = self._strip_think(report.get("answer", ""))
        contexts = report.get("contexts", [])
        ground_truth = report.get("ground_truth", "")
        gt_contexts = report.get("ground_truth_contexts", [])

        return (
            f"## 案例 {rank}: {category}\n"
            f"**问题**: {q}\n"
            f"**得分**: Faithfulness={float(scores.get('faithfulness', 0.0)):.2f}, "
            f"Recall={float(scores.get('context_recall', 0.0)):.2f}\n"
            f"**诊断模型**: {model}\n\n"
            f"### 🤖 模型生成答案\n{self._clip_text(answer, max_len=1000)}\n\n"
            f"### 📚 检索引用信息（Top-3）\n{self._format_context_blocks(contexts, max_items=3, each_max_len=500)}\n\n"
            f"### ✅ 标准答案（Ground Truth）\n{self._clip_text(ground_truth, max_len=1000)}\n\n"
            f"### 📖 标准参考上下文（Top-2）\n{self._format_context_blocks(gt_contexts, max_items=2, each_max_len=400)}\n\n"
            f"### 🩺 LLM 专家诊断\n{diagnosis}\n\n"
        )

    def _build_report_markdown(self, eval_run_id: str, reports: List[Dict]) -> str:
        head = f"# 🚨 RAG 坏案例深度诊断报告\n\n**评估运行ID**: `{eval_run_id}`\n\n"
        if not reports:
            return head + "（无坏案例）\n"
        body = []
        for i, report in enumerate(reports, start=1):
            body.append(self._build_case_markdown(report, i) + "---\n")
        return head + "\n".join(body)

    def _save_diagnosis_results(self, eval_run_id: str, reports: List[Dict]) -> int:
        """将坏案例诊断结果写入 rag_eval_diagnoses。"""
        if not reports:
            return 0

        # 关键点：诊断结果与评估结果共享 eval_run_id，支持后续追踪同一轮运行。
        rows = []
        now = datetime.utcnow()
        for rank, report in enumerate(reports, start=1):
            scores = report.get("scores", {}) or {}
            rows.append(
                {
                    "eval_run_id": eval_run_id,
                    "sample_id": str(report.get("sample_id", report.get("id", ""))),
                    "sample_batch_id": int(report.get("sample_batch_id", 1) or 1),
                    "bad_case_rank": rank,
                    "predicted_category": str(report.get("predicted_category", "")),
                    "diagnosis_model": str(report.get("diagnosis_model", "")),
                    "diagnosis": self._format_diagnosis_text(report.get("diagnosis", "")),
                    "diagnosis_markdown": self._build_case_markdown(report, rank),
                    "question": str(report.get("question", "")),
                    "answer": self._strip_think(str(report.get("answer", ""))),
                    "contexts": self._to_text_list(report.get("contexts", [])),
                    "ground_truth": str(report.get("ground_truth", "")),
                    "ground_truth_contexts": self._to_text_list(report.get("ground_truth_contexts", [])),
                    "faithfulness": float(scores.get("faithfulness", 0.0) or 0.0),
                    "answer_relevancy": float(scores.get("answer_relevancy", 0.0) or 0.0),
                    "context_precision": float(scores.get("context_precision", 0.0) or 0.0),
                    "context_recall": float(scores.get("context_recall", 0.0) or 0.0),
                    "avg_score": float(scores.get("avg_score", 0.0) or 0.0),
                    "created_at": now,
                }
            )

        pg = get_postgres_client()
        with pg.get_session() as session:
            session.bulk_insert_mappings(RagEvalDiagnosis, rows)
            session.commit()
        return len(rows)

    def to_markdown(self, path: str) -> str:
        """将最近一次分析报告写入指定 markdown 文件。"""
        target = Path(path)
        target.parent.mkdir(parents=True, exist_ok=True)
        with open(target, "w", encoding="utf-8") as f:
            f.write(self.report or "# 🚨 RAG 坏案例深度诊断报告\n\n（空报告）\n")
        return str(target)

    def analyze(
        self,
        limit: int = 200,
        batch_ids: Optional[List[int]] = None,
        eval_run_id: Optional[str] = None,
        bad_case_top_k: int = 5,
    ) -> Dict:
        """
        一体化分析入口：
        1) 从评估结果表 rag_eval_results 按 batch_ids / eval_run_id 读取数据
        2) 分析坏案例并写入诊断表 rag_eval_diagnoses
        3) 生成完整 markdown 报告到 self.report
        """
        pg = get_postgres_client()
        with pg.get_session() as session:
            q = session.query(RagEvalResult)
            if batch_ids:
                q = q.filter(RagEvalResult.sample_batch_id.in_(batch_ids))
            if eval_run_id:
                q = q.filter(RagEvalResult.eval_run_id == eval_run_id)
            q = q.order_by(desc(RagEvalResult.created_at))
            rows = q.limit(limit).all()

        if not rows:
            run_id = eval_run_id or f"diag_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}"
            self.report = self._build_report_markdown(eval_run_id=run_id, reports=[])
            return {
                "eval_run_id": run_id,
                "eval_count": 0,
                "diagnosis_count": 0,
                "df_results": pd.DataFrame(),
                "reports": [],
            }

        # 若未指定 eval_run_id，默认取最近一条记录所属的运行批次，避免混入多次运行数据。
        selected_run_id = eval_run_id or rows[0].eval_run_id
        selected_rows = [r for r in rows if r.eval_run_id == selected_run_id]
        df = pd.DataFrame(
            [
                {
                    "sample_id": r.sample_id,
                    "sample_batch_id": r.sample_batch_id,
                    "question": r.question,
                    "answer": r.answer,
                    "contexts": r.contexts or [],
                    "ground_truth": r.ground_truth,
                    "ground_truth_contexts": r.ground_truth_contexts or [],
                    "faithfulness": r.faithfulness,
                    "answer_relevancy": r.answer_relevancy,
                    "context_precision": r.context_precision,
                    "context_recall": r.context_recall,
                }
                for r in selected_rows
            ]
        )

        if df is None or df.empty:
            self.report = self._build_report_markdown(eval_run_id=selected_run_id, reports=[])
            return {
                "eval_run_id": selected_run_id,
                "eval_count": 0,
                "diagnosis_count": 0,
                "df_results": df,
                "reports": [],
            }

        reports = self.analyze_bad_cases_sync(df_results=df, top_k=bad_case_top_k)
        saved_diag = self._save_diagnosis_results(eval_run_id=selected_run_id, reports=reports)
        self.report = self._build_report_markdown(eval_run_id=selected_run_id, reports=reports)

        logger.info(
            "分析完成 eval_run_id=%s source_eval_rows=%s diagnosis_saved=%s",
            selected_run_id,
            len(df),
            saved_diag,
        )
        return {
            "eval_run_id": selected_run_id,
            "eval_count": int(len(df)),
            "diagnosis_count": int(saved_diag),
            "df_results": df,
            "reports": reports,
        }

    def _auto_categorize_error(self, row: Dict) -> str:
        """基于指标 heuristic 自动归类。"""
        if row["context_recall"] < 0.5:
            return "检索失败 (Retrieval Failure)"
        if row["context_precision"] < 0.5:
            return "排序失败 (Ranking Failure)"
        if row["faithfulness"] < 0.6:
            return "幻觉问题 (Hallucination)"
        if row["answer_relevancy"] < 0.6:
            return "答非所问 (Irrelevance)"
        return "混合问题 / 需人工复核"

test/run_evaluation.py

"""
运行评估脚本（固定配置版）
直接修改下方 CONFIG 字典后执行：
python -m src.test.run_evaluation
"""

import logging
from datetime import datetime
from pathlib import Path

from src.augmented.analyst import RAGAnalyst


CONFIG = {
    "limit": 15,
    "batch_ids": [1],  # 按批次评估
    # 可选：指定某次评估运行ID；不填时默认使用最近一次 eval_run_id
    "eval_run_id": None,
    "bad_case_top_k": 3,  # 坏案例诊断数量
    "save_markdown_report": True,
    "report_dir": "C:\\Users\\pdnbplus\\Documents\\python全系列\\AIAgent开发\\data\\eval",
}


def main():
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")

    analyst = RAGAnalyst()
    result = analyst.analyze(
        limit=CONFIG["limit"],
        batch_ids=CONFIG.get("batch_ids"),
        eval_run_id=CONFIG.get("eval_run_id"),
        bad_case_top_k=CONFIG["bad_case_top_k"],
    )

    eval_run_id = result["eval_run_id"]
    print(f"\n=== 评估运行 ID ===\n{eval_run_id}")
    print(f"评估结果入库条数: {result['eval_count']}")
    print(f"诊断结果入库条数: {result['diagnosis_count']}")

    if CONFIG.get("save_markdown_report", False):
        report_dir = Path(CONFIG["report_dir"])
        report_name = f"bad_cases_report_{datetime.now().strftime('%Y%m%d_%H%M')}.md"
        report_path = report_dir / report_name
        saved_path = analyst.to_markdown(str(report_path))
        print(f"报告已输出: {saved_path}")

if __name__ == "__main__":
    main()

执行脚本

python -m src.test.run_evaluation

在这里插入图片描述

📂 当前项目结构解析

./augmented/ 📂
├── __init__.py         📦 # 包导出入口，导出核心类 (DatasetGenerator, RAGEvaluator, RAGAnalyst 等)
├── analyst.py          📊 # 结果诊断：查询评估结果，诊断坏案例并入库，自动生成完整报告
├── config.py           ⚙️ # 负责模块运行配置（chunk 数量、最小长度、策略开关、策略参数 JSON、LLM 配置文件路径）
├── data_generator.py   🏭 # 主编排入口：Milvus 抽取 -> 策略构任务 -> 调 LLM -> 解析校验 -> 组装样本 -> 写 PostgreSQL
├── evaluator.py        📝 # RAG 效果评估器：加载测试集 -> 调用 Pipeline 推理 -> 计算 Ragas 指标 -> 写 PostgreSQL
├── llm_endpoints.json  🔌 # LLM 端点配置文件
├── llm_router.py       🚦 # LLM 调用路由层（故障转移、模型分发）
├── models.py           🗃️ # Pydantic 数据模型定义（输入/输出结构、数据库映射对象）
├── prompts.py          💬 # Prompt 模板中心（任务提示词配置）
├── sinks.py            🚠 # 落库适配层 (数据汇：PostgreSQL 写入接口)
├── sources.py          📥 # 数据源适配层 (数据源：Milvus/文件读取接口)
└── strategies.py       ♟️ # 策略模式核心（定义不同的数据增强/改写算法实现）

🌊 系统数据流向架构图

红色线为数据流向示意：

在这里插入图片描述

# 语言模型 # python
Logo
亚马逊云科技技术品牌专区

更多推荐

cover

一次 API 调用搞定时序预测：时序大模型 TimechoAI 使用与能力全解析

avatar 亚马逊云科技技术品牌专区
cover

智能软件工程AI4SE（十六）——需求工程工具Jira

avatar 亚马逊云科技技术品牌专区

AI Agent 面试题 353：A2A协议的服务发现和注册机制详解

A2A 协议 是 AI Agent 技术体系中的重要组成部分。简单来说，它涉及到 Agent 如何在 工具使用与函数调用 层面实现智能化的行为和决策。在实际应用中，A2A 协议 的核心目标是让 Agent 能够更加高效、准确地完成特定任务。这需要我们深入理解其底层原理和实现机制。从学术角度来看，A2A 协议 的研究可以追溯到人工智能的早期阶段。早在 1950 年代，Alan Turing 就提出了

avatar 亚马逊云科技技术品牌专区