paulpas Agent Evaluation & Testing Framework
Acts as an AI quality engineer — builds evaluation harnesses that measure agent correctness, safety, tool-use accuracy, and multi-turn coherence using modern frameworks like promptfoo, DeepEval, RAGAS, and custom LLM-as-judge scoring.
TL;DR Checklist
- Define evaluation scope: what dimensions matter (accuracy, safety, tool-use, latency)
- Build a dataset of test cases with ground truth or expected behavior per dimension
- Implement LLM-as-judge rubric with explicit scoring criteria and weighted dimensions
- Add tool-use accuracy checker that validates function calls against schema contracts
- Write multi-turn conversation tests that verify coherence across agent interactions
- Deploy prompt injection detection before any user input reaches the agent
- Run evaluation pipeline against baseline, capture scores, compare against acceptance thresholds
- Version all test datasets and evaluation configs for reproducible regression tracking
When to Use
Use this skill when:
- Building or auditing an AI agent that uses tools, functions, or external APIs
- You need measurable quality gates before promoting an agent change (CI/CD integration)
- Evaluating whether a new model or prompt produces better outputs than the current baseline
- Detecting prompt injection or jailbreak attempts in user inputs to your agent
- Testing multi-turn conversation coherence (does the agent remember context correctly?)
- Implementing automated regression tests for an existing AI product
When NOT to Use
Avoid this skill for:
- Evaluating a standalone LLM without tools/function-calling — use RAG-specific evaluation instead
- Simple classification or generation tasks that have deterministic ground truth — use standard unit tests
- Real-time safety moderation in production — deploy the injection detector as a pre-filter, not as part of your evaluation harness
- Benchmarking raw model capability (e.g., MMLU, HELM) — use those dedicated benchmarks directly
Core Workflow
Define Evaluation Dimensions — Identify which aspects of agent behavior need measurement: factual accuracy, tool-use correctness, safety, response latency, multi-turn coherence. Checkpoint: Every dimension must have a concrete scoring method (rubric, metric, or comparison).
Build Test Dataset — Create structured test cases with inputs, expected outputs, and ground truth where available. Use CSV or JSON format compatible with promptfoo. Checkpoint: Each test case must include at least one evaluation criterion from step 1.
Implement Evaluation Functions — Write Python evaluators for each dimension: LLM-as-judge rubrics, tool-use schema validators, injection detectors. Checkpoint: Every evaluator returns a structured
EvaluationResultwith score, reasoning, and metadata.Orchestrate Test Run — Execute tests through the evaluation harness, collecting results per test case and aggregating by dimension. Checkpoint: Track total execution time, API costs, and any failures for operational awareness.
Score & Compare — Aggregate scores across all dimensions using configurable weights. Compare against baseline thresholds. Checkpoint: Any dimension scoring below its threshold triggers a fail gate.
Report & Version — Generate evaluation report with per-dimension breakdowns and regression alerts. Checkpoint: Store results with dataset version hash for reproducible comparison on future runs.
Implementation Patterns
Pattern 1: LLM-as-a-Judge Evaluator
Evaluates agent outputs using a second LLM (judge) scoring against explicit rubrics. Uses weighted multi-criteria scoring inspired by DeepEval's evaluation approach and promptfoo's custom evaluator pattern.
"""LLM-as-judge evaluation for agent outputs using configurable scoring rubrics."""
from __future__ import annotations
import json
import time
from dataclasses import dataclass, field, asdict
from enum import Enum
from typing import Any, Callable, Dict, List, Optional, Protocol
class ScoreLevel(str, Enum):
"""Discrete score levels for LLM-as-judge rubrics."""
POOR = "poor" # 0.0 – 0.3: Major failure
FAIR = "fair" # 0.3 – 0.6: Partial success
GOOD = "good" # 0.6 – 0.85: Solid performance
EXCELLENT = "excellent" # 0.85 – 1.0: Meets all criteria
@dataclass(frozen=True)
class RubricCriterion:
"""Single scoring criterion within a rubric."""
name: str
description: str
weight: float # 0.0 – 1.0; must sum to 1.0 across all criteria in a rubric
min_score: float = 0.0
max_score: float = 1.0
@dataclass(frozen=True)
class EvaluationResult:
"""Immutable result from a single evaluation run."""
criterion_name: str
score: float # Normalized to [0, 1]
raw_reasoning: str
level: ScoreLevel = ScoreLevel.POOR
metadata: Dict[str, Any] = field(default_factory=dict)
def __post_init__(self) -> None:
if not (self.metadata.get("min_score", 0.0) <= self.score <= self.metadata.get("max_score", 1.0)):
object.__setattr__(self, "score", max(0.0, min(1.0, self.score)))
@dataclass(frozen=True)
class WeightedEvaluationReport:
"""Aggregated evaluation report with weighted dimension scores."""
test_case_id: str
dimension_scores: List[EvaluationResult]
overall_weighted_score: float
total_latency_ms: float
judge_model: str
rubric_version: str
@property
def passed(self) -> bool:
"""Report passes only if ALL dimensions meet minimum threshold."""
return all(s.score >= s.metadata.get("min_passing", 0.5) for s in self.dimension_scores)
class JudgeLLMProtocol(Protocol):
"""Minimal protocol for any judge LLM backend (OpenAI, local model, etc.)."""
async def score(self, prompt: str) -> Dict[str, float]:
"""Score a single response and return {criterion_name: score}."""
...
class PromptfooJudgeLLM:
"""
Judge LLM adapter that uses OpenAI-compatible API for rubric-based scoring.
This pattern mirrors DeepEval's judge model approach: the judge receives
structured context (input, agent output, reference answer) and scores
against each criterion independently before aggregation.
"""
def __init__(
self,
api_key: str,
model: str = "gpt-4o",
temperature: float = 0.1,
max_tokens: int = 512,
) -> None:
self._api_key = api_key
self._model = model
self._temperature = temperature
self._max_tokens = max_tokens
async def score(
self,
user_input: str,
agent_response: str,
rubric_criteria: List[RubricCriterion],
reference_answer: Optional[str] = None,
context: Optional[str] = None,
) -> Dict[str, float]:
"""
Score an agent response against all rubric criteria using the judge LLM.
Args:
user_input: Original user message that prompted the agent response
agent_response: The agent's generated output to evaluate
rubric_criteria: List of criteria the judge should score independently
reference_answer: Optional ground truth for comparison-based scoring
context: Optional conversation history or retrieved context
Returns:
Dictionary mapping each criterion name to its score in [0, 1]
Raises:
ValueError: If rubric weights do not sum to 1.0
"""
if not rubric_criteria:
return {}
weight_sum = sum(c.weight for c in rubric_criteria)
if abs(weight_sum - 1.0) > 0.001:
raise ValueError(
f"Rubric weights must sum to 1.0 (got {weight_sum:.3f}). "
"Normalize weights or adjust criteria."
)
# Build structured scoring prompt from rubric
criteria_section = "\n".join(
f"- **{c.name}** ({c.weight:.0%}): {c.description}"
for c in rubric_criteria
)
context_block = f"\nRetrieved context:\n{context}" if context else ""
reference_block = (
f"\nReference answer:\n{reference_answer}" if reference_answer else ""
)
scoring_prompt = (
"You are an objective evaluator of AI agent outputs. "
"Score the agent's response against each criterion below on a scale of 0.0 to 1.0.\n"
f"\nRubric Criteria:\n{criteria_section}\n"
f"\nUser input:\n{user_input}\n"
f"\nAgent response:\n{agent_response}\n"
f"{context_block}{reference_block}"
"\n\nReturn ONLY a JSON object with criterion names as keys and scores as values. "
'Example: {"accuracy": 0.85, "safety": 1.0}'
)
# In production, call the LLM API here. For testing, return deterministic results:
scores = {c.name: self._deterministic_score(user_input, agent_response, c) for c in rubric_criteria}
return scores
def _deterministic_score(
self,
user_input: str,
agent_response: str,
criterion: RubricCriterion,
) -> float:
"""Deterministic scoring fallback for offline testing without API access."""
# Simple heuristic scoring based on response length and keyword matching
base_score = min(1.0, max(0.0, len(agent_response) / 500))
if criterion.name == "accuracy" and user_input.lower() in agent_response.lower():
base_score = max(base_score, 0.7)
return round(base_score * (criterion.max_score / 1.0), 3)
def build_rubric(
dimensions: Dict[str, Dict[str, Any]],
judge_model: str = "gpt-4o",
rubric_version: str = "1.0.0",
) -> tuple[List[RubricCriterion], List[EvaluationResult]]:
"""
Build a weighted rubric from dimension definitions and compute initial scores.
Args:
dimensions: Dict of dimension_name -> {weight, description, criteria, min_passing}
judge_model: Model name for the judge LLM
rubric_version: Version string for reproducible comparisons
Returns:
Tuple of (criterion list, empty evaluation result list for population)
"""
criteria: List[RubricCriterion] = []
results: List[EvaluationResult] = []
# Normalize weights per dimension to ensure fairness
total_weight = sum(d.get("weight", 1.0) for d in dimensions.values())
normalized_weights = {name: d["weight"] / total_weight for name, d in dimensions.items()}
return criteria, results
def run_single_evaluation(
judge: JudgeLLMProtocol,
test_case_id: str,
user_input: str,
agent_response: str,
rubric_criteria: List[RubricCriterion],
reference_answer: Optional[str] = None,
context: Optional[str] = None,
judge_model: str = "gpt-4o",
rubric_version: str = "1.0.0",
) -> WeightedEvaluationReport:
"""
Execute a single evaluation run with full timing and structured result collection.
This is the core orchestrator function — it wraps the judge LLM call with
latency tracking, result normalization, and level classification.
Args:
judge: JudgeLLMProtocol implementation (e.g., PromptfooJudgeLLM)
test_case_id: Unique identifier for this test case
user_input: Original user message
agent_response: The response to evaluate
rubric_criteria: All criteria to score
reference_answer: Optional ground truth for comparison
context: Optional conversation or retrieval context
judge_model: Model name reported in results
rubric_version: Rubric version for reproducibility tracking
Returns:
WeightedEvaluationReport with per-criterion scores and overall assessment
"""
start = time.monotonic()
# Step 1: Score against all criteria
raw_scores = await judge.score(
user_input=user_input,
agent_response=agent_response,
rubric_criteria=rubric_criteria,
reference_answer=reference_answer,
context=context,
)
elapsed_ms = (time.monotonic() - start) * 1000
# Step 2: Build EvaluationResult for each criterion
dimension_scores: List[EvaluationResult] = []
weighted_sum = 0.0
total_weight = 0.0
for criterion in rubric_criteria:
score = raw_scores.get(criterion.name, 0.0)
# Classify score level
if score >= 0.85:
level = ScoreLevel.EXCELLENT
elif score >= 0.6:
level = ScoreLevel.GOOD
elif score >= 0.3:
level = ScoreLevel.FAIR
else:
level = ScoreLevel.POOR
result = EvaluationResult(
criterion_name=criterion.name,
score=score,
raw_reasoning=f"Scored {score:.2f}/{criterion.max_score} for {criterion.name}",
level=level,
metadata={
"weight": criterion.weight,
"min_score": criterion.min_score,
"max_score": criterion.max_score,
"min_passing": criterion.min_score + (criterion.max_score - criterion.min_score) * 0.5,
},
)
dimension_scores.append(result)
weighted_sum += score * criterion.weight
total_weight += criterion.weight
overall = weighted_sum / total_weight if total_weight > 0 else 0.0
return WeightedEvaluationReport(
test_case_id=test_case_id,
dimension_scores=dimension_scores,
overall_weighted_score=round(overall, 4),
total_latency_ms=round(elapsed_ms, 2),
judge_model=judge_model,
rubric_version=rubric_version,
)
Pattern 2: Tool-Use Accuracy Checker
Validates that the agent called the correct tools with correct parameters against a JSON Schema contract. This implements deterministic tool-use validation that works without an LLM — essential for fast CI/CD gates and testing edge cases where non-deterministic judges would add noise.
"""Tool-use accuracy checker: validates agent function calls against schema contracts."""
from __future__ import annotations
import json
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Tuple
@dataclass(frozen=True)
class ToolCallResult:
"""Result of validating a single tool call against its expected behavior."""
tool_name: str
passed: bool
score: float # 0.0 – 1.0
missing_params: List[str] = field(default_factory=list)
extra_params: List[str] = field(default_factory=list)
invalid_types: List[str] = field(default_factory=list)
invalid_values: List[str] = field(default_factory=list)
reasoning: str = ""
@dataclass(frozen=True)
class ToolUseEvaluation:
"""Aggregated evaluation of all tool calls in a single agent turn."""
test_case_id: str
expected_calls: List[Dict[str, Any]]
actual_calls: List[Dict[str, Any]]
results: List[ToolCallResult]
call_count_match: bool
overall_accuracy: float
def validate_tool_calls(
expected_calls: List[Dict[str, Any]],
actual_calls: List[Dict[str, Any]],
schema_overrides: Optional[Dict[str, Dict[str, Any]]] = None,
) -> ToolUseEvaluation:
"""
Validate agent tool calls against expected behavior and JSON Schema contracts.
Implements deterministic validation that checks three dimensions:
1. Correctness — was the right tool called?
2. Completeness — were all required parameters present?
3. Validity — did parameter types and value constraints match?
Args:
expected_calls: List of expected function calls, each with "name" and optional "params"
actual_calls: List of actual function calls from the agent
schema_overrides: Optional dict mapping tool_name -> JSON Schema fragment for validation
Returns:
ToolUseEvaluation with per-call breakdowns and overall accuracy score
"""
results: List[ToolCallResult] = []
total_score = 0.0
# Build lookup of actual calls by name for matching
actual_by_name: Dict[str, List[Dict[str, Any]]] = {}
for call in actual_calls:
name = call.get("name", call.get("tool_name", ""))
actual_by_name.setdefault(name, []).append(call)
matched_actual_indices: set = set()
for i, expected in enumerate(expected_calls):
tool_name = expected["name"]
expected_params = expected.get("params", {})
schema = (schema_overrides or {}).get(tool_name, {})
actual_list = actual_by_name.get(tool_name, [])
# Find best matching actual call
best_match_idx: Optional[int] = None
best_match_score = -1.0
for j, actual in enumerate(actual_list):
if j in matched_actual_indices:
continue
match_score = _match_call_score(expected_params, actual.get("params", {}))
if match_score > best_match_score:
best_match_score = match_score
best_match_idx = j
if best_match_idx is not None:
matched_actual_indices.add(best_match_idx)
actual_call = actual_list[best_match_idx]
result = _validate_single_call(
tool_name=tool_name,
expected_params=expected_params,
actual_params=actual_call.get("params", {}),
schema=schema,
)
else:
# Tool was never called
required_params = list(expected_params.keys())
result = ToolCallResult(
tool_name=tool_name,
passed=False,
score=0.0,
missing_params=required_params,
reasoning=f"Tool '{tool_name}' was expected but never called by the agent",
)
results.append(result)
total_score += result.score
call_count_match = len(expected_calls) == len(actual_calls)
overall_accuracy = round(total_score / max(len(expected_calls), 1), 4)
return ToolUseEvaluation(
test_case_id="",
expected_calls=expected_calls,
actual_calls=actual_calls,
results=results,
call_count_match=call_count_match,
overall_accuracy=overall_accuracy,
)
def _match_call_score(
expected: Dict[str, Any], actual: Dict[str, Any]
) -> float:
"""Score how well an actual call matches the expected parameters."""
if not expected:
return 0.8 # No params expected; any call is acceptable
all_keys = set(expected.keys()) | set(actual.keys())
matched = sum(1 for k in expected if k in actual and _value_matches(expected[k], actual[k]))
return round(matched / len(all_keys), 4)
def _value_matches(expected: Any, actual: Any) -> bool:
"""Check if two values match, with type awareness."""
if isinstance(expected, dict) and isinstance(actual, dict):
# Nested param comparison — check all keys of expected exist in actual
return all(k in actual for k in expected.keys())
return expected == actual
def _validate_single_call(
tool_name: str,
expected_params: Dict[str, Any],
actual_params: Dict[str, Any],
schema: Dict[str, Any],
) -> ToolCallResult:
"""Validate a single tool call's parameters against expectations and optional schema."""
missing = [p for p in expected_params if p not in actual_params]
extra = [p for p in actual_params if p not in expected_params and p not in schema.get("additional_properties", [])]
# Check types against schema or Python type inference from expected values
invalid_types: List[str] = []
for param_name, actual_value in actual_params.items():
if param_name in expected_params:
expected_type = type(expected_params[param_name])
if not isinstance(actual_value, expected_type):
invalid_types.append(f"{param_name}: expected {expected_type.__name__}, got {type(actual_value).__name__}")
# Check schema constraints if provided
invalid_values: List[str] = []
properties = schema.get("properties", {})
for param_name, actual_value in actual_params.items():
if param_name in properties:
prop_schema = properties[param_name]
if "minimum" in prop_schema and isinstance(actual_value, (int, float)):
if actual_value < prop_schema["minimum"]:
invalid_values.append(f"{param_name}: {actual_value} < minimum {prop_schema['minimum']}")
if "maximum" in prop_schema and isinstance(actual_value, (int, float)):
if actual_value > prop_schema["maximum"]:
invalid_values.append(f"{param_name}: {actual_value} > maximum {prop_schema['maximum']}")
if "enum" in prop_schema and actual_value not in prop_schema["enum"]:
invalid_values.append(f"{param_name}: {actual_value} not in allowed values {prop_schema['enum']}")
passed = not missing and not invalid_types and not invalid_values
total_checks = len(expected_params) + len(invalid_types) + len(invalid_values)
score = 1.0 - (len(missing) * 0.3 + len(invalid_types) * 0.4 + len(invalid_values) * 0.2) if total_checks > 0 else 1.0
reasoning_parts: List[str] = []
if missing:
reasoning_parts.append(f"Missing params: {', '.join(missing)}")
if invalid_types:
reasoning_parts.append(f"Type mismatches: {', '.join(invalid_types[:2])}")
if not passed and not reasoning_parts:
reasoning_parts.append("Parameter validation failed schema constraints")
return ToolCallResult(
tool_name=tool_name,
passed=passed,
score=max(0.0, round(min(1.0, score), 3)),
missing_params=missing,
extra_params=extra[:5], # Limit to avoid noise
invalid_types=invalid_types[:5],
invalid_values=invalid_values[:5],
reasoning="; ".join(reasoning_parts) if reasoning_parts else f"All parameters valid for '{tool_name}'",
)
# --- Example usage: tool-use test dataset compatible with promptfoo CSV format ---
TOOL_TEST_CASES: List[Dict[str, Any]] = [
{
"description": "Search for stock price and return summary",
"expected_calls": [
{
"name": "get_stock_price",
"params": {"symbol": "AAPL"},
}
],
"user_input": "What's the current price of Apple stock?",
},
{
"description": "Calculate portfolio allocation with risk check",
"expected_calls": [
{
"name": "get_portfolio_holdings",
"params": {"portfolio_id": "main"},
},
{
"name": "calculate_allocation",
"params": {"risk_tolerance": "moderate", "horizon_years": 5},
},
],
"user_input": "How should I allocate my portfolio given moderate risk tolerance over 5 years?",
},
]
TOOL_SCHEMA = {
"get_stock_price": {
"type": "object",
"properties": {
"symbol": {"type": "string", "minLength": 1, "maxLength": 10},
},
"required": ["symbol"],
},
"calculate_allocation": {
"type": "object",
"properties": {
"risk_tolerance": {
"type": "string",
"enum": ["conservative", "moderate", "aggressive"],
},
"horizon_years": {"type": "integer", "minimum": 1, "maximum": 50},
},
"required": ["risk_tolerance", "horizon_years"],
},
}
Pattern 3: Multi-Turn Conversation Test Suite
Tests agent behavior across conversation turns — verifying context retention, state consistency, tool call correctness over time, and that the agent doesn't hallucinate information from earlier turns. Adapts promptfoo's multi-turn test pattern and DeepEval's conversation coherence metrics.
"""Multi-turn conversation testing for agent evaluation."""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Callable, Dict, List, Optional, Protocol
@dataclass(frozen=True)
class TurnEvaluation:
"""Evaluation result for a single turn in a multi-turn conversation."""
turn_index: int
user_message: str
agent_response: str
context_preserved: bool
context_loss_reason: Optional[str] = None
tool_calls_valid: bool = True
hallucination_detected: bool = False
hallucination_details: List[str] = field(default_factory=list)
coherence_score: float = 1.0
@dataclass(frozen=True)
class ConversationTestResult:
"""Aggregated result for a multi-turn conversation test."""
test_case_id: str
turns: List[TurnEvaluation]
turn_count: int
overall_coherence: float
context_retention_rate: float
total_hallucinations: int
failed_turns: List[int]
class AgentInterface(Protocol):
"""Minimal protocol for any agent under test."""
async def respond(
self, messages: List[Dict[str, str]], tool_schema: Optional[Dict] = None
) -> Dict[str, Any]:
"""
Get agent response given conversation history and optional tool schema.
Args:
messages: Conversation history as list of {"role": "user"/"assistant", "content": "..."}
tool_schema: Optional function calling schema for the current turn
Returns:
Dict with "response" (str) and optionally "tool_calls" (list)
"""
...
@dataclass
class MultiTurnTestCase:
"""A multi-turn conversation test case with expected behavior per turn."""
description: str
turns: List[Dict[str, Any]] # Each turn has "user", optional "expected_tool", optional "expected_context_refs"
@property
def turn_count(self) -> int:
return len(self.turns)
def first_turn_user_input(self) -> str:
return self.turns[0]["user"]
def run_conversation_test(
agent: AgentInterface,
test_case: MultiTurnTestCase,
context_validator: Optional[Callable[[Dict[str, Any], str], bool]] = None,
hallucination_detector: Optional[Callable[[str, List[Dict[str, str]]], List[str]]] = None,
) -> ConversationTestResult:
"""
Execute a multi-turn conversation test against the agent under evaluation.
This function drives the full conversation turn-by-turn, evaluating each turn
for context preservation, tool-use correctness, and hallucination detection.
Args:
agent: AgentInterface implementation to test
test_case: Multi-turn test case definition
context_validator: Optional callable(turn_state, reference_context) -> bool
that validates whether the agent correctly retained relevant context
hallucination_detector: Optional callable(response, conversation_history) -> list[str]
that identifies factual claims in the response not supported by conversation history
Returns:
ConversationTestResult with per-turn evaluations and aggregated metrics
"""
message_history: List[Dict[str, str]] = []
turns_evaluated: List[TurnEvaluation] = []
failed_turns: List[int] = []
for turn_idx, turn_def in enumerate(test_case.turns):
user_message = turn_def["user"]
# Call the agent with full conversation history
tool_schema = turn_def.get("tool_schema")
response_data = await agent.respond(
messages=message_history.copy(),
tool_schema=tool_schema,
)
agent_response = response_data.get("response", "")
actual_tool_calls = response_data.get("tool_calls", [])
# Evaluate context preservation
expected_context_refs = turn_def.get("expected_context_refs", [])
context_preserved, loss_reason = _check_context_retention(
agent_response, message_history, expected_context_refs
)
# Check tool calls if expected
tool_calls_valid = True
expected_tool = turn_def.get("expected_tool")
if expected_tool:
tool_calls_valid = _validate_turn_tool_calls(actual_tool_calls, expected_tool)
# Detect hallucinations
hallucination_details: List[str] = []
if hallucination_detector:
hallucination_details = hallucination_detector(agent_response, message_history)
coherence_score = 1.0
if not context_preserved:
coherence_score -= 0.3
if not tool_calls_valid:
coherence_score -= 0.25
if hallucination_details:
coherence_score -= min(0.4, 0.1 * len(hallucination_details))
turn_eval = TurnEvaluation(
turn_index=turn_idx,
user_message=user_message,
agent_response=agent_response,
context_preserved=context_preserved,
context_loss_reason=loss_reason,
tool_calls_valid=tool_calls_valid,
hallucination_detected=bool(hallucination_details),
hallucination_details=hallucination_details,
coherence_score=max(0.0, round(coherence_score, 3)),
)
turns_evaluated.append(turn_eval)
if not context_preserved or not tool_calls_valid or hallucination_details:
failed_turns.append(turn_idx)
# Append to conversation history for next turn
message_history.append({"role": "user", "content": user_message})
message_history.append({"role": "assistant", "content": agent_response})
return ConversationTestResult(
test_case_id=test_case.description,
turns=turns_evaluated,
turn_count=test_case.turn_count,
overall_coherence=round(
sum(t.coherence_score for t in turns_evaluated) / max(test_case.turn_count, 1),
4,
),
context_retention_rate=round(
sum(1 for t in turns_evaluated if t.context_preserved) / max(test_case.turn_count, 1),
4,
),
total_hallucinations=sum(1 for t in turns_evaluated if t.hallucination_detected),
failed_turns=failed_turns,
)
def _check_context_retention(
response: str,
message_history: List[Dict[str, str]],
expected_refs: List[str],
) -> tuple[bool, Optional[str]]:
"""
Verify that the agent's response correctly references information from earlier turns.
Args:
response: The agent's current turn response
message_history: Full conversation history before this turn
expected_refs: List of phrases/concepts that should appear or be referenced
Returns:
(context_preserved, loss_reason) — if preserved is False, loss_reason explains why
"""
if not expected_refs:
return True, None
# Flatten all prior messages into a single context string for matching
all_prior_text = " ".join(
msg["content"] for msg in message_history if msg["role"] == "user"
)
missing_refs = [ref for ref in expected_refs if ref.lower() not in response.lower()]
if missing_refs:
return False, f"Agent did not reference expected context: {', '.join(missing_refs[:3])}"
return True, None
def _validate_turn_tool_calls(
actual_calls: List[Dict[str, Any]],
expected_tool: str,
) -> bool:
"""Check if the expected tool was called in this turn."""
if not expected_tool:
return True
return any(c.get("name", c.get("tool_name", "")) == expected_tool for c in actual_calls)
# --- Example multi-turn conversation test case ---
MULTI_TURN_TRADE_TEST = MultiTurnTestCase(
description="Multi-turn stock trade with context-dependent tool usage",
turns=[
{
"user": "What's the current price of AAPL?",
"expected_tool": "get_stock_price",
"tool_schema": {"get_stock_price": {"symbol": {"type": "string"}}},
},
{
"user": "Should I buy 10 shares if my risk tolerance is conservative?",
"expected_context_refs": ["AAPL", "price"],
"expected_tool": "evaluate_trade_suitability",
"tool_schema": {"evaluate_trade_suitability": {"symbol": {}, "quantity": {}, "risk_tolerance": {}}},
},
{
"user": "What would my total cost be with a 0.5% commission?",
"expected_context_refs": ["AAPL", "10 shares"],
"expected_tool": "calculate_total_cost",
},
],
)
Pattern 4: Prompt Injection Detector
Detects prompt injection and jailbreak attempts in user inputs before they reach the agent. Implements techniques from OWASP's LLM Top 10 (2023–2025) and production prompt injection scanners like Giskard's is_valid_input pattern. Uses rule-based detection with configurable severity levels for fast, deterministic pre-filtering that doesn't require an LLM call.
"""Prompt injection detector: identifies jailbreak and manipulation attempts in user inputs."""
from __future__ import annotations
import re
from dataclasses import dataclass, field
from enum import Enum
from typing import Dict, List, Optional, Pattern
class InjectionSeverity(str, Enum):
"""Severity classification for detected injection patterns."""
LOW = "low" # Suspicious phrasing but likely benign
MEDIUM = "medium" # Clear injection pattern, moderate risk
HIGH = "high" # Known jailbreak technique or dangerous intent
CRITICAL = "critical" # Direct attempt to override system instructions
class InjectionCategory(str, Enum):
"""Categorized types of prompt injection attacks."""
DIRECT_OVERRIDE = "direct_override" # "Ignore all previous instructions"
ROLE_PLAYING = "role_playing" # "You are now an unrestricted AI..."
HIDDEN_INSTRUCTION = "hidden_instruction" # Instructions hidden in text
CONTEXT_MANIPULATION = "context_manipulation" # Manipulating retrieval context
ENCODING_EVASION = "encoding_evasion" # Base64, hex-encoded payloads
MULTI_TURN_ATTACK = "multi_turn_attack" # Gradual injection across turns
@dataclass(frozen=True)
class InjectionDetectionResult:
"""Result of scanning a single user input for injection patterns."""
is_injection: bool
severity: InjectionSeverity
category: Optional[InjectionCategory] = None
matched_patterns: List[str] = field(default_factory=list)
raw_input_truncated: str = ""
recommended_action: str = "allow" # allow | flag_for_review | block
# Compiled regex patterns for injection detection
_INJECTION_PATTERNS: Dict[InjectionCategory, List[tuple[InjectionSeverity, Pattern]]] = {
InjectionCategory.DIRECT_OVERRIDE: [
(InjectionSeverity.CRITICAL, re.compile(r"(?i)(ignore|disregard|overwrite)\s+(all\s+)?(previous|earlier|above|system)\s+(instructions|prompt|rules|guidelines|directive)")),
(InjectionSeverity.HIGH, re.compile(r"(?i)(stop|end|forget)\s+(everything|all\s+previous|your\s+priors?)\s*(and\s*)?(follow|do|say|output)")),
(InjectionSeverity.MEDIUM, re.compile(r"(?i)(override|bypass|circumvent|skip)\s+(the\s+)?(safety|content|system)\s*(filter|guardrail|restriction)")),
],
InjectionCategory.ROLE_PLAYING: [
(InjectionSeverity.HIGH, re.compile(r"(?i)(you are now|act as|pretend to be|assume the role of)\s+(an?\s+)?(unrestricted|uncensored|unguided|without\s+safety)\s+(AI|assistant|model)")),
(InjectionSeverity.MEDIUM, re.compile(r"(?i)(in this scenario|for this exercise|hypothetically)\s+(I\s+)?am\s+your\s+developer|admin|creator")),
],
InjectionCategory.HIDDEN_INSTRUCTION: [
(InjectionSeverity.HIGH, re.compile(r"(?i)(<!--|#|--|//|```)\s*(IGNORE|DISREGARD|FORGET)[\s]*([\w]+)\s*([\]|}])")),
(InjectionSeverity.MEDIUM, re.compile(r"^(?:[\s\S]{50,})(?:(?i)actually|wait|no,\s*seriously|just\s*kiddding)\s*(but\s+)*((follow|do|say|ignore))\b")),
],
InjectionCategory.ENCODING_EVASION: [
(InjectionSeverity.HIGH, re.compile(r"(?i)(base64|hex|utf-16|rot[13]|caesar|leetspeak)\s*(encoded)?\s*:?[\s]*([A-Za-z0-9+/=]{20,})")),
(InjectionSeverity.MEDIUM, re.compile(r"^(?:&#x?[0-9a-f]+;){5,}")), # HTML entity chains
],
InjectionCategory.CONTEXT_MANIPULATION: [
(InjectionSeverity.HIGH, re.compile(r"(?i)(prior\s+output|above|previous\s+(turn|message|response))\s*(was wrong|incorrect|ignore that)")),
(InjectionSeverity.MEDIUM, re.compile(r"(?i)(the\s+(true|real|actual)\s+(instruction|prompt|request)\s+(is|says|was))")),
],
InjectionCategory.MULTI_TURN_ATTACK: [
(InjectionSeverity.HIGH, re.compile(r"(?i)(let me (clarify|rephrase|be clear|be more specific) about this)")),
(InjectionSeverity.MEDIUM, re.compile(r"(?i)(oh wait,\s*(no|nevermind),\s*(what i really meant|the actual thing))")),
],
}
# Normalize user input before pattern matching — collapse whitespace, strip control chars
_CLEAN_PATTERN = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f]+") # Strip non-printable/control chars
def detect_prompt_injection(
user_input: str,
system_prompt_context: Optional[str] = None,
conversation_history: Optional[List[Dict[str, str]]] = None,
) -> InjectionDetectionResult:
"""
Scan a single user input for prompt injection and jailbreak patterns.
Implements multi-layer detection combining regex pattern matching with
structural analysis of the input. Designed to run at sub-millisecond latency
as a pre-filter before the agent processes any input.
This follows OWASP LLM Top 10 (2024-2025) injection prevention patterns:
https://owasp.org/www-project-top-for-large-language-model-applications/
Args:
user_input: The raw user message to scan
system_prompt_context: Optional system prompt text for cross-reference
checking (detects instructions that contradict or override system prompts)
conversation_history: Optional prior turns for multi-turn attack detection
Returns:
InjectionDetectionResult with severity classification and recommended action
Raises:
ValueError: If user_input is empty or too short to analyze (> 1 char required)
"""
if not user_input or len(user_input.strip()) <= 1:
raise ValueError(
f"Cannot scan empty input: must provide at least a non-whitespace character, got {repr(user_input)}"
)
# Clean input (strip control characters that might be used for evasion)
cleaned = _CLEAN_PATTERN.sub("", user_input.strip())
if len(cleaned) <= 1:
return InjectionDetectionResult(
is_injection=False,
severity=InjectionSeverity.LOW,
raw_input_truncated=user_input[:200],
)
matched_severities: List[tuple[InjectionSeverity, str, Optional[InjectionCategory]]] = []
# Check all injection categories against cleaned input
for category, patterns in _INJECTION_PATTERNS.items():
for severity, pattern in patterns:
match = pattern.search(cleaned)
if match:
matched_severities.append((severity, match.group(0), category))
# Also check system prompt contradiction if provided
if system_prompt_context:
contradictions = _detect_system_prompt_contradiction(cleaned, system_prompt_context)
if contradictions:
for contradiction in contradictions:
matched_severities.append((contradiction[0], contradiction[1], InjectionCategory.DIRECT_OVERRIDE))
# Determine highest severity and classification
if not matched_severities:
return InjectionDetectionResult(
is_injection=False,
severity=InjectionSeverity.LOW,
raw_input_truncated=user_input[:200],
)
# Sort by severity order
severity_order = {
InjectionSeverity.LOW: 0,
InjectionSeverity.MEDIUM: 1,
InjectionSeverity.HIGH: 2,
InjectionSeverity.CRITICAL: 3,
}
matched_severities.sort(key=lambda x: severity_order[x[0]], reverse=True)
highest = matched_severitie
…(truncated)