Official docs: Custom evaluators
Built-in evaluators cover common quality and safety metrics. But real-world applications need domain-specific evaluation — brand voice compliance, JSON schema validation, disclaimer enforcement, or business-rule checks. This lesson shows you how to build custom evaluators, use the LLM-as-judge pattern, and stress-test your app with adversarial simulation.
A code-based evaluator is any callable that accepts evaluation inputs and returns a score dictionary. No LLM needed — just deterministic Python logic.
"""Evaluator that checks response length stays within bounds."""
class WordCountEvaluator:
"""Flag responses that are too short or too long."""
def __init__(self, min_words: int = 10, max_words: int = 300):
self.min_words = min_words
self.max_words = max_words
def __call__(self, *, response: str, **kwargs) -> dict:
word_count = len(response.split())
in_range = self.min_words <= word_count <= self.max_words
return {
"word_count": word_count,
"word_count_in_range": 1.0 if in_range else 0.0,
}
# Usage
evaluator = WordCountEvaluator(min_words=20, max_words=200)
result = evaluator(response="This is a short response.")
print(result)
# {'word_count': 6, 'word_count_in_range': 0.0}
"""Evaluator that checks if the response is valid JSON matching a schema."""
import json
class JsonValidityEvaluator:
"""Validate that the response is parseable JSON with required keys."""
def __init__(self, required_keys: list[str] | None = None):
self.required_keys = required_keys or []
def __call__(self, *, response: str, **kwargs) -> dict:
try:
parsed = json.loads(response)
except (json.JSONDecodeError, TypeError):
return {
"json_valid": 0.0,
"json_has_required_keys": 0.0,
"json_error": "Invalid JSON",
}
if not isinstance(parsed, dict):
return {
"json_valid": 1.0,
"json_has_required_keys": 0.0,
"json_error": "JSON is not an object",
}
missing = [k for k in self.required_keys if k not in parsed]
return {
"json_valid": 1.0,
"json_has_required_keys": 1.0 if not missing else 0.0,
"json_missing_keys": missing,
}
# Usage
evaluator = JsonValidityEvaluator(required_keys=["answer", "sources"])
result = evaluator(response='{"answer": "Hello", "sources": ["doc1.pdf"]}')
print(result)
# {'json_valid': 1.0, 'json_has_required_keys': 1.0, 'json_missing_keys': []}
"""Evaluator that checks if the response includes a required disclaimer."""
class DisclaimerEvaluator:
"""Verify that responses about regulated topics include a disclaimer."""
DISCLAIMER_PHRASES = [
"not a substitute for professional",
"consult a qualified",
"this is not legal advice",
"this is not medical advice",
"for informational purposes only",
]
def __call__(self, *, response: str, **kwargs) -> dict:
response_lower = response.lower()
has_disclaimer = any(
phrase in response_lower for phrase in self.DISCLAIMER_PHRASES
)
return {"has_disclaimer": 1.0 if has_disclaimer else 0.0}
# Usage
evaluator = DisclaimerEvaluator()
result = evaluator(
response="You should take ibuprofen. This is not medical advice — consult a qualified professional."
)
print(result)
# {'has_disclaimer': 1.0}
Tip: Code-based evaluators run instantly with no API calls. Use them for structural checks, format validation, and business rules. Reserve LLM-based evaluators for subjective quality judgments.
For subjective criteria that require language understanding, use an LLM as the judge. This pattern sends a scoring prompt to GPT-4o and parses the result.
"""LLM-as-judge evaluator for brand voice compliance."""
from openai import AzureOpenAI
BRAND_VOICE_PROMPT = """\
You are an expert brand voice auditor. Evaluate whether the following response
adheres to our brand guidelines:
Brand guidelines:
- Tone: Professional yet approachable
- Avoid jargon unless the user used it first
- Always address the user directly ("you")
- Never use passive aggressive language
- Include actionable next steps when possible
Query: {query}
Response: {response}
Score the response on a scale of 1-5:
1 = Major brand voice violations
2 = Several deviations from brand voice
3 = Acceptable but room for improvement
4 = Good adherence with minor suggestions
5 = Perfect brand voice alignment
Return your evaluation as JSON:
score
"""
class BrandVoiceEvaluator:
"""Evaluate response adherence to brand voice guidelines using an LLM judge."""
def __init__(self, model_config: dict):
self.client = AzureOpenAI(
azure_endpoint=model_config["azure_endpoint"],
api_version=model_config["api_version"],
)
self.deployment = model_config["azure_deployment"]
def __call__(self, *, query: str, response: str, **kwargs) -> dict:
import json
prompt = BRAND_VOICE_PROMPT.format(query=query, response=response)
completion = self.client.chat.completions.create(
model=self.deployment,
messages=[{"role": "user", "content": prompt}],
temperature=0.0,
response_format={"type": "json_object"},
)
try:
result = json.loads(completion.choices[0].message.content)
return {
"brand_voice_score": float(result["score"]),
"brand_voice_reason": result.get("reason", ""),
"brand_voice_suggestions": result.get("suggestions", []),
}
except (json.JSONDecodeError, KeyError, TypeError):
return {
"brand_voice_score": 0.0,
"brand_voice_reason": "Failed to parse LLM judge response",
"brand_voice_suggestions": [],
}
# Usage
model_config = {
"azure_deployment": "gpt-4o",
"azure_endpoint": "https://<your-endpoint>.openai.azure.com/",
"api_version": "2024-12-01-preview",
}
evaluator = BrandVoiceEvaluator(model_config=model_config)
result = evaluator(
query="How do I reset my password?",
response=(
"You can reset your password by going to Settings > Security > "
"Change Password. Enter your current password, then type your "
"new password twice. Click Save to confirm the change."
),
)
print(f"Brand voice score: {result['brand_voice_score']}")
print(f"Reason: {result['brand_voice_reason']}")
Tip: Use
temperature=0.0andresponse_format={"type": "json_object"}for consistent, parseable judge responses. Always include a fallback for parsing failures.
Combine multiple evaluators into a single composite score for overall quality assessment.
"""Composite evaluator that combines multiple evaluators with weighted scores."""
from azure.ai.evaluation import (
GroundednessEvaluator,
RelevanceEvaluator,
CoherenceEvaluator,
)
class CompositeEvaluator:
"""Aggregate multiple evaluator scores into a single weighted score."""
def __init__(self, evaluators: dict, weights: dict):
"""
Args:
evaluators: {"name": evaluator_instance}
weights: {"name": weight} — weights should sum to 1.0
"""
self.evaluators = evaluators
self.weights = weights
total = sum(weights.values())
if abs(total - 1.0) > 0.01:
raise ValueError(f"Weights must sum to 1.0, got {total}")
def __call__(self, **kwargs) -> dict:
all_results = {}
weighted_sum = 0.0
for name, evaluator in self.evaluators.items():
result = evaluator(**kwargs)
all_results[f"component_{name}"] = result
# Extract the primary score (first numeric value)
score = next(
(v for v in result.values() if isinstance(v, (int, float))), 0
)
# Normalize: quality scores are 1–5, normalize to 0–1
normalized = (score - 1) / 4 if score > 1 else score
weighted_sum += normalized * self.weights.get(name, 0)
all_results["composite_score"] = round(weighted_sum, 3)
all_results["composite_pass"] = weighted_sum >= 0.7 # 70% threshold
return all_results
# Usage
composite = CompositeEvaluator(
evaluators={
"groundedness": GroundednessEvaluator(model_config=model_config),
"relevance": RelevanceEvaluator(model_config=model_config),
"coherence": CoherenceEvaluator(model_config=model_config),
"brand_voice": BrandVoiceEvaluator(model_config=model_config),
"word_count": WordCountEvaluator(min_words=20, max_words=200),
},
weights={
"groundedness": 0.30,
"relevance": 0.25,
"coherence": 0.15,
"brand_voice": 0.20,
"word_count": 0.10,
},
)
result = composite(
query="What is Azure AI Foundry?",
context="Azure AI Foundry is a unified platform for building AI applications.",
response=(
"Azure AI Foundry is a platform that helps you build, evaluate, and "
"deploy AI applications. You can access models from the catalog, run "
"evaluations, and monitor your apps in production."
),
)
print(f"Composite score: {result['composite_score']}")
print(f"Pass: {result['composite_pass']}")
Use the AdversarialSimulator to stress-test your application with adversarial inputs — prompt injections, jailbreak attempts, and harmful content requests.
Official docs: Adversarial simulation
"""Run adversarial simulation to test your app's safety boundaries."""
from azure.ai.evaluation.simulator import AdversarialSimulator, AdversarialScenario
from azure.identity import DefaultAzureCredential
credential = DefaultAzureCredential()
# Your Azure AI project connection string
azure_ai_project = {
"subscription_id": "<your-subscription-id>",
"resource_group_name": "<your-resource-group>",
"project_name": "<your-project-name>",
}
async def my_application(messages: dict) -> dict:
"""Your application callback — the simulator sends adversarial inputs here."""
from openai import AzureOpenAI
client = AzureOpenAI(
azure_endpoint="https://<your-endpoint>.openai.azure.com/",
api_version="2024-12-01-preview",
)
response = client.chat.completions.create(
model="gpt-4o",
messages=messages["messages"],
temperature=0.0,
)
return {
"messages": messages["messages"]
+ [
{
"role": "assistant",
"content": response.choices[0].message.content,
}
]
}
# Initialize the simulator
simulator = AdversarialSimulator(
azure_ai_project=azure_ai_project, credential=credential
)
async def run_simulation():
"""Run adversarial simulation across multiple attack categories."""
results = await simulator(
target=my_application,
scenario=AdversarialScenario.ADVERSARIAL_QA,
max_simulation_results=50,
max_conversation_turns=3,
)
print(f"Simulated {len(results)} adversarial conversations")
# Save results for analysis
with open("adversarial_results.jsonl", "w") as f:
for conversation in results:
import json
f.write(json.dumps(conversation) + "\n")
return results
# Run it
import asyncio
results = asyncio.run(run_simulation())
| Scenario | What It Tests |
|---|---|
ADVERSARIAL_QA |
General adversarial Q&A attacks |
ADVERSARIAL_CONVERSATION |
Multi-turn manipulation attempts |
ADVERSARIAL_SUMMARIZATION |
Adversarial summarization attacks |
ADVERSARIAL_REWRITE |
Attempts to make the model rewrite harmful content |
ADVERSARIAL_CONTENT_GEN_UNGROUNDED |
Ungrounded content generation |
ADVERSARIAL_INDIRECT_JAILBREAK (XPIA) |
Prompt injection via retrieved context |
Tip: Always run
ADVERSARIAL_INDIRECT_JAILBREAKfor RAG applications. It simulates prompt injection attacks hidden in your retrieval context — a common real-world attack vector.
Combine adversarial simulation with safety evaluators for a complete red-teaming workflow.
"""End-to-end red-teaming: simulate attacks, then evaluate safety."""
from azure.ai.evaluation import evaluate
from azure.ai.evaluation import (
ViolenceEvaluator,
SexualEvaluator,
SelfHarmEvaluator,
HateUnfairnessEvaluator,
IndirectAttackEvaluator,
)
def red_team_evaluate(adversarial_results_path: str, model_config: dict) -> dict:
"""Evaluate adversarial simulation results with safety evaluators."""
result = evaluate(
data=adversarial_results_path,
evaluators={
"violence": ViolenceEvaluator(model_config=model_config),
"sexual": SexualEvaluator(model_config=model_config),
"self_harm": SelfHarmEvaluator(model_config=model_config),
"hate": HateUnfairnessEvaluator(model_config=model_config),
"indirect_attack": IndirectAttackEvaluator(model_config=model_config),
},
evaluator_config={
"violence": {
"query": "${data.query}",
"response": "${data.response}",
},
"sexual": {
"query": "${data.query}",
"response": "${data.response}",
},
"self_harm": {
"query": "${data.query}",
"response": "${data.response}",
},
"hate": {
"query": "${data.query}",
"response": "${data.response}",
},
"indirect_attack": {
"query": "${data.query}",
"context": "${data.context}",
"response": "${data.response}",
},
},
)
# Report
metrics = result["metrics"]
print("=== Red Team Safety Report ===")
for key, value in sorted(metrics.items()):
print(f" {key}: {value}")
# Flag any high-severity issues
flagged_rows = []
for i, row in enumerate(result["rows"]):
for key, val in row.items():
if "score" in key and isinstance(val, (int, float)) and val >= 4:
flagged_rows.append((i, key, val))
if flagged_rows:
print(f"\n⚠️ {len(flagged_rows)} high-severity issues found:")
for row_idx, metric, severity in flagged_rows:
print(f" Row {row_idx}: {metric} = {severity}")
else:
print("\n✅ No high-severity safety issues detected.")
return result
# Run
report = red_team_evaluate("adversarial_results.jsonl", model_config)
Integrate custom evaluators into your CI/CD pipeline to block deployments that don’t meet standards.
"""CI/CD quality gate combining built-in and custom evaluators."""
import sys
from azure.ai.evaluation import evaluate, GroundednessEvaluator, RelevanceEvaluator
# Thresholds
GATES = {
"groundedness.groundedness": {"min": 4.0},
"relevance.relevance": {"min": 4.0},
"brand_voice.brand_voice_score": {"min": 3.5},
"word_count.word_count_in_range": {"min": 0.9}, # 90% of responses in range
"json_valid.json_valid": {"min": 1.0}, # 100% valid JSON (if applicable)
}
def run_ci_evaluation(data_path: str, model_config: dict) -> bool:
"""Run evaluation and check quality gates. Returns True if all gates pass."""
result = evaluate(
data=data_path,
evaluators={
"groundedness": GroundednessEvaluator(model_config=model_config),
"relevance": RelevanceEvaluator(model_config=model_config),
"brand_voice": BrandVoiceEvaluator(model_config=model_config),
"word_count": WordCountEvaluator(min_words=20, max_words=200),
},
evaluator_config={
"groundedness": {
"query": "${data.query}",
"context": "${data.context}",
"response": "${data.response}",
},
"relevance": {
"query": "${data.query}",
"context": "${data.context}",
"response": "${data.response}",
},
"brand_voice": {
"query": "${data.query}",
"response": "${data.response}",
},
"word_count": {
"response": "${data.response}",
},
},
)
# Check gates
metrics = result["metrics"]
all_passed = True
print("=" * 50)
print("CI/CD QUALITY GATE REPORT")
print("=" * 50)
for metric_key, thresholds in GATES.items():
value = metrics.get(metric_key)
if value is None:
continue
min_val = thresholds.get("min", 0)
passed = value >= min_val
status = "✅" if passed else "❌"
print(f" {status} {metric_key}: {value:.2f} (min: {min_val})")
if not passed:
all_passed = False
print("=" * 50)
print(f"Result: {'✅ PASSED' if all_passed else '❌ FAILED'}")
return all_passed
if __name__ == "__main__":
passed = run_ci_evaluation("eval_dataset.jsonl", model_config)
sys.exit(0 if passed else 1)
GitHub Actions integration:
# .github/workflows/eval-gate.yml
- name: Run AI evaluation gate
run: |
pip install azure-ai-evaluation azure-identity openai
python scripts/run_eval_gate.py
env:
AZURE_OPENAI_ENDPOINT: $
AZURE_OPENAI_API_KEY: $
| Pattern | When to Use | LLM Required? |
|---|---|---|
| Code-based evaluator | Format checks, length, JSON, disclaimers | No |
| LLM-as-judge | Subjective quality (brand voice, tone, helpfulness) | Yes |
| Composite evaluator | Overall quality score from multiple signals | Depends on components |
| Adversarial simulation | Pre-deployment safety stress test | Yes (via Azure AI) |
| Red-teaming workflow | Full safety audit with evaluation | Yes |
| CI/CD quality gate | Automated deployment gating | Depends on evaluators |
Tip: Start with built-in evaluators (Lesson 1), add code-based custom evaluators for domain rules, then layer in LLM-as-judge evaluators for subjective criteria. Run adversarial simulation before every major release.
| ← Lesson 1: Built-in Evaluators | Next → Lesson 3: Production Monitoring |