How to Test Autonomous Agents: Evaluation Frameworks for Production Reliability

import json
import asyncio
from dataclasses import dataclass, field
from typing import Callable, Any
from langsmith import Client as LangSmithClient
from ragas import evaluate as ragas_evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_recall
from datasets import Dataset
from braintrust import Eval, Score
from openai import OpenAI


@dataclass
class GoldenExample:
    input: str
    expected_output: str
    expected_tool_calls: list[str]  # ordered list of expected tool names
    context_docs: list[str]  # relevant docs (for RAG eval)
    metadata: dict = field(default_factory=dict)


@dataclass
class EvalResult:
    input: str
    actual_output: str
    actual_tool_calls: list[str]
    retrieved_context: list[str]
    task_complete: bool
    tool_call_accuracy: float
    faithfulness_score: float
    relevancy_score: float
    latency_ms: float


class AgentEvaluationHarness:
    """
    Evaluation harness combining RAGAS metrics, tool call accuracy,
    and Braintrust human eval integration.
    """

    def __init__(self, agent_fn: Callable, golden_dataset_path: str):
        self.agent_fn = agent_fn  # async fn(input: str) -> dict
        self.golden_dataset = self._load_golden(golden_dataset_path)
        self.langsmith = LangSmithClient()
        self.openai = OpenAI()

    def _load_golden(self, path: str) -> list[GoldenExample]:
        with open(path) as f:
            raw = json.load(f)
        return [GoldenExample(**ex) for ex in raw["examples"]]

    def _tool_call_accuracy(self, expected: list[str], actual: list[str]) -> float:
        if not expected:
            return 1.0 if not actual else 0.0
        correct = sum(1 for e, a in zip(expected, actual) if e == a)
        coverage = correct / len(expected)
        length_penalty = max(0.0, 1.0 - abs(len(actual) - len(expected)) * 0.1)
        return coverage * length_penalty

    def _llm_judge_task_complete(self, input_: str, output: str, expected: str) -> bool:
        prompt = f"""Does the following agent output successfully address the task?
Task input: {input_}
Expected behaviour: {expected}
Actual output: {output}
Answer only YES or NO."""
        resp = self.openai.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            temperature=0,
            max_tokens=5,
        )
        return resp.choices[0].message.content.strip().upper() == "YES"

    async def run_single(self, example: GoldenExample) -> EvalResult:
        import time
        start = time.perf_counter()
        result = await self.agent_fn(example.input)
        latency_ms = (time.perf_counter() - start) * 1000

        actual_output = result.get("output", "")
        actual_tools = result.get("tool_calls", [])
        retrieved_ctx = result.get("retrieved_context", [])

        tool_acc = self._tool_call_accuracy(example.expected_tool_calls, actual_tools)
        task_ok = self._llm_judge_task_complete(
            example.input, actual_output, example.expected_output
        )

        # RAGAS evaluation for RAG quality
        ragas_data = Dataset.from_dict({
            "question": [example.input],
            "answer": [actual_output],
            "contexts": [retrieved_ctx],
            "ground_truth": [example.expected_output],
        })
        ragas_result = ragas_evaluate(
            ragas_data, metrics=[faithfulness, answer_relevancy, context_recall]
        )

        return EvalResult(
            input=example.input,
            actual_output=actual_output,
            actual_tool_calls=actual_tools,
            retrieved_context=retrieved_ctx,
            task_complete=task_ok,
            tool_call_accuracy=tool_acc,
            faithfulness_score=ragas_result["faithfulness"],
            relevancy_score=ragas_result["answer_relevancy"],
            latency_ms=latency_ms,
        )

    async def run_full_eval(self) -> dict:
        results = await asyncio.gather(
            *[self.run_single(ex) for ex in self.golden_dataset]
        )
        n = len(results)
        summary = {
            "n": n,
            "task_completion_rate": sum(r.task_complete for r in results) / n,
            "avg_tool_call_accuracy": sum(r.tool_call_accuracy for r in results) / n,
            "avg_faithfulness": sum(r.faithfulness_score for r in results) / n,
            "avg_relevancy": sum(r.relevancy_score for r in results) / n,
            "p50_latency_ms": sorted(r.latency_ms for r in results)[n // 2],
            "p99_latency_ms": sorted(r.latency_ms for r in results)[int(n * 0.99)],
        }
        print(json.dumps(summary, indent=2))
        return summary


# Usage — wire in your agent function
async def my_agent(input_text: str) -> dict:
    # Replace with your actual agent call
    return {"output": "...", "tool_calls": ["search_kb", "route_ticket"],
            "retrieved_context": ["doc1 text", "doc2 text"]}


if __name__ == "__main__":
    harness = AgentEvaluationHarness(
        agent_fn=my_agent,
        golden_dataset_path="eval/golden_dataset_v3.json"
    )
    asyncio.run(harness.run_full_eval())