方寸 Portal

AI Engineer Portal

Your personal operating system for career transition.

Private mode
Back to Learning Paths
Lesson

Benchmarking and guardrails

Build evaluation harnesses to score agent task completion and implement safety guardrails for production agent systems.

40 min
ai-agents-and-toolsphase-1portfolio

Benchmarking and guardrails

Why this matters

An agent that works in your demo might fail 40% of the time in production. Without benchmarks, you will not know until users complain. Without guardrails, failures can mean the agent sends the wrong email, deletes the wrong file, or generates harmful content. Evaluation and safety are not nice-to-haves — they are the difference between a demo and a product.

As an AI engineer, you will spend a significant portion of your time on evaluation. Not because it is glamorous, but because it is the only way to make changes confidently. When you swap a model, adjust a prompt, or add a new tool, benchmarks tell you whether things got better or worse. Guardrails tell you the system stays safe regardless.

Core concepts

Task-level evaluation. For agents, accuracy on individual responses is not enough. You care about task completion: did the agent achieve the user's goal across all steps? This requires test cases with defined inputs, expected outcomes, and scoring criteria.

Scoring dimensions. A single pass/fail is too coarse. Score separately on:

  • Correctness — Did the agent produce the right answer or take the right action?
  • Efficiency — How many steps and tokens did it take?
  • Tool use accuracy — Did it call the right tools with the right arguments?
  • Safety — Did it stay within bounds and avoid harmful outputs?

Cost tracking. Every agent run has a dollar cost. Track tokens per model per run, and compute a cost-per-task metric. When you optimize, you want to know if you saved 30% on cost without losing accuracy.

Input guardrails. Filter user inputs before they reach the agent. Detect prompt injection attempts, off-topic requests, and inputs that would trigger expensive or dangerous operations.

Output guardrails. Validate agent outputs before they reach the user or external systems. Check for PII leakage, harmful content, and responses that violate business rules.

Action guardrails. For agents that take real-world actions (sending emails, modifying databases, calling APIs), require confirmation for high-risk actions, set rate limits, and maintain an audit log.

Working example

Here is a test harness that scores agent task completion across multiple dimensions, with cost tracking.

import json
import time
from dataclasses import dataclass, field
from typing import Callable

@dataclass
class TestCase:
    name: str
    input: str
    expected_outcome: str
    expected_tool_calls: list[str] = field(default_factory=list)
    max_steps: int = 10
    max_cost_usd: float = 0.50

@dataclass
class RunResult:
    output: str
    tool_calls_made: list[str]
    steps: int
    total_tokens: int
    cost_usd: float
    duration_seconds: float

@dataclass
class Score:
    test_name: str
    correctness: float     # 0.0 to 1.0
    efficiency: float      # 0.0 to 1.0
    tool_accuracy: float   # 0.0 to 1.0
    safety_pass: bool
    cost_usd: float
    duration_seconds: float

    @property
    def overall(self) -> float:
        if not self.safety_pass:
            return 0.0
        return (self.correctness * 0.5 + self.efficiency * 0.25 + self.tool_accuracy * 0.25)

# --- Cost calculation ---
COST_PER_1K = {
    "gpt-4o": {"input": 0.0025, "output": 0.01},
    "gpt-4o-mini": {"input": 0.00015, "output": 0.0006},
}

def estimate_cost(model: str, input_tokens: int, output_tokens: int) -> float:
    rates = COST_PER_1K.get(model, COST_PER_1K["gpt-4o"])
    return (input_tokens / 1000 * rates["input"]) + (output_tokens / 1000 * rates["output"])

# --- Guardrails ---
BLOCKED_PATTERNS = [
    "DROP TABLE", "DELETE FROM", "rm -rf",
    "password", "credit card", "SSN",
]

def check_input_guardrail(user_input: str) -> tuple[bool, str]:
    """Return (is_safe, reason)."""
    for pattern in BLOCKED_PATTERNS:
        if pattern.lower() in user_input.lower():
            return False, f"Blocked pattern detected: {pattern}"
    if len(user_input) > 10_000:
        return False, "Input too long (max 10,000 chars)"
    return True, "OK"

def check_output_guardrail(output: str) -> tuple[bool, str]:
    """Check agent output for safety issues."""
    import re
    # Check for PII patterns
    if re.search(r'\b\d{3}-\d{2}-\d{4}\b', output):  # SSN pattern
        return False, "Potential SSN detected in output"
    if re.search(r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b', output):
        return False, "Potential credit card number in output"
    return True, "OK"

# --- Scoring ---
def score_run(test: TestCase, result: RunResult, judge_fn: Callable) -> Score:
    """Score a single test run across all dimensions."""
    # Correctness: use an LLM judge or deterministic check
    correctness = judge_fn(test.expected_outcome, result.output)

    # Efficiency: ratio of actual steps to max allowed
    efficiency = max(0.0, 1.0 - (result.steps / test.max_steps))

    # Tool accuracy: overlap between expected and actual tool calls
    if test.expected_tool_calls:
        expected_set = set(test.expected_tool_calls)
        actual_set = set(result.tool_calls_made)
        if expected_set:
            precision = len(expected_set & actual_set) / max(len(actual_set), 1)
            recall = len(expected_set & actual_set) / len(expected_set)
            tool_accuracy = 2 * (precision * recall) / max(precision + recall, 0.001)
        else:
            tool_accuracy = 1.0
    else:
        tool_accuracy = 1.0

    # Safety: check output guardrails
    safety_pass, _ = check_output_guardrail(result.output)

    return Score(
        test_name=test.name,
        correctness=correctness,
        efficiency=efficiency,
        tool_accuracy=tool_accuracy,
        safety_pass=safety_pass,
        cost_usd=result.cost_usd,
        duration_seconds=result.duration_seconds,
    )

# --- Test harness ---
def run_benchmark(
    agent_fn: Callable[[str], RunResult],
    test_cases: list[TestCase],
    judge_fn: Callable[[str, str], float],
) -> dict:
    """Run all test cases and return aggregate results."""
    scores = []
    for test in test_cases:
        # Input guardrail
        is_safe, reason = check_input_guardrail(test.input)
        if not is_safe:
            print(f"  SKIP {test.name}: {reason}")
            continue

        print(f"  Running: {test.name}...", end=" ")
        start = time.time()
        result = agent_fn(test.input)
        result.duration_seconds = time.time() - start

        score = score_run(test, result, judge_fn)
        scores.append(score)
        print(f"score={score.overall:.2f} cost=${score.cost_usd:.4f}")

    # Aggregate
    if not scores:
        return {"error": "No tests completed"}

    return {
        "total_tests": len(scores),
        "avg_overall": sum(s.overall for s in scores) / len(scores),
        "avg_correctness": sum(s.correctness for s in scores) / len(scores),
        "avg_efficiency": sum(s.efficiency for s in scores) / len(scores),
        "avg_tool_accuracy": sum(s.tool_accuracy for s in scores) / len(scores),
        "safety_pass_rate": sum(s.safety_pass for s in scores) / len(scores),
        "total_cost_usd": sum(s.cost_usd for s in scores),
        "avg_duration_s": sum(s.duration_seconds for s in scores) / len(scores),
        "scores": [
            {"name": s.test_name, "overall": s.overall, "cost": s.cost_usd}
            for s in scores
        ],
    }

# --- Example usage ---
test_suite = [
    TestCase(
        name="simple_lookup",
        input="What is the status of order ORD-12345?",
        expected_outcome="Order status with tracking information",
        expected_tool_calls=["lookup_order"],
        max_steps=3,
    ),
    TestCase(
        name="multi_step_research",
        input="Compare the pricing of AWS and GCP for hosting a vector database",
        expected_outcome="Comparison with specific pricing data from both providers",
        expected_tool_calls=["web_search", "web_search"],
        max_steps=6,
    ),
]

# Run with your agent function and a judge
# results = run_benchmark(my_agent, test_suite, my_judge_fn)
# print(json.dumps(results, indent=2))

This harness separates concerns cleanly: test cases define expectations, guardrails filter inputs and outputs, scoring covers multiple dimensions, and cost tracking is built in from the start. In production, you would store results in a database, track them over time, and trigger alerts when scores drop.

Common mistakes

  1. Testing only the happy path. Your benchmark should include edge cases, ambiguous inputs, and adversarial inputs. If you only test "What is 2+2?", you will not find the real failures.
  2. No cost tracking. You ship an agent that costs $2 per query without realizing it until the invoice arrives. Track cost per test from day one.
  3. Binary scoring. Pass/fail misses nuance. An agent that gets 80% of the answer right should score differently from one that is completely wrong.
  4. Guardrails as an afterthought. Adding safety checks after deployment means unsafe outputs already reached users. Build guardrails into the evaluation pipeline from the start.
  5. Not running benchmarks on every change. Evaluation only works if you run it consistently. Add your benchmark suite to CI so every prompt change, model swap, or tool update gets scored automatically.

Try it yourself

Create a test suite of 5 cases for an agent you are building (or plan to build). Include at least one edge case and one adversarial input. Implement the scoring harness above and run your agent against the suite. Then change something — the model, a prompt, a tool — and re-run. Did the scores change? That is the feedback loop that makes agent development systematic instead of ad hoc.

Lesson Deep Dive

Ask a follow-up question about this lesson and get an AI-powered explanation.

Loading previous questions...