So in the last article we created a very simple agent which takes location as an input from the user and uses the necessary tools to provide a response to the user. But let’s be honest—the response was pretty bland. Large Language Models (LLMs) are known for their engaging, conversational nature, so let’s take things up a notch. This time, we’ll create our first multi-agent system, allowing us to generate richer, more dynamic responses for our users.

Why Multi Agent?

It's very important to answer "why" before making any system design decisions (or even life decisions 🤷‍♂️) to ensure we're not using a sword to cut a carrot.

First, let's review what an Agent is. An agent is a highly skilled helper that knows how to perform certain tasks very well to satisfy its master. If you teach this agent many tasks, as you add more capabilities over time, it might become difficult to manage and could lower the quality of its work due to task confusion. This is similar to the teams you work with now—some engineers are great at coding but not as good at presenting, while managers might excel at planning but not at marketing, etc.

The leaner your agent is, the better it will perform the task at hand, and it will be easier to manage. However, there are drawbacks. The more agents you create, the more latency there will be in providing a response to the end user, as you need to identify the right agent for the task, and that agent might rely on another agent to complete it. So, there's no right or wrong answer; it comes down to understanding the pros and cons and making a decision based on that. Now that you know the why, let's start cooking 👨‍🍳

Objective

We will create a response agent that will take JSON input from the weather agent to generate a response for the user.

High Level Flow

Response Agent

from pydantic_ai import Agent

response_agent = Agent(
    'openai:gpt-4o',
    system_prompt=(
        'Provide a detailed weather description based on the response from the weather agent. '
        'Include temperature, humidity, wind speed, and any notable conditions like rain, snow, or storms. '
        'Describe how the weather might feel to a person, such as whether it is comfortable, chilly, or humid.'
    ),
    retries=2,
    deps_type=str,
)

Weather Agent

from __future__ import annotations as _annotations
from dataclasses import dataclass
import logfire
from httpx import AsyncClient
from pydantic_ai import Agent, ModelRetry, RunContext

from response_generator_agent import response_agent

@dataclass
class Deps:
    client: AsyncClient
    weather_api_key: str | None
    geo_api_key: str | None


weather_agent = Agent(
    'openai:gpt-4o',
    system_prompt=(
        'Use the `get_lat_lng` tool to get the latitude and longitude of the locations, '
        'then use the `get_weather` tool to get the weather, '
        'then send the response from the `get_weather` tool as JSON'
    ),
    deps_type=Deps,
    retries=2,
)


@weather_agent.tool
async def get_lat_lng(
    ctx: RunContext[Deps], location_description: str
) -> dict[str, float]:
    """Get the latitude and longitude of a location.

    Args:
        ctx: The context.
        location_description: A description of a location.
    """
    if ctx.deps.geo_api_key is None:
        # if no API key is provided, return a dummy response (London)
        return {'lat': 51.1, 'lng': -0.1}

    params = {
        'q': location_description,
        'api_key': ctx.deps.geo_api_key,
    }
    with logfire.span('calling geocode API', params=params) as span:
        r = await ctx.deps.client.get('https://geocode.maps.co/search', params=params)
        r.raise_for_status()
        data = r.json()
        span.set_attribute('response', data)

    if data:
        return {'lat': data[0]['lat'], 'lng': data[0]['lon']}
    else:
        raise ModelRetry('Could not find the location')


@weather_agent.tool
async def get_weather(ctx: RunContext[Deps], lat: float, lng: float) -> dict[str, str]:
    """Get the weather at a location.

    Args:
        ctx: The context.
        lat: Latitude of the location.
        lng: Longitude of the location.
    """
    if ctx.deps.weather_api_key is None:
        # if no API key is provided, return a dummy response
        return {'temperature': '21 °C', 'description': 'Sunny'}

    params = {
        'apikey': ctx.deps.weather_api_key,
        'location': f'{lat},{lng}',
        'units': 'metric',
    }
    with logfire.span('calling weather API', params=params) as span:
        r = await ctx.deps.client.get(
            'https://api.tomorrow.io/v4/weather/realtime', params=params
        )
        r.raise_for_status()
        data = r.json()
        span.set_attribute('response', data)

    values = data['data']['values']
    # https://docs.tomorrow.io/reference/data-layers-weather-codes
    code_lookup = {
        1000: 'Clear, Sunny',
        1100: 'Mostly Clear',
        1101: 'Partly Cloudy',
        1102: 'Mostly Cloudy',
        1001: 'Cloudy',
        2000: 'Fog',
        2100: 'Light Fog',
        4000: 'Drizzle',
        4001: 'Rain',
        4200: 'Light Rain',
        4201: 'Heavy Rain',
        5000: 'Snow',
        5001: 'Flurries',
        5100: 'Light Snow',
        5101: 'Heavy Snow',
        6000: 'Freezing Drizzle',
        6001: 'Freezing Rain',
        6200: 'Light Freezing Rain',
        6201: 'Heavy Freezing Rain',
        7000: 'Ice Pellets',
        7101: 'Heavy Ice Pellets',
        7102: 'Light Ice Pellets',
        8000: 'Thunderstorm',
    }
    return {
        **values,
        'description': code_lookup.get(values['weatherCode'], 'Unknown'),
    }

Main File

import asyncio
import os
from httpx import AsyncClient
import logfire
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

from weather_agent import weather_agent, Deps
from response_generator_agent import response_agent

# 'if-token-present' means nothing will be sent (and the example will work) if you don't have logfire configured
logfire.configure(send_to_logfire='if-token-present')


async def main():
    async with AsyncClient() as client:
        # Get location from user input
        location = input("Enter a location: ")
        # create a free API key at https://www.tomorrow.io/weather-api/
        weather_api_key = os.getenv('WEATHER_API_KEY')
        # create a free API key at https://geocode.maps.co/
        geo_api_key = os.getenv('GEO_API_KEY')
        deps = Deps(
            client=client, weather_api_key=weather_api_key, geo_api_key=geo_api_key
        )
        result = await weather_agent.run(
            f'What is the weather like in {location}?', deps=deps
        )
        response = await response_agent.run(
            f'Please generate a response based on the weather data for {location}: {result.data}',
            deps=result.data,
        )
        print('Response:', response.data)


if __name__ == "__main__":
    asyncio.run(main())

Output

The code for the ABC of multi agent is available here.

ABC of Multi Agent