python -m venv venv
source venv/bin/activate
pip install pytest openai pandas langsmith
export OPENAI_API_KEY="sk-your-key"
export LANGCHAIN_API_KEY="ls-your-key"

# test_cases/manager.py
from pydantic import BaseModel
from typing import List, Dict, Optional
import json

class TestCase(BaseModel):
    id: str
    input: str
    expected_output: Optional[str] = None
    expected_tool: Optional[str] = None
    tags: List[str] = []
    difficulty: str = "medium"

class TestCaseManager:
    def __init__(self):
        self.test_cases: List[TestCase] = []

    def add_test_case(self, test_case: TestCase) -> None:
        self.test_cases.append(test_case)

    def load_from_file(self, file_path: str) -> None:
        with open(file_path, "r") as f:
            data = json.load(f)
        for item in data:
            self.test_cases.append(TestCase(**item))

    def save_to_file(self, file_path: str) -> None:
        data = [tc.model_dump() for tc in self.test_cases]
        with open(file_path, "w") as f:
            json.dump(data, f, indent=2)

    def get_by_tag(self, tag: str) -> List[TestCase]:
        return [tc for tc in self.test_cases if tag in tc.tags]

    def get_by_difficulty(self, difficulty: str) -> List[TestCase]:
        return [tc for tc in self.test_cases if tc.difficulty == difficulty]

    def create_test_suite(
        self, tags: List[str] = None, difficulty: str = None, max_tests: int = None
    ) -> List[TestCase]:
        suite = self.test_cases
        if tags:
            suite = [tc for tc in suite if any(t in tc.tags for t in tags)]
        if difficulty:
            suite = [tc for tc in suite if tc.difficulty == difficulty]
        if max_tests:
            suite = suite[:max_tests]
        return suite

# tracing/tracer.py
from typing import Dict, List, Any
import time
from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class TraceStep:
    step_id: int
    name: str
    input_data: Any
    output_data: Any
    duration_ms: float
    tokens_used: int = 0
    cost: float = 0.0
    metadata: Dict = field(default_factory=dict)

class ExecutionTracer:
    def __init__(self):
        self.traces: List[Dict] = []
        self.current_trace: Dict = None
        self.steps: List[TraceStep] = []

    def start_trace(self, test_case_id: str, input_data: str) -> None:
        self.current_trace = {
            "id": f"trace_{datetime.now().strftime('%Y%m%d_%H%M%S')}",
            "test_case_id": test_case_id,
            "input": input_data,
            "start_time": time.time(),
            "steps": [],
            "status": "running",
        }
        self.steps = []

    def add_step(
        self,
        name: str,
        input_data: Any,
        output_data: Any,
        duration_ms: float,
        tokens_used: int = 0,
        cost: float = 0.0,
    ) -> None:
        step = TraceStep(
            step_id=len(self.steps) + 1,
            name=name,
            input_data=input_data,
            output_data=output_data,
            duration_ms=duration_ms,
            tokens_used=tokens_used,
            cost=cost,
        )
        self.steps.append(step)

    def end_trace(self, output: str, success: bool = True) -> None:
        if not self.current_trace:
            return
        self.current_trace["output"] = output
        self.current_trace["success"] = success
        self.current_trace["end_time"] = time.time()
        self.current_trace["duration_ms"] = (
            self.current_trace["end_time"] - self.current_trace["start_time"]
        ) * 1000
        self.current_trace["steps"] = [
            {
                "step_id": s.step_id,
                "name": s.name,
                "duration_ms": s.duration_ms,
                "tokens_used": s.tokens_used,
                "cost": s.cost,
            }
            for s in self.steps
        ]
        self.current_trace["total_tokens"] = sum(s.tokens_used for s in self.steps)
        self.current_trace["total_cost"] = sum(s.cost for s in self.steps)
        self.traces.append(self.current_trace)
        self.current_trace = None
        self.steps = []

    def get_traces(self) -> List[Dict]:
        return self.traces

# metrics/calculator.py
from typing import Dict, List
import re

class MetricsCalculator:
    def calculate_accuracy(
        self, predicted: str, expected: str, method: str = "exact"
    ) -> float:
        if method == "exact":
            return 1.0 if predicted.strip() == expected.strip() else 0.0
        elif method == "contains":
            return 1.0 if expected.lower() in predicted.lower() else 0.0
        elif method == "semantic":
            return self._semantic_similarity(predicted, expected)
        return 0.0

    def _semantic_similarity(self, text1: str, text2: str) -> float:
        words1 = set(text1.lower().split())
        words2 = set(text2.lower().split())
        intersection = words1 & words2
        union = words1 | words2
        return len(intersection) / len(union) if union else 0.0

    def calculate_metrics(self, traces: List[Dict]) -> Dict:
        if not traces:
            return {"total": 0}
        total = len(traces)
        successes = sum(1 for t in traces if t.get("success"))
        avg_duration = sum(t.get("duration_ms", 0) for t in traces) / total
        avg_tokens = sum(t.get("total_tokens", 0) for t in traces) / total
        total_cost = sum(t.get("total_cost", 0) for t in traces)
        return {
            "total_tests": total,
            "successes": successes,
            "failures": total - successes,
            "success_rate": successes / total * 100,
            "avg_duration_ms": round(avg_duration, 2),
            "avg_tokens": round(avg_tokens, 2),
            "total_cost": round(total_cost, 4),
            "p50_duration": self._percentile([t.get("duration_ms", 0) for t in traces], 50),
            "p95_duration": self._percentile([t.get("duration_ms", 0) for t in traces], 95),
        }

    def _percentile(self, data: List[float], percentile: int) -> float:
        sorted_data = sorted(data)
        index = int(len(sorted_data) * percentile / 100)
        return sorted_data[min(index, len(sorted_data) - 1)]

    def compare_results(self, results1: Dict, results2: Dict) -> Dict:
        comparison = {}
        for key in results1:
            if key in results2 and isinstance(results1[key], (int, float)):
                diff = results2[key] - results1[key]
                pct = (diff / results1[key] * 100) if results1[key] != 0 else 0
                comparison[key] = {"baseline": results1[key], "comparison": results2[key], "diff": diff, "pct_change": pct}
        return comparison

# framework.py
from test_cases.manager import TestCaseManager, TestCase
from tracing.tracer import ExecutionTracer
from metrics.calculator import MetricsCalculator
from typing import Dict, List, Callable

class AgentEvaluationFramework:
    def __init__(self):
        self.test_manager = TestCaseManager()
        self.tracer = ExecutionTracer()
        self.metrics = MetricsCalculator()
        self.results: List[Dict] = []

    def evaluate_agent(
        self,
        agent_func: Callable,
        test_cases: List[TestCase] = None,
    ) -> Dict:
        if test_cases is None:
            test_cases = self.test_manager.test_cases
        for tc in test_cases:
            self.tracer.start_trace(tc.id, tc.input)
            start = time.time()
            try:
                output = agent_func(tc.input)
                duration = (time.time() - start) * 1000
                self.tracer.add_step("agent_execution", tc.input, output, duration)
                accuracy = self.metrics.calculate_accuracy(
                    output, tc.expected_output or ""
                )
                self.tracer.end_trace(output, success=accuracy > 0.5)
                self.results.append({
                    "test_case_id": tc.id,
                    "input": tc.input,
                    "expected": tc.expected_output,
                    "actual": output,
                    "accuracy": accuracy,
                    "success": accuracy > 0.5,
                })
            except Exception as e:
                self.tracer.end_trace(str(e), success=False)
                self.results.append({
                    "test_case_id": tc.id,
                    "input": tc.input,
                    "expected": tc.expected_output,
                    "actual": str(e),
                    "accuracy": 0.0,
                    "success": False,
                })
        return self.metrics.calculate_metrics(self.tracer.get_traces())

    def generate_report(self) -> str:
        metrics = self.metrics.calculate_metrics(self.tracer.get_traces())
        report = "# Agent Evaluation Report\n\n"
        report += f"## Summary\n"
        report += f"- Total Tests: {metrics['total_tests']}\n"
        report += f"- Success Rate: {metrics['success_rate']:.1f}%\n"
        report += f"- Avg Duration: {metrics['avg_duration_ms']:.0f}ms\n"
        report += f"- Avg Tokens: {metrics['avg_tokens']:.0f}\n"
        report += f"- Total Cost: ${metrics['total_cost']:.4f}\n\n"
        report += "## Failed Tests\n"
        for r in self.results:
            if not r["success"]:
                report += f"- {r['test_case_id']}: Expected '{r['expected'][:50]}' got '{r['actual'][:50]}'\n"
        return report

import pytest
from framework import AgentEvaluationFramework
from test_cases.manager import TestCase

def test_framework():
    framework = AgentEvaluationFramework()
    framework.test_manager.add_test_case(TestCase(
        id="test1", input="hello", expected_output="hi there"
    ))
    def mock_agent(x):
        return "hi there"
    results = framework.evaluate_agent(mock_agent)
    assert results["success_rate"] == 100.0

# main.py
from framework import AgentEvaluationFramework
from test_cases.manager import TestCase

def main():
    framework = AgentEvaluationFramework()
    framework.test_manager.add_test_case(TestCase(
        id="t1", input="What is 2+2?", expected_output="4"
    ))
    def my_agent(q):
        return "4"
    results = framework.evaluate_agent(my_agent)
    print(framework.generate_report())

if __name__ == "__main__":
    main()