# Agent Evaluation Testing > Builds evaluation harnesses for AI agents — LLM-as-judge scoring, tool-use accuracy validation, multi-turn conversation testing, and prompt injection detection in production-ready Python. - Skill: `paulpas/agent-evaluation-testing` (Agent Skill) - Install (CLI): `npx skillmds@latest add paulpas/agent-evaluation-testing` - Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/agent-evaluation-testing/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: AI & ML - License: MIT - Author: paulpas (https://skillmd.com/u/paulpas) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/paulpas/agent-evaluation-testing --- # 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 1. **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). 2. **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. 3. **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 `EvaluationResult` with score, reasoning, and metadata. 4. **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. 5. **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. 6. **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. ```python """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. ```python """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. ```python """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. ```python """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)(