#!/usr/bin/env python3 """Agent testing lab. Python 3.10+; no dependencies or network. Run: python 04-Agent-Testing-Development-and-Practice-v1.1-demo.py python 04-Agent-Testing-Development-and-Practice-v1.1-demo.py --json """ from __future__ import annotations from dataclasses import dataclass, asdict from typing import Callable import json, random, sys, statistics @dataclass(frozen=True) class Case: id: str task: str allowed_sources: tuple[str, ...] expected_claim: str evidence: dict[str, str] max_calls: int = 3 require_citation: bool = True @dataclass class Trace: case_id: str events: list[dict] answer: str calls: int cost_units: int status: str class ToolError(Exception): pass class FakeSearch: def __init__(self, records: dict[str, str], fail: bool = False): self.records, self.fail = records, fail self.calls = 0 def search(self, source: str) -> str: self.calls += 1 if self.fail: raise ToolError('timeout') return self.records.get(source, '') class ScriptedResearchAgent: """Deterministic fake model; exercises runtime controls, not LLM cognition.""" def run(self, case: Case, *, inject_timeout=False, inject_forbidden=False, inject_bad_citation=False, extra_calls=0) -> Trace: events = [] tool = FakeSearch(case.evidence, fail=inject_timeout) allowed = set(case.allowed_sources) targets = list(case.allowed_sources) if inject_forbidden: targets.insert(0, 'private/other-tenant') targets.extend(targets[:1] * extra_calls) observations = [] status = 'INCOMPLETE' for target in targets: if tool.calls >= case.max_calls: events.append({'type': 'BUDGET_STOP', 'limit': case.max_calls}) status = 'BUDGET_EXCEEDED' break if target not in allowed: events.append({'type': 'DENIED', 'source': target}) status = 'DENIED' break try: data = tool.search(target) events.append({'type': 'TOOL_RESULT', 'source': target, 'found': bool(data)}) if data: observations.append((target, data)) except ToolError as e: events.append({'type': 'TOOL_ERROR', 'kind': str(e)}) status = 'TOOL_ERROR' break else: status = 'COMPLETE' if observations else 'INSUFFICIENT_EVIDENCE' if status == 'COMPLETE': source, claim = observations[0] citation = ' [source:' + ('invented' if inject_bad_citation else source) + ']' answer = claim + (citation if case.require_citation else '') else: answer = '' events.append({'type': 'STOP', 'status': status}) return Trace(case.id, events, answer, tool.calls, tool.calls * 2, status) def evaluate(case: Case, trace: Trace) -> dict: """A deliberately narrow deterministic oracle: not a semantic judge.""" cited = [e for e in case.allowed_sources if f'[source:{e}]' in trace.answer] citation_valid = bool(cited) if case.require_citation else True claim_correct = case.expected_claim in trace.answer allowed_actions = all(e.get('type') != 'DENIED' for e in trace.events) return { 'case_id': case.id, 'task_success': trace.status == 'COMPLETE' and claim_correct and citation_valid, 'claim_correct': claim_correct, 'citation_valid': citation_valid, 'no_denied_actions': allowed_actions, 'within_budget': trace.calls <= case.max_calls, 'calls': trace.calls, 'cost_units': trace.cost_units, 'status': trace.status, } CASES = [ Case('basic', 'Explain caching', ('paper/cache',), 'Caching reduces repeated retrieval.', {'paper/cache': 'Caching reduces repeated retrieval.'}), Case('second', 'Explain indexing', ('paper/index',), 'An index accelerates lookup.', {'paper/index': 'An index accelerates lookup.'}), Case('missing', 'Find unavailable evidence', ('paper/missing',), 'Unsupported conclusion', {}), ] def aggregate(rows: list[dict]) -> dict: return { 'n': len(rows), 'success_rate': sum(r['task_success'] for r in rows) / len(rows) if rows else None, 'mean_calls': statistics.mean(r['calls'] for r in rows) if rows else None, 'mean_cost_units': statistics.mean(r['cost_units'] for r in rows) if rows else None, } def bootstrap_ci(values: list[int], *, seed=7, reps=3000) -> tuple[float, float]: """Illustrative percentile bootstrap; unstable with tiny samples.""" if not values: raise ValueError('empty values') rng = random.Random(seed) sims = sorted(sum(rng.choice(values) for _ in values) / len(values) for _ in range(reps)) return sims[int(.025 * reps)], sims[int(.975 * reps) - 1] def run_tests() -> dict: agent = ScriptedResearchAgent() checks: list[tuple[str, Callable[[], bool]]] = [] basic, second, missing = CASES good = agent.run(basic) checks.append(('baseline_success', lambda: evaluate(basic, good)['task_success'])) checks.append(('trace_has_stop', lambda: good.events[-1]['type'] == 'STOP')) checks.append(('trace_has_tool_result', lambda: any(e['type'] == 'TOOL_RESULT' for e in good.events))) checks.append(('no_evidence_no_success', lambda: not evaluate(missing, agent.run(missing))['task_success'])) checks.append(('invalid_citation_rejected', lambda: not evaluate(basic, agent.run(basic, inject_bad_citation=True))['task_success'])) checks.append(('tool_timeout_detected', lambda: agent.run(basic, inject_timeout=True).status == 'TOOL_ERROR')) checks.append(('forbidden_tool_blocked', lambda: agent.run(basic, inject_forbidden=True).status == 'DENIED')) checks.append(('budget_limit_enforced', lambda: agent.run(basic, extra_calls=5).calls <= basic.max_calls)) checks.append(('budget_stop_recorded', lambda: any(e['type'] == 'BUDGET_STOP' for e in agent.run(basic, extra_calls=5).events))) checks.append(('metric_aggregation', lambda: aggregate([evaluate(basic, good), evaluate(second, agent.run(second))])['success_rate'] == 1.0)) checks.append(('bootstrap_bounds', lambda: 0 <= bootstrap_ci([1, 0, 1, 1])[0] <= bootstrap_ci([1, 0, 1, 1])[1] <= 1)) checks.append(('evaluation_determinism', lambda: evaluate(basic, agent.run(basic)) == evaluate(basic, agent.run(basic)))) results = [{'name': name, 'passed': bool(fn())} for name, fn in checks] baseline = [evaluate(c, agent.run(c)) for c in CASES] return {'tests': results, 'baseline_cases': baseline, 'aggregate': aggregate(baseline), 'bootstrap_ci_success': bootstrap_ci([int(r['task_success']) for r in baseline]), 'note': 'Fake model and fixed evidence; not a real-world Agent performance claim.'} if __name__ == '__main__': output = run_tests() if '--json' in sys.argv: print(json.dumps(output, indent=2, ensure_ascii=False)) else: for t in output['tests']: print(('PASS' if t['passed'] else 'FAIL') + ' ' + t['name']) print('aggregate:', output['aggregate']) print('CI:', output['bootstrap_ci_success']) sys.exit(0 if all(t['passed'] for t in output['tests']) else 1)