AI Agent量化评估：LLM-as-Judge、人工标注、A/B测试全解，智能体多维评测实践21.0

import os
import json
from typing import List, Dict
from openai import OpenAI

# 接入腾讯混元大模型（TokenHub平台）
api_key = os.environ.get('TENCENT_API_KEY')
llm_client = OpenAI(
    api_key=api_key,
    base_url="https://tokenhub.tencentmaas.com/v1",
)
model_name = "hy3-preview"

def llm_invoke(prompt: str) -> str:
    """调用腾讯混元大模型进行LLM-as-Judge评测"""
    response = llm_client.chat.completions.create(
        model=model_name,
        messages=[
            {"role": "system", "content": "你是一个专业的AI Agent评测裁判。请严格按照评测规则输出JSON格式结果，不要输出任何多余内容。"},
            {"role": "user", "content": prompt},
        ],
        temperature=0.1,
    )
    return response.choices[0].message.content

# 基础单样本评测
def single_judge_evaluate(user_query: str, agent_process: str, agent_result: str) -> Dict:
    """单条样本LLM裁判评测"""
    prompt = f"""
    你是专业的AI Agent评测裁判，严格依据Agent评测四维标准打分：
    评测维度：功能准确性、逻辑连贯性、内容真实性、场景适配性
    满分100，85分以上合格，60-85分待优化，60分以下不合格

    用户指令：{user_query}
    Agent执行链路：{agent_process}
    Agent最终输出：{agent_result}
    请输出JSON结果：score、grade、reason
    """
    res = llm_invoke(prompt)
    return json.loads(res)

# 偏差修正：位置交换复测
def cross_evaluate(sample1: Dict, sample2: Dict) -> List[Dict]:
    """交叉评测，抵消顺序偏差"""
    res1 = single_judge_evaluate(**sample1)
    res2 = single_judge_evaluate(**sample2)
    # 交换样本再次评测
    res1_swap = single_judge_evaluate(**sample2)
    res2_swap = single_judge_evaluate(**sample1)
    # 取两次平均分
    final1_score = (res1["score"] + res2_swap["score"]) / 2
    final2_score = (res2["score"] + res1_swap["score"]) / 2
    res1["score"], res2["score"] = final1_score, final2_score
    return [res1, res2]

# 批量离线评测
def batch_evaluate(test_samples: List[Dict]) -> List[Dict]:
    """批量评测样本，输出完整评测结果"""
    batch_result = []
    for sample in test_samples:
        judge_res = single_judge_evaluate(
            user_query=sample["user_query"],
            agent_process=sample["agent_process"],
            agent_result=sample["agent_result"]
        )
        batch_result.append({**sample, **judge_res})
    return batch_result

# 评测结果聚合统计
def aggregate_evaluate_result(batch_res: List[Dict]) -> Dict:
    """批量结果聚合，统计平均分、合格率、缺陷分布"""
    total_score = [item["score"] for item in batch_res]
    pass_count = len([s for s in total_score if s >= 85])
    return {
        "sample_count": len(batch_res),
        "avg_score": round(sum(total_score)/len(total_score), 2),
        "pass_rate": round(pass_count/len(batch_res), 2),
        "defect_top": "细节冗余、轻微幻觉"
    }

# 测试运行
if __name__ == "__main__":
    # 模拟业务测试样本
    test_case = [
        {
            "user_query": "统计本周办公文件整理数量并生成报表",
            "agent_process": "解析需求->调用文件查询工具->统计数据->生成报表",
            "agent_result": "本周共整理办公文件28份，分类明细完整，报表生成完成"
        }
    ]
    # 批量评测
    batch_result = batch_evaluate(test_case)
    # 聚合分析
    stat_result = aggregate_evaluate_result(batch_result)
    print("单样本评测结果：", batch_result)
    print("批量统计结果：", stat_result)

import os
import json
from typing import List, Dict
from openai import OpenAI

# 接入腾讯混元大模型（TokenHub平台）
api_key = os.environ.get('TENCENT_API_KEY')
llm_client = OpenAI(
    api_key=api_key,
    base_url="https://tokenhub.tencentmaas.com/v1",
)
model_name = "hy3-preview"

def llm_judge(user_query: str, agent_process: str, agent_result: str) -> Dict:
    """调用腾讯混元大模型进行LLM-as-Judge评测"""
    prompt = f"""
    你是专业的AI Agent评测裁判，严格依据Agent评测四维标准打分：
    评测维度：功能准确性、逻辑连贯性、内容真实性、场景适配性
    满分100，85分以上合格，60-85分待优化，60分以下不合格

    用户指令：{user_query}
    Agent执行链路：{agent_process}
    Agent最终输出：{agent_result}
    请仅输出JSON格式：{{"score": 总分, "grade": "A/B/C/D", "reason": "扣分原因"}}
    """
    response = llm_client.chat.completions.create(
        model=model_name,
        messages=[
            {"role": "system", "content": "你是一个专业的AI Agent评测裁判。请严格按照评测规则输出JSON格式结果，不要输出任何多余内容。"},
            {"role": "user", "content": prompt},
        ],
        temperature=0.1,
    )
    content = response.choices[0].message.content
    # 清理可能的Markdown代码块包裹
    if content.startswith("```"):
        content = content.strip("```json").strip("```").strip()
    return json.loads(content)

# 模拟人工标注标准答案数据集
def build_manual_benchmark() -> List[Dict]:
    """构建人工标注基准数据集"""
    return [
        {
            "sample_id": "case_001",
            "user_query": "统计本周办公文件整理数量并生成报表",
            "standard_score": 95,
            "standard_reason": "需求完成完整、数据真实、无幻觉",
            "is_border_case": False
        },
        {
            "sample_id": "case_002",
            "user_query": "排查系统今日异常报错并给出解决方案",
            "standard_score": 88,
            "standard_reason": "异常排查完整，解决方案可行，缺少细节补充",
            "is_border_case": True
        }
    ]

# 机器与人工结果偏差比对
def judge_manual_compare(judge_batch_res: List[Dict], manual_data: List[Dict]) -> List[Dict]:
    """自动化评测结果与人工标注比对，统计偏差"""
    compare_res = []
    manual_dict = {item["user_query"]: item for item in manual_data}
    for judge_item in judge_batch_res:
        query = judge_item["user_query"]
        if query not in manual_dict:
            continue
        manual_item = manual_dict[query]
        # 计算分数偏差
        score_diff = abs(judge_item["score"] - manual_item["standard_score"])
        compare_res.append({
            "user_query": query,
            "machine_score": judge_item["score"],
            "manual_score": manual_item["standard_score"],
            "score_diff": score_diff,
            "is_error": score_diff > 10,  # 偏差超10分判定为机器误判
            "is_border_case": manual_item["is_border_case"]
        })
    return compare_res

# 筛选需人工复核样本（低分+高偏差+边界场景）
def screen_review_samples(compare_res: List[Dict]) -> List[Dict]:
    """筛选重点复核样本，实现轻量化标注"""
    review_samples = [
        item for item in compare_res
        if item["score_diff"] > 8 or item["is_border_case"] or item["machine_score"] < 80
    ]
    return review_samples

# 沉淀标准化评测数据集
def save_benchmark_dataset(compare_res: List[Dict], path: str = "agent_benchmark.json"):
    """更新并保存基准数据集"""
    with open(path, "w", encoding="utf-8") as f:
        json.dump(compare_res, f, ensure_ascii=False, indent=2)

# 测试运行
if __name__ == "__main__":
    # 1. 加载人工标注数据
    manual_data = build_manual_benchmark()
    # 2. 调用混元大模型进行机器评测
    judge_res = []
    test_samples = [
        {
            "user_query": "统计本周办公文件整理数量并生成报表",
            "agent_process": "解析需求->调用文件查询工具->统计数据->生成报表",
            "agent_result": "本周共整理办公文件28份，分类明细完整，报表生成完成"
        },
        {
            "user_query": "排查系统今日异常报错并给出解决方案",
            "agent_process": "需求解析->调用日志检索工具->异常归类分析->生成方案",
            "agent_result": "发现3条异常报错，已定位根因并给出修复建议，方案可行"
        }
    ]
    for sample in test_samples:
        res = llm_judge(
            user_query=sample["user_query"],
            agent_process=sample["agent_process"],
            agent_result=sample["agent_result"]
        )
        judge_res.append({"user_query": sample["user_query"], **res})
        print(f"[{sample['user_query']}] 机器评测结果: score={res['score']}, grade={res['grade']}")
    # 3. 偏差比对
    compare_result = judge_manual_compare(judge_res, manual_data)
    # 4. 筛选复核样本
    review_list = screen_review_samples(compare_result)
    # 5. 沉淀数据集
    save_benchmark_dataset(compare_result)
    print("偏差比对结果：", compare_result)
    print("待人工复核样本：", review_list)

import os
import random
import numpy as np
from scipy import stats
from typing import Dict, List
from openai import OpenAI

# 接入腾讯混元大模型（TokenHub平台）
api_key = os.environ.get('TENCENT_API_KEY')
llm_client = OpenAI(
    api_key=api_key,
    base_url="https://tokenhub.tencentmaas.com/v1",
)
model_name = "hy3-preview"

# 模拟线上流量拆分与数据采集
def ab_test_flow_split(total_count: int) -> tuple[List[float], List[float]]:
    """
    随机均匀拆分流量，模拟对照组（旧版本）、实验组（新版本）指标数据
    指标：任务完成成功率（0-1）
    """
    # 对照组：旧版本Agent成功率分布
    control_group = [round(random.uniform(0.82, 0.92), 4) for _ in range(total_count)]
    # 实验组：新版本Agent成功率分布（优化后略高）
    test_group = [round(random.uniform(0.86, 0.96), 4) for _ in range(total_count)]
    return control_group, test_group

# 核心指标统计
def calc_ab_metrics(control_data: List[float], test_data: List[float]) -> Dict:
    """计算两组核心指标：均值、提升率、标准差"""
    ctrl_mean = np.mean(control_data)
    test_mean = np.mean(test_data)
    improve_rate = round((test_mean - ctrl_mean) / ctrl_mean * 100, 2)
    return {
        "control_avg_success": round(ctrl_mean, 4),
        "test_avg_success": round(test_mean, 4),
        "improve_rate": f"{improve_rate}%",
        "control_std": round(np.std(control_data), 4),
        "test_std": round(np.std(test_data), 4)
    }

# 统计学显著性检验（T检验）
def significance_test(control_data: List[float], test_data: List[float], alpha: float = 0.05) -> Dict:
    """
    独立样本T检验，判断版本差异是否显著
    p < alpha 代表差异显著，迭代有效
    """
    t_stat, p_value = stats.ttest_ind(test_data, control_data)
    return {
        "t_statistic": round(t_stat, 4),
        "p_value": round(p_value, 4),
        "alpha_threshold": alpha,
        "is_significant": p_value < alpha
    }

# A/B测试最终决策（混元大模型分析）
def ab_test_decision(metrics: Dict, sig_res: Dict) -> str:
    """调用腾讯混元大模型，基于统计指标输出版本迭代分析与决策建议"""
    prompt = f"""
    你是一个专业的A/B测试数据分析专家。请根据以下统计结果，输出版本迭代决策建议：

    【核心指标】
    - 对照组平均成功率：{metrics['control_avg_success']}
    - 实验组平均成功率：{metrics['test_avg_success']}
    - 提升率：{metrics['improve_rate']}
    - 对照组标准差：{metrics['control_std']}
    - 实验组标准差：{metrics['test_std']}

    【显著性检验】
    - T统计量：{sig_res['t_statistic']}
    - P值：{sig_res['p_value']}
    - 显著性阈值(α)：{sig_res['alpha_threshold']}
    - 是否显著：{sig_res['is_significant']}

    请从以下维度分析并给出决策建议（控制在200字以内）：
    1. 统计显著性判断
    2. 业务效果评估（提升幅度是否具备实际价值）
    3. 最终决策建议（全量上线/灰度扩量/驳回重优化）

    仅输出分析结论文本，不要JSON格式。
    """
    response = llm_client.chat.completions.create(
        model=model_name,
        messages=[
            {"role": "system", "content": "你是一个专业的A/B测试数据分析专家，请给出简洁、有依据的版本迭代决策建议。"},
            {"role": "user", "content": prompt},
        ],
        temperature=0.3,
    )
    return response.choices[0].message.content.strip()

# 测试运行
if __name__ == "__main__":
    # 模拟1000条线上业务流量
    control_data, test_data = ab_test_flow_split(1000)
    # 指标计算
    ab_metrics = calc_ab_metrics(control_data, test_data)
    # 显著性检验
    sig_result = significance_test(control_data, test_data)
    # 最终决策
    final_decision = ab_test_decision(ab_metrics, sig_result)

    print("A/B测试核心指标：", ab_metrics)
    print("统计学检验结果：", sig_result)
    print("大模型版本迭代决策：", final_decision)