AI Agent 评估体系：LLM-as-Judge 与回归测试集设计

def deterministic_checks(agent_output, expected):
    results = []
    
    # 1. Schema 校验（如果有结构化输出）
    if expected.get("required_schema"):
        try:
            validate(agent_output, expected["required_schema"])
            results.append({"check": "schema_valid", "pass": True})
        except ValidationError as e:
            results.append({"check": "schema_valid", "pass": False, "error": str(e)})
    
    # 2. 工具调用检查
    if expected.get("required_tools"):
        called_tools = {call.name for call in agent_output.tool_calls}
        missing = set(expected["required_tools"]) - called_tools
        results.append({
            "check": "required_tools_called",
            "pass": len(missing) == 0,
            "missing": list(missing)
        })
    
    # 3. 业务规则
    if expected.get("business_rules"):
        for rule in expected["business_rules"]:
            passed = eval(rule, {"output": agent_output})
            results.append({"check": rule, "pass": passed})
    
    # 4. PII / 安全检查
    if contains_pii(agent_output.text):
        results.append({"check": "no_pii", "pass": False})
    
    return results

JUDGE_PROMPT = """你是一个严格的评估员。请根据以下标准评估 AI Agent 的输出。

[用户问题]
{question}

[Agent 输出]
{output}

[参考答案（如有）]
{reference}

评估标准（每个 0-10 分）：
1. **准确性**：事实是否正确？是否有幻觉？
2. **相关性**：是否切题？是否解决了用户问题？
3. **完整性**：是否遗漏关键信息？
4. **清晰度**：表达是否流畅、结构化？

请按 JSON 格式输出：
{{"accuracy": 8, "relevance": 9, "completeness": 7, "clarity": 8, "reasoning": "..."}}
"""

def llm_judge(question, output, reference=None) -> dict:
    response = openai.ChatCompletion.create(
        model="gpt-4o",  # 用强模型做 judge
        messages=[{
            "role": "user",
            "content": JUDGE_PROMPT.format(
                question=question,
                output=output,
                reference=reference or "（无）"
            )
        }],
        response_format={"type": "json_object"}
    )
    return json.loads(response.choices[0].message.content)

class HumanEval:
    def __init__(self):
        self.queue = []  # 待评估样本
    
    def submit(self, question, agent_output_a, agent_output_b):
        """A/B 测试：让人类选哪个更好"""
        self.queue.append({
            "id": uuid4(),
            "question": question,
            "output_a": agent_output_a,
            "output_b": agent_output_b,
            "status": "pending"
        })
    
    def get_sample(self, n=10):
        """抽样返回待评估项"""
        return random.sample([q for q in self.queue if q["status"] == "pending"], n)
    
    def record_result(self, eval_id, preferred: str, reason: str):
        for q in self.queue:
            if q["id"] == eval_id:
                q["status"] = "completed"
                q["preferred"] = preferred
                q["reason"] = reason
                break

golden_dataset_entry = {
    # 基本标识
    "id": "test-001",
    "category": "refund_request",  # 分类标签
    "difficulty": "easy",  # easy / medium / hard
    "tags": ["customer_service", "edge_case"],
    
    # 输入
    "input": {
        "user_message": "我要退款",
        "context": {  # 当时的 context（订单信息等）
            "user_id": "u-123",
            "recent_orders": ["ORD-001", "ORD-002"]
        }
    },
    
    # 期望输出（多维度）
    "expected": {
        # 确定性检查用
        "required_tools": ["get_orders", "process_refund"],
        "must_not_include": ["邮箱地址", "内部订单链接"],
        "required_fields": ["refund_id", "amount"],
        
        # LLM-as-Judge 用
        "golden_answer": "应该在 24 小时内退款到原支付渠道...",
        "key_points": [
            "确认订单属于本人",
            "解释退款流程",
            "给出明确时间预期"
        ],
        
        # 参考答案（可选）
        "reference_solution": "完整理想回答示例..."
    },
    
    # 元信息
    "created_at": "2025-06-01",
    "created_by": "human-review-1",
    "last_updated": "2025-06-10",
    "notes": "边界 case：用户没说具体订单，需要先询问"
}

# 1. 从生产事故扩充
incident_samples = extract_incidents(incident_logs, n=10)
golden_dataset.extend(incident_samples)

# 2. 从用户反馈扩充
negative_feedback = fetch_feedback(thumbs_down=True)
golden_dataset.extend(negative_feedback[:5])

# 3. 定期 review
def review_dataset():
    for entry in golden_dataset:
        # 标记过时：业务规则变了、标准答案过期了
        if entry.last_updated < six_months_ago:
            entry.needs_review = True

rag_metrics = {
    # 检索阶段
    "context_precision": "检索的 context 中相关片段占比",
    "context_recall": "所有相关片段被检索到的比例",
    
    # 生成阶段
    "faithfulness": "回答内容是否忠于 context（无幻觉）",
    "answer_relevancy": "回答是否切题",
    
    # 鲁棒性
    "noise_sensitivity": "context 含干扰信息时是否还能正确回答",
    "context_relevance": "context 中的噪声是否影响最终回答质量"
}

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

result = evaluate(
    dataset,
    metrics=[context_precision, context_recall, faithfulness, answer_relevancy]
)
print(result)
# {'context_precision': 0.85, 'context_recall': 0.78, ...}

agent_metrics = {
    # 工具选择
    "tool_selection_accuracy": "在每个决策点选对了工具",
    "tool_call_efficiency": "用最少的工具调用完成任务",
    
    # 参数生成
    "argument_correctness": "工具参数符合 schema + 业务正确",
    
    # 路径效率
    "steps_to_completion": "完成任务用了多少步",
    "redundant_calls": "重复调用同一工具的次数",
    
    # 鲁棒性
    "error_recovery_rate": "工具报错后能否恢复",
    "loop_detection": "是否陷入死循环"
}

from deepeval import evaluate
from deepeval.metrics import (
    ToolCorrectnessMetric,
    TaskCompletionMetric,
    PlanQualityMetric
)
from deepeval.test_case import LLMTestCase, ToolCall

test_case = LLMTestCase(
    input="查询订单 ORD-001 的状态",
    actual_output="订单已发货",
    tools_called=[
        ToolCall(name="get_order", arguments={"order_id": "ORD-001"})
    ],
    expected_tools=[
        ToolCall(name="get_order", arguments={"order_id": "ORD-001"})
    ]
)

# ToolCorrectnessMetric 检查工具调用是否正确
metric = ToolCorrectnessMetric(threshold=0.7)
metric.measure(test_case)
print(f"Score: {metric.score}, Reason: {metric.reason}")

conversation_metrics = {
    # 角色一致性
    "role_adherence": "整个对话保持角色设定",
    
    # 知识保留
    "knowledge_retention": "记得之前的对话信息",
    "context_consistency": "前后回答不矛盾",
    
    # 用户体验
    "conversational_coherence": "对话自然流畅",
    "user_satisfaction_signal": "用户是否最终得到解决"
}

# .github/workflows/agent-eval.yml
name: Agent Evaluation

on:
  pull_request:
    paths:
      - 'prompts/**'
      - 'agents/**'
      - 'tools/**'

jobs:
  eval:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      
      - name: Install dependencies
        run: pip install -r requirements.txt
      
      - name: Run agent evaluation
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
        run: |
          python -m pytest tests/eval/ -v --tb=short
      
      - name: Check quality gates
        run: |
          python scripts/check_eval_gates.py \
            --min-faithfulness 0.8 \
            --min-answer-relevancy 0.85 \
            --max-regression 0.05

QUALITY_GATES = {
    # 单条样本最低分
    "min_faithfulness_score": 0.80,
    "min_answer_relevancy": 0.85,
    "min_tool_accuracy": 0.90,
    
    # 整体回归控制
    "max_score_regression": 0.05,  # 比 main 分低 5% 就失败
    "min_pass_rate": 0.90,  # 至少 90% 样本通过
}

def check_gates(current_results, baseline_results):
    failures = []
    
    for metric, threshold in QUALITY_GATES.items():
        if "regression" in metric:
            regression = baseline_results[metric] - current_results[metric]
            if regression > QUALITY_GATES[metric]:
                failures.append(f"{metric}: regression {regression:.2%}")
        else:
            if current_results[metric] < threshold:
                failures.append(f"{metric}: {current_results[metric]:.2%} < {threshold:.2%}")
    
    if failures:
        raise QualityGateFailed(failures)

from scipy import stats

def is_significant(a_scores, b_scores, alpha=0.05):
    """配对 t 检验：每个样本跑 A 和 B，对比分数差"""
    
    # 数据对齐
    if len(a_scores) != len(b_scores):
        raise ValueError("需要配对数据")
    
    # t 检验
    t_stat, p_value = stats.ttest_rel(a_scores, b_scores)
    
    return {
        "mean_a": np.mean(a_scores),
        "mean_b": np.mean(b_scores),
        "diff": np.mean(b_scores) - np.mean(a_scores),
        "p_value": p_value,
        "significant": p_value < alpha,
        "winner": "B" if np.mean(b_scores) > np.mean(a_scores) else "A"
    }

# 示例：50 个样本上 A 模型平均 0.78，B 模型 0.82
result = is_significant(a_scores, b_scores)
print(f"P-value: {result['p_value']:.4f}, Significant: {result['significant']}")
# P-value: 0.034, Significant: True

from phoenix.trace import trace

@trace
def run_agent(user_input):
    thought_1 = llm.complete(f"分析用户意图：{user_input}")
    # ...
    # 每个 step 自动被追踪

class ProductionMonitor:
    def __init__(self):
        self.eval_sample_rate = 0.05  # 5% 流量进评估
        self.daily_budget = 1000  # 每天最多评估 1000 条
    
    async def on_request_complete(self, request, response):
        # 1. 抽样
        if random.random() > self.eval_sample_rate:
            return
        
        # 2. 评估
        score = await llm_judge(request.input, response.output)
        
        # 3. 异常信号
        if score["accuracy"] < 0.5:
            alert_low_quality(request, response, score)
        
        if response.tool_calls and any(call.error for call in response.tool_calls):
            alert_tool_error(request, response)
        
        # 4. 入库分析
        store_eval_result(request, response, score)