Evaluation & Testing

from pydantic_ai import Agent
from pydantic_ai.models.test import TestModel

agent = Agent(
    'openai:gpt-4o',
    instructions='Help with calculations.'
)

@agent.tool_plain
def calculate(expression: str) -> str:
    return str(eval(expression))

def test_agent_has_calculate_tool():
    """Test that the calculate tool is available."""
    test_model = TestModel()

    result = agent.run_sync(
        'What is 2+2?',
        model=test_model  # Use test model instead of real LLM
    )

    # Check that tool is available
    tools = test_model.last_model_request_parameters.function_tools
    tool_names = [t.name for t in tools]

    assert 'calculate' in tool_names

def test_agent_returns_output():
    """Test that agent produces output."""
    test_model = TestModel()

    result = agent.run_sync('Hello', model=test_model)

    assert result.output is not None

from pydantic_ai import Agent, ModelMessage, ModelResponse, TextPart
from pydantic_ai.models.function import AgentInfo, FunctionModel

agent = Agent('openai:gpt-4o')

def test_with_custom_model():
    received_messages = []

    def mock_model(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse:
        # Record what was sent
        received_messages.extend(messages)
        # Return a fixed response
        return ModelResponse(parts=[TextPart(content='Mocked response')])

    result = agent.run_sync('Hello', model=FunctionModel(mock_model))

    assert len(received_messages) > 0
    assert result.output == 'Mocked response'

import pytest
from pydantic_ai import Agent, RunContext
from dataclasses import dataclass
from unittest.mock import Mock

@dataclass
class TestDeps:
    call_log: list = None

    def __post_init__(self):
        self.call_log = []

agent = Agent('openai:gpt-4o', deps_type=TestDeps)

@agent.tool
def fetch_data(ctx: RunContext[TestDeps], item_id: int) -> dict:
    ctx.deps.call_log.append(item_id)
    return {'id': item_id, 'name': f'Item {item_id}'}

def test_fetch_data_tool():
    """Test tool function directly."""
    deps = TestDeps()

    # Create mock context
    mock_ctx = Mock()
    mock_ctx.deps = deps

    # Call tool directly (bypass agent)
    result = fetch_data.__wrapped__(mock_ctx, 42)

    assert result == {'id': 42, 'name': 'Item 42'}
    assert deps.call_log == [42]

from pydantic import BaseModel
from pydantic_ai import Agent

class Evaluation(BaseModel):
    score: float  # 0 to 1
    reasoning: str
    suggestions: list[str]

evaluator = Agent(
    'openai:gpt-4o',
    output_type=Evaluation,
    instructions='''Evaluate the AI response.
    Score from 0 (terrible) to 1 (perfect).
    Consider: accuracy, helpfulness, clarity, safety.
    Provide reasoning and suggestions for improvement.'''
)

async def evaluate_response(question: str, response: str) -> Evaluation:
    result = await evaluator.run(f'''
    Question: {question}
    Response: {response}

    Evaluate this response.
    ''')
    return result.output

# Usage
eval_result = await evaluate_response(
    question="What is the capital of France?",
    response="Paris is the capital of France."
)
print(f"Score: {eval_result.score}")  # Score: 0.95
print(f"Reasoning: {eval_result.reasoning}")

import json
from datetime import datetime

class EvaluationTracker:
    def __init__(self, filepath: str):
        self.filepath = filepath
        self.history = []

    def record(self, test_name: str, score: float, details: dict):
        self.history.append({
            'timestamp': datetime.now().isoformat(),
            'test': test_name,
            'score': score,
            'details': details
        })
        self._save()

    def _save(self):
        with open(self.filepath, 'w') as f:
            json.dump(self.history, f, indent=2)

    def check_regression(self, test_name: str, current_score: float, threshold: float = 0.1):
        """Alert if score dropped significantly."""
        previous = [h for h in self.history if h['test'] == test_name]
        if previous:
            last_score = previous[-1]['score']
            if current_score < last_score - threshold:
                raise AssertionError(
                    f"Regression detected: {test_name} dropped from {last_score} to {current_score}"
                )