17 | 评估方法论——怎么知道 Agent 做得好不好？

import json
from typing import Dict, List, Tuple


def rule_based_evaluate(agent_output: str, expected: Dict) -> Dict[str, bool]:
    """基于规则的 Agent 输出评估。

    Args:
        agent_output: Agent 返回的原始文本
        expected: 期望结果的字典

    Returns:
        各规则检查项的通过情况
    """
    results = {}

    # 规则 1: 输出是否为合法 JSON
    try:
        parsed = json.loads(agent_output)
        results["valid_json"] = True
    except json.JSONDecodeError:
        results["valid_json"] = False
        results["has_name"] = False
        results["has_rating"] = False
        results["has_address"] = False
        results["rating_in_range"] = False
        return results

    # 规则 2: 必填字段是否齐全
    required_fields = ["name", "rating", "address"]
    for field in required_fields:
        results[f"has_{field}"] = field in parsed

    # 规则 3: 评分是否在合理范围 (1.0 - 5.0)
    if "rating" in parsed:
        try:
            rating = float(parsed["rating"])
            results["rating_in_range"] = 1.0 <= rating <= 5.0
        except (ValueError, TypeError):
            results["rating_in_range"] = False

    # 规则 4: 餐厅名是否包含预期关键词
    if "name" in parsed and "expected_keywords" in expected:
        name = str(parsed["name"]).lower()
        results["name_matches"] = any(
            kw.lower() in name for kw in expected["expected_keywords"]
        )

    return results


# 使用示例
output = '{"name": "海的寿司", "rating": 4.5, "address": "海淀区中关村大街1号"}'
expected = {"expected_keywords": ["寿司", "日料", "海鲜"]}

scores = rule_based_evaluate(output, expected)
for rule, passed in scores.items():
    status = "PASS" if passed else "FAIL"
    print(f"  {rule}: {status}")
# 输出:
#   valid_json: PASS
#   has_name: PASS
#   has_rating: PASS
#   has_address: PASS
#   rating_in_range: PASS
#   name_matches: PASS

"""LLM-as-Judge 评估实现。

使用前请确保已安装 openai 库：pip install openai
并设置环境变量 OPENAI_API_KEY。
"""

import json
from openai import OpenAI


# 评分维度的 Prompt 模板
JUDGE_PROMPT = """你是一个专业的 AI 输出质量评估员。请对以下 Agent 回答进行评分。

## 用户问题
{question}

## Agent 的回答
{answer}

## 参考答案（如有）
{reference}

## 评分维度
请从以下 4 个维度评分（每个维度 1-5 分）：

1. **准确性 (accuracy)**: 回答中的事实信息是否正确？
   - 5分: 所有信息完全正确
   - 3分: 大部分正确，有小错误
   - 1分: 包含严重事实错误

2. **完整性 (completeness)**: 回答是否覆盖了用户问题的所有方面？
   - 5分: 全面覆盖，没有遗漏
   - 3分: 覆盖了主要方面，有少量遗漏
   - 1分: 大量遗漏，回答不完整

3. **相关性 (relevance)**: 回答是否切题？有没有答非所问？
   - 5分: 完全切题，每句话都跟问题相关
   - 3分: 基本切题，有少量跑题内容
   - 1分: 大量不相关内容

4. **自然度 (naturalness)**: 回答的语言是否自然流畅？
   - 5分: 像人类专家写的，流畅自然
   - 3分: 能看懂，但有些地方生硬
   - 1分: 语法混乱或机器味很重

## 输出格式
请严格以 JSON 格式输出：
{{
    "accuracy": <1-5>,
    "completeness": <1-5>,
    "relevance": <1-5>,
    "naturalness": <1-5>,
    "overall": <1-5>,
    "reasoning": "<用一两句话解释你的评分理由>"
}}
"""


def llm_judge_evaluate(
    question: str,
    answer: str,
    reference: str = "无",
    model: str = "gpt-4o",
) -> dict:
    """用 LLM-as-Judge 方法评估 Agent 回答质量。

    Args:
        question: 用户的原始问题
        answer: Agent 生成的回答
        reference: 参考答案（可选）
        model: 用作 Judge 的模型名称

    Returns:
        包含各维度评分的字典
    """
    client = OpenAI()

    prompt = JUDGE_PROMPT.format(
        question=question,
        answer=answer,
        reference=reference,
    )

    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        temperature=0.1,  # 低温度保证评分稳定
        response_format={"type": "json_object"},
    )

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


def batch_evaluate(test_cases: list[dict], model: str = "gpt-4o") -> dict:
    """批量评估多个测试用例。

    Args:
        test_cases: 测试用例列表，每个包含 question、answer、reference
        model: Judge 模型

    Returns:
        汇总评估结果
    """
    all_scores = []

    for i, case in enumerate(test_cases):
        score = llm_judge_evaluate(
            question=case["question"],
            answer=case["answer"],
            reference=case.get("reference", "无"),
            model=model,
        )
        all_scores.append(score)
        print(f"[{i+1}/{len(test_cases)}] overall={score['overall']}/5 - {score['reasoning'][:50]}")

    # 汇总统计
    n = len(all_scores)
    summary = {
        "total_cases": n,
        "avg_accuracy": sum(s["accuracy"] for s in all_scores) / n,
        "avg_completeness": sum(s["completeness"] for s in all_scores) / n,
        "avg_relevance": sum(s["relevance"] for s in all_scores) / n,
        "avg_naturalness": sum(s["naturalness"] for s in all_scores) / n,
        "avg_overall": sum(s["overall"] for s in all_scores) / n,
    }
    return summary


# 使用示例
if __name__ == "__main__":
    test_cases = [
        {
            "question": "Python 的 GIL 是什么？它对多线程有什么影响？",
            "answer": (
                "GIL（全局解释器锁）是 CPython 中的一把互斥锁，"
                "它确保同一时刻只有一个线程执行 Python 字节码。"
                "这意味着在 CPU 密集型任务中，多线程并不能真正并行执行。"
                "但对于 I/O 密集型任务（如网络请求），GIL 会在等待 I/O 时释放，"
                "所以多线程仍然有效。如果需要真正的并行，可以用 multiprocessing 模块。"
            ),
            "reference": "GIL 是 CPython 的全局解释器锁，限制同一时刻只有一个线程执行字节码。",
        },
        {
            "question": "帮我推荐 3 个适合初学者的 Python 项目练手",
            "answer": (
                "1. 待办事项 CLI 工具：用 argparse 实现命令行增删改查，"
                "学习文件操作和命令行参数解析。\n"
                "2. 天气查询脚本：调用 OpenWeather API，学习 HTTP 请求和 JSON 解析。\n"
                "3. Markdown 转 HTML 转换器：用正则表达式实现基本的 Markdown 语法转换，"
                "学习字符串处理和正则。"
            ),
            "reference": "无",
        },
    ]

    summary = batch_evaluate(test_cases)
    print("\n--- 评估汇总 ---")
    for key, value in summary.items():
        if isinstance(value, float):
            print(f"  {key}: {value:.2f}")
        else:
            print(f"  {key}: {value}")

"""使用 Ragas 框架评估 RAG 系统。

安装依赖：pip install ragas datasets
"""

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


def build_eval_dataset() -> Dataset:
    """构建评估数据集。

    实际项目中，这些数据来自你的 RAG 系统运行日志。
    """
    data = {
        "question": [
            "公司的远程办公政策是什么？",
            "如何申请出差报销？",
            "新产品 X 的核心功能有哪些？",
        ],
        "answer": [
            # Agent 的实际回答
            "公司允许每周远程办公 2 天，需要直属领导审批。",
            "出差报销需要填写报销单，附上发票，提交给财务部，一般 5 个工作日到账。",
            "产品 X 支持实时协作、自动备份、以及 AI 辅助写作功能。",
        ],
        "contexts": [
            # RAG 检索到的上下文
            [
                "公司远程办公政策（2025 版）：员工每周可申请远程办公 2 天，"
                "需提前一天向直属领导提交申请并获得批准。",
                "公司考勤制度：迟到 30 分钟以上记为旷工半天。",  # 不相关的上下文
            ],
            [
                "差旅报销流程：员工出差结束后 7 个工作日内提交报销申请，"
                "需填写《差旅报销单》并附上所有原始发票。"
                "财务部审核通过后，款项将在 5 个工作日内打入工资卡。",
            ],
            [
                "产品 X 功能列表：1. 实时多人协作编辑；2. 自动云备份（每 5 分钟）；"
                "3. AI 辅助写作（支持摘要生成、续写、翻译）。",
                "产品 X 定价：基础版免费，专业版 99 元/月。",  # 部分相关
            ],
        ],
        "ground_truth": [
            # 标准答案（人工标注）
            "每周可远程办公 2 天，需直属领导提前审批。",
            "出差结束后 7 个工作日内提交报销单和发票，财务 5 个工作日处理。",
            "核心功能包括实时协作、自动备份、AI 辅助写作。",
        ],
    }
    return Dataset.from_dict(data)


def run_ragas_evaluation():
    """运行 Ragas 评估并输出结果。"""
    dataset = build_eval_dataset()

    # 执行评估
    results = evaluate(
        dataset=dataset,
        metrics=[
            faithfulness,
            answer_relevance,
            context_precision,
            context_recall,
        ],
    )

    # 打印总体得分
    print("=== Ragas 评估结果 ===\n")
    print(f"  Faithfulness:        {results['faithfulness']:.3f}")
    print(f"  Answer Relevance:    {results['answer_relevance']:.3f}")
    print(f"  Context Precision:   {results['context_precision']:.3f}")
    print(f"  Context Recall:      {results['context_recall']:.3f}")

    # 逐条查看详细结果
    print("\n=== 逐条明细 ===\n")
    df = results.to_pandas()
    for i, row in df.iterrows():
        print(f"Q{i+1}: {row['question'][:30]}...")
        print(f"  Faithfulness:      {row['faithfulness']:.2f}")
        print(f"  Answer Relevance:  {row['answer_relevance']:.2f}")
        print(f"  Context Precision: {row['context_precision']:.2f}")
        print(f"  Context Recall:    {row['context_recall']:.2f}")
        print()

    return results


if __name__ == "__main__":
    run_ragas_evaluation()

"""评估流水线集成示例——在 CI 中自动运行评估。

这个脚本可以在 GitHub Actions / GitLab CI 中运行，
将评估结果与基线对比，决定是否通过。
"""

import json
import sys
from pathlib import Path


def load_baseline() -> dict:
    """加载基线评估分数。"""
    baseline_path = Path("eval_baseline.json")
    if not baseline_path.exists():
        print("WARNING: 未找到基线文件，跳过对比")
        return {}
    return json.loads(baseline_path.read_text())


def run_evaluation_pipeline(
    test_cases_path: str = "eval_test_cases.json",
    baseline_path: str = "eval_baseline.json",
    threshold: float = 0.05,
) -> bool:
    """运行评估流水线并与基线对比。

    Args:
        test_cases_path: 测试用例文件路径
        baseline_path: 基线分数文件路径
        threshold: 允许的最大分数下降幅度（如 0.05 表示 5%）

    Returns:
        True 表示评估通过，False 表示未通过
    """
    # 这里调用你实际的评估函数
    # current_scores = run_your_agent_evaluation(test_cases_path)
    # 示例分数：
    current_scores = {
        "task_completion": 0.84,
        "tool_accuracy": 0.90,
        "faithfulness": 0.92,
        "avg_overall_llm_judge": 4.2,
    }

    baseline = load_baseline()

    if not baseline:
        # 首次运行，保存为基线
        Path(baseline_path).write_text(json.dumps(current_scores, indent=2))
        print("首次运行，已保存基线分数")
        return True

    # 对比基线
    passed = True
    print("\n=== 评估对比 ===")
    for metric, current in current_scores.items():
        base = baseline.get(metric, 0)
        diff = current - base
        diff_pct = (diff / base * 100) if base else 0
        status = "PASS" if diff >= -threshold else "FAIL"

        if status == "FAIL":
            passed = False

        print(
            f"  {metric}: {current:.3f} (baseline: {base:.3f}, "
            f"diff: {diff_pct:+.1f}%) [{status}]"
        )

    # 更新基线（如果通过）
    if passed:
        Path(baseline_path).write_text(json.dumps(current_scores, indent=2))
        print("\n评估通过，已更新基线分数")
    else:
        print(f"\n评估未通过！分数下降超过 {threshold*100:.0f}%，请检查改动")

    return passed


if __name__ == "__main__":
    success = run_evaluation_pipeline()
    sys.exit(0 if success else 1)