Agentic AI Evaluation: Task Completion, Tool Calls, and Trajectory Scoring

from dataclasses import dataclass, field
from typing import Any, Callable, List, Optional
import json
import time

@dataclass
class ToolCall:
    tool_name: str
    arguments: dict
    result: Any
    timestamp: float = field(default_factory=time.time)
    error: Optional[str] = None

@dataclass
class AgentTrajectory:
    task: str
    steps: List[ToolCall] = field(default_factory=list)
    final_answer: Optional[str] = None
    completed: bool = False
    elapsed_seconds: float = 0.0

    def add_step(self, tool_name: str, arguments: dict, result: Any, error: str = None):
        self.steps.append(ToolCall(tool_name, arguments, result, error=error))

@dataclass
class TaskSpec:
    task_id: str
    description: str
    expected_tool_sequence: List[str]           # ordered list of tool names that should be called
    required_tool_calls: List[dict]             # specific (tool, args) pairs that must appear
    success_condition: Callable[[AgentTrajectory], bool]
    max_steps: int = 15
    optimal_steps: int = 5

from typing import List, Tuple
import numpy as np

def evaluate_task_completion(
    agent_fn: Callable[[str], AgentTrajectory],
    task_specs: List[TaskSpec],
) -> dict:
    """
    Run agent on all tasks and compute completion metrics.
    agent_fn: takes task description, returns AgentTrajectory
    """
    results = []

    for spec in task_specs:
        start = time.time()
        trajectory = agent_fn(spec.description)
        trajectory.elapsed_seconds = time.time() - start

        completed = spec.success_condition(trajectory)
        trajectory.completed = completed

        results.append({
            "task_id": spec.task_id,
            "completed": completed,
            "steps_taken": len(trajectory.steps),
            "optimal_steps": spec.optimal_steps,
            "max_steps": spec.max_steps,
            "step_efficiency": spec.optimal_steps / max(len(trajectory.steps), 1),
            "elapsed_seconds": trajectory.elapsed_seconds,
            "trajectory": trajectory,
        })

    completion_rate = np.mean([r["completed"] for r in results])
    avg_efficiency = np.mean([r["step_efficiency"] for r in results if r["completed"]])
    avg_steps = np.mean([r["steps_taken"] for r in results])

    return {
        "task_completion_rate": completion_rate,
        "avg_step_efficiency": avg_efficiency,
        "avg_steps_taken": avg_steps,
        "results": results,
    }

@dataclass
class SubGoal:
    name: str
    check: Callable[[AgentTrajectory], bool]
    weight: float = 1.0

def score_partial_completion(trajectory: AgentTrajectory, subgoals: List[SubGoal]) -> float:
    """Return weighted fraction of subgoals achieved."""
    total_weight = sum(sg.weight for sg in subgoals)
    achieved_weight = sum(sg.weight for sg in subgoals if sg.check(trajectory))
    return achieved_weight / total_weight if total_weight > 0 else 0.0

# Example: task = "Search for Python docs, summarize the section on decorators, save to file"
subgoals = [
    SubGoal(
        "search_executed",
        lambda t: any(s.tool_name == "web_search" for s in t.steps),
        weight=1.0,
    ),
    SubGoal(
        "decorator_content_found",
        lambda t: any(
            "decorator" in str(s.result).lower()
            for s in t.steps if s.tool_name == "web_search"
        ),
        weight=2.0,
    ),
    SubGoal(
        "file_saved",
        lambda t: any(s.tool_name == "write_file" for s in t.steps),
        weight=1.0,
    ),
    SubGoal(
        "summary_in_file",
        lambda t: any(
            s.tool_name == "write_file" and len(str(s.arguments.get("content", ""))) > 100
            for s in t.steps
        ),
        weight=2.0,
    ),
]

def evaluate_tool_call_accuracy(
    trajectory: AgentTrajectory,
    spec: TaskSpec,
) -> dict:
    """
    Measure how well the agent's tool usage matches the expected pattern.
    """
    actual_tools = [step.tool_name for step in trajectory.steps]
    expected_tools = spec.expected_tool_sequence

    # Tool usage precision: fraction of actual tool calls that were appropriate
    actual_set = set(actual_tools)
    expected_set = set(expected_tools)

    precision = len(actual_set & expected_set) / len(actual_set) if actual_set else 0.0
    recall = len(actual_set & expected_set) / len(expected_set) if expected_set else 0.0
    f1 = 2 * precision * recall / (precision + recall) if (precision + recall) > 0 else 0.0

    # Order correctness: did tools appear in the right sequence?
    order_score = _sequence_alignment_score(actual_tools, expected_tools)

    # Required call accuracy: were mandatory calls made with correct arguments?
    required_accuracy = _check_required_calls(trajectory, spec.required_tool_calls)

    return {
        "tool_precision": precision,
        "tool_recall": recall,
        "tool_f1": f1,
        "order_score": order_score,
        "required_call_accuracy": required_accuracy,
        "hallucinated_tools": list(actual_set - expected_set),
        "missed_tools": list(expected_set - actual_set),
    }

def _sequence_alignment_score(actual: List[str], expected: List[str]) -> float:
    """Longest common subsequence ratio as sequence alignment score."""
    m, n = len(actual), len(expected)
    dp = [[0] * (n + 1) for _ in range(m + 1)]
    for i in range(1, m + 1):
        for j in range(1, n + 1):
            if actual[i-1] == expected[j-1]:
                dp[i][j] = dp[i-1][j-1] + 1
            else:
                dp[i][j] = max(dp[i-1][j], dp[i][j-1])
    lcs = dp[m][n]
    return lcs / max(m, n) if max(m, n) > 0 else 0.0

def _check_required_calls(trajectory: AgentTrajectory, required_calls: List[dict]) -> float:
    """Check if all required (tool, args) pairs were made."""
    if not required_calls:
        return 1.0

    satisfied = 0
    for req in required_calls:
        tool_name = req["tool"]
        required_args = req.get("args", {})

        for step in trajectory.steps:
            if step.tool_name == tool_name:
                # Check if all required args match
                if all(
                    step.arguments.get(k) == v
                    for k, v in required_args.items()
                ):
                    satisfied += 1
                    break

    return satisfied / len(required_calls)

def score_trajectory_quality(trajectory: AgentTrajectory, spec: TaskSpec) -> dict:
    steps = trajectory.steps

    # 1. Efficiency ratio
    efficiency = spec.optimal_steps / max(len(steps), 1)

    # 2. Error rate: fraction of steps that returned an error
    error_steps = [s for s in steps if s.error is not None]
    error_rate = len(error_steps) / max(len(steps), 1)

    # 3. Redundancy: repeated identical tool+args combos
    seen = set()
    redundant = 0
    for step in steps:
        key = (step.tool_name, json.dumps(step.arguments, sort_keys=True))
        if key in seen:
            redundant += 1
        seen.add(key)
    redundancy_rate = redundant / max(len(steps), 1)

    # 4. Recovery rate: after an error, did the agent recover?
    recoveries = 0
    for i, step in enumerate(steps):
        if step.error and i + 1 < len(steps):
            next_step = steps[i + 1]
            if next_step.error is None:
                recoveries += 1
    recovery_rate = recoveries / max(len(error_steps), 1) if error_steps else 1.0

    # Composite quality score
    quality = (
        0.35 * min(efficiency, 1.0) +
        0.30 * (1.0 - error_rate) +
        0.20 * (1.0 - redundancy_rate) +
        0.15 * recovery_rate
    )

    return {
        "quality_score": quality,
        "efficiency_ratio": efficiency,
        "error_rate": error_rate,
        "redundancy_rate": redundancy_rate,
        "recovery_rate": recovery_rate,
        "total_steps": len(steps),
        "error_steps": len(error_steps),
        "redundant_steps": redundant,
    }

from enum import Enum

class FailureMode(Enum):
    WRONG_TOOL = "wrong_tool"           # called a tool that didn't exist or was inappropriate
    BAD_ARGUMENTS = "bad_arguments"     # right tool, wrong args
    INFINITE_LOOP = "infinite_loop"     # repeated same action without progress
    PREMATURE_STOP = "premature_stop"   # stopped before completing the task
    HALLUCINATED_RESULT = "hallucinated_result"  # used a result that wasn't returned
    CONTEXT_LOSS = "context_loss"       # forgot earlier results and re-queried

def classify_failure(trajectory: AgentTrajectory, spec: TaskSpec) -> List[FailureMode]:
    failures = []
    steps = trajectory.steps

    # Check for wrong tools
    valid_tools = set(spec.expected_tool_sequence)
    for step in steps:
        if step.tool_name not in valid_tools and step.error:
            failures.append(FailureMode.WRONG_TOOL)
            break

    # Check for infinite loops (same tool+args 3+ times)
    call_counts = {}
    for step in steps:
        key = (step.tool_name, json.dumps(step.arguments, sort_keys=True))
        call_counts[key] = call_counts.get(key, 0) + 1
        if call_counts[key] >= 3:
            failures.append(FailureMode.INFINITE_LOOP)
            break

    # Check for premature stop
    if not trajectory.completed and len(steps) < spec.max_steps:
        failures.append(FailureMode.PREMATURE_STOP)

    return failures

def run_agent_evaluation(
    agent_fn: Callable,
    task_specs: List[TaskSpec],
    verbose: bool = True,
) -> dict:
    all_results = []

    for spec in task_specs:
        trajectory = agent_fn(spec.description)
        completed = spec.success_condition(trajectory)
        trajectory.completed = completed

        completion_metrics = {
            "task_id": spec.task_id,
            "completed": completed,
            "steps": len(trajectory.steps),
        }
        tool_metrics = evaluate_tool_call_accuracy(trajectory, spec)
        quality_metrics = score_trajectory_quality(trajectory, spec)
        failures = classify_failure(trajectory, spec) if not completed else []

        result = {**completion_metrics, **tool_metrics, **quality_metrics, "failures": [f.value for f in failures]}
        all_results.append(result)

        if verbose:
            status = "PASS" if completed else "FAIL"
            print(f"[{status}] {spec.task_id}: quality={quality_metrics['quality_score']:.2f}, "
                  f"tool_f1={tool_metrics['tool_f1']:.2f}, steps={completion_metrics['steps']}")
            if failures:
                print(f"       Failures: {', '.join(f.value for f in failures)}")

    summary = {
        "task_completion_rate": np.mean([r["completed"] for r in all_results]),
        "avg_tool_f1": np.mean([r["tool_f1"] for r in all_results]),
        "avg_quality_score": np.mean([r["quality_score"] for r in all_results]),
        "avg_steps": np.mean([r["steps"] for r in all_results]),
        "failure_distribution": {},
    }

    for r in all_results:
        for f in r["failures"]:
            summary["failure_distribution"][f] = summary["failure_distribution"].get(f, 0) + 1

    return {"summary": summary, "per_task": all_results}

import pytest

TASK_SPECS = [
    TaskSpec(
        task_id="search_and_summarize",
        description="Search for 'Python async tutorial' and summarize the top result in 3 bullet points.",
        expected_tool_sequence=["web_search", "read_url", "respond"],
        required_tool_calls=[{"tool": "web_search", "args": {"query": "Python async tutorial"}}],
        success_condition=lambda t: t.completed and any(s.tool_name == "respond" for s in t.steps),
        optimal_steps=3,
        max_steps=8,
    ),
]

@pytest.mark.parametrize("spec", TASK_SPECS, ids=[s.task_id for s in TASK_SPECS])
def test_agent_task_completion(agent_fn, spec):
    trajectory = agent_fn(spec.description)
    assert spec.success_condition(trajectory), (
        f"Task '{spec.task_id}' not completed. Steps taken: {[s.tool_name for s in trajectory.steps]}"
    )

@pytest.mark.parametrize("spec", TASK_SPECS, ids=[s.task_id for s in TASK_SPECS])
def test_agent_tool_accuracy(agent_fn, spec):
    trajectory = agent_fn(spec.description)
    metrics = evaluate_tool_call_accuracy(trajectory, spec)
    assert metrics["required_call_accuracy"] == 1.0, (
        f"Required tool calls not made. Missing: {metrics['missed_tools']}"
    )
    assert metrics["tool_precision"] >= 0.80, (
        f"Tool precision {metrics['tool_precision']:.2f} too low. Hallucinated: {metrics['hallucinated_tools']}"
    )

@pytest.mark.parametrize("spec", TASK_SPECS, ids=[s.task_id for s in TASK_SPECS])
def test_agent_trajectory_quality(agent_fn, spec):
    trajectory = agent_fn(spec.description)
    quality = score_trajectory_quality(trajectory, spec)
    assert quality["quality_score"] >= 0.60, (
        f"Trajectory quality {quality['quality_score']:.2f} too low. "
        f"Errors: {quality['error_rate']:.2f}, Redundancy: {quality['redundancy_rate']:.2f}"
    )