LLM评估体系

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_recall,
    context_precision,
)
from datasets import Dataset

# 准备评估数据集
# 每条数据需要：问题、检索到的文档、模型回答、参考答案（context_recall 需要）
eval_data = {
    "question": [
        "Flink 的 Checkpoint 和 Savepoint 有什么区别？",
        "什么是 Kafka 的消费者组？",
        "Spark 的 RDD 和 DataFrame 有什么区别？",
    ],
    "answer": [
        # 模型实际生成的回答
        "Checkpoint 是 Flink 自动触发的容错机制，用于故障恢复。Savepoint 是手动触发的状态快照，专用于版本升级和迁移。",
        "消费者组是 Kafka 的消费端抽象，同组内的消费者共同消费一个 Topic 的所有分区，每个分区只被组内一个消费者消费。",
        "RDD 是 Flink 底层的数据结构，不支持 SQL。DataFrame 是基于 RDD 的高层抽象，支持 SQL 查询且有 Schema。",
    ],
    "contexts": [
        # 检索返回的文档列表（每道题对应一个文档列表）
        [
            "Apache Flink 的 Checkpoint 是一种容错机制，由 Flink 运行时自动触发...",
            "Savepoint 是用户手动触发的全局一致性快照...",
        ],
        [
            "Kafka 消费者组（Consumer Group）允许多个消费者实例共同消费同一个 Topic...",
        ],
        [
            "Spark 的 RDD（弹性分布式数据集）是最基础的数据抽象...",
            "DataFrame 是 Spark 1.3 引入的高层 API，提供了类似 SQL 的操作接口...",
        ],
    ],
    "ground_truth": [
        # 参考答案（Context Recall 需要用到）
        "Checkpoint 自动触发用于故障恢复，Savepoint 手动触发用于版本升级和状态迁移。两者都是全局一致性快照。",
        "消费者组使得多个消费者可以协作消费一个 Topic，实现负载均衡，每个分区在同一时刻只能被组内一个消费者消费。",
        "RDD 是低层 API，类型安全但缺少优化；DataFrame 有 Schema，支持 SQL，Catalyst 优化器自动优化执行计划。",
    ],
}

dataset = Dataset.from_dict(eval_data)

# 执行评估（RAGAS 内部会调用 LLM 来判断语义匹配）
results = evaluate(
    dataset=dataset,
    metrics=[
        context_precision,
        context_recall,
        faithfulness,
        answer_relevancy,
    ],
    # 默认使用 gpt-3.5-turbo，可以换成更便宜的模型
    # llm=ChatOpenAI(model="gpt-4o-mini"),
)

print(results)
# {'context_precision': 0.88, 'context_recall': 0.75,
#  'faithfulness': 0.92, 'answer_relevancy': 0.85}

# 转成 DataFrame 详细分析
df = results.to_pandas()
print(df"question", "faithfulness", "answer_relevancy")

import json
from pathlib import Path
from datetime import datetime

def run_automated_evaluation(
    test_cases_path: str,
    output_path: str,
    pass_threshold: dict = None,
) -> bool:
    """
    运行自动化评估，返回是否通过质量门禁
    可集成到 GitHub Actions 等 CI/CD 流水线
    """
    if pass_threshold is None:
        pass_threshold = {
            "faithfulness": 0.85,        # 幻觉率不超过 15%
            "answer_relevancy": 0.80,    # 回答相关性不低于 80%
            "context_precision": 0.75,   # 检索精度不低于 75%
        }

    # 加载测试用例
    with open(test_cases_path, encoding="utf-8") as f:
        test_cases = json.load(f)

    dataset = Dataset.from_dict(test_cases)
    results = evaluate(
        dataset=dataset,
        metrics=[faithfulness, answer_relevancy, context_precision],
    )

    scores = {
        "faithfulness": results["faithfulness"],
        "answer_relevancy": results["answer_relevancy"],
        "context_precision": results["context_precision"],
    }

    # 判断是否通过
    passed = all(scores[k] >= pass_threshold[k] for k in pass_threshold)

    # 输出评估报告
    report = {
        "timestamp": datetime.now().isoformat(),
        "scores": scores,
        "thresholds": pass_threshold,
        "passed": passed,
        "failed_metrics": [
            k for k, v in scores.items()
            if v < pass_threshold.get(k, 0)
        ],
    }

    with open(output_path, "w", encoding="utf-8") as f:
        json.dump(report, f, ensure_ascii=False, indent=2)

    print(f"评估{'通过' if passed else '未通过'}：{scores}")
    return passed

# CI/CD 中使用
# if not run_automated_evaluation("test_cases.json", "eval_report.json"):
#     sys.exit(1)  # 失败则阻断部署

from openai import OpenAI
import json
from typing import Optional

client = OpenAI()

SCORING_PROMPT_TEMPLATE = """你是一个专业的 AI 质量评估专家。

请从以下四个维度评估【助手回答】的质量，每个维度打 1-5 分：

**评分维度**：
1. **正确性（Correctness）**：答案是否事实准确？
   - 5分：完全准确，无错误
   - 3分：基本准确，有小瑕疵
   - 1分：存在明显错误或幻觉

2. **相关性（Relevancy）**：答案是否切题？
   - 5分：完全切题，直接回答了问题
   - 3分：基本相关，但有离题部分
   - 1分：答非所问

3. **完整性（Completeness）**：答案是否覆盖了问题的主要方面？
   - 5分：全面覆盖，无明显遗漏
   - 3分：覆盖主要方面，有次要遗漏
   - 1分：严重不完整

4. **清晰度（Clarity）**：答案是否易于理解？
   - 5分：结构清晰，表达准确
   - 3分：基本清晰，有些地方可以更好
   - 1分：混乱或难以理解

**用户问题**：
{question}

**参考信息**（如果有）：
{context}

**助手回答**：
{answer}

请严格按照以下 JSON 格式输出，不要有其他内容：
{{
  "correctness": <1-5分>,
  "relevancy": <1-5分>,
  "completeness": <1-5分>,
  "clarity": <1-5分>,
  "overall": <四项平均值，保留一位小数>,
  "reasoning": "<简要说明打分理由，100字以内>",
  "issues": ["<存在的问题1>", "<存在的问题2>"]
}}"""

def llm_judge_score(
    question: str,
    answer: str,
    context: str = "",
    judge_model: str = "gpt-4o-mini",
) -> dict:
    """
    使用 LLM 对单条输出打分
    返回各维度分数和理由
    """
    prompt = SCORING_PROMPT_TEMPLATE.format(
        question=question,
        context=context if context else "（无参考信息）",
        answer=answer,
    )

    response = client.chat.completions.create(
        model=judge_model,
        messages=[{"role": "user", "content": prompt}],
        response_format={"type": "json_object"},
        temperature=0,  # 评估需要确定性，不要随机
    )

    return json.loads(response.choices[0].message.content)

# 使用示例
score = llm_judge_score(
    question="什么是 Kafka 的消费者组？",
    answer="Kafka 消费者组是一组共同消费 Topic 的消费者，每个分区只能被组内一个消费者消费，实现负载均衡。",
    context="Kafka Consumer Group 是 Kafka 消费端的核心抽象...",
)
print(score)
# {
#   "correctness": 5,
#   "relevancy": 5,
#   "completeness": 4,
#   "clarity": 5,
#   "overall": 4.8,
#   "reasoning": "答案准确、切题，略缺少 Rebalance 机制的说明",
#   "issues": ["未提及 Rebalance 触发场景"]
# }

PAIRWISE_PROMPT_TEMPLATE = """你是一个专业的 AI 质量评估专家。

请比较以下两个回答哪个更好。

**用户问题**：{question}

**回答 A**：
{answer_a}

**回答 B**：
{answer_b}

评估标准（按重要性排序）：
1. 事实正确性（最重要）
2. 问题相关性
3. 表达清晰度
4. 内容完整性

注意：避免长度偏见，不要因为某个回答更长就认为它更好。

请输出 JSON：
{{
  "winner": "A" 或 "B" 或 "tie",
  "confidence": "high" 或 "medium" 或 "low",
  "reasoning": "<比较理由，150字以内>",
  "a_strengths": ["<A的优点>"],
  "b_strengths": ["<B的优点>"]
}}"""

def pairwise_compare(
    question: str,
    answer_a: str,
    answer_b: str,
    judge_model: str = "gpt-4o-mini",
    swap_and_average: bool = True,  # 交换顺序再评估一次，缓解位置偏见
) -> dict:
    """
    对比两个答案，判断哪个更好
    swap_and_average=True 时：正序 + 逆序各评估一次，取综合结论
    """
    def single_compare(a: str, b: str) -> dict:
        prompt = PAIRWISE_PROMPT_TEMPLATE.format(
            question=question, answer_a=a, answer_b=b
        )
        response = client.chat.completions.create(
            model=judge_model,
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"},
            temperature=0,
        )
        return json.loads(response.choices[0].message.content)

    result_1 = single_compare(answer_a, answer_b)

    if not swap_and_average:
        return result_1

    # 交换顺序再评估（注意：winner 需要取反）
    result_2 = single_compare(answer_b, answer_a)

    # 汇总两次结果
    winner_votes = {"A": 0, "B": 0, "tie": 0}

    if result_1["winner"] == "A":
        winner_votes["A"] += 1
    elif result_1["winner"] == "B":
        winner_votes["B"] += 1
    else:
        winner_votes["tie"] += 1

    # 第二次评估中 A/B 是反的
    if result_2["winner"] == "A":  # A 是原始的 B
        winner_votes["B"] += 1
    elif result_2["winner"] == "B":  # B 是原始的 A
        winner_votes["A"] += 1
    else:
        winner_votes["tie"] += 1

    final_winner = max(winner_votes, key=winner_votes.get)

    return {
        "winner": final_winner,
        "vote_counts": winner_votes,
        "confidence": "high" if winner_votes[final_winner] == 2 else "low",
        "round1": result_1,
        "round2": result_2,
    }

import concurrent.futures
from tqdm import tqdm

def batch_llm_judge(
    eval_cases: list[dict],
    judge_model: str = "gpt-4o-mini",
    max_workers: int = 5,  # 并发数，注意 API 限速
) -> list[dict]:
    """
    批量评估，使用线程池并发加速
    """
    def evaluate_single(case: dict) -> dict:
        score = llm_judge_score(
            question=case["question"],
            answer=case["answer"],
            context=case.get("context", ""),
            judge_model=judge_model,
        )
        return {**case, "judge_score": score}

    results = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(evaluate_single, case): case for case in eval_cases}

        for future in tqdm(
            concurrent.futures.as_completed(futures),
            total=len(eval_cases),
            desc="LLM 评估中"
        ):
            try:
                results.append(future.result())
            except Exception as e:
                print(f"评估失败：{e}")

    # 汇总统计
    all_scores = [r["judge_score"]["overall"] for r in results if "judge_score" in r]
    print(f"\n评估完成：{len(results)} 条")
    print(f"平均综合分：{sum(all_scores)/len(all_scores):.2f}/5.0")
    print(f"高质量（>4分）比例：{sum(1 for s in all_scores if s > 4)/len(all_scores):.1%}")

    return results

# 评估任务分发系统（简化版）
from dataclasses import dataclass, field
from datetime import datetime
import random
import json

@dataclass
class EvalTask:
    task_id: str
    question: str
    answer: str
    context: str = ""
    assigned_to: str = ""
    completed: bool = False
    human_score: dict = field(default_factory=dict)
    created_at: str = field(default_factory=lambda: datetime.now().isoformat())

class HumanEvalWorkflow:
    """
    人工评估任务管理
    - 随机分配评估任务
    - 同一条数据由 2 人独立评估（用于计算标注一致性）
    - 分歧较大时触发第三人裁决
    """

    def __init__(self, evaluators: list[str], agreement_threshold: float = 0.8):
        self.evaluators = evaluators
        self.agreement_threshold = agreement_threshold
        self.tasks: list[EvalTask] = []

    def add_tasks(self, eval_cases: list[dict]) -> None:
        """添加评估任务，重要数据分配 2 个标注者"""
        for case in eval_cases:
            task = EvalTask(
                task_id=f"eval_{len(self.tasks):04d}",
                question=case["question"],
                answer=case["answer"],
                context=case.get("context", ""),
            )
            self.tasks.append(task)

    def assign_tasks(self) -> dict[str, list[str]]:
        """
        为每个评估者分配任务
        每条数据分配给 2 个不同的评估者
        """
        assignments: dict[str, list[str]] = {e: [] for e in self.evaluators}

        for task in self.tasks:
            # 随机选 2 个评估者
            assigned = random.sample(self.evaluators, min(2, len(self.evaluators)))
            for evaluator in assigned:
                assignments[evaluator].append(task.task_id)

        return assignments

    def compute_inter_rater_agreement(
        self,
        scores_a: list[float],
        scores_b: list[float],
    ) -> float:
        """
        计算两个标注者之间的一致性（Cohen's Kappa 近似）
        实际生产中建议使用 sklearn.metrics.cohen_kappa_score
        """
        agreements = sum(
            1 for a, b in zip(scores_a, scores_b)
            if abs(a - b) <= 1  # 相差不超过 1 分算一致
        )
        return agreements / len(scores_a) if scores_a else 0.0

    def flag_disagreements(self, task_id: str, scores: list[float]) -> bool:
        """检查是否需要第三人裁决"""
        if len(scores) < 2:
            return False
        return abs(scores[0] - scores[1]) > 2  # 分差超过 2 分，需要裁决

def smart_sampling(
    production_logs: list[dict],
    sample_size: int = 200,
) -> list[dict]:
    """
    策略性采样：困难案例重点评估
    """
    samples = []

    # 1. 随机样本（20%）：代表正态分布，用于总体质量估计
    random_sample = random.sample(production_logs, int(sample_size * 0.2))
    samples.extend([(log, "random") for log in random_sample])

    # 2. 低自信案例（30%）：模型表达了不确定性的输出
    uncertainty_keywords = ["我不确定", "可能", "据我所知", "不太清楚"]
    uncertain_logs = [
        log for log in production_logs
        if any(kw in log.get("answer", "") for kw in uncertainty_keywords)
    ]
    uncertain_sample = random.sample(
        uncertain_logs, min(int(sample_size * 0.3), len(uncertain_logs))
    )
    samples.extend([(log, "uncertain") for log in uncertain_sample])

    # 3. 用户负反馈案例（30%）：被用户踩的输出，最有价值
    thumbs_down_logs = [
        log for log in production_logs
        if log.get("user_feedback", 1) == 0  # 0 = 踩
    ]
    thumbs_down_sample = random.sample(
        thumbs_down_logs, min(int(sample_size * 0.3), len(thumbs_down_logs))
    )
    samples.extend([(log, "thumbs_down") for log in thumbs_down_sample])

    # 4. 长上下文案例（20%）：复杂输入往往更容易出幻觉
    long_context_logs = [
        log for log in production_logs
        if len(log.get("context", "")) > 2000
    ]
    long_sample = random.sample(
        long_context_logs, min(int(sample_size * 0.2), len(long_context_logs))
    )
    samples.extend([(log, "long_context") for log in long_sample])

    return [{"log": log, "sample_type": t} for log, t in samples[:sample_size]]

def mine_hard_cases_from_logs(
    langfuse_client,
    days_back: int = 7,
    hard_case_criteria: dict = None,
) -> list[dict]:
    """
    从 LangFuse 日志中自动挖掘困难案例
    """
    if hard_case_criteria is None:
        hard_case_criteria = {
            "min_latency_seconds": 5,    # 超时的请求（可能上下文过长）
            "user_thumbs_down": True,    # 用户踩过的
            "low_faithfulness": 0.7,     # RAGAS 评分低的
        }

    hard_cases = []

    # 从 LangFuse 获取最近 N 天的 Trace
    from datetime import datetime, timedelta
    end = datetime.now()
    start = end - timedelta(days=days_back)

    traces = langfuse_client.fetch_traces(
        from_timestamp=start,
        to_timestamp=end,
        limit=1000,
    ).data

    for trace in traces:
        is_hard = False
        reasons = []

        # 检查延迟
        if trace.latency and trace.latency > hard_case_criteria["min_latency_seconds"] * 1000:
            is_hard = True
            reasons.append("high_latency")

        # 检查用户反馈
        scores = [s for s in (trace.scores or []) if s.name == "user_feedback"]
        if scores and scores[0].value < 0.5:
            is_hard = True
            reasons.append("user_thumbs_down")

        # 检查 RAGAS 分数
        faithfulness_scores = [s for s in (trace.scores or []) if s.name == "faithfulness"]
        if faithfulness_scores and faithfulness_scores[0].value < hard_case_criteria["low_faithfulness"]:
            is_hard = True
            reasons.append("low_faithfulness")

        if is_hard:
            hard_cases.append({
                "trace_id": trace.id,
                "question": trace.input.get("query", "") if trace.input else "",
                "answer": trace.output.get("answer", "") if trace.output else "",
                "hard_reasons": reasons,
            })

    return hard_cases

import json
from pathlib import Path
from datetime import datetime

class GoldenDataset:
    """
    黄金标注数据集管理
    - 存储经过人工验证的高质量问答对
    - 用于定期评估模型退化
    - 版本化管理，支持数据集迭代
    """

    def __init__(self, dataset_path: str = "golden_dataset.json"):
        self.path = Path(dataset_path)
        self.data = self._load()

    def _load(self) -> list[dict]:
        if self.path.exists():
            with open(self.path, encoding="utf-8") as f:
                return json.load(f)
        return []

    def add(
        self,
        question: str,
        ground_truth_answer: str,
        context: list[str],
        annotator: str,
        difficulty: str = "medium",  # easy / medium / hard
        domain: str = "general",
    ) -> None:
        entry = {
            "id": f"gd_{len(self.data):04d}",
            "question": question,
            "ground_truth": ground_truth_answer,
            "contexts": context,
            "metadata": {
                "annotator": annotator,
                "difficulty": difficulty,
                "domain": domain,
                "created_at": datetime.now().isoformat(),
            }
        }
        self.data.append(entry)
        self._save()

    def _save(self) -> None:
        with open(self.path, "w", encoding="utf-8") as f:
            json.dump(self.data, f, ensure_ascii=False, indent=2)

    def get_by_difficulty(self, difficulty: str) -> list[dict]:
        return [d for d in self.data if d["metadata"]["difficulty"] == difficulty]

    def export_for_ragas(self) -> dict:
        """导出为 RAGAS 评估格式"""
        return {
            "question": [d["question"] for d in self.data],
            "ground_truth": [d["ground_truth"] for d in self.data],
            "contexts": [d["contexts"] for d in self.data],
        }

def evaluation_gate(
    new_prompt_version: str,
    current_prompt_version: str,
    golden_dataset_path: str,
    min_improvement: float = 0.02,  # 至少提升 2%
) -> dict:
    """
    评估门禁：自动化决策新版本是否可以上线
    """
    golden_dataset = GoldenDataset(golden_dataset_path)

    # 对两个版本分别生成回答
    current_answers = generate_answers(
        golden_dataset.data, prompt_version=current_prompt_version
    )
    new_answers = generate_answers(
        golden_dataset.data, prompt_version=new_prompt_version
    )

    # RAGAS 评估
    current_scores = run_ragas_eval(current_answers, golden_dataset)
    new_scores = run_ragas_eval(new_answers, golden_dataset)

    # 综合得分对比
    current_overall = sum(current_scores.values()) / len(current_scores)
    new_overall = sum(new_scores.values()) / len(new_scores)

    improvement = new_overall - current_overall
    should_deploy = improvement >= min_improvement

    return {
        "current_version": current_prompt_version,
        "new_version": new_prompt_version,
        "current_overall": current_overall,
        "new_overall": new_overall,
        "improvement": improvement,
        "should_deploy": should_deploy,
        "scores_detail": {
            "current": current_scores,
            "new": new_scores,
        },
        "recommendation": (
            f"建议上线：新版本综合分提升 {improvement:.1%}"
            if should_deploy
            else f"不建议上线：提升不足（{improvement:.1%} < {min_improvement:.1%}）"
        ),
    }