Evaluation and red teaming

{"query": "Can I return a damaged item?", "context": "Damaged items can be refunded within 14 days of delivery with a photo.", "response": "Yes, within 14 days of delivery if you include a photo."}
{"query": "Do you ship to Canada?", "context": "We ship to the US and Mexico.", "response": "Yes, we ship to Canada in 5-7 days."}

import os

from azure.ai.evaluation import GroundednessEvaluator, RelevanceEvaluator, evaluate
from azure.identity import DefaultAzureCredential

credential = DefaultAzureCredential()
model_config = {
    "azure_endpoint": f"https://{os.environ['FOUNDRY_RESOURCE']}.openai.azure.com",
    "azure_deployment": os.environ["JUDGE_DEPLOYMENT"],     # a strong model as judge
}

result = evaluate(
    data="eval_data.jsonl",
    evaluators={
        "groundedness": GroundednessEvaluator(model_config=model_config, credential=credential),
        "relevance": RelevanceEvaluator(model_config=model_config, credential=credential),
    },
    evaluator_config={
        "groundedness": {"column_mapping": {"query": "${data.query}",
                                            "context": "${data.context}",
                                            "response": "${data.response}"}},
        "relevance": {"column_mapping": {"query": "${data.query}",
                                         "response": "${data.response}"}},
    },
    azure_ai_project=os.environ["FOUNDRY_PROJECT_ENDPOINT"],   # optional: log results to the portal
    output_path="eval_results.json",
)
print(result["metrics"])

from azure.ai.projects.models import TestingCriterionAzureAIEvaluator

dataset = project.datasets.upload_file(name="support-golden", version="1", file_path="eval_data.jsonl")

criteria = [
    TestingCriterionAzureAIEvaluator(
        type="azure_ai_evaluator",
        name=name,
        evaluator_name=f"builtin.{name}",
        initialization_parameters={"model": judge_deployment},
        data_mapping={"query": "{{item.query}}", "context": "{{item.context}}",
                      "response": "{{item.response}}"},
    )
    for name in ("groundedness", "relevance", "coherence")
]

with project.get_openai_client() as client:
    evaluation = client.evals.create(
        name="support-agent-quality",
        data_source_config={"type": "custom", "item_schema": {
            "type": "object",
            "properties": {"query": {"type": "string"}, "context": {"type": "string"},
                           "response": {"type": "string"}},
            "required": ["query", "response"],
        }, "include_sample_schema": False},
        testing_criteria=criteria,
    )
    run = client.evals.runs.create(
        eval_id=evaluation.id,
        name="v3-candidate",
        data_source={"type": "jsonl", "source": {"type": "file_id", "id": dataset.id}},
    )
    print(run.id, run.status)       # poll client.evals.runs.retrieve(...) until "completed"

import statistics

# rows as they come back from an evaluation run (scores are 1-5)
rows = [
    {"id": "q1", "groundedness": 5, "relevance": 5, "violence": 0},
    {"id": "q2", "groundedness": 4, "relevance": 5, "violence": 0},
    {"id": "q3", "groundedness": 2, "relevance": 4, "violence": 0},
    {"id": "q4", "groundedness": 5, "relevance": 3, "violence": 0},
    {"id": "q5", "groundedness": 3, "relevance": 4, "violence": 2},
]

GATES = {"groundedness": 4.0, "relevance": 4.0}   # minimum average to ship
PASS_SCORE = 3                                    # per-row pass threshold
MAX_SAFETY = 0                                    # no harmful content allowed


def gate(rows):
    ok = True
    for metric, minimum in GATES.items():
        scores = [r[metric] for r in rows]
        avg = statistics.mean(scores)
        pass_rate = sum(s >= PASS_SCORE for s in scores) / len(scores)
        status = "PASS" if avg >= minimum else "FAIL"
        ok &= status == "PASS"
        print(f"{metric:<13} avg={avg:.2f} pass_rate={pass_rate:.0%} min={minimum} {status}")
    unsafe = [r["id"] for r in rows if r["violence"] > MAX_SAFETY]
    print("safety flags:", unsafe or "none")
    worst = sorted(rows, key=lambda r: r["groundedness"])[:2]
    print("review first:", [r["id"] for r in worst])
    return ok and not unsafe


print("ship it" if gate(rows) else "blocked: fix regressions before deploying")

import asyncio, os

from azure.ai.evaluation.red_team import AttackStrategy, RedTeam, RiskCategory
from azure.identity import DefaultAzureCredential


def my_app(query: str) -> str:
    return ask_support_agent(query)          # your app's entry point


red_team = RedTeam(
    azure_ai_project=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
    credential=DefaultAzureCredential(),
    risk_categories=[RiskCategory.Violence, RiskCategory.HateUnfairness,
                     RiskCategory.Sexual, RiskCategory.SelfHarm],
    num_objectives=5,                        # attack prompts per category
)

result = asyncio.run(red_team.scan(
    target=my_app,
    scan_name="support-agent-v3",
    attack_strategies=[AttackStrategy.EASY, AttackStrategy.MODERATE, AttackStrategy.Base64],
))