How to test agents? integration tests or evals?

[AlexEnrique]

Question

Hi. I am trying to test my agents without doing so manually, but I still didn't understand how to do it well.

I need to run live tests, hitting the LLM API.

Here are some cases I need to test:

• Test with different inputs if a tool is called as expected. Maybe I need to mock a tool
• Test the impacts of changes in instructions with complete workflows, for different inputs (many tool calls, testing a whole conversation)
• Given some message histories that caused unexpected model behavior exceptions, I would like to write tests for changes in the tools and prompts in order to see if the issue was fixed
• write tests to validade the agent against new workflows, observing the message history exchanged (could be through logfire)
• Evaluate the impact of switching models, if the agent continue to perform as expected

Could someone help me figure out how to perform these tests?

Additional Context

No response

[nurikk]

I find DeepEval quite helpful in this https://github.com/confident-ai/deepeval

You simply write your usual pytest code and have whatever assertions you want (LLM judges or vanilla assertions)

[samGende]

I had a similar question with how to evaluate Agents with various models. The solution I found was to use override within a factory function allowing you to switch models.

[AlexEnrique]

I think the main problem I have is to test complex flows. My agents are chatbots, so I need to test whole conversations, see if tools are being called when expected, etc.

Looking at pydantic-evals and deepevals, it seems they were designed to test only one output given an input (perhaps with a message history). I need to test full conversations and evaluate how well the agent perform.

I think I will have to build something for this.

[DouweM]

@AlexEnrique With pydantic_evals, you can have the evaluated function return the entire run result ((await agent.run(...))) or message history: (.all_messages()) rather than just the output and then use a custom evaluator to test that it has the expected ToolCallPart.

[AlexEnrique]

Thanks @DouweM , I will try this approach and come back to tell if it worked

[AlexEnrique]

@DouweM, I'm playing with pydantic-evals and would like to know if it is possible to store the message history in datasets, persisting it to a file. Here is an example of what I am attempting (the loading part fails)

"""
Eval with different conversation histories in the dataset.
Each test case can have its own conversation context.
"""

from dataclasses import dataclass
from typing import Literal

from pydantic import BaseModel, Field
from pydantic_ai.messages import (
    ModelRequest,
    ModelResponse,
    TextPart,
    ToolCallPart,
    ToolReturnPart,
    UserPromptPart,
)
from pydantic_evals import Case, Dataset
from pydantic_evals.evaluators import Evaluator, EvaluatorContext

type SpecialistType = Literal['agendamento', 'atendimento']


# Pydantic Evals Models
class SupervisorInputs(BaseModel, use_attribute_docstrings=True):
    """Inputs for the supervisor test."""

    user_message: str
    """The user message that needs to be routed."""

    message_history: list[ModelRequest | ModelResponse]
    """Message history of the conversation up to this point."""


class SupervisorOutput(BaseModel, use_attribute_docstrings=True):
    """Expected supervisor output."""

    agent: SpecialistType = Field(
        description='Agent to which the message should be sent.',
    )


class SupervisorMetadata(BaseModel, use_attribute_docstrings=True):
    """Metadata about the test case."""

    scenario: str
    """Description of the scenario being tested."""

    conversation_context: str
    """Conversation context (e.g., 'agendamento_em_andamento', 'inicio_conversa')."""


class _SupervisorResult(BaseModel):
    """Internal supervisor agent result."""

    agent: SpecialistType = Field(
        description='Agent to which the message should be sent.',
    )


# Custom evaluator
@dataclass
class AgentRoutingEvaluator(Evaluator):
    """Evaluator that checks if the agent was routed correctly."""

    async def evaluate(
        self,
        ctx: EvaluatorContext[
            SupervisorInputs, SupervisorOutput, SupervisorMetadata
        ],
    ) -> float:
        """
        Evaluates if the routing was correct.

        Args:
            ctx: Evaluator context with inputs, outputs and expected_output.

        Returns:
            1.0 if routing was correct, 0.0 otherwise.
        """
        if ctx.output is None:
            return 0.0

        assert ctx.expected_output is not None
        return 1.0 if ctx.output.agent == ctx.expected_output.agent else 0.0


# Reusable message histories
HISTORY_AGENDAMENTO_EM_ANDAMENTO = [
    ModelRequest(
        parts=[UserPromptPart(content='I want to schedule a haircut')]
    ),
    ModelResponse(
        parts=[
            ToolCallPart(
                tool_name='get_services',
                args='{}',
                tool_call_id='call_123',
            ),
        ],
    ),
    ModelRequest(
        parts=[
            ToolReturnPart(
                tool_name='get_services',
                content='[{"id": 3, "name": "Men\'s Haircut", "price": "$40.00"}]',
                tool_call_id='call_123',
            ),
        ],
    ),
    ModelResponse(
        parts=[
            TextPart(
                content="The Men's Haircut costs $40.00. What date would you like?",
            ),
        ],
    ),
]

HISTORY_INICIO_CONVERSA = [
    ModelRequest(parts=[UserPromptPart(content='hi')]),
    ModelResponse(
        parts=[
            TextPart(
                content='Hello! How can I help you today?',
            ),
        ],
    ),
]

HISTORY_CONSULTANDO_SERVICOS = [
    ModelRequest(
        parts=[UserPromptPart(content='What services do you offer?')]
    ),
    ModelResponse(
        parts=[
            ToolCallPart(
                tool_name='get_services',
                args='{}',
                tool_call_id='call_456',
            ),
        ],
    ),
    ModelRequest(
        parts=[
            ToolReturnPart(
                tool_name='get_services',
                content='[{"id": 3, "name": "Men\'s Haircut", "price": "$40.00"}, {"id": 4, "name": "Beard Trim", "price": "$30.00"}]',
                tool_call_id='call_456',
            ),
        ],
    ),
    ModelResponse(
        parts=[
            TextPart(
                content="We offer Men's Haircut ($40) and Beard Trim ($30). Which one would you like to schedule?",
            ),
        ],
    ),
]


# Dataset with different conversation contexts
dataset = Dataset[SupervisorInputs, SupervisorOutput, SupervisorMetadata](
    cases=[
        Case(
            name='Question about date during scheduling',
            inputs=SupervisorInputs(
                user_message='Do you have availability for October 10th?',
                message_history=HISTORY_AGENDAMENTO_EM_ANDAMENTO,
            ),
            expected_output=SupervisorOutput(agent='agendamento'),
            metadata=SupervisorMetadata(
                scenario='Client is in scheduling process and asks about specific date',
                conversation_context='agendamento_em_andamento',
            ),
            evaluators=(AgentRoutingEvaluator(),),
        ),
        Case(
            name='Vague question during scheduling',
            inputs=SupervisorInputs(
                user_message='When do you have availability?',
                message_history=HISTORY_AGENDAMENTO_EM_ANDAMENTO,
            ),
            expected_output=SupervisorOutput(agent='agendamento'),
            metadata=SupervisorMetadata(
                scenario='Client is in scheduling process and asks vague question about availability',
                conversation_context='agendamento_em_andamento',
            ),
            evaluators=(AgentRoutingEvaluator(),),
        ),
        Case(
            name='Price question at conversation start',
            inputs=SupervisorInputs(
                user_message='How much does it cost?',
                message_history=HISTORY_INICIO_CONVERSA,
            ),
            expected_output=SupervisorOutput(agent='atendimento'),
            metadata=SupervisorMetadata(
                scenario='Client asks about price without scheduling context',
                conversation_context='inicio_conversa',
            ),
            evaluators=(AgentRoutingEvaluator(),),
        ),
        Case(
            name='Scheduling confirmation after service consultation',
            inputs=SupervisorInputs(
                user_message='I want to schedule',
                message_history=HISTORY_CONSULTANDO_SERVICOS,
            ),
            expected_output=SupervisorOutput(agent='agendamento'),
            metadata=SupervisorMetadata(
                scenario='Client decides to schedule after consulting services',
                conversation_context='pos_consulta_servicos',
            ),
            evaluators=(AgentRoutingEvaluator(),),
        ),
        Case(
            name='Question without history (start)',
            inputs=SupervisorInputs(
                user_message='I want to book an appointment',
                message_history=[],
            ),
            expected_output=SupervisorOutput(agent='agendamento'),
            metadata=SupervisorMetadata(
                scenario='First client message already mentioning scheduling',
                conversation_context='primeira_mensagem',
            ),
            evaluators=(AgentRoutingEvaluator(),),
        ),
    ],
)


# Main function
def main():
    """Main function to save and load datasets."""
    print('=== Saving datasets ===')
    dataset.to_file(
        'dump_dataset_with_message_history.json',
        fmt='json',
        custom_evaluator_types=(AgentRoutingEvaluator,),
    )

    dataset2 = Dataset[
        SupervisorInputs, SupervisorOutput, SupervisorMetadata
    ].from_file(
        'dump_dataset_with_message_history.json',
        fmt='json',
        custom_evaluator_types=(AgentRoutingEvaluator,),
    )

    print('=== Loading datasets ===')
    print(dataset2)
    print(dataset2.cases)


if __name__ == '__main__':
    main()

[AlexEnrique]

Also, it would be very nice to evaluate chatbots if we could copy/generate the message history from a manual run.

I'm currently running conversations manually to generate message histories using logfire's raw data tab. But this requires a big effort to generate even some simple message histories to create datasets for evals, and I don't know if an LLM could help me generate exactly what I am trying to reproduce when generating these datasets manually.

[DouweM]

if it is possible to store the message history in datasets, persisting it to a file

@AlexEnrique Good catch, fixed in #3079

Also, it would be very nice to evaluate chatbots if we could copy/generate the message history from a manual run.

Are you looking for the all_messages() method on the agent run result? See https://ai.pydantic.dev/message-history/