references/evaluation-guide.md
# Evaluation Guide
Comprehensive guide for evaluating AI agent performance using Microsoft Foundry.
## Evaluation Workflow
1. **Prepare Dataset** - Create JSONL test data
2. **Upload Dataset** - Push to Microsoft Foundry
3. **Define Evaluators** - Select built-in or create custom
4. **Run Evaluation** - Execute against your agent
5. **Analyze Results** - Review scores and iterate
## Test Dataset Format
Create a JSONL file with input/output pairs:
```jsonl
{"query": "What is the capital of France?", "expected_response": "Paris"}
{"query": "Calculate 15 * 23", "expected_response": "345"}
{"query": "Summarize the benefits of AI", "response": "AI improves efficiency..."}
```
## Built-in Evaluators
### Agent Evaluators
| Evaluator | Purpose | Data Mapping |
|-----------|---------|--------------|
| `builtin.intent_resolution` | Was intent correctly identified? | query, response |
| `builtin.task_adherence` | Were instructions followed? | query, response, instructions |
| `builtin.task_completion` | Was task completed end-to-end? | query, response |
| `builtin.tool_call_accuracy` | Were tools used correctly? | query, response, tool_calls |
| `builtin.tool_selection` | Were right tools chosen? | query, response, available_tools |
### Quality Evaluators
| Evaluator | Purpose | Data Mapping |
|-----------|---------|--------------|
| `builtin.coherence` | Natural text flow? | query, response |
| `builtin.fluency` | Grammar correct? | response |
| `builtin.groundedness` | Claims substantiated? (RAG) | query, response, context |
| `builtin.relevance` | Answers key points? (RAG) | query, response |
## Complete Evaluation Example
```python
from azure.identity import DefaultAzureCredential
from azure.ai.projects import AIProjectClient
from openai.types.eval_create_params import DataSourceConfigCustom
from openai.types.evals.create_eval_jsonl_run_data_source_param import (
CreateEvalJSONLRunDataSourceParam, SourceFileID
)
import os
import time
endpoint = os.getenv("FOUNDRY_PROJECT_ENDPOINT")
model_deployment = os.getenv("MODEL_DEPLOYMENT_NAME")
with (
DefaultAzureCredential() as credential,
AIProjectClient(endpoint=endpoint, credential=credential) as project_client,
project_client.get_openai_client() as openai_client,
):
# 1. Upload Dataset
dataset = project_client.datasets.upload_file(
name="eval-data",
version="1",
file_path="data.jsonl"
)
# 2. Define Data Schema
data_source_config = DataSourceConfigCustom({
"type": "custom",
"item_schema": {
"type": "object",
"properties": {
"query": {"type": "string"},
"response": {"type": "string"}
},
"required": ["query", "response"]
},
"include_sample_schema": True
})
# 3. Define Evaluators
testing_criteria = [
{
"type": "azure_ai_evaluator",
"name": "coherence",
"evaluator_name": "builtin.coherence",
"data_mapping": {
"query": "{{item.query}}",
"response": "{{item.response}}"
},
"initialization_parameters": {"deployment_name": model_deployment}
},
{
"type": "azure_ai_evaluator",
"name": "relevance",
"evaluator_name": "builtin.relevance",
"data_mapping": {
"query": "{{item.query}}",
"response": "{{item.response}}"
},
"initialization_parameters": {"deployment_name": model_deployment}
}
]
# 4. Create Evaluation
evaluation = openai_client.evals.create(
name="agent-eval",
data_source_config=data_source_config,
testing_criteria=testing_criteria
)
# 5. Run Evaluation
run = openai_client.evals.runs.create(
eval_id=evaluation.id,
name="eval-run",
data_source=CreateEvalJSONLRunDataSourceParam(
type="jsonl",
source=SourceFileID(type="file_id", id=dataset.id)
)
)
# 6. Wait for Completion
while run.status not in ["completed", "failed"]:
run = openai_client.evals.runs.retrieve(run_id=run.id, eval_id=evaluation.id)
time.sleep(3)
print(f"Status: {run.status}")
print(f"Report: {run.report_url}")
```
## Custom Evaluators
### Code-Based Evaluator (Objective Metrics)
```python
code_evaluator = project_client.evaluators.create_version(
name="response_length_check",
evaluator_version={
"name": "response_length_check",
"definition": {
"type": "CODE",
"code_text": """
def grade(sample, item):
length = len(item.get("response", ""))
if 100 <= length <= 500:
return 1.0
elif length < 100:
return 0.5
else:
return 0.7
""",
"input_schema": {
"type": "object",
"properties": {
"response": {"type": "string"}
},
"required": ["response"]
},
"output_schema": {
"type": "number"
}
}
}
)
```
### Prompt-Based Evaluator (Subjective Metrics)
```python
prompt_evaluator = project_client.evaluators.create_version(
name="friendliness_check",
evaluator_version={
"name": "friendliness_check",
"definition": {
"type": "PROMPT",
"prompt_text": """
Rate the friendliness of this response on a scale of 1-5:
Query: {{query}}
Response: {{response}}
Consider:
- Tone (warm, welcoming, helpful)
- Language (polite, respectful)
- Helpfulness (goes above and beyond)
Output JSON only: {"result": <int 1-5>, "reason": "<brief explanation>"}
""",
"input_schema": {
"type": "object",
"properties": {
"query": {"type": "string"},
"response": {"type": "string"}
},
"required": ["query", "response"]
},
"output_schema": {
"type": "object",
"properties": {
"result": {"type": "integer"},
"reason": {"type": "string"}
}
}
}
}
)
```
## Quality Thresholds
| Metric | Minimum | Target | Notes |
|--------|---------|--------|-------|
| Coherence | 3.5 | 4.0+ | Scale 1-5 |
| Fluency | 4.0 | 4.5+ | Scale 1-5 |
| Relevance | 3.5 | 4.0+ | Scale 1-5 |
| Task Completion | 0.7 | 0.9+ | Scale 0-1 |
| Tool Accuracy | 0.8 | 0.95+ | Scale 0-1 |
## Troubleshooting
| Issue | Cause | Solution |
|-------|-------|----------|
| Dataset upload fails | Invalid JSONL | Validate JSON format line-by-line |
| Evaluator not found | Wrong name | Use exact `builtin.*` names |
| Low scores | Misaligned expectations | Review and refine agent instructions |
| Timeout | Large dataset | Split into smaller batches |
references/model-change-test-automation.md
# Model Change Test Automation
> Every AI agent should prove it works on more than one model before going to production.
> If your agent only works on GPT-5.1, you don't have a product - you have a dependency.
---
## Why Multi-Model Testing Is Non-Negotiable
### The Single-Model Trap
Most teams build and test against exactly one model. This creates hidden risks:
| Risk | What Happens | How Multi-Model Testing Prevents It |
|------|-------------|-------------------------------------|
| **Provider outage** | Agent goes down entirely | You know which backup model works |
| **Silent model update** | Quality degrades without code changes | Baseline comparison catches regressions |
| **Vendor lock-in** | Can't negotiate pricing, can't migrate | You've already validated alternatives |
| **Model deprecation** | Scramble migration under pressure | Replacement is pre-tested and ready |
| **Cost optimization** | Overpaying for a tier you don't need | Data shows cheaper model passes your bar |
### The Multi-Model Testing Mindset
```
WRONG: "Test on one model, deploy"
Build Agent -> Test on GPT-5.1 -> Deploy -> Hope it works
RIGHT: "Test on multiple, deploy the best"
Build Agent -> Test on [GPT-5.1, Claude Opus, O3, GPT-5.1-mini]
-> Compare scores
-> Pick primary + designate fallback
-> Deploy with confidence
-> Re-run monthly to catch drift
```
---
## Test Matrix Design
### Dimensions
A proper model comparison test covers four dimensions:
```
Models (rows) Evaluators (columns) Datasets (layers) Scenarios (depth)
Example:
4 models 5 evaluators 2 datasets 3 scenarios = 120 evaluation runs
```
### Model Selection Strategy
Choose models that cover your risk surface:
| Slot | Purpose | Example | Why |
|------|---------|---------|-----|
| **Primary** | Production model | gpt-5.1-2026-01-15 | Current production choice |
| **Challenger** | Next candidate | gpt-5.2-2026-02-01 | Newer version to evaluate |
| **Fallback** | Backup if primary fails | claude-opus-4-5 | Different provider for resilience |
| **Budget** | Cost-optimized option | gpt-5.1-mini | Can cheaper model pass the bar? |
| **Reasoning** | Complex task specialist | o3 | Worth the cost for hard tasks? |
**Minimum viable matrix**: Primary + one alternative from a different provider.
### Evaluator Selection
Run the same evaluators across all models:
| Evaluator | Type | Purpose |
|-----------|------|---------|
| `builtin.task_completion` | AI-assisted | Does the agent complete the task? |
| `builtin.coherence` | AI-assisted | Is the response well-structured? |
| `builtin.relevance` | AI-assisted | Does it address the user's query? |
| Custom: `format_compliance` | Code-based | Does output match expected JSON schema? |
| Custom: `tool_accuracy` | Code-based | Did the agent call the right tools? |
| Custom: `latency_check` | Code-based | Response time within SLA? |
| Custom: `cost_check` | Code-based | Token usage within budget? |
### Dataset Strategy
| Dataset | Size | Purpose | Update Cadence |
|---------|------|---------|----------------|
| **Core regression** | 50-100 cases | Critical path scenarios | Per release |
| **Edge cases** | 20-30 cases | Ambiguous/tricky inputs | Monthly |
| **Production sample** | 50 cases | Real user queries (anonymized) | Weekly refresh |
| **Adversarial** | 10-20 cases | Jailbreak, injection, off-topic | Quarterly |
---
## Comparison Report Format
### Standard Output Structure
Every model comparison run should produce a standardized report:
```json
{
"report_id": "compare-2026-02-11-001",
"timestamp": "2026-02-11T14:30:00Z",
"dataset": "core-regression-v3",
"dataset_size": 75,
"models": [
{
"name": "gpt-5.1-2026-01-15",
"role": "primary",
"scores": {
"task_completion": 0.92,
"coherence": 4.3,
"relevance": 4.1,
"format_compliance": 0.97,
"tool_accuracy": 0.95,
"avg_latency_ms": 1200,
"avg_tokens": 850,
"estimated_cost_per_1k": 2.89
}
},
{
"name": "claude-opus-4-5",
"role": "challenger",
"scores": {
"task_completion": 0.89,
"coherence": 4.5,
"relevance": 4.0,
"format_compliance": 0.93,
"tool_accuracy": 0.91,
"avg_latency_ms": 1800,
"avg_tokens": 920,
"estimated_cost_per_1k": 9.20
}
}
],
"comparison": {
"winner_by_metric": {
"task_completion": "gpt-5.1-2026-01-15",
"coherence": "claude-opus-4-5",
"relevance": "gpt-5.1-2026-01-15",
"format_compliance": "gpt-5.1-2026-01-15",
"cost_efficiency": "gpt-5.1-2026-01-15"
},
"recommendation": "Keep gpt-5.1 as primary. Claude wins on coherence but costs 3.2x more.",
"alerts": [
"claude-opus-4-5 format_compliance dropped below 0.95 threshold"
]
},
"thresholds": {
"task_completion": { "min": 0.85, "target": 0.90 },
"coherence": { "min": 3.5, "target": 4.0 },
"format_compliance": { "min": 0.95, "target": 0.98 }
}
}
```
### Comparison Table (Human-Readable)
```
+======================================================================+
| Metric | GPT-5.1 | Claude Opus | O3 | Threshold |
======================================================================
| Task Completion | 0.92 [PASS] | 0.89 [PASS] | 0.94 [PASS] | 0.85 |
| Coherence | 4.3 [PASS] | 4.5 [PASS] | 4.1 [PASS] | 3.5 |
| Format Compliance | 0.97 [PASS] | 0.93 [WARN] | 0.98 [PASS] | 0.95 |
| Tool Accuracy | 0.95 [PASS] | 0.91 [WARN] | 0.96 [PASS] | 0.90 |
| Avg Latency (ms) | 1200 | 1800 | 3500 | 5000 |
| Cost per 1K queries | $2.89 | $9.20 | $3.50 | $15 |
======================================================================
| RECOMMENDATION | PRIMARY [PASS]| FALLBACK [WARN] | REASONING | |
+======================================================================+
```
---
## CI/CD Integration
### Pipeline Architecture
```
-----------------------------------------------------------------
| Model Comparison Pipeline |
+-----------------------------------------------------------------+
| |
| Trigger: Schedule (weekly) | PR (model config changed) | Manual|
| |
| ---------- ------------------ ---------------------- |
| | Load |-> | Run eval suite |-> | Generate comparison | |
| | model | | per model | | report | |
| | matrix | | (parallel) | | | |
| ---------- ------------------ ----------+----------- |
| | |
| ---------------------- |
| | Gate: All models meet | |
| | minimum thresholds? | |
| -----------+----------- |
| ---------+---------- |
| YES| NO| |
| -------- ------- |
| | Pass [PASS] | | Fail [FAIL] | |
| | Upload | | Alert | |
| | report | | team | |
| --------- -------- |
-----------------------------------------------------------------
```
### GitHub Actions Workflow
```yaml
name: Model Comparison Test
on:
schedule:
- cron: '0 6 * * 1' # Weekly Monday 6am UTC
workflow_dispatch:
inputs:
models:
description: 'Comma-separated model list (or "all")'
default: 'all'
dataset:
description: 'Dataset to use'
default: 'core-regression'
push:
paths:
- 'config/models.yaml'
- 'evaluation/**'
jobs:
compare-models:
runs-on: ubuntu-latest
strategy:
matrix:
model: [gpt-5.1-2026-01-15, claude-opus-4-5, gpt-5.1-mini, o3]
fail-fast: false # Run ALL models even if one fails
steps:
- uses: actions/checkout@v4
- name: Setup Python
uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: Install dependencies
run: pip install agent-framework-azure-ai azure-ai-projects azure-identity
- name: Run evaluation for ${{ matrix.model }}
env:
FOUNDRY_ENDPOINT: ${{ secrets.FOUNDRY_ENDPOINT }}
FOUNDRY_API_KEY: ${{ secrets.FOUNDRY_API_KEY }}
EVAL_MODEL: ${{ matrix.model }}
EVAL_DATASET: evaluation/core-regression.jsonl
run: python scripts/run-model-eval.py
- name: Upload results
uses: actions/upload-artifact@v4
with:
name: eval-${{ matrix.model }}
path: evaluation/results/
compare-results:
needs: compare-models
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Download all results
uses: actions/download-artifact@v4
with:
path: evaluation/results/
- name: Generate comparison report
run: python scripts/run-model-comparison.py --results-dir evaluation/results/
- name: Check thresholds
run: python scripts/run-model-comparison.py --check-gates --fail-on-regression
- name: Comment on PR (if applicable)
if: github.event_name == 'pull_request'
uses: actions/github-script@v7
with:
script: |
const fs = require('fs');
const report = fs.readFileSync('evaluation/comparison-report.md', 'utf8');
github.rest.issues.createComment({
issue_number: context.issue.number,
owner: context.repo.owner,
repo: context.repo.repo,
body: report
});
```
### Model Configuration File
```yaml
# config/models.yaml - Single source of truth for model test matrix
models:
primary:
name: gpt-5.1-2026-01-15
deployment: gpt51-prod
provider: azure
role: primary
challenger:
name: gpt-5.2-2026-02-01
deployment: gpt52-staging
provider: azure
role: challenger
fallback:
name: claude-opus-4-5
deployment: claude-fallback
provider: azure # via Foundry model catalog
role: fallback
budget:
name: gpt-5.1-mini
deployment: gpt51-mini-prod
provider: azure
role: budget
thresholds:
# Minimum scores for any model to be considered viable
task_completion: 0.85
coherence: 3.5
relevance: 3.5
format_compliance: 0.95
tool_accuracy: 0.90
max_latency_ms: 5000
max_cost_per_1k: 15.00
# Regression: max allowed drop from baseline
max_regression_pct: 10
evaluation:
datasets:
- name: core-regression
path: evaluation/core-regression.jsonl
required: true
- name: edge-cases
path: evaluation/edge-cases.jsonl
required: false
- name: production-sample
path: evaluation/production-sample.jsonl
required: false
evaluators:
- builtin.task_completion
- builtin.coherence
- builtin.relevance
- custom.format_compliance
- custom.tool_accuracy
schedule:
comparison: weekly # Full matrix comparison
regression: on-push # Primary model only, on code changes
canary: daily # Lightweight smoke test on primary
```
---
## Implementation Patterns
### Pattern 1: Parametric Agent (Model-Agnostic Design)
Design your agent so the model is injected, not hardcoded:
```python
"""Model-agnostic agent that accepts any model at runtime."""
import os
from dataclasses import dataclass
@dataclass
class ModelConfig:
"""Model configuration - injected at runtime, not hardcoded."""
name: str
deployment: str
endpoint: str
api_key: str
temperature: float = 0.7
max_tokens: int = 4096
@classmethod
def from_env(cls, prefix: str = "AGENT") -> "ModelConfig":
"""Load model config from environment variables."""
return cls(
name=os.getenv(f"{prefix}_MODEL", "gpt-5.1"),
deployment=os.getenv(f"{prefix}_DEPLOYMENT", "gpt51-prod"),
endpoint=os.getenv(f"{prefix}_ENDPOINT", ""),
api_key=os.getenv(f"{prefix}_API_KEY", ""),
temperature=float(os.getenv(f"{prefix}_TEMPERATURE", "0.7")),
max_tokens=int(os.getenv(f"{prefix}_MAX_TOKENS", "4096")),
)
class AgentRunner:
"""Runs agent with any model - key for multi-model testing."""
def __init__(self, model_config: ModelConfig, system_prompt: str, tools: list):
self.config = model_config
self.system_prompt = system_prompt
self.tools = tools
self._client = self._create_client()
def _create_client(self):
"""Create model client - abstracted to support any provider."""
from agent_framework.openai import OpenAIChatClient
return OpenAIChatClient(
model=self.config.deployment,
api_key=self.config.api_key,
endpoint=self.config.endpoint,
)
async def run(self, query: str) -> dict:
"""Run agent and return structured result with metadata."""
import time
start = time.perf_counter()
response = await self._client.chat(
messages=[
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": query},
],
temperature=self.config.temperature,
max_tokens=self.config.max_tokens,
)
elapsed_ms = (time.perf_counter() - start) * 1000
return {
"model": self.config.name,
"query": query,
"response": response.content,
"latency_ms": round(elapsed_ms),
"tokens_used": getattr(response, "usage", {}).get("total_tokens", 0),
}
```
### Pattern 2: Evaluation Runner (Multi-Model)
```python
"""Run same evaluation suite against multiple models."""
import asyncio
import json
import yaml
from pathlib import Path
async def run_model_comparison(
config_path: str = "config/models.yaml",
dataset_path: str = "evaluation/core-regression.jsonl",
output_dir: str = "evaluation/results",
) -> dict:
"""Run evaluation suite against all models in config."""
# Load configuration
with open(config_path) as f:
config = yaml.safe_load(f)
# Load dataset
dataset = []
with open(dataset_path) as f:
for line in f:
dataset.append(json.loads(line.strip()))
# Run each model
results = {}
for role, model_cfg in config["models"].items():
print(f"\n--- Evaluating: {model_cfg['name']} ({role}) ---")
model_config = ModelConfig(
name=model_cfg["name"],
deployment=model_cfg["deployment"],
endpoint=os.getenv("FOUNDRY_ENDPOINT"),
api_key=os.getenv("FOUNDRY_API_KEY"),
)
runner = AgentRunner(
model_config=model_config,
system_prompt=AGENT_SYSTEM_PROMPT, # Your agent's prompt
tools=AGENT_TOOLS, # Your agent's tools
)
model_results = []
for item in dataset:
result = await runner.run(item["query"])
result["expected"] = item.get("expected_response", "")
model_results.append(result)
results[model_cfg["name"]] = {
"role": role,
"results": model_results,
}
# Save individual results
output = Path(output_dir)
output.mkdir(parents=True, exist_ok=True)
for model_name, data in results.items():
safe_name = model_name.replace("/", "-").replace(" ", "-")
with open(output / f"{safe_name}.json", "w") as f:
json.dump(data, f, indent=2)
return results
```
### Pattern 3: Comparison Report Generator
```python
"""Generate comparison report from multi-model eval results."""
import json
from pathlib import Path
def generate_comparison(results_dir: str, thresholds: dict) -> dict:
"""Compare all model results and generate report."""
results_path = Path(results_dir)
all_results = {}
# Load all result files
for result_file in results_path.glob("*.json"):
with open(result_file) as f:
data = json.load(f)
model_name = result_file.stem
all_results[model_name] = data
# Calculate aggregate scores per model
comparison = {"models": [], "alerts": []}
for model_name, data in all_results.items():
results = data["results"]
n = len(results)
scores = {
"model": model_name,
"role": data["role"],
"dataset_size": n,
"avg_latency_ms": sum(r["latency_ms"] for r in results) / n,
"avg_tokens": sum(r["tokens_used"] for r in results) / n,
"format_compliance": sum(
1 for r in results if _check_format(r["response"])
) / n,
}
# Check thresholds
for metric, threshold in thresholds.items():
value = scores.get(metric, None)
if value is not None and isinstance(threshold, (int, float)):
if metric.startswith("max_") and value > threshold:
comparison["alerts"].append(
f"{model_name}: {metric} = {value} exceeds {threshold}"
)
elif not metric.startswith("max_") and value < threshold:
comparison["alerts"].append(
f"{model_name}: {metric} = {value:.3f} below {threshold}"
)
comparison["models"].append(scores)
# Determine winner per metric
comparison["winner_by_metric"] = {}
numeric_metrics = ["format_compliance", "avg_latency_ms", "avg_tokens"]
for metric in numeric_metrics:
if metric.startswith("avg_latency") or metric.startswith("avg_token"):
# Lower is better
winner = min(comparison["models"], key=lambda m: m.get(metric, float("inf")))
else:
# Higher is better
winner = max(comparison["models"], key=lambda m: m.get(metric, 0))
comparison["winner_by_metric"][metric] = winner["model"]
return comparison
def _check_format(response: str) -> bool:
"""Check if response matches expected format. Customize per agent."""
try:
json.loads(response)
return True
except (json.JSONDecodeError, TypeError):
# Not all agents return JSON - customize this check
return len(response.strip()) > 0
```
---
## Cost Management
### Budget-Aware Testing
Running evals across 4+ models on 100+ test cases gets expensive. Manage costs:
| Strategy | How | Savings |
|----------|-----|---------|
| **Tiered datasets** | Full suite weekly, subset daily | 5x reduction |
| **Sampling** | Random 20% for non-critical runs | 5x reduction |
| **Caching** | Cache deterministic responses (temp=0) | 2-3x reduction |
| **Budget caps** | Set per-model token limits in config | Prevents runaway |
| **Smart scheduling** | Full comparison weekly, regression on PR only | 4x reduction |
### Cost Estimation Formula
```
Cost per run = (model_cost_per_1M_tokens avg_tokens_per_query dataset_size) / 1,000,000
Example:
GPT-5.1: $3.44/1M 850 tokens 100 queries = $0.29
Claude Opus: $10/1M 920 tokens 100 queries = $0.92
O3: $3.5/1M 1100 tokens 100 queries = $0.39
GPT-5.1-mini: $0.30/1M 700 tokens 100 queries = $0.02
Total per weekly run: ~$1.62
Monthly: ~$6.50
```
---
## Decision Framework
### When to Switch Primary Model
```
Should you switch your primary model?
1. New model scores HIGHER on task_completion?
+- No -> Keep current primary
- Yes -> Continue...
2. Passes ALL threshold checks?
+- No -> Not ready (address failures first)
- Yes -> Continue...
3. Format compliance 95%?
+- No -> Prompt adjustments needed for new model
- Yes -> Continue...
4. Cost acceptable? (within 2x of current)
+- No -> Consider for premium tier only
- Yes -> Continue...
5. Latency acceptable? (within 1.5x of current)
+- No -> Consider for async/batch only
- Yes -> SWITCH to new primary
6. Run 48-hour canary before full cutover
```
### Model Lifecycle States
```
CANDIDATE -> TESTING -> QUALIFIED -> CANARY -> PRIMARY -> DEPRECATED -> RETIRED
CANDIDATE: New model added to test matrix
TESTING: Running through evaluation suite
QUALIFIED: Passes all thresholds, ready for canary
CANARY: Serving 5-10% production traffic
PRIMARY: Full production traffic
DEPRECATED: Still available but being phased out
RETIRED: Removed from test matrix
```
---
## Anti-Patterns
| Anti-Pattern | Why It Fails | Fix |
|-------------|-------------|-----|
| "We tested on GPT-5.1 so we're good" | Single point of failure | Test on 2+ models minimum |
| Testing models sequentially by hand | Inconsistent, slow, error-prone | Automate with CI pipeline |
| Different prompts per model | Can't compare fairly | Same prompt, same dataset |
| Ignoring cost in comparison | Cheapest model might pass the bar | Include cost as a metric |
| One-time comparison | Models change, new options emerge | Schedule weekly/monthly runs |
| Testing without thresholds | No objective pass/fail criteria | Define min scores upfront |
| Skipping format compliance | New model may output differently | Always test structured output |
| Not testing tool calling | Models differ in function-calling | Include tool accuracy metric |
---
## Quick Start Checklist
### Phase 1: Foundation (Day 1)
- [ ] Design agent with model injection (Pattern 1)
- [ ] Create core regression dataset (50+ cases)
- [ ] Define quality thresholds per metric
- [ ] Create `config/models.yaml` with 2+ models
### Phase 2: Automation (Week 1)
- [ ] Implement evaluation runner script
- [ ] Generate first comparison report
- [ ] Set up CI pipeline (GitHub Actions or Azure DevOps)
- [ ] Store baseline scores for primary model
### Phase 3: Operations (Ongoing)
- [ ] Schedule weekly full comparison runs
- [ ] Add regression check to PR pipeline
- [ ] Update datasets with production samples monthly
- [ ] Review and refresh model matrix quarterly
- [ ] Document model decisions and switch history
---
## Related
- [Multi-Model Patterns](multi-model-patterns.md) - Routing and fallback for production
- [Model Drift & Judge Patterns](model-drift-judge-patterns.md) - Detecting degradation over time
- [Evaluation Guide](evaluation-guide.md) - Microsoft Foundry evaluation SDK usage
references/model-drift-judge-patterns.md
# Model Change, Data Drift & Judge LLM Patterns
> The three most common production blind spots in AI agent development.
> Most teams build agents that work on day one and silently degrade by week four.
---
## 1. Model Change Management
### The Problem
When you switch models (e.g., `gpt-4o` -> `gpt-5.1`) or the provider updates a model version silently, agent behavior changes **without any code change**. Your tests pass, your CI is green, but:
- Output format shifts (JSON keys reordered, casing changes)
- Tone and verbosity change
- Tool calling patterns differ (different models have different function-calling biases)
- Reasoning quality changes (some tasks improve, others regress)
- Token usage and latency shift
### Decision Tree
```
Model change detected?
+- Planned change (you chose to switch)?
| +- Run full evaluation suite BEFORE switching
| +- Compare scores: old model vs new model
| +- Check: structured output format unchanged?
| +- Check: tool calling accuracy maintained?
| - Only deploy if all thresholds met
+- Provider silent update (same model name, new version)?
| +- Monitor evaluation scores over time (weekly cadence)
| +- Alert on score drops > 10% from baseline
| - Pin model version if provider supports it
- Multi-model setup (different models for different tasks)?
+- Test each model independently
+- Test the composition (Model A -> Model B handoff)
- Document which model does what and why
```
### Model Change Checklist
- [ ] **Pin model versions** - Use `gpt-5.1-2026-01-15`, not just `gpt-5.1`
- [ ] **Maintain evaluation baseline** - Store scores from current model as `baseline.json`
- [ ] **Run A/B evaluation** - Compare new model against baseline before switching
- [ ] **Test structured outputs** - Verify JSON schema compliance didn't break
- [ ] **Test tool calling** - Verify function-calling accuracy maintained
- [ ] **Test edge cases** - Models differ most on ambiguous/tricky inputs
- [ ] **Check cost/latency** - New model may have different pricing or speed
- [ ] **Document the change** - Record why you switched and evaluation results
### Model Configuration Best Practices
```python
# [FAIL] Bad: Implicit model, no version pinning
client = OpenAIChatClient(model="gpt-5.1")
# [PASS] Good: Explicit version, configurable, documented
MODEL_CONFIG = {
"model": os.getenv("AGENT_MODEL", "gpt-5.1-2026-01-15"),
"temperature": 0.7,
"max_tokens": 4096,
"model_version_pinned": True, # Document intent
"last_evaluated": "2026-02-01", # When was this model last evaluated?
"baseline_scores": "evaluation/baseline-gpt51.json", # Where are baseline scores?
}
```
### Model Migration Workflow
```
1. BASELINE -> Run eval suite on current model, save scores as baseline
2. CANDIDATE -> Deploy new model in shadow mode (log responses, don't serve)
3. COMPARE -> Run same eval suite on candidate, compare against baseline
4. THRESHOLD -> All metrics within acceptable range? (5% typically)
5. CANARY -> Route 5-10% traffic to new model, monitor live metrics
6. PROMOTE -> Switch fully if canary succeeds for 48+ hours
7. DOCUMENT -> Update model config, baseline file, and changelog
```
---
## 2. Data Drift Detection
### The Problem
Your agent is tested on sample queries during development. In production, users send:
- **Different topics** than your test set covered
- **Different languages** or formatting
- **Adversarial inputs** (jailbreak attempts, edge cases)
- **Longer/shorter inputs** than expected
- **Changed domain context** (new products, updated policies)
Over weeks, the distribution of real inputs diverges from your test data. Agent quality degrades silently.
### Decision Tree
```
Monitoring agent inputs?
+- No -> Set up input logging immediately
| +- Log: query length, topic classification, language
| +- Log: tool selection distribution
| - Log: response satisfaction signals (if available)
+- Yes -> Analyzing drift?
| +- Compare production input distribution vs test dataset
| +- Flag queries with no similar test case (novelty detection)
| +- Track topic distribution shifts week-over-week
| - Track failure rate by input category
- Drift detected?
+- Update test dataset with representative production samples
+- Re-run evaluation on updated dataset
+- Adjust agent instructions if needed
- Add guardrails for unexpected input categories
```
### Data Drift Checklist
- [ ] **Log all production inputs** - At minimum: query text, timestamp, response, latency
- [ ] **Classify inputs** - Categorize by topic/intent to track distribution
- [ ] **Sample production data weekly** - Pull random sample for manual review
- [ ] **Compare distributions** - Production inputs vs evaluation dataset
- [ ] **Track failure patterns** - Group low-quality responses by input characteristics
- [ ] **Update eval dataset quarterly** - Add new representative production queries
- [ ] **Monitor out-of-domain queries** - Detect inputs your agent wasn't designed for
### Drift Signals to Monitor
| Signal | What It Means | Action |
|--------|--------------|--------|
| Query length shifting | Users are asking differently | Update test cases |
| New topic clusters | Agent is being used for unintended purposes | Add guardrails or expand scope |
| Increasing tool call failures | Input format changed | Update tool schemas |
| Declining satisfaction scores | Overall quality degrading | Full diagnosis needed |
| Rising latency | Queries getting more complex | Optimize or add caching |
| Language mix changing | New user demographics | Add multilingual testing |
### Drift Detection Implementation
```python
"""Lightweight drift detector for agent inputs."""
import json
from collections import Counter
from datetime import datetime, timedelta
class DriftDetector:
"""Compare production input patterns against baseline."""
def __init__(self, baseline_path: str):
with open(baseline_path) as f:
self.baseline = json.load(f)
def check_length_drift(self, recent_queries: list[str]) -> dict:
"""Check if query lengths have shifted."""
baseline_avg = self.baseline.get("avg_query_length", 100)
current_avg = sum(len(q) for q in recent_queries) / len(recent_queries)
drift_pct = abs(current_avg - baseline_avg) / baseline_avg * 100
return {
"metric": "query_length",
"baseline": baseline_avg,
"current": current_avg,
"drift_percent": round(drift_pct, 1),
"alert": drift_pct > 25, # > 25% shift = alert
}
def check_topic_drift(self, recent_topics: list[str]) -> dict:
"""Check if topic distribution has shifted."""
baseline_dist = self.baseline.get("topic_distribution", {})
current_dist = dict(Counter(recent_topics))
# Normalize
total = sum(current_dist.values())
current_pct = {k: v / total for k, v in current_dist.items()}
# Find new topics not in baseline
new_topics = set(current_pct.keys()) - set(baseline_dist.keys())
return {
"metric": "topic_distribution",
"new_topics": list(new_topics),
"alert": len(new_topics) > 0,
}
def save_snapshot(self, queries: list[str], topics: list[str], path: str):
"""Save current distribution as new baseline."""
snapshot = {
"timestamp": datetime.now().isoformat(),
"avg_query_length": sum(len(q) for q in queries) / len(queries),
"query_count": len(queries),
"topic_distribution": dict(Counter(topics)),
}
with open(path, "w") as f:
json.dump(snapshot, f, indent=2)
```
---
## 3. Judge LLM Implementation
### The Problem
When you use an LLM to evaluate another LLM's outputs (LLM-as-judge), you need the judge to be:
- **Consistent** - Same input should get same score
- **Calibrated** - Scores should mean what you think they mean
- **Grounded** - Judging on specific criteria, not vibes
- **Validated** - The judge itself needs to be tested
Most teams either skip evaluation entirely, or implement a judge so vague it's useless.
### Decision Tree
```
Need to evaluate agent quality?
+- Objective metric? (exact match, count, format check)
| - Use code-based evaluator (no LLM needed)
+- Subjective metric? (quality, tone, helpfulness)
| +- Use LLM-as-judge with structured rubric
| +- Define explicit scoring criteria per level
| - Validate judge with known-answer set
- Critical decision? (safety, compliance, accuracy)
+- Use multiple judges (judge ensemble)
+- Include human review sample
- Cross-validate judge agreement rate
```
### Judge Anti-Patterns
| Anti-Pattern | Why It Fails | Fix |
|-------------|-------------|-----|
| "Rate 1-5" with no rubric | Judge has no criteria -> random scores | Define what each score level means |
| Same model as agent and judge | Self-evaluation bias | Use different model for judging |
| No judge validation | Don't know if judge is reliable | Test judge on known-answer pairs |
| Single judge for everything | Different aspects need different criteria | Separate judges per dimension |
| Binary pass/fail | Loses nuance, hard to improve | Use 1-5 scale with rubric per level |
| No inter-rater agreement check | Judge may be inconsistent | Run same input 3x, check variance |
### Proper Judge Implementation
```python
"""Structured judge evaluator with rubric and validation."""
JUDGE_PROMPT = """You are evaluating an AI agent's response quality.
RUBRIC - Score each dimension on 1-5:
**Accuracy** (Is the information correct?):
5: Fully accurate, all facts verified
4: Mostly accurate, minor imprecision
3: Partially accurate, some errors but core is right
2: Significant errors that mislead the user
1: Fundamentally wrong or fabricated
**Completeness** (Does it address the full query?):
5: Addresses all aspects with appropriate depth
4: Addresses most aspects, minor gaps
3: Addresses core question but misses secondary points
2: Partially addresses the question
1: Does not address the query
**Helpfulness** (Is the response actionable?):
5: Directly actionable, user can proceed immediately
4: Helpful with minor clarification needed
3: Somewhat helpful but requires additional research
2: Minimally helpful, mostly filler
1: Not helpful, confusing, or harmful
INPUT:
Query: {query}
Response: {response}
Context: {context}
OUTPUT (JSON only):
{{
"accuracy": <int 1-5>,
"completeness": <int 1-5>,
"helpfulness": <int 1-5>,
"overall": <float, weighted average>,
"reasoning": "<2-3 sentence justification>"
}}
"""
```
### Judge Validation Process
Every judge LLM needs its own validation:
```
1. CREATE KNOWN-ANSWER SET
- 20-30 examples with human-assigned "gold" scores
- Include clear good (5), clear bad (1), and ambiguous (3) cases
- Have 2+ humans score independently for agreement baseline
2. RUN JUDGE ON KNOWN-ANSWER SET
- Score all 20-30 examples with your judge prompt
- Run 3x to check consistency (variance < 0.5 on 1-5 scale)
3. MEASURE JUDGE QUALITY
- Cohen's Kappa vs human scores (target > 0.6 = substantial agreement)
- Mean Absolute Error (target < 0.8 on 1-5 scale)
- Check for position bias (does order of examples affect scores?)
- Check for length bias (do longer responses get higher scores?)
4. ITERATE
- If agreement is low, refine the rubric
- If variance is high, add more specific criteria
- If biased, add de-biasing instructions to prompt
```
### Judge Ensemble Pattern
For critical evaluations, use multiple judges:
```python
"""Judge ensemble: majority vote from 3 independent judges."""
JUDGE_MODELS = [
{"model": "gpt-5.1", "role": "primary"},
{"model": "claude-opus-4-5", "role": "secondary"},
{"model": "gpt-5.1", "role": "tiebreaker", "temperature": 0.3},
]
async def ensemble_judge(query: str, response: str) -> dict:
"""Run 3 judges and take weighted average."""
scores = []
for judge_config in JUDGE_MODELS:
score = await run_single_judge(
query=query,
response=response,
model=judge_config["model"],
temperature=judge_config.get("temperature", 0.0),
)
scores.append(score)
# Aggregate
return {
"accuracy": sum(s["accuracy"] for s in scores) / len(scores),
"completeness": sum(s["completeness"] for s in scores) / len(scores),
"helpfulness": sum(s["helpfulness"] for s in scores) / len(scores),
"judge_agreement": max(s["overall"] for s in scores) - min(s["overall"] for s in scores),
"individual_scores": scores,
}
```
---
## Integration Checklist
### Before Launch
- [ ] Model version pinned (not just model name)
- [ ] Evaluation baseline saved (`baseline.json`)
- [ ] Judge validated on known-answer set (agreement > 0.6)
- [ ] Input logging enabled
- [ ] Drift detection alerts configured
### Weekly Operations
- [ ] Review drift metrics dashboard
- [ ] Sample and review 10 random production queries
- [ ] Check evaluation scores haven't dropped
- [ ] Review judge consistency (if custom judges)
### On Model Change
- [ ] Run full evaluation against baseline
- [ ] Compare all metric dimensions (not just overall)
- [ ] Test structured output format compliance
- [ ] 48-hour canary before full rollout
- [ ] Update baseline after successful migration
### Quarterly
- [ ] Update evaluation dataset with production samples
- [ ] Re-validate judge on expanded known-answer set
- [ ] Review and prune unused model configurations
- [ ] Audit drift detection thresholds
---
## Quick Reference
| Concern | Detection | Prevention |
|---------|-----------|-----------|
| **Model change** | Eval score comparison, A/B testing | Pin versions, maintain baselines |
| **Data drift** | Distribution monitoring, novelty detection | Regular eval dataset updates |
| **Judge reliability** | Known-answer validation, consistency checks | Structured rubrics, ensembles |
---
**Related**: [Evaluation Guide](evaluation-guide.md) - [Tracing & Evaluation](tracing-and-evaluation.md)
references/multi-model-patterns.md
# Multi-Model Agent Patterns
## Multi-Model Patterns
### Environment Configuration
Use a `.env` file for local development (always add to `.gitignore`):
```env
# .env.example - Copy to .env and fill in values
# Required
FOUNDRY_ENDPOINT=https://your-resource.services.ai.azure.com
FOUNDRY_API_KEY=your-api-key-here
MODEL_DEPLOYMENT_NAME=gpt-4o
# Optional: Multi-model setup
MODEL_FAST=gpt-4o-mini
MODEL_REASONING=o3
MODEL_EMBEDDING=text-embedding-3-large
# Optional: Observability
APPLICATIONINSIGHTS_CONNECTION_STRING=
```
### Model Routing
Route requests to different models based on task complexity:
```python
import os
MODELS = {
"fast": os.environ.get("MODEL_FAST", "gpt-4o-mini"), # Simple tasks, low latency
"standard": os.environ.get("MODEL_DEPLOYMENT_NAME", "gpt-4o"), # General purpose
"reasoning": os.environ.get("MODEL_REASONING", "o3"), # Complex analysis
}
def select_model(task_type: str) -> str:
"""Select model based on task complexity."""
routing = {
"classification": "fast",
"summarization": "fast",
"code_generation": "standard",
"architecture_review": "reasoning",
"complex_analysis": "reasoning",
}
tier = routing.get(task_type, "standard")
return MODELS[tier]
```
### Fallback Chains
Implement fallback when a model is unavailable or rate-limited:
```python
async def call_with_fallback(prompt: str, models: list[str]) -> str:
"""Try models in order, falling back on failure."""
for model in models:
try:
return await client.complete(model=model, prompt=prompt)
except (RateLimitError, ServiceUnavailableError):
continue
raise AllModelsUnavailableError("All models in fallback chain failed")
# Usage: prefer fast, fall back to standard
result = await call_with_fallback(prompt, ["gpt-4o-mini", "gpt-4o"])
```
### Cost Optimization
| Tier | Model | Use Case | Relative Cost |
|------|-------|----------|---------------|
| Fast | gpt-4o-mini | Classification, routing, simple Q&A | $ |
| Standard | gpt-4o | Code generation, summarization | $$ |
| Reasoning | o3 | Complex analysis, multi-step reasoning | $$$$ |
**Guidelines**:
- Default to the **fast** tier; escalate only when quality requires it
- Cache frequent prompts/responses where deterministic
- Monitor token usage per model with tracing (see Observability section)
---
references/orchestration-patterns.md
# Orchestration Patterns
Guide to multi-agent orchestration patterns using Microsoft Agent Framework.
> **[WARN] Prompt Management Rule**: In all patterns below, `instructions` are shown inline for brevity. In production, **ALWAYS** load prompts from separate files: `Path("prompts/{agent}.md").read_text()`. See [SKILL.md](../SKILL.md#prompt--template-file-management).
## Pattern Overview
| Pattern | Use Case | Complexity |
|---------|----------|------------|
| Sequential | Step-by-step processing | Low |
| Parallel | Independent tasks | Medium |
| Conditional | Decision-based routing | Medium |
| Group Chat | Collaborative discussion | High |
| Fan-out/Fan-in | Distribute and aggregate | High |
| Human-in-the-Loop | Approval workflows | Medium |
## Sequential Workflow
Agents execute in order, passing results to the next.
```python
from pathlib import Path
from agent_framework.workflows import SequentialWorkflow
# Load prompts from files - NEVER embed as inline strings in production
researcher = {
"name": "Researcher",
"instructions": Path("prompts/researcher.md").read_text(encoding="utf-8")
}
writer = {
"name": "Writer",
"instructions": Path("prompts/writer.md").read_text(encoding="utf-8")
}
editor = {
"name": "Editor",
"instructions": Path("prompts/editor.md").read_text(encoding="utf-8")
}
# Create workflow
workflow = SequentialWorkflow(
agents=[researcher, writer, editor],
handoff_strategy="on_completion"
)
# Execute
result = await workflow.run(
query="Write a report on AI trends in 2026"
)
```
## Parallel Workflow
Multiple agents work simultaneously on different tasks.
```python
from agent_framework.workflows import ParallelWorkflow
# Define parallel agents
market_analyst = {
"name": "Market Analyst",
"instructions": "Analyze market trends and opportunities."
}
tech_analyst = {
"name": "Tech Analyst",
"instructions": "Analyze technical landscape and innovations."
}
risk_analyst = {
"name": "Risk Analyst",
"instructions": "Identify and assess potential risks."
}
# Create parallel workflow
workflow = ParallelWorkflow(
agents=[market_analyst, tech_analyst, risk_analyst],
aggregator={
"name": "Aggregator",
"instructions": "Combine all analyses into a comprehensive report."
}
)
# Execute (all agents run in parallel, then aggregator combines)
result = await workflow.run(
query="Comprehensive analysis of AI startup landscape"
)
```
## Conditional Workflow
Route to different agents based on conditions.
```python
from agent_framework.workflows import ConditionalWorkflow
# Define specialized agents
support_agent = {
"name": "Support Agent",
"instructions": "Handle customer support inquiries."
}
sales_agent = {
"name": "Sales Agent",
"instructions": "Handle sales and pricing questions."
}
technical_agent = {
"name": "Technical Agent",
"instructions": "Handle technical questions and troubleshooting."
}
# Define routing logic
def route_query(query: str) -> str:
query_lower = query.lower()
if any(word in query_lower for word in ["price", "buy", "purchase", "cost"]):
return "sales"
elif any(word in query_lower for word in ["error", "bug", "fix", "issue"]):
return "technical"
else:
return "support"
# Create conditional workflow
workflow = ConditionalWorkflow(
router=route_query,
agents={
"support": support_agent,
"sales": sales_agent,
"technical": technical_agent
}
)
# Execute
result = await workflow.run(
query="I'm getting an error when I try to login"
) # Routes to technical_agent
```
## Group Chat
Multiple agents collaborate through conversation.
```python
from agent_framework.workflows import GroupChat
# Define participants
ceo = {
"name": "CEO",
"instructions": "Provide strategic direction and final decisions."
}
cto = {
"name": "CTO",
"instructions": "Advise on technical feasibility and architecture."
}
cfo = {
"name": "CFO",
"instructions": "Advise on budget and financial implications."
}
moderator = {
"name": "Moderator",
"instructions": "Keep discussion focused and summarize decisions."
}
# Create group chat
chat = GroupChat(
participants=[ceo, cto, cfo],
moderator=moderator,
max_rounds=5,
termination_condition="consensus_reached"
)
# Execute
result = await chat.run(
topic="Should we invest in building an AI-powered product?"
)
```
## Fan-out/Fan-in
Distribute work across multiple agents, then aggregate.
```python
from agent_framework.workflows import FanOutFanIn
# Define worker agents (can be dynamically created)
def create_analyzer(section: str):
return {
"name": f"Section_{section}_Analyzer",
"instructions": f"Analyze the {section} section thoroughly."
}
sections = ["introduction", "methodology", "results", "conclusion"]
analyzers = [create_analyzer(s) for s in sections]
# Aggregator combines all results
aggregator = {
"name": "Report Aggregator",
"instructions": "Synthesize all section analyses into a cohesive review."
}
# Create fan-out/fan-in workflow
workflow = FanOutFanIn(
workers=analyzers,
aggregator=aggregator,
distribute_strategy="round_robin" # or "random", "load_balanced"
)
# Execute
result = await workflow.run(
document="<full paper content>",
task="Review this research paper"
)
```
## Human-in-the-Loop
Include human approval or input in the workflow.
```python
from agent_framework.workflows import HumanInTheLoop
# Define agent
code_generator = {
"name": "Code Generator",
"instructions": "Generate code based on requirements."
}
# Human approval callback
async def require_approval(output: str, context: dict) -> tuple[bool, str]:
# In production, this would send to a human reviewer
# For now, auto-approve if code looks valid
if "def " in output or "class " in output:
return True, "Code looks valid"
else:
return False, "Please regenerate with proper Python syntax"
# Create workflow with human gate
workflow = HumanInTheLoop(
agent=code_generator,
approval_gate=require_approval,
max_retries=3
)
# Execute
result = await workflow.run(
requirements="Create a function to validate email addresses"
)
```
## Loop with Reflection
Agent iterates on its own output using reflection.
```python
from agent_framework.workflows import ReflectiveLoop
# Define worker and critic
writer = {
"name": "Writer",
"instructions": "Write content based on the brief."
}
critic = {
"name": "Critic",
"instructions": "Review the content and provide specific improvement suggestions."
}
# Create reflective loop
workflow = ReflectiveLoop(
worker=writer,
critic=critic,
max_iterations=3,
stop_condition=lambda feedback: "excellent" in feedback.lower()
)
# Execute
result = await workflow.run(
brief="Write a compelling product description for an AI assistant"
)
```
## Best Practices
### Choosing the Right Pattern
| Scenario | Recommended Pattern |
|----------|-------------------|
| Processing pipeline | Sequential |
| Independent analysis | Parallel |
| Customer service routing | Conditional |
| Brainstorming/Planning | Group Chat |
| Large document analysis | Fan-out/Fan-in |
| High-risk decisions | Human-in-the-Loop |
| Quality improvement | Loop with Reflection |
### Performance Considerations
1. **Parallel when possible** - Independent tasks should run concurrently
2. **Minimize handoffs** - Each handoff adds latency
3. **Set iteration limits** - Prevent infinite loops
4. **Use appropriate models** - Simpler agents can use faster/cheaper models
5. **Cache intermediate results** - Avoid redundant processing
6. **Store prompts in files** - Load from `prompts/` directory, never inline
### Error Handling
```python
from agent_framework.workflows import SequentialWorkflow, WorkflowError
try:
result = await workflow.run(query="...")
except WorkflowError as e:
print(f"Workflow failed at step {e.failed_step}: {e.message}")
# Access partial results
partial = e.partial_results
# Retry from failed step
result = await workflow.resume(from_step=e.failed_step)
```
### Monitoring
Enable tracing to visualize workflow execution:
```python
from agent_framework.observability import configure_otel_providers
configure_otel_providers(
vs_code_extension_port=4317,
enable_sensitive_data=True
)
# Now run your workflow - traces will show agent interactions
```
Open trace viewer: `Ctrl+Shift+P` -> `AI Toolkit: Open Trace Viewer`
references/tracing-and-evaluation.md
# AI Agent Tracing & Evaluation Patterns
## Observability (Tracing)
### Setup OpenTelemetry
```python
from agent_framework.observability import configure_otel_providers
# Before running agent - must open trace viewer first!
configure_otel_providers(
vs_code_extension_port=4317, # AI Toolkit gRPC port
enable_sensitive_data=True
)
```
**Open Trace Viewer**: `Ctrl+Shift+P` -> `AI Toolkit: Open Trace Viewer`
[WARN] **CRITICAL**: Open trace viewer BEFORE running your agent.
---
## Evaluation
### Workflow
1. Upload dataset (JSONL)
2. Define evaluators (built-in or custom)
3. Create evaluation
4. Run evaluation
5. Analyze results
### Prerequisites
```bash
pip install "azure-ai-projects>=2.0.0b2"
```
### Built-in Evaluators
**Agent Evaluators**:
- `builtin.intent_resolution` - Intent correctly identified?
- `builtin.task_adherence` - Instructions followed?
- `builtin.task_completion` - Task completed end-to-end?
- `builtin.tool_call_accuracy` - Tools used correctly?
- `builtin.tool_selection` - Right tools chosen?
**Quality Evaluators**:
- `builtin.coherence` - Natural text flow?
- `builtin.fluency` - Grammar correct?
- `builtin.groundedness` - Claims substantiated? (RAG)
- `builtin.relevance` - Answers key points? (RAG)
### Evaluation Example
```python
from azure.identity import DefaultAzureCredential
from azure.ai.projects import AIProjectClient
from openai.types.eval_create_params import DataSourceConfigCustom
from openai.types.evals.create_eval_jsonl_run_data_source_param import (
CreateEvalJSONLRunDataSourceParam, SourceFileID
)
endpoint = os.getenv("FOUNDRY_PROJECT_ENDPOINT")
model_deployment = os.getenv("MODEL_DEPLOYMENT_NAME")
with (
DefaultAzureCredential() as credential,
AIProjectClient(endpoint=endpoint, credential=credential) as project_client,
project_client.get_openai_client() as openai_client,
):
# 1. Upload Dataset
dataset = project_client.datasets.upload_file(
name="eval-data",
version="1",
file_path="data.jsonl"
)
# 2. Define Data Schema
data_source_config = DataSourceConfigCustom({
"type": "custom",
"item_schema": {
"type": "object",
"properties": {
"query": {"type": "string"},
"response": {"type": "string"}
},
"required": ["query", "response"]
},
"include_sample_schema": True
})
# 3. Define Evaluators
testing_criteria = [
{
"type": "azure_ai_evaluator",
"name": "coherence",
"evaluator_name": "builtin.coherence",
"data_mapping": {
"query": "{{item.query}}",
"response": "{{item.response}}"
},
"initialization_parameters": {"deployment_name": model_deployment}
}
]
# 4. Create Evaluation
evaluation = openai_client.evals.create(
name="agent-eval",
data_source_config=data_source_config,
testing_criteria=testing_criteria
)
# 5. Run Evaluation
run = openai_client.evals.runs.create(
eval_id=evaluation.id,
name="eval-run",
data_source=CreateEvalJSONLRunDataSourceParam(
type="jsonl",
source=SourceFileID(type="file_id", id=dataset.id)
)
)
# 6. Wait for Completion
while run.status not in ["completed", "failed"]:
run = openai_client.evals.runs.retrieve(run_id=run.id, eval_id=evaluation.id)
time.sleep(3)
print(f"Report: {run.report_url}")
```
### Custom Evaluators
**Code-based** (objective metrics):
```python
code_evaluator = project_client.evaluators.create_version(
name="response_length_check",
evaluator_version={
"name": "response_length_check",
"definition": {
"type": "CODE",
"code_text": """
def grade(sample, item):
length = len(item.get("response", ""))
return 1.0 if 100 <= length <= 500 else 0.5
""",
# ... schema omitted for brevity
}
}
)
```
**Prompt-based** (subjective metrics):
```python
prompt_evaluator = project_client.evaluators.create_version(
name="friendliness_check",
evaluator_version={
"name": "friendliness_check",
"definition": {
"type": "PROMPT",
"prompt_text": """
Rate friendliness (1-5):
Query: {{query}}
Response: {{response}}
Output JSON: {"result": <int>, "reason": "<text>"}
""",
# ... schema omitted for brevity
}
}
)
```
---
scripts/check-model-drift.ps1
<#
.SYNOPSIS
Validate AI agent project for model change, data drift, and judge LLM readiness.
.DESCRIPTION
Checks an AI agent project for the three most common production blind spots:
1. Model Change Management - pinned versions, baselines, migration plans
2. Data Drift Detection - input logging, distribution tracking, eval freshness
3. Judge LLM Quality - rubric presence, validation data, consistency checks
Scans source code, config files, and evaluation artifacts.
.PARAMETER Path
Root of the agent project to validate. Defaults to current directory.
.PARAMETER Strict
Treat warnings as failures (exit code > 0).
.EXAMPLE
./check-model-drift.ps1
./check-model-drift.ps1 -Path ./my-agent -Strict
#>
param(
[string]$Path = ".",
[switch]$Strict
)
$ErrorActionPreference = "Stop"
$script:Passed = 0
$script:Warned = 0
$script:Failed = 0
function Write-Check {
param([string]$Name, [string]$Status, [string]$Detail = "")
switch ($Status) {
"PASS" {
Write-Host " [PASS] $Name" -ForegroundColor Green
$script:Passed++
}
"WARN" {
Write-Host " [WARN] $Name" -ForegroundColor Yellow
if ($Detail) { Write-Host " $Detail" -ForegroundColor DarkYellow }
$script:Warned++
}
"FAIL" {
Write-Host " [FAIL] $Name" -ForegroundColor Red
if ($Detail) { Write-Host " $Detail" -ForegroundColor DarkRed }
$script:Failed++
}
}
}
$Root = Resolve-Path $Path -ErrorAction SilentlyContinue
if (-not $Root) {
Write-Host "Error: Path '$Path' not found." -ForegroundColor Red
exit 1
}
# Collect source files
$pyFiles = Get-ChildItem -Path $Root -Filter "*.py" -Recurse -ErrorAction SilentlyContinue |
Where-Object { $_.FullName -notmatch '(\.venv|venv|__pycache__|node_modules|\.git)' }
$csFiles = Get-ChildItem -Path $Root -Filter "*.cs" -Recurse -ErrorAction SilentlyContinue |
Where-Object { $_.FullName -notmatch '(bin|obj|\.git)' }
$configFiles = @()
$configFiles += Get-ChildItem -Path $Root -Include "*.json","*.yaml","*.yml","*.toml","*.env*" -Recurse -ErrorAction SilentlyContinue |
Where-Object { $_.FullName -notmatch '(\.venv|venv|node_modules|\.git|bin|obj)' }
$allSource = @($pyFiles) + @($csFiles) | Where-Object { $_ -ne $null }
Write-Host ""
Write-Host "============================================================" -ForegroundColor Cyan
Write-Host " Model Change, Data Drift & Judge LLM Validator" -ForegroundColor Cyan
Write-Host " Path: $Root" -ForegroundColor DarkGray
Write-Host "============================================================" -ForegroundColor Cyan
# ================================================================
# SECTION 1: MODEL CHANGE MANAGEMENT
# ================================================================
Write-Host ""
Write-Host "--- 1. Model Change Management ---" -ForegroundColor White
# Check: Model version pinned (not just generic name)
$hasPinnedVersion = $false
$hasGenericModel = $false
$genericModelFiles = @()
foreach ($file in $allSource + $configFiles) {
$content = Get-Content $file.FullName -Raw -ErrorAction SilentlyContinue
if (-not $content) { continue }
# Pinned: model name with date suffix like gpt-5.1-2026-01-15
if ($content -match '(gpt-\d[\w.-]+-\d{4}-\d{2}-\d{2}|claude-[\w.]+-\d{8}|model_version.*pinned)') {
$hasPinnedVersion = $true
}
# Generic: just "gpt-5.1" or "gpt-4o" without date pin
if ($content -match 'model["\s:=]+["'']?(gpt-\d[\w.]*|claude-[\w.-]+|o\d+(-\w+)?)["'']?' -and
$content -notmatch '(gpt-\d[\w.-]+-\d{4}-\d{2}-\d{2})') {
$hasGenericModel = $true
$genericModelFiles += $file.Name
}
}
if ($hasPinnedVersion) {
Write-Check "Model version pinned (date-stamped)" "PASS"
} elseif ($hasGenericModel) {
$fileList = ($genericModelFiles | Select-Object -Unique) -join ", "
Write-Check "Model version pinned" "WARN" "Generic model names found in: $fileList. Pin with date suffix (e.g., gpt-5.1-2026-01-15)"
} else {
Write-Check "Model version pinned" "WARN" "No model references found to validate"
}
# Check: Model config is externalized (env var or config file)
$modelExternalized = $false
foreach ($file in $allSource) {
$content = Get-Content $file.FullName -Raw -ErrorAction SilentlyContinue
if (-not $content) { continue }
if ($content -match '(os\.getenv\([''"].*MODEL|Environment\.GetEnvironmentVariable\([''"].*MODEL|config\[.*model|\.env.*MODEL)') {
$modelExternalized = $true
break
}
}
if ($modelExternalized) {
Write-Check "Model config externalized (env/config)" "PASS"
} elseif ($hasGenericModel) {
Write-Check "Model config externalized" "FAIL" "Model name appears hardcoded. Use env vars (AGENT_MODEL, FOUNDRY_MODEL)"
} else {
Write-Check "Model config externalized" "WARN" "Could not determine model configuration approach"
}
# Check: Evaluation baseline exists
$baselineFiles = Get-ChildItem -Path $Root -Include "baseline*.json","baseline*.jsonl","eval-baseline*" -Recurse -ErrorAction SilentlyContinue |
Where-Object { $_.FullName -notmatch '(\.venv|node_modules|\.git)' }
if ($baselineFiles.Count -gt 0) {
Write-Check "Evaluation baseline file exists ($($baselineFiles.Count) found)" "PASS"
} else {
Write-Check "Evaluation baseline file" "FAIL" "No baseline*.json found. Run eval and save scores before deploying"
}
# Check: Model migration documented
$hasMigrationDocs = $false
foreach ($file in ($allSource + $configFiles)) {
$content = Get-Content $file.FullName -Raw -ErrorAction SilentlyContinue
if (-not $content) { continue }
if ($content -match '(last_evaluated|model_migration|migration_notes|model.*changelog|model.*history)') {
$hasMigrationDocs = $true
break
}
}
if ($hasMigrationDocs) {
Write-Check "Model migration tracking documented" "PASS"
} else {
Write-Check "Model migration tracking" "WARN" "Add last_evaluated date and migration notes to model config"
}
# ================================================================
# SECTION 2: DATA DRIFT DETECTION
# ================================================================
Write-Host ""
Write-Host "--- 2. Data Drift Detection ---" -ForegroundColor White
# Check: Input logging present
$hasInputLogging = $false
foreach ($file in $allSource) {
$content = Get-Content $file.FullName -Raw -ErrorAction SilentlyContinue
if (-not $content) { continue }
if ($content -match '(log.*query|log.*input|log.*request|logger.*user.*message|log_input|track_input|record_query)') {
$hasInputLogging = $true
break
}
}
if ($hasInputLogging) {
Write-Check "Input logging implemented" "PASS"
} elseif ($allSource.Count -gt 0) {
Write-Check "Input logging" "FAIL" "Log all production inputs (query, timestamp, response latency)"
} else {
Write-Check "Input logging" "WARN" "No source files to check"
}
# Check: Evaluation dataset exists and is fresh
$evalDatasets = Get-ChildItem -Path $Root -Include "*.jsonl" -Recurse -ErrorAction SilentlyContinue |
Where-Object { $_.FullName -notmatch '(\.venv|node_modules|\.git)' -and $_.Name -match '(eval|test|dataset)' }
if ($evalDatasets.Count -gt 0) {
$oldest = $evalDatasets | Sort-Object LastWriteTime | Select-Object -First 1
$ageInDays = ((Get-Date) - $oldest.LastWriteTime).Days
if ($ageInDays -gt 90) {
Write-Check "Evaluation dataset freshness" "WARN" "$($oldest.Name) is $ageInDays days old. Update with recent production samples (target: < 90 days)"
} else {
Write-Check "Evaluation dataset exists and fresh ($ageInDays days old)" "PASS"
}
} else {
Write-Check "Evaluation dataset" "FAIL" "No eval/test dataset found. Create evaluation/*.jsonl with representative queries"
}
# Check: Drift monitoring code
$hasDriftMonitoring = $false
foreach ($file in $allSource) {
$content = Get-Content $file.FullName -Raw -ErrorAction SilentlyContinue
if (-not $content) { continue }
if ($content -match '(drift|distribution.*compare|novelty.*detect|topic.*cluster|input.*stats|query.*stats|DriftDetector)') {
$hasDriftMonitoring = $true
break
}
}
if ($hasDriftMonitoring) {
Write-Check "Drift monitoring implemented" "PASS"
} else {
Write-Check "Drift monitoring" "WARN" "Implement input distribution tracking to detect data drift over time"
}
# Check: Out-of-domain handling
$hasOODHandling = $false
foreach ($file in $allSource) {
$content = Get-Content $file.FullName -Raw -ErrorAction SilentlyContinue
if (-not $content) { continue }
if ($content -match '(out.*of.*domain|unsupported.*topic|I.*can.*not.*help|outside.*scope|guardrail|boundary.*check|off.topic)') {
$hasOODHandling = $true
break
}
}
if ($hasOODHandling) {
Write-Check "Out-of-domain handling" "PASS"
} else {
Write-Check "Out-of-domain handling" "WARN" "Add guardrails for inputs outside agent's intended scope"
}
# ================================================================
# SECTION 3: JUDGE LLM IMPLEMENTATION
# ================================================================
Write-Host ""
Write-Host "--- 3. Judge LLM Implementation ---" -ForegroundColor White
# Check: Evaluator/judge exists
$hasEvaluator = $false
$hasCustomJudge = $false
foreach ($file in $allSource) {
$content = Get-Content $file.FullName -Raw -ErrorAction SilentlyContinue
if (-not $content) { continue }
if ($content -match '(Evaluator|evaluator|evaluate\(|evals\.create|judge|builtin\.\w+)') {
$hasEvaluator = $true
}
if ($content -match '(PROMPT.*type|prompt_text.*Rate|rubric|scoring.*criteria|score.*1.*5|judge.*prompt)') {
$hasCustomJudge = $true
}
}
if ($hasEvaluator) {
Write-Check "Evaluator/judge configured" "PASS"
} else {
Write-Check "Evaluator/judge" "FAIL" "No evaluation setup found. Implement LLM-as-judge or use builtin evaluators"
}
# Check: Judge has structured rubric (not just "rate 1-5")
if ($hasCustomJudge) {
Write-Check "Judge rubric defined (structured scoring criteria)" "PASS"
} elseif ($hasEvaluator) {
Write-Check "Judge rubric" "WARN" "Using evaluator without visible custom rubric. Ensure scoring criteria are explicit"
} else {
Write-Check "Judge rubric" "WARN" "No judge rubric found. Define what each score level means"
}
# Check: Judge validation data (known-answer set)
$judgeValidation = Get-ChildItem -Path $Root -Include "*.jsonl","*.json" -Recurse -ErrorAction SilentlyContinue |
Where-Object { $_.FullName -notmatch '(\.venv|node_modules|\.git)' -and $_.Name -match '(judge.*valid|gold.*standard|known.*answer|judge.*test|annotated)' }
if ($judgeValidation.Count -gt 0) {
Write-Check "Judge validation dataset (known-answer set)" "PASS"
} else {
Write-Check "Judge validation dataset" "WARN" "Create judge-validation.jsonl with 20-30 human-scored examples to validate judge accuracy"
}
# Check: Different model for judge vs agent
$useDifferentJudgeModel = $false
foreach ($file in $allSource) {
$content = Get-Content $file.FullName -Raw -ErrorAction SilentlyContinue
if (-not $content) { continue }
if ($content -match '(judge.*model|JUDGE_MODEL|evaluator.*model|secondary.*model|judge_deployment)') {
$useDifferentJudgeModel = $true
break
}
}
if ($useDifferentJudgeModel) {
Write-Check "Separate model for judge vs agent" "PASS"
} elseif ($hasEvaluator) {
Write-Check "Separate judge model" "WARN" "Consider using a different model for evaluation to avoid self-evaluation bias"
} else {
Write-Check "Separate judge model" "WARN" "No evaluator found to check"
}
# Check: Multi-dimensional evaluation (not just one score)
$hasMultiDimEval = $false
foreach ($file in $allSource) {
$content = Get-Content $file.FullName -Raw -ErrorAction SilentlyContinue
if (-not $content) { continue }
# Look for multiple evaluator dimensions
$dimensions = 0
if ($content -match 'coherence') { $dimensions++ }
if ($content -match 'relevance') { $dimensions++ }
if ($content -match 'fluency') { $dimensions++ }
if ($content -match 'groundedness') { $dimensions++ }
if ($content -match 'accuracy') { $dimensions++ }
if ($content -match 'completeness') { $dimensions++ }
if ($content -match 'helpfulness') { $dimensions++ }
if ($content -match 'task_completion') { $dimensions++ }
if ($dimensions -ge 2) {
$hasMultiDimEval = $true
break
}
}
if ($hasMultiDimEval) {
Write-Check "Multi-dimensional evaluation (2+ metrics)" "PASS"
} elseif ($hasEvaluator) {
Write-Check "Multi-dimensional evaluation" "WARN" "Add multiple evaluation dimensions (accuracy, completeness, helpfulness, etc.)"
} else {
Write-Check "Multi-dimensional evaluation" "WARN" "No evaluator to analyze"
}
# ================================================================
# SUMMARY
# ================================================================
Write-Host ""
Write-Host "============================================================" -ForegroundColor Cyan
Write-Host " Results: $($script:Passed) passed, $($script:Warned) warnings, $($script:Failed) failed" -ForegroundColor White
Write-Host "============================================================" -ForegroundColor Cyan
$exitCode = $script:Failed
if ($Strict) { $exitCode += $script:Warned }
if ($exitCode -eq 0) {
Write-Host " Agent is resilient to model change, drift, and judge issues!" -ForegroundColor Green
} elseif ($script:Failed -eq 0) {
Write-Host " No critical failures, but address warnings for production hardening." -ForegroundColor Yellow
} else {
Write-Host " Critical gaps found. Address FAIL items before production deployment." -ForegroundColor Red
}
Write-Host ""
Write-Host " Reference: model-drift-judge-patterns.md" -ForegroundColor DarkGray
Write-Host ""
exit $exitCode
scripts/run-model-comparison.py
#!/usr/bin/env python3
"""Run evaluation suite against multiple models and generate comparison report.
Reads model configuration from config/models.yaml (or --config), runs the same
evaluation dataset against each model, and produces a comparison report in JSON
and Markdown.
Usage:
python run-model-comparison.py
python run-model-comparison.py --config path/to/models.yaml --dataset evaluation/core.jsonl
python run-model-comparison.py --results-dir evaluation/results/
python run-model-comparison.py --results-dir evaluation/results/ --check-gates
python run-model-comparison.py --results-dir evaluation/results/ --check-gates --fail-on-regression
Requirements:
pip install pyyaml
pip install agent-framework-azure-ai # only needed for --run mode
"""
from __future__ import annotations
import argparse
import json
import os
import sys
import time
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
@dataclass
class ModelSpec:
"""One model in the comparison matrix."""
name: str
deployment: str
role: str
provider: str = "azure"
@dataclass
class Thresholds:
"""Minimum pass criteria."""
task_completion: float = 0.85
coherence: float = 3.5
relevance: float = 3.5
format_compliance: float = 0.95
tool_accuracy: float = 0.90
max_latency_ms: float = 5000
max_cost_per_1k: float = 15.0
max_regression_pct: float = 10.0
@dataclass
class ModelResult:
"""Aggregated scores for one model."""
name: str
role: str
dataset_size: int = 0
task_completion: float = 0.0
coherence: float = 0.0
relevance: float = 0.0
format_compliance: float = 0.0
tool_accuracy: float = 0.0
avg_latency_ms: float = 0.0
avg_tokens: float = 0.0
estimated_cost_per_1k: float = 0.0
passed: bool = True
failures: list[str] = field(default_factory=list)
def load_config(path: str) -> dict[str, Any]:
"""Load models.yaml configuration."""
try:
import yaml
except ImportError:
print("ERROR: PyYAML required. Install: pip install pyyaml", file=sys.stderr)
sys.exit(1)
config_path = Path(path)
if not config_path.exists():
print(f"ERROR: Config not found: {path}", file=sys.stderr)
print(
"Create config/models.yaml with model matrix. "
"See model-change-test-automation.md",
file=sys.stderr,
)
sys.exit(1)
with open(config_path, encoding="utf-8") as config_file:
return yaml.safe_load(config_file)
def load_thresholds(config: dict) -> Thresholds:
"""Extract thresholds from config, with defaults."""
raw = config.get("thresholds", {})
return Thresholds(
**{key: value for key, value in raw.items() if hasattr(Thresholds, key)}
)
def load_models(config: dict) -> list[ModelSpec]:
"""Extract model specs from config."""
models = []
for role, spec in config.get("models", {}).items():
models.append(
ModelSpec(
name=spec["name"],
deployment=spec.get("deployment", spec["name"]),
role=role,
provider=spec.get("provider", "azure"),
)
)
return models
def load_dataset(path: str) -> list[dict]:
"""Load JSONL evaluation dataset."""
dataset_path = Path(path)
if not dataset_path.exists():
print(f"ERROR: Dataset not found: {path}", file=sys.stderr)
sys.exit(1)
items = []
with open(dataset_path, encoding="utf-8") as dataset_file:
for lineno, line in enumerate(dataset_file, 1):
line = line.strip()
if not line:
continue
try:
items.append(json.loads(line))
except json.JSONDecodeError as error:
print(
f"WARNING: Invalid JSON on line {lineno}: {error}",
file=sys.stderr,
)
return items
async def run_single_model(
model: ModelSpec,
dataset: list[dict],
system_prompt: str,
) -> list[dict]:
"""Run dataset through a single model. Returns per-query results."""
try:
from agent_framework.openai import OpenAIChatClient
except ImportError:
print(
"ERROR: agent-framework not installed. "
"Use --results-dir to compare pre-existing results.",
file=sys.stderr,
)
sys.exit(1)
client = OpenAIChatClient(
model=model.deployment,
api_key=os.getenv("FOUNDRY_API_KEY", ""),
endpoint=os.getenv("FOUNDRY_ENDPOINT", ""),
)
results = []
for index, item in enumerate(dataset):
query = item.get("query", item.get("input", ""))
start = time.perf_counter()
try:
response = await client.chat(
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": query},
]
)
elapsed_ms = (time.perf_counter() - start) * 1000
content = (
response.content if hasattr(response, "content") else str(response)
)
tokens = (
getattr(response, "usage", {}).get("total_tokens", 0)
if hasattr(response, "usage")
else 0
)
results.append(
{
"index": index,
"query": query,
"response": content,
"expected": item.get("expected_response", item.get("response", "")),
"latency_ms": round(elapsed_ms),
"tokens_used": tokens,
"error": None,
}
)
except Exception as error:
elapsed_ms = (time.perf_counter() - start) * 1000
results.append(
{
"index": index,
"query": query,
"response": "",
"expected": item.get("expected_response", ""),
"latency_ms": round(elapsed_ms),
"tokens_used": 0,
"error": str(error),
}
)
if (index + 1) % 10 == 0 or index == len(dataset) - 1:
print(f" [{model.name}] {index + 1}/{len(dataset)} queries complete")
return results
async def run_all_models(
models: list[ModelSpec],
dataset: list[dict],
output_dir: Path,
system_prompt: str = "You are a helpful assistant.",
) -> None:
"""Run evaluation for all models and save results."""
output_dir.mkdir(parents=True, exist_ok=True)
for model in models:
print(f"\n{'=' * 60}")
print(f" Evaluating: {model.name} ({model.role})")
print(f"{'=' * 60}")
results = await run_single_model(model, dataset, system_prompt)
safe_name = model.name.replace("/", "-").replace(" ", "-")
output_file = output_dir / f"{safe_name}.json"
with open(output_file, "w", encoding="utf-8") as result_file:
json.dump(
{
"model": model.name,
"role": model.role,
"provider": model.provider,
"timestamp": datetime.now(timezone.utc).isoformat(),
"dataset_size": len(dataset),
"results": results,
},
result_file,
indent=2,
)
print(f" Saved: {output_file}")
def aggregate_scores(data: dict) -> ModelResult:
"""Calculate aggregate metrics from raw per-query results."""
results = data.get("results", [])
total = len(results)
if total == 0:
return ModelResult(name=data["model"], role=data.get("role", "unknown"))
successful = [result for result in results if not result.get("error")]
success_count = len(successful) if successful else 1
return ModelResult(
name=data["model"],
role=data.get("role", "unknown"),
dataset_size=total,
task_completion=round(len(successful) / total, 3),
format_compliance=round(
sum(
1
for result in successful
if len(result.get("response", "").strip()) > 10
)
/ success_count,
3,
),
avg_latency_ms=round(
sum(result["latency_ms"] for result in results) / total, 1
),
avg_tokens=round(
sum(result.get("tokens_used", 0) for result in successful) / success_count,
1,
),
)
def compare_models(results_dir: str, thresholds: Thresholds) -> dict[str, Any]:
"""Load all result files and generate comparison report."""
results_path = Path(results_dir)
if not results_path.exists():
print(f"ERROR: Results directory not found: {results_dir}", file=sys.stderr)
sys.exit(1)
result_files = list(results_path.glob("*.json"))
if not result_files:
print(f"ERROR: No .json result files in {results_dir}", file=sys.stderr)
sys.exit(1)
models: list[ModelResult] = []
alerts: list[str] = []
for result_file in sorted(result_files):
with open(result_file, encoding="utf-8") as file_handle:
data = json.load(file_handle)
scores = aggregate_scores(data)
checks = [
(
"task_completion",
scores.task_completion,
thresholds.task_completion,
"min",
),
(
"format_compliance",
scores.format_compliance,
thresholds.format_compliance,
"min",
),
("avg_latency_ms", scores.avg_latency_ms, thresholds.max_latency_ms, "max"),
]
for metric, value, threshold, direction in checks:
if direction == "min" and value < threshold:
msg = (
f"{scores.name}: {metric} = {value:.3f} "
f"below threshold {threshold}"
)
alerts.append(msg)
scores.failures.append(msg)
scores.passed = False
elif direction == "max" and value > threshold:
msg = (
f"{scores.name}: {metric} = {value:.1f} "
f"exceeds threshold {threshold}"
)
alerts.append(msg)
scores.failures.append(msg)
scores.passed = False
models.append(scores)
winners = {}
if models:
viable = [model for model in models if model.passed] or models
winners["task_completion"] = max(
viable, key=lambda model: model.task_completion
).name
winners["format_compliance"] = max(
viable, key=lambda model: model.format_compliance
).name
winners["latency"] = min(viable, key=lambda model: model.avg_latency_ms).name
winners["token_efficiency"] = min(
viable,
key=lambda model: model.avg_tokens or float("inf"),
).name
return {
"report_id": f"compare-{datetime.now(timezone.utc).strftime('%Y%m%d-%H%M%S')}",
"timestamp": datetime.now(timezone.utc).isoformat(),
"models_tested": len(models),
"models": [asdict(model) for model in models],
"winner_by_metric": winners,
"alerts": alerts,
"all_passed": all(model.passed for model in models),
}
def generate_markdown_report(report: dict) -> str:
"""Generate human-readable Markdown comparison report."""
lines = [
"# Model Comparison Report",
"",
f"**Generated**: {report['timestamp']} ",
f"**Report ID**: {report['report_id']} ",
f"**Models Tested**: {report['models_tested']}",
"",
"## Results",
"",
"| Model | Role | Task Completion | Format Compliance | Avg Latency | Avg Tokens | Status |",
"|-------|------|---------------:|------------------:|------------:|-----------:|--------|",
]
for model in report["models"]:
status = "[PASS] PASS" if model["passed"] else "[FAIL] FAIL"
lines.append(
f"| {model['name']} | {model['role']} | {model['task_completion']:.3f} | "
f"{model['format_compliance']:.3f} | {model['avg_latency_ms']:.0f}ms | "
f"{model['avg_tokens']:.0f} | {status} |"
)
if report.get("winner_by_metric"):
lines.extend(["", "## Best By Metric", ""])
for metric, winner in report["winner_by_metric"].items():
lines.append(f"- **{metric}**: {winner}")
if report.get("alerts"):
lines.extend(["", "## Alerts", ""])
for alert in report["alerts"]:
lines.append(f"- [WARN] {alert}")
lines.append("")
if report["all_passed"]:
lines.append("## Verdict: [PASS] All models meet minimum thresholds")
else:
failed = [model["name"] for model in report["models"] if not model["passed"]]
lines.append(
f"## Verdict: [FAIL] {len(failed)} model(s) failed threshold checks"
)
for name in failed:
lines.append(f" - {name}")
lines.append("")
return "\n".join(lines)
def save_reports(report: dict, output_dir: str) -> None:
"""Save JSON and Markdown reports."""
output = Path(output_dir)
output.mkdir(parents=True, exist_ok=True)
json_path = output / "comparison-report.json"
with open(json_path, "w", encoding="utf-8") as json_file:
json.dump(report, json_file, indent=2)
print(f" JSON report: {json_path}")
md_path = output / "comparison-report.md"
md_content = generate_markdown_report(report)
with open(md_path, "w", encoding="utf-8") as markdown_file:
markdown_file.write(md_content)
print(f" Markdown report: {md_path}")
def print_summary(report: dict) -> None:
"""Print summary to terminal."""
print("\n" + "=" * 60)
print(" MODEL COMPARISON SUMMARY")
print("=" * 60)
for model in report["models"]:
status = "PASS" if model["passed"] else "FAIL"
print(f"\n [{status}] {model['name']} ({model['role']})")
print(f" Task Completion: {model['task_completion']:.3f}")
print(f" Format Compliance: {model['format_compliance']:.3f}")
print(f" Avg Latency: {model['avg_latency_ms']:.0f}ms")
print(f" Avg Tokens: {model['avg_tokens']:.0f}")
if model["failures"]:
for failure in model["failures"]:
print(f" [!] {failure}")
if report.get("alerts"):
print(f"\n ALERTS ({len(report['alerts'])})")
for alert in report["alerts"]:
print(f" - {alert}")
print("\n" + "=" * 60)
if report["all_passed"]:
print(" [PASS] All models meet minimum thresholds")
else:
print(" [FAIL] Some models failed threshold checks")
print("=" * 60 + "\n")
def check_regression(
report: dict,
baseline_path: str | None,
max_regression_pct: float,
) -> bool:
"""Check if primary model regressed from baseline. Returns True if OK."""
if not baseline_path:
for candidate in [
"evaluation/baseline.json",
"baseline.json",
"evaluation/results/baseline.json",
]:
if Path(candidate).exists():
baseline_path = candidate
break
if not baseline_path or not Path(baseline_path).exists():
print(" No baseline found - skipping regression check")
return True
with open(baseline_path, encoding="utf-8") as baseline_file:
baseline = json.load(baseline_file)
primary = next(
(model for model in report["models"] if model["role"] == "primary"), None
)
if not primary:
print(" No primary model found in results - skipping regression check")
return True
baseline_scores = baseline.get("scores", baseline)
regression_found = False
for metric in ["task_completion", "format_compliance"]:
baseline_val = baseline_scores.get(metric, 0)
current_val = primary.get(metric, 0)
if baseline_val > 0:
drop_pct = ((baseline_val - current_val) / baseline_val) * 100
if drop_pct > max_regression_pct:
print(
f" REGRESSION: {metric} dropped {drop_pct:.1f}% "
f"(baseline: {baseline_val:.3f} -> current: {current_val:.3f})"
)
regression_found = True
else:
print(
f" {metric}: {current_val:.3f} "
f"(baseline: {baseline_val:.3f}) - OK"
)
return not regression_found
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="Run multi-model evaluation comparison for AI agents",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
python run-model-comparison.py --config config/models.yaml --dataset evaluation/core-regression.jsonl
python run-model-comparison.py --results-dir evaluation/results/
python run-model-comparison.py --results-dir evaluation/results/ --check-gates --fail-on-regression
""",
)
parser.add_argument(
"--config",
default="config/models.yaml",
help="Path to model matrix config (default: config/models.yaml)",
)
parser.add_argument(
"--dataset",
default="evaluation/core-regression.jsonl",
help="Path to evaluation dataset JSONL",
)
parser.add_argument(
"--results-dir",
default="evaluation/results",
help="Directory for per-model result files",
)
parser.add_argument(
"--output-dir",
default="evaluation",
help="Directory for comparison reports",
)
parser.add_argument(
"--system-prompt", help="System prompt for the agent (or path to .txt file)"
)
parser.add_argument(
"--check-gates",
action="store_true",
help="Check thresholds and exit with code 1 on failure",
)
parser.add_argument(
"--fail-on-regression",
action="store_true",
help="Exit 1 if primary model regressed from baseline",
)
parser.add_argument("--baseline", default=None, help="Path to baseline scores JSON")
parser.add_argument(
"--skip-eval",
action="store_true",
help="Skip evaluation, only compare existing results",
)
return parser.parse_args()
def main() -> int:
args = parse_args()
exit_code = 0
config = {}
config_path = Path(args.config)
if config_path.exists():
config = load_config(args.config)
thresholds = load_thresholds(config)
if not args.skip_eval and config.get("models") and Path(args.dataset).exists():
import asyncio
models = load_models(config)
dataset = load_dataset(args.dataset)
system_prompt = "You are a helpful assistant."
if args.system_prompt:
prompt_path = Path(args.system_prompt)
if prompt_path.exists():
system_prompt = prompt_path.read_text(encoding="utf-8")
else:
system_prompt = args.system_prompt
print(f"\nRunning evaluation: {len(models)} models {len(dataset)} queries")
asyncio.run(
run_all_models(
models=models,
dataset=dataset,
output_dir=Path(args.results_dir),
system_prompt=system_prompt,
)
)
results_path = Path(args.results_dir)
if results_path.exists() and list(results_path.glob("*.json")):
print("\nGenerating comparison report...")
report = compare_models(args.results_dir, thresholds)
save_reports(report, args.output_dir)
print_summary(report)
if args.check_gates and not report["all_passed"]:
print("GATE CHECK FAILED: Not all models meet thresholds")
exit_code = 1
if args.fail_on_regression:
if not check_regression(
report, args.baseline, thresholds.max_regression_pct
):
print("REGRESSION CHECK FAILED: Primary model regressed from baseline")
exit_code = 1
else:
print(f"\nNo results found in {args.results_dir}")
print("Run with a valid --config and --dataset to generate results,")
print("or provide --results-dir pointing to existing .json result files.")
exit_code = 1
return exit_code
if __name__ == "__main__":
sys.exit(main())
scripts/scaffold-agent.py
#!/usr/bin/env python3
"""Scaffold an AI agent project with Agent Framework boilerplate.
Generates a production-ready agent project structure with:
- Agent Framework client setup (Python or .NET)
- OpenTelemetry tracing configuration
- Evaluation harness template
- Environment variable template (.env)
- Project configuration (pyproject.toml or .csproj)
Usage:
python scaffold-agent.py --name my-agent
python scaffold-agent.py --name my-agent --runtime dotnet
python scaffold-agent.py --name my-agent --pattern multi-agent
python scaffold-agent.py --name my-agent --with-eval --with-mcp
"""
import argparse
import os
import sys
from pathlib import Path
from datetime import datetime
def create_file(path: Path, content: str) -> None:
"""Create a file with content, making parent directories as needed."""
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(content, encoding="utf-8")
print(f" Created: {path}")
def scaffold_python_agent(
root: Path, name: str, pattern: str, with_eval: bool, with_mcp: bool
) -> None:
"""Generate Python agent project structure."""
# pyproject.toml
deps = [
'"agent-framework-azure-ai>=0.1.0"',
'"azure-identity>=1.15.0"',
'"opentelemetry-api>=1.20.0"',
'"opentelemetry-sdk>=1.20.0"',
'"python-dotenv>=1.0.0"',
]
if with_eval:
deps.append('"azure-ai-evaluation>=1.0.0"')
if with_mcp:
deps.append('"agent-framework-mcp>=0.1.0"')
deps_str = ",\n ".join(deps)
create_file(
root / "pyproject.toml",
f"""[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project]
name = "{name}"
version = "0.1.0"
description = "AI Agent built with Microsoft Agent Framework"
requires-python = ">=3.11"
dependencies = [
{deps_str},
]
[project.optional-dependencies]
dev = [
"pytest>=8.0",
"pytest-asyncio>=0.23.0",
"ruff>=0.4.0",
]
[tool.ruff]
target-version = "py311"
line-length = 120
[tool.ruff.lint]
select = ["E", "F", "I", "UP", "B", "SIM"]
[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
""",
)
# .env template
create_file(
root / ".env.template",
"""# AI Agent Environment Variables
# Copy to .env and fill in values
# Microsoft Foundry / Azure OpenAI
FOUNDRY_ENDPOINT=https://your-project.services.ai.azure.com
FOUNDRY_API_KEY=your-api-key-here
FOUNDRY_MODEL=gpt-5.1
# OpenTelemetry (optional - for tracing)
OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317
OTEL_SERVICE_NAME={name}
# Application
LOG_LEVEL=INFO
MAX_TURNS=10
""",
)
# .env (gitignored placeholder)
create_file(
root / ".gitignore",
"""# Environment
.env
.env.local
# Python
__pycache__/
*.py[cod]
*.egg-info/
dist/
build/
.venv/
venv/
# IDE
.vscode/
.idea/
# Traces and outputs
traces/
outputs/
""",
)
# Main agent module
if pattern == "single":
agent_code = _python_single_agent(name)
elif pattern == "multi-agent":
agent_code = _python_multi_agent(name)
elif pattern == "sequential":
agent_code = _python_sequential_workflow(name)
else:
agent_code = _python_single_agent(name)
create_file(root / "src" / name.replace("-", "_") / "__init__.py", "")
create_file(root / "src" / name.replace("-", "_") / "agent.py", agent_code)
# Tracing setup
create_file(
root / "src" / name.replace("-", "_") / "tracing.py",
f"""\"\"\"OpenTelemetry tracing configuration for {name}.\"\"\"
import os
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
# Agent Framework auto-instrumentation
from agent_framework.openai import AIInferenceInstrumentor
def setup_tracing() -> None:
\"\"\"Initialize OpenTelemetry tracing with Agent Framework instrumentation.
Call this BEFORE creating any agent or client instances.
\"\"\"
provider = TracerProvider()
# Configure exporter based on environment
endpoint = os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT")
if endpoint:
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
exporter = OTLPSpanExporter(endpoint=endpoint)
provider.add_span_processor(SimpleSpanProcessor(exporter))
trace.set_tracer_provider(provider)
# Instrument Agent Framework (auto-captures LLM calls)
AIInferenceInstrumentor().instrument()
print(f"Tracing initialized for {{os.getenv('OTEL_SERVICE_NAME', '{name}')}}")
""",
)
# Main entry point
create_file(
root / "src" / name.replace("-", "_") / "main.py",
f"""\"\"\"Entry point for {name} agent.\"\"\"
import asyncio
import os
from dotenv import load_dotenv
from .tracing import setup_tracing
from .agent import run_agent
async def main() -> None:
\"\"\"Initialize and run the agent.\"\"\"
load_dotenv()
setup_tracing()
query = os.getenv("AGENT_QUERY", "Hello! What can you help me with?")
result = await run_agent(query)
print(f"\\nAgent response:\\n{{result}}")
if __name__ == "__main__":
asyncio.run(main())
""",
)
# Tests
create_file(root / "tests" / "__init__.py", "")
create_file(
root / "tests" / "test_agent.py",
f"""\"\"\"Tests for {name} agent.\"\"\"
import pytest
@pytest.mark.asyncio
async def test_agent_responds():
\"\"\"Agent should return a non-empty response.\"\"\"
from {name.replace("-", "_")}.agent import run_agent
# Note: Requires valid credentials in .env
# For CI, mock the client or use a test endpoint
# result = await run_agent("Hello")
# assert result is not None
# assert len(result) > 0
pytest.skip("Requires valid AI endpoint credentials")
""",
)
# Evaluation harness (optional)
if with_eval:
create_file(
root / "evaluation" / "evaluate.py",
f"""\"\"\"Evaluation harness for {name} agent.\"\"\"
import json
from pathlib import Path
from azure.ai.evaluation import evaluate
from azure.ai.evaluation import (
CoherenceEvaluator,
FluencyEvaluator,
GroundednessEvaluator,
RelevanceEvaluator,
)
def run_evaluation():
\"\"\"Run evaluation against test dataset.\"\"\"
# Load test dataset
dataset_path = Path(__file__).parent / "test_dataset.jsonl"
if not dataset_path.exists():
print("Create evaluation/test_dataset.jsonl with test cases first.")
print("Format: {{\\"query\\": \\"...\\"," "\\"expected\\": \\"...\\"," "\\"context\\": \\"...\\"}}")
return
results = evaluate(
data=str(dataset_path),
evaluators={{
"coherence": CoherenceEvaluator(),
"fluency": FluencyEvaluator(),
"groundedness": GroundednessEvaluator(),
"relevance": RelevanceEvaluator(),
}},
output_path="evaluation/results.json",
)
print("Evaluation Results:")
print(json.dumps(results, indent=2))
if __name__ == "__main__":
run_evaluation()
""",
)
create_file(
root / "evaluation" / "test_dataset.jsonl",
"""{"query": "What is the capital of France?", "expected": "Paris", "context": "France is a country in Europe."}
{"query": "What is 2 + 2?", "expected": "4", "context": "Basic arithmetic question."}
""",
)
# MCP server template (optional)
if with_mcp:
create_file(
root / "src" / name.replace("-", "_") / "mcp_tools.py",
f"""\"\"\"MCP tool definitions for {name} agent.\"\"\"
from agent_framework.mcp import MCPServer, tool
class {name.replace("-", "").title()}Tools(MCPServer):
\"\"\"MCP tools exposed by this agent.\"\"\"
@tool(description="Example tool that echoes input")
async def echo(self, message: str) -> str:
\"\"\"Echo the input message back.\"\"\"
return f"Echo: {{message}}"
@tool(description="Get current timestamp")
async def get_timestamp(self) -> str:
\"\"\"Return current UTC timestamp.\"\"\"
from datetime import datetime, timezone
return datetime.now(timezone.utc).isoformat()
""",
)
# README
create_file(
root / "README.md",
f"""# {name}
AI Agent built with [Microsoft Agent Framework](https://github.com/microsoft/agent-framework).
## Setup
```bash
# Create virtual environment
python -m venv .venv
source .venv/bin/activate # Linux/Mac
# .venv\\Scripts\\activate # Windows
# Install dependencies
pip install -e ".[dev]"
# Configure environment
cp .env.template .env
# Edit .env with your Foundry credentials
```
## Run
```bash
python -m {name.replace("-", "_")}.main
```
## Test
```bash
pytest
```
{"## Evaluate" if with_eval else ""}
{"```bash" if with_eval else ""}
{"python evaluation/evaluate.py" if with_eval else ""}
{"```" if with_eval else ""}
## Architecture
- **Pattern**: {pattern}
- **Runtime**: Python 3.11+
- **Framework**: Microsoft Agent Framework
- **Tracing**: OpenTelemetry (auto-instrumented)
{"- **MCP**: Tool server enabled" if with_mcp else ""}
""",
)
def _python_single_agent(name: str) -> str:
"""Generate single agent pattern code."""
return f"""\"\"\"Single agent implementation for {name}.\"\"\"
import os
from agent_framework.openai import OpenAIChatClient
async def run_agent(query: str) -> str:
\"\"\"Run the agent with a single query.\"\"\"
client = OpenAIChatClient(
model=os.getenv("FOUNDRY_MODEL", "gpt-5.1"),
api_key=os.getenv("FOUNDRY_API_KEY"),
endpoint=os.getenv("FOUNDRY_ENDPOINT"),
)
agent = {{
"name": "{name}",
"instructions": \"\"\"You are a helpful AI assistant.
TASK: Answer the user's question accurately and concisely.
CONSTRAINTS:
- Be factual and cite sources when possible
- Say "I don't know" if uncertain
- Keep responses under 500 words
\"\"\",
"tools": [],
}}
response = await client.chat(
messages=[{{"role": "user", "content": query}}],
agent=agent,
)
return response.content
"""
def _python_multi_agent(name: str) -> str:
"""Generate multi-agent pattern code."""
return f"""\"\"\"Multi-agent orchestration for {name}.\"\"\"
import os
from agent_framework.openai import OpenAIChatClient
from agent_framework.workflows import GroupChatWorkflow
async def run_agent(query: str) -> str:
\"\"\"Run multi-agent workflow with group chat orchestration.\"\"\"
client = OpenAIChatClient(
model=os.getenv("FOUNDRY_MODEL", "gpt-5.1"),
api_key=os.getenv("FOUNDRY_API_KEY"),
endpoint=os.getenv("FOUNDRY_ENDPOINT"),
)
researcher = {{
"name": "Researcher",
"instructions": "You research topics thoroughly. Provide factual information with sources.",
"tools": [],
}}
writer = {{
"name": "Writer",
"instructions": "You write clear, engaging content based on research provided.",
"tools": [],
}}
reviewer = {{
"name": "Reviewer",
"instructions": "You review content for accuracy, clarity, and completeness. Suggest improvements.",
"tools": [],
}}
workflow = GroupChatWorkflow(
agents=[researcher, writer, reviewer],
client=client,
max_turns=10,
termination_condition="approval",
)
result = await workflow.run(query=query)
return result.final_output
"""
def _python_sequential_workflow(name: str) -> str:
"""Generate sequential workflow pattern code."""
return f"""\"\"\"Sequential workflow for {name}.\"\"\"
import os
from agent_framework.openai import OpenAIChatClient
from agent_framework.workflows import SequentialWorkflow
async def run_agent(query: str) -> str:
\"\"\"Run agents in sequence: research -> analyze -> summarize.\"\"\"
client = OpenAIChatClient(
model=os.getenv("FOUNDRY_MODEL", "gpt-5.1"),
api_key=os.getenv("FOUNDRY_API_KEY"),
endpoint=os.getenv("FOUNDRY_ENDPOINT"),
)
researcher = {{
"name": "Researcher",
"instructions": "Gather comprehensive information about the topic.",
}}
analyzer = {{
"name": "Analyzer",
"instructions": "Analyze the research and identify key insights and patterns.",
}}
summarizer = {{
"name": "Summarizer",
"instructions": "Create a concise executive summary from the analysis.",
}}
workflow = SequentialWorkflow(
agents=[researcher, analyzer, summarizer],
client=client,
handoff_strategy="on_completion",
)
result = await workflow.run(query=query)
return result.final_output
"""
def scaffold_dotnet_agent(root: Path, name: str, pattern: str, with_eval: bool) -> None:
"""Generate .NET agent project structure."""
safe_name = name.replace("-", ".")
namespace = safe_name.replace(".", "").title()
# .csproj
create_file(
root / f"{safe_name}.csproj",
f"""<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>net9.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Microsoft.Agents.AI.AzureAI" Version="*-*" />
<PackageReference Include="Microsoft.Agents.AI.Workflows" Version="*-*" />
<PackageReference Include="Microsoft.Extensions.Configuration.Json" Version="9.*" />
<PackageReference Include="Microsoft.Extensions.Configuration.EnvironmentVariables" Version="9.*" />
<PackageReference Include="Azure.Identity" Version="1.*" />
<PackageReference Include="OpenTelemetry" Version="1.*" />
<PackageReference Include="OpenTelemetry.Exporter.OpenTelemetryProtocol" Version="1.*" />
</ItemGroup>
</Project>
""",
)
# Program.cs
create_file(
root / "Program.cs",
f"""using Microsoft.Agents.AI.AzureAI;
using OpenTelemetry;
using OpenTelemetry.Trace;
// Setup tracing
using var tracerProvider = Sdk.CreateTracerProviderBuilder()
.AddSource("{name}")
.AddOtlpExporter()
.Build();
// Configure client
var client = new OpenAIChatClient(
model: Environment.GetEnvironmentVariable("FOUNDRY_MODEL") ?? "gpt-5.1",
apiKey: Environment.GetEnvironmentVariable("FOUNDRY_API_KEY")!,
endpoint: new Uri(Environment.GetEnvironmentVariable("FOUNDRY_ENDPOINT")!)
);
// Run agent
var response = await client.ChatAsync(
messages: [new {{ Role = "user", Content = "Hello! What can you help me with?" }}],
agent: new
{{
Name = "{name}",
Instructions = "You are a helpful AI assistant.",
Tools = Array.Empty<object>()
}}
);
Console.WriteLine($"Response: {{response.Content}}");
""",
)
# appsettings.json
create_file(
root / "appsettings.json",
"""{
"Foundry": {
"Endpoint": "",
"Model": "gpt-5.1"
},
"Logging": {
"LogLevel": {
"Default": "Information"
}
}
}
""",
)
# .gitignore
create_file(
root / ".gitignore",
"""bin/
obj/
.vs/
*.user
appsettings.Development.json
.env
""",
)
# README
create_file(
root / "README.md",
f"""# {name}
AI Agent built with [Microsoft Agent Framework](https://github.com/microsoft/agent-framework) (.NET).
## Setup
```bash
dotnet restore
# Set environment variables or edit appsettings.json
```
## Run
```bash
dotnet run
```
""",
)
def main():
parser = argparse.ArgumentParser(
description="Scaffold an AI agent project with Agent Framework"
)
parser.add_argument("--name", required=True, help="Project name (kebab-case)")
parser.add_argument(
"--runtime",
choices=["python", "dotnet"],
default="python",
help="Runtime platform (default: python)",
)
parser.add_argument(
"--pattern",
choices=["single", "multi-agent", "sequential"],
default="single",
help="Agent pattern (default: single)",
)
parser.add_argument(
"--with-eval",
action="store_true",
help="Include evaluation harness template",
)
parser.add_argument(
"--with-mcp",
action="store_true",
help="Include MCP server template (Python only)",
)
parser.add_argument(
"--output",
type=str,
default=None,
help="Output directory (default: ./<name>)",
)
args = parser.parse_args()
root = Path(args.output or args.name).resolve()
if root.exists() and any(root.iterdir()):
print(f"Error: Directory '{root}' already exists and is not empty.")
sys.exit(1)
print(f"\nScaffolding AI agent project: {args.name}")
print(f" Runtime: {args.runtime}")
print(f" Pattern: {args.pattern}")
print(f" Output: {root}\n")
if args.runtime == "python":
scaffold_python_agent(
root, args.name, args.pattern, args.with_eval, args.with_mcp
)
else:
scaffold_dotnet_agent(root, args.name, args.pattern, args.with_eval)
print(f"\n[OK] Agent project scaffolded at: {root}")
print(f"\nNext steps:")
if args.runtime == "python":
print(f" cd {args.name}")
print(f" python -m venv .venv && .venv/Scripts/activate")
print(f" pip install -e '.[dev]'")
print(f" cp .env.template .env # Fill in credentials")
print(f" python -m {args.name.replace('-', '_')}.main")
else:
print(f" cd {args.name}")
print(f" dotnet restore")
print(f" # Set FOUNDRY_ENDPOINT and FOUNDRY_API_KEY env vars")
print(f" dotnet run")
if __name__ == "__main__":
main()
scripts/validate-agent-checklist.ps1
<#
.SYNOPSIS
Validates an AI agent project against the production checklist.
.DESCRIPTION
Programmatically checks the production readiness checklist from the
ai-agent-development SKILL.md. Scans for:
- Hardcoded secrets/API keys
- Tracing/observability setup
- Error handling patterns
- Evaluation dataset presence
- Security best practices
- Environment variable usage
.PARAMETER Path
Root of the agent project to validate. Defaults to current directory.
.PARAMETER Strict
Treat warnings as failures.
.EXAMPLE
./validate-agent-checklist.ps1
./validate-agent-checklist.ps1 -Path ./my-agent -Strict
#>
param(
[string]$Path = ".",
[switch]$Strict
)
$ErrorActionPreference = "Stop"
$script:Passed = 0
$script:Warned = 0
$script:Failed = 0
function Write-Check {
param([string]$Name, [string]$Status, [string]$Detail = "")
switch ($Status) {
"PASS" {
Write-Host " [PASS] $Name" -ForegroundColor Green
$script:Passed++
}
"WARN" {
Write-Host " [WARN] $Name" -ForegroundColor Yellow
if ($Detail) { Write-Host " $Detail" -ForegroundColor DarkYellow }
$script:Warned++
}
"FAIL" {
Write-Host " [FAIL] $Name" -ForegroundColor Red
if ($Detail) { Write-Host " $Detail" -ForegroundColor DarkRed }
$script:Failed++
}
}
}
$Root = Resolve-Path $Path -ErrorAction SilentlyContinue
if (-not $Root) {
Write-Host "Error: Path '$Path' not found." -ForegroundColor Red
exit 1
}
Write-Host ""
Write-Host "-----------------------------------------------------" -ForegroundColor Cyan
Write-Host " AI Agent Production Checklist Validator" -ForegroundColor Cyan
Write-Host " Path: $Root" -ForegroundColor DarkGray
Write-Host "-----------------------------------------------------" -ForegroundColor Cyan
# Collect all source files
$pyFiles = Get-ChildItem -Path $Root -Filter "*.py" -Recurse -ErrorAction SilentlyContinue | Where-Object { $_.FullName -notmatch '(\.venv|venv|__pycache__|node_modules|\.git)' }
$csFiles = Get-ChildItem -Path $Root -Filter "*.cs" -Recurse -ErrorAction SilentlyContinue | Where-Object { $_.FullName -notmatch '(bin|obj|\.git)' }
$allFiles = @($pyFiles) + @($csFiles) | Where-Object { $_ -ne $null }
# --- 1. Development Checks ---------------------------------------
Write-Host ""
Write-Host "[TASK] Development" -ForegroundColor White
# Check: No hardcoded secrets
$secretPatterns = @(
'api[_-]?key\s*=\s*["\x27][A-Za-z0-9]',
'password\s*=\s*["\x27][^$\{]',
'secret\s*=\s*["\x27][A-Za-z0-9]',
'sk-[A-Za-z0-9]{20,}',
'Bearer\s+[A-Za-z0-9\-._~+/]+=*'
)
$secretsFound = $false
foreach ($file in $allFiles) {
$content = Get-Content $file.FullName -Raw -ErrorAction SilentlyContinue
if (-not $content) { continue }
foreach ($pattern in $secretPatterns) {
if ($content -match $pattern) {
# Skip template/example files
if ($file.Name -match '\.(template|example|sample)') { continue }
if ($content -match '(your-api-key|placeholder|example|CHANGE_ME)') { continue }
$secretsFound = $true
break
}
}
}
if ($secretsFound) {
Write-Check "No hardcoded secrets" "FAIL" "Found potential hardcoded credentials in source files"
} else {
Write-Check "No hardcoded secrets" "PASS"
}
# Check: Error handling present
$hasErrorHandling = $false
foreach ($file in $allFiles) {
$content = Get-Content $file.FullName -Raw -ErrorAction SilentlyContinue
if (-not $content) { continue }
if ($content -match '(try\s*:|try\s*\{|except\s|catch\s*\(|retry|max_retries|RetryPolicy)') {
$hasErrorHandling = $true
break
}
}
if ($hasErrorHandling) {
Write-Check "Error handling with retries" "PASS"
} elseif ($allFiles.Count -gt 0) {
Write-Check "Error handling with retries" "WARN" "No try/catch or retry patterns found"
} else {
Write-Check "Error handling with retries" "WARN" "No source files found"
}
# Check: Environment variables for config
$usesEnvVars = $false
foreach ($file in $allFiles) {
$content = Get-Content $file.FullName -Raw -ErrorAction SilentlyContinue
if (-not $content) { continue }
if ($content -match '(os\.getenv|os\.environ|Environment\.GetEnvironmentVariable|\.env|dotenv|IConfiguration)') {
$usesEnvVars = $true
break
}
}
if ($usesEnvVars) {
Write-Check "Credentials via environment/config" "PASS"
} elseif ($allFiles.Count -gt 0) {
Write-Check "Credentials via environment/config" "FAIL" "No env var usage found - secrets may be hardcoded"
} else {
Write-Check "Credentials via environment/config" "WARN" "No source files to check"
}
# Check: .env.template or appsettings.json exists
$hasEnvTemplate = (Test-Path "$Root/.env.template") -or (Test-Path "$Root/.env.example") -or (Test-Path "$Root/appsettings.json")
if ($hasEnvTemplate) {
Write-Check "Environment template exists" "PASS"
} else {
Write-Check "Environment template exists" "WARN" "Add .env.template or appsettings.json for onboarding"
}
# --- 2. Observability Checks -------------------------------------
Write-Host ""
Write-Host "[OBS] Observability" -ForegroundColor White
# Check: Tracing setup
$hasTracing = $false
foreach ($file in $allFiles) {
$content = Get-Content $file.FullName -Raw -ErrorAction SilentlyContinue
if (-not $content) { continue }
if ($content -match '(opentelemetry|AIInferenceInstrumentor|TracerProvider|AddOtlpExporter|instrument\(\))') {
$hasTracing = $true
break
}
}
if ($hasTracing) {
Write-Check "OpenTelemetry tracing configured" "PASS"
} elseif ($allFiles.Count -gt 0) {
Write-Check "OpenTelemetry tracing configured" "FAIL" "No OpenTelemetry/tracing setup found - critical for debugging"
} else {
Write-Check "OpenTelemetry tracing configured" "WARN" "No source files to check"
}
# Check: Structured logging
$hasLogging = $false
foreach ($file in $allFiles) {
$content = Get-Content $file.FullName -Raw -ErrorAction SilentlyContinue
if (-not $content) { continue }
if ($content -match '(logging\.|Logger|ILogger|logger\.|log\.|structlog|serilog)') {
$hasLogging = $true
break
}
}
if ($hasLogging) {
Write-Check "Structured logging configured" "PASS"
} else {
Write-Check "Structured logging configured" "WARN" "Consider adding structured logging for production"
}
# --- 3. Evaluation Checks ----------------------------------------
Write-Host ""
Write-Host "[CHART] Evaluation" -ForegroundColor White
# Check: Evaluation dataset exists
$hasEvalData = (Test-Path "$Root/evaluation") -or
(Get-ChildItem -Path $Root -Filter "*.jsonl" -Recurse -ErrorAction SilentlyContinue | Where-Object { $_.Name -match '(eval|test|dataset)' }).Count -gt 0
if ($hasEvalData) {
Write-Check "Evaluation dataset exists" "PASS"
} else {
Write-Check "Evaluation dataset exists" "WARN" "Create evaluation/ directory with test_dataset.jsonl"
}
# Check: Evaluators configured
$hasEvaluators = $false
foreach ($file in $allFiles) {
$content = Get-Content $file.FullName -Raw -ErrorAction SilentlyContinue
if (-not $content) { continue }
if ($content -match '(Evaluator|evaluate\(|azure.ai.evaluation|EvaluatorConfig)') {
$hasEvaluators = $true
break
}
}
if ($hasEvaluators) {
Write-Check "Evaluators defined" "PASS"
} else {
Write-Check "Evaluators defined" "WARN" "Add evaluation with CoherenceEvaluator, RelevanceEvaluator, etc."
}
# --- 4. Security Checks ------------------------------------------
Write-Host ""
Write-Host "[SEC] Security" -ForegroundColor White
# Check: .gitignore excludes .env
$gitignore = "$Root/.gitignore"
if (Test-Path $gitignore) {
$gitignoreContent = Get-Content $gitignore -Raw
if ($gitignoreContent -match '\.env') {
Write-Check ".env excluded from git" "PASS"
} else {
Write-Check ".env excluded from git" "FAIL" "Add .env to .gitignore"
}
} else {
Write-Check ".gitignore exists" "WARN" "Create .gitignore with .env exclusion"
}
# Check: Max turns / termination condition
$hasTermination = $false
foreach ($file in $allFiles) {
$content = Get-Content $file.FullName -Raw -ErrorAction SilentlyContinue
if (-not $content) { continue }
if ($content -match '(max_turns|MaxTurns|termination|max_iterations|MaxIterations)') {
$hasTermination = $true
break
}
}
if ($hasTermination) {
Write-Check "Agent termination conditions" "PASS"
} else {
Write-Check "Agent termination conditions" "WARN" "Set max_turns to prevent infinite agent loops"
}
# Check: Input validation
$hasInputValidation = $false
foreach ($file in $allFiles) {
$content = Get-Content $file.FullName -Raw -ErrorAction SilentlyContinue
if (-not $content) { continue }
if ($content -match '(validate|sanitize|strip\(|\.strip|input.*check|len\(.*\)|MaxLength|StringLength)') {
$hasInputValidation = $true
break
}
}
if ($hasInputValidation) {
Write-Check "Input validation present" "PASS"
} else {
Write-Check "Input validation present" "WARN" "Validate and sanitize user inputs before sending to LLM"
}
# --- 5. Operations Checks ----------------------------------------
Write-Host ""
Write-Host "[OPS] Operations" -ForegroundColor White
# Check: README exists
if (Test-Path "$Root/README.md") {
Write-Check "README.md exists" "PASS"
} else {
Write-Check "README.md exists" "WARN" "Add README with setup, run, and architecture docs"
}
# Check: Tests exist
$testFiles = @()
$testFiles += Get-ChildItem -Path $Root -Filter "test_*.py" -Recurse -ErrorAction SilentlyContinue | Where-Object { $_.FullName -notmatch '\.venv' }
$testFiles += Get-ChildItem -Path $Root -Filter "*_test.py" -Recurse -ErrorAction SilentlyContinue | Where-Object { $_.FullName -notmatch '\.venv' }
$testFiles += Get-ChildItem -Path $Root -Filter "*Tests.cs" -Recurse -ErrorAction SilentlyContinue | Where-Object { $_.FullName -notmatch '(bin|obj)' }
if ($testFiles.Count -gt 0) {
Write-Check "Tests exist ($($testFiles.Count) test files)" "PASS"
} else {
Write-Check "Tests exist" "WARN" "Add tests for agent behavior and tool functions"
}
# --- Summary -----------------------------------------------------
Write-Host ""
Write-Host "-----------------------------------------------------" -ForegroundColor Cyan
Write-Host " Results: $($script:Passed) passed, $($script:Warned) warnings, $($script:Failed) failed" -ForegroundColor White
Write-Host "-----------------------------------------------------" -ForegroundColor Cyan
$exitCode = $script:Failed
if ($Strict) { $exitCode += $script:Warned }
if ($exitCode -eq 0) {
Write-Host " [PASS] Agent project is production-ready!" -ForegroundColor Green
} else {
Write-Host " [WARN] Address the issues above before deploying." -ForegroundColor Yellow
}
Write-Host ""
exit $exitCode
SKILL.md
---
name: ai-agent-development
description: 'Build production-ready AI agents with Microsoft Foundry and Agent Framework. Use when creating AI agents, selecting LLM models, implementing agent orchestration, adding tracing/observability, or evaluating agent quality. Covers agent architecture, model selection, multi-agent workflows, and production deployment.'
---
# AI Agent Development
> **Purpose**: Build production-ready AI agents with Microsoft Foundry and Agent Framework.
> **Scope**: Agent architecture, model selection, orchestration, observability, evaluation.
---
## When to Use This Skill
- Building AI agents with Microsoft Foundry or Agent Framework
- Selecting LLM models for agent scenarios
- Implementing multi-agent orchestration workflows
- Adding tracing and observability to AI agents
- Evaluating agent quality and response accuracy
## Decision Tree
```
Need an AI agent?
+-- Simple request-response? -> Single agent with tools
+-- Multi-step reasoning? -> Chain-of-thought agent with planner
+-- Multiple specialized domains? -> Multi-agent orchestration
+-- Human approval needed? -> Human-in-the-loop workflow
+-- High reliability required? -> Reflection + self-correction loop
+-- Real-time streaming? -> Async event-driven agent architecture
```
## Prerequisites
- Python 3.14+ or .NET 10+
- agent-framework-azure-ai package
- Microsoft Foundry workspace with deployed model
## Quick Start
### Installation
**Python** (Recommended):
```bash
pip install agent-framework-azure-ai --pre # --pre required during preview
```
**.NET**:
```bash
dotnet add package Microsoft.Agents.AI.AzureAI --prerelease
dotnet add package Microsoft.Agents.AI.Workflows --prerelease
```
### Model Selection
**Top Production Models** (Microsoft Foundry):
| Model | Best For | Context | Cost/1M |
|-------|----------|---------|---------|
| **gpt-5.2** | Enterprise agents, structured outputs | 200K/100K | TBD |
| **gpt-5.1-codex-max** | Agentic coding workflows | 272K/128K | $3.44 |
| **claude-opus-4-5** | Complex agents, coding, computer use | 200K/64K | $10 |
| **gpt-5.1** | Multi-step reasoning | 200K/100K | $3.44 |
| **o3** | Advanced reasoning | 200K/100K | $3.5 |
**Deploy Model**: `Ctrl+Shift+P` -> `AI Toolkit: Deploy Model`
---
## Agent Patterns
### Single Agent
```python
from pathlib import Path
from agent_framework.openai import OpenAIChatClient
# Load prompt from file - NEVER embed prompts as inline strings
prompt = Path("prompts/assistant.md").read_text(encoding="utf-8")
client = OpenAIChatClient(
model="gpt-5.1",
api_key=os.getenv("FOUNDRY_API_KEY"),
endpoint=os.getenv("FOUNDRY_ENDPOINT")
)
agent = {
"name": "Assistant",
"instructions": prompt, # Loaded from prompts/assistant.md
"tools": [] # Add tools as needed
}
response = await client.chat(
messages=[{"role": "user", "content": "Hello"}],
agent=agent
)
```
### Multi-Agent Orchestration
```python
from pathlib import Path
from agent_framework.workflows import SequentialWorkflow
# Each agent loads its prompt from a dedicated file
researcher = {
"name": "Researcher",
"instructions": Path("prompts/researcher.md").read_text(encoding="utf-8")
}
writer = {
"name": "Writer",
"instructions": Path("prompts/writer.md").read_text(encoding="utf-8")
}
workflow = SequentialWorkflow(
agents=[researcher, writer],
handoff_strategy="on_completion"
)
result = await workflow.run(query="Write about AI agents")
```
**Advanced Patterns**: Search [github.com/microsoft/agent-framework](https://github.com/microsoft/agent-framework) for:
- Group Chat, Concurrent, Conditional, Loop
- Human-in-the-Loop, Reflection, Fan-out/Fan-in
- MCP, Multimodal, Custom Executors
---
## Core Rules
### Prompt & Template File Management
> **RULE**: NEVER embed prompts or output templates as inline strings in code. Always store them as separate files.
**Why**: Prompts are content, not code. Separating them enables:
- Version control diffs that show exactly what changed in a prompt
- Non-developer editing (PMs, prompt engineers) without touching code
- A/B testing different prompts without code changes
- Reuse across agents, languages, and test harnesses
- Clear separation of concerns (logic vs. content)
**Directory Convention**:
```
project/
prompts/ # All system/agent prompts
assistant.md # One file per agent role
researcher.md
writer.md
reviewer.md
templates/ # Output templates used by agents
report-template.md # Structured output templates
email-template.md
summary-template.md
config/
models.yaml # Model configuration
```
**Loading Pattern**:
```python
from pathlib import Path
# Load prompt
prompt = Path("prompts/assistant.md").read_text(encoding="utf-8")
# Load output template and inject into prompt
template = Path("templates/report-template.md").read_text(encoding="utf-8")
prompt_with_template = f"{prompt}\n\n## Output Format\n{template}"
```
**Rules**:
- MUST store all system prompts in `prompts/` directory as `.md` or `.txt` files
- MUST store output format templates in `templates/` directory
- MUST NOT embed prompt text longer than one sentence directly in code
- SHOULD use Markdown format for prompts (readable, supports structure)
- SHOULD name files after the agent role: `prompts/{agent-name}.md`
- SHOULD include a brief comment header in each prompt file (purpose, version, model target)
- MAY use template variables (`{variable}`) for dynamic content injected at runtime
### Development
[PASS] **DO**:
- Plan agent architecture before coding (Research -> Design -> Implement)
- Use Microsoft Foundry models for production
- Implement tracing from day one
- Test with evaluation datasets before deployment
- Use structured outputs for reliable agent responses
- Implement error handling and retry logic
- Version your agents and track changes
- **Store all prompts as separate files in `prompts/` directory**
- **Store output templates as separate files in `templates/` directory**
[FAIL] **DON'T**:
- Hardcode API keys or endpoints
- Embed prompts or output templates as multi-line strings in code
- Skip tracing setup (critical for debugging)
- Deploy without evaluation
- Use GitHub models in production (free tier has limits)
- Ignore token limits and context windows
- Mix agent logic with business logic
### Security
- Store credentials in environment variables or Azure Key Vault
- Validate all tool inputs and outputs
- Implement rate limiting for agent APIs
- Log agent actions for audit trails
- Use role-based access control (RBAC) for Foundry resources
- Review OWASP Top 10 for AI: [owasp.org/AI-Security-and-Privacy-Guide](https://owasp.org/www-project-ai-security-and-privacy-guide/)
### Performance
- Cache model responses when appropriate
- Use batch processing for multiple requests
- Monitor token usage and costs
- Implement timeout handling
- Use async/await for I/O operations
- Consider model size vs. latency tradeoffs
### Monitoring
- Track key metrics: latency, success rate, token usage, cost
- Set up alerts for failures and anomalies
- Use structured logging with context
- Integrate with Azure Monitor / Application Insights
- Review traces regularly for optimization opportunities
---
## Production Checklist
**Development**
- [ ] Agent architecture documented
- [ ] Model selected and deployed
- [ ] Tools/plugins implemented and tested
- [ ] Error handling with retries
- [ ] Structured outputs configured
- [ ] No hardcoded secrets
- [ ] All prompts stored as separate files in `prompts/` (not inline in code)
- [ ] All output templates stored in `templates/` (not inline in code)
**Model Change Management (MANDATORY)**
- [ ] Model version pinned explicitly (e.g., `gpt-5.1-2026-01-15`)
- [ ] Model version configurable via environment variable
- [ ] Evaluation baseline saved for current model
- [ ] A/B evaluation run before any model switch
- [ ] Structured output schema verified after model change
- [ ] Tool/function-calling accuracy verified after model change
- [ ] Model change documented in changelog with eval results
- [ ] Weekly evaluation monitoring configured for drift detection
- [ ] Alert threshold set for score drops > 10% from baseline
**Model Change Test Automation (MANDATORY)**
- [ ] Agent designed as model-agnostic (model injected via config)
- [ ] `config/models.yaml` defines model test matrix with thresholds
- [ ] Tested against 2 models (primary + fallback from different provider)
- [ ] Multi-model comparison pipeline in CI/CD (weekly + on model config change)
- [ ] Deployment gated on threshold checks (CI fails on regression)
- [ ] Validated fallback model designated and documented
- [ ] Comparison report generated per run (JSON + human-readable)
- [ ] Cost and latency evaluators included alongside quality metrics
**Observability**
- [ ] OpenTelemetry tracing enabled
- [ ] Trace viewer tested
- [ ] Structured logging implemented
- [ ] Metrics collection configured
**Evaluation**
- [ ] Evaluation dataset created
- [ ] Evaluators defined (built-in + custom)
- [ ] Evaluation runs passing
- [ ] Results meet quality thresholds
- [ ] Multi-model comparison run (2+ models tested)
- [ ] Fallback model validated and documented
- [ ] Model comparison baseline saved
**Security & Compliance**
- [ ] Credentials in Key Vault/env vars
- [ ] Input validation implemented
- [ ] RBAC configured
- [ ] Audit logging enabled
- [ ] OWASP AI Top 10 reviewed
**Operations**
- [ ] Health checks implemented
- [ ] Rate limiting configured
- [ ] Monitoring alerts set up
- [ ] Deployment strategy defined
- [ ] Rollback plan documented
- [ ] Cost monitoring enabled
---
## Anti-Patterns
- **Inline prompt strings**: Embedding prompts as multi-line strings in code -> Store in `prompts/` directory as separate files
- **Unpinned model versions**: Using `gpt-4o` without date suffix -> Pin explicitly (e.g., `gpt-5.1-2026-01-15`)
- **No evaluation before deploy**: Shipping agents without running eval datasets -> Gate deployment on quality thresholds
- **Monolithic agent**: One agent handling all domains and tasks -> Split into specialized agents with clear handoffs
- **Ignoring token costs**: No monitoring of per-request token usage -> Track tokens per component and set budgets
- **Missing error recovery**: No retry or fallback on LLM failures -> Implement retries with backoff and fallback models
- **Skipping tracing setup**: Deploying without observability -> Enable OpenTelemetry tracing from day one
---
## Resources
**Official Documentation**:
- Agent Framework: [github.com/microsoft/agent-framework](https://github.com/microsoft/agent-framework)
- Microsoft Foundry: [ai.azure.com](https://ai.azure.com)
- Azure AI Projects SDK: [learn.microsoft.com/python/api/overview/azure/ai-projects](https://learn.microsoft.com/python/api/overview/azure/ai-projects)
- OpenTelemetry: [opentelemetry.io](https://opentelemetry.io)
**AI Toolkit**:
- Model Catalog: `Ctrl+Shift+P` -> `AI Toolkit: Model Catalog`
- Trace Viewer: `Ctrl+Shift+P` -> `AI Toolkit: Open Trace Viewer`
- Playground: `Ctrl+Shift+P` -> `AI Toolkit: Model Playground`
**Security**:
- OWASP AI Security: [owasp.org/AI-Security-and-Privacy-Guide](https://owasp.org/www-project-ai-security-and-privacy-guide/)
- Azure Security Best Practices: [learn.microsoft.com/azure/security](https://learn.microsoft.com/azure/security)
---
**Related**: [AGENTS.md](../../../../AGENTS.md) for agent behavior guidelines - [Skills.md](../../../../Skills.md) for general production practices
**Last Updated**: January 17, 2026
## Scripts
| Script | Purpose | Usage |
|--------|---------|-------|
| [`scaffold-agent.py`](scripts/scaffold-agent.py) | Scaffold AI agent project (Python/.NET) with tracing & eval | `python scripts/scaffold-agent.py --name my-agent [--pattern multi-agent] [--with-eval]` |
| [`validate-agent-checklist.ps1`](scripts/validate-agent-checklist.ps1) | Validate agent project against production checklist | `./scripts/validate-agent-checklist.ps1 [-Path ./my-agent] [-Strict]` |
| [`check-model-drift.ps1`](scripts/check-model-drift.ps1) | Validate model pinning, data drift signals, and judge LLM readiness | `./scripts/check-model-drift.ps1 [-Path ./my-agent] [-Strict]` |
| [`run-model-comparison.py`](scripts/run-model-comparison.py) | Run eval suite against multiple models and generate comparison report | `python scripts/run-model-comparison.py --config config/models.yaml --dataset evaluation/core.jsonl` |
## Troubleshooting
| Issue | Solution |
|-------|----------|
| Model not found | Verify model deployment in Foundry portal and check endpoint URL |
| Tracing not appearing | Ensure AIInferenceInstrumentor().instrument() called before agent creation |
| Agent loops indefinitely | Set max_turns limit and add termination conditions |
## References
- [Tracing And Evaluation](references/tracing-and-evaluation.md)
- [Multi Model Patterns](references/multi-model-patterns.md)
- [Model Drift And Judge Patterns](references/model-drift-judge-patterns.md)
- [Model Change Test Automation](references/model-change-test-automation.md)