Weather Agent

Weather Agent

여러 개의 도구를 등록해 두고, LLM이 순서대로 도구를 호출해야 답을 낼 수 있는 에이전트를 만들어 볼게요. 사용자가 "런던과 윌트셔의 날씨가 어때?"라고 묻는다면, 에이전트는 먼저 get_lat_lng 도구로 장소의 위도·경도를 알아낸 뒤, 그 값을 get_weather 도구에 넘겨 날씨를 조회해야 해요. 이 예시는 Pydantic AI에서 도구를 여러 개 쓰는 것, 에이전트 의존성, 스트리밍 텍스트 응답, 그리고 Gradio로 UI를 만드는 것까지 보여줍니다.

출처: 공식문서

예시가 보여주는 것

예시 실행하기

이 예시를 제대로 실행하려면 API 키 두 개를 추가하는 게 좋아요. (어느 하나가 없더라도 코드는 더미 데이터로 대체되므로, 필수는 아니에요):

의존성을 설치하고 환경 변수를 설정했다면, 아래처럼 실행해요.

python -m pydantic_ai_examples.weather_agent

uv run -m pydantic_ai_examples.weather_agent

예시 코드

weather_agent.py

from __future__ import annotations as _annotations

import asyncio
from dataclasses import dataclass
from typing import Any

import logfire
from httpx import AsyncClient
from pydantic import BaseModel

from pydantic_ai import Agent, RunContext

# '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')
logfire.instrument_pydantic_ai()


@dataclass
class Deps:
    client: AsyncClient


weather_agent = Agent(
    'openai:gpt-5-mini',
    # 'Be concise, reply with one sentence.' is enough for some models (like openai) to use
    # the below tools appropriately, but others like anthropic and gemini require a bit more direction.
    instructions='Be concise, reply with one sentence.',
    deps_type=Deps,
    retries=2,
)


class LatLng(BaseModel):
    lat: float
    lng: float


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

    Args:
        ctx: The context.
        location_description: A description of a location.
    """
    # NOTE: the response here will be random, and is not related to the location description.
    r = await ctx.deps.client.get(
        'https://demo-endpoints.pydantic.workers.dev/latlng',
        params={'location': location_description},
    )
    r.raise_for_status()
    return LatLng.model_validate_json(r.content)


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

    Args:
        ctx: The context.
        lat: Latitude of the location.
        lng: Longitude of the location.
    """
    # NOTE: the responses here will be random, and are not related to the lat and lng.
    temp_response, descr_response = await asyncio.gather(
        ctx.deps.client.get(
            'https://demo-endpoints.pydantic.workers.dev/number',
            params={'min': 10, 'max': 30},
        ),
        ctx.deps.client.get(
            'https://demo-endpoints.pydantic.workers.dev/weather',
            params={'lat': lat, 'lng': lng},
        ),
    )
    temp_response.raise_for_status()
    descr_response.raise_for_status()
    return {
        'temperature': f'{temp_response.text} °C',
        'description': descr_response.text,
    }


async def main():
    async with AsyncClient() as client:
        logfire.instrument_httpx(client, capture_all=True)
        deps = Deps(client=client)
        result = await weather_agent.run(
            'What is the weather like in London and in Wiltshire?', deps=deps
        )
        print('Response:', result.output)


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

주의해서 볼 부분이 하나 있어요. get_lat_lng은 장소 설명을 파라미터로 받아 좌표를 반환하는데, 이 데모 환경의 응답은 무작위라 실제 위치와 관계없어요. get_weather는 LatLng를 받아 온도와 날씨 설명을 반환해요. 두 도구 모두 ctx.deps.client로 HTTP 클라이언트에 접근하는데, 이게 바로 에이전트 의존성이에요. deps_type=Deps로 의존성 타입을 선언하고, run을 호출할 때 deps=deps로 실제 값을 넘겨주는 구조예요.

UI 실행하기

Gradio는 Python만으로 AI 웹 애플리케이션을 만들게 해주는 프레임워크예요. 채팅 컴포넌트와 에이전트 지원이 내장돼 있어서, UI 전체를 Python 파일 하나로 구현할 수 있어요. 날씨 에이전트 UI는 아래처럼 생겼어요.

pip install gradio>=6.7.0
python/uv-run -m pydantic_ai_examples.weather_agent_gradio

UI 코드

weather_agent_gradio.py

from __future__ import annotations as _annotations

import json

from httpx import AsyncClient
from pydantic import BaseModel

from pydantic_ai import ToolCallPart, ToolReturnPart
from pydantic_ai_examples.weather_agent import Deps, weather_agent

try:
    import gradio as gr
except ImportError as e:
    raise ImportError(
        'Please install gradio with `pip install gradio`. You must use python>=3.10.'
    ) from e

TOOL_TO_DISPLAY_NAME = {'get_lat_lng': 'Geocoding API', 'get_weather': 'Weather API'}

client = AsyncClient()
deps = Deps(client=client)


async def stream_from_agent(prompt: str, chatbot: list[dict], past_messages: list):
    chatbot.append({'role': 'user', 'content': prompt})
    yield gr.Textbox(interactive=False, value=''), chatbot, gr.skip()
    async with weather_agent.run_stream(
        prompt, deps=deps, message_history=past_messages
    ) as result:
        for message in result.new_messages():
            for call in message.parts:
                if isinstance(call, ToolCallPart):
                    call_args = call.args_as_json_str()
                    metadata = {
                        'title': f'🛠️ Using {TOOL_TO_DISPLAY_NAME[call.tool_name]}',
                    }
                    if call.tool_call_id is not None:
                        metadata['id'] = call.tool_call_id

                    gr_message = {
                        'role': 'assistant',
                        'content': 'Parameters: ' + call_args,
                        'metadata': metadata,
                    }
                    chatbot.append(gr_message)
                if isinstance(call, ToolReturnPart):
                    for gr_message in chatbot:
                        if (gr_message.get('metadata') or {}).get(
                            'id', ''
                        ) == call.tool_call_id:
                            if isinstance(call.content, BaseModel):
                                json_content = call.content.model_dump_json()
                            else:
                                json_content = json.dumps(call.content)
                            gr_message['content'] += f'\nOutput: {json_content}'
                yield gr.skip(), chatbot, gr.skip()
        chatbot.append({'role': 'assistant', 'content': ''})
        async for message in result.stream_text():
            chatbot[-1]['content'] = message
            yield gr.skip(), chatbot, gr.skip()
        past_messages = result.all_messages()

        yield gr.Textbox(interactive=True), gr.skip(), past_messages


async def handle_retry(chatbot, past_messages: list, retry_data: gr.RetryData):
    new_history = chatbot[: retry_data.index]
    previous_prompt = chatbot[retry_data.index]['content']
    past_messages = past_messages[: retry_data.index]
    async for update in stream_from_agent(previous_prompt, new_history, past_messages):
        yield update


def undo(chatbot, past_messages: list, undo_data: gr.UndoData):
    new_history = chatbot[: undo_data.index]
    past_messages = past_messages[: undo_data.index]
    return chatbot[undo_data.index]['content'], new_history, past_messages


def select_data(message: gr.SelectData) -> str:
    return message.value['text']


with gr.Blocks() as demo:
    gr.HTML(
        """
<div style="display: flex; justify-content: center; align-items: center; gap: 2rem; padding: 1rem; width: 100%">
    <img src="https://pydantic.dev/docs/ai/img/logo-white.svg" style="max-width: 200px; height: auto">
    <div>
        <h1 style="margin: 0 0 1rem 0">Weather Assistant</h1>
        <h3 style="margin: 0 0 0.5rem 0">
            This assistant answer your weather questions.
        </h3>
    </div>
</div>
"""
    )
    past_messages = gr.State([])
    chatbot = gr.Chatbot(
        label='Packing Assistant',
        avatar_images=(None, 'https://pydantic.dev/docs/ai/img/logo-white.svg'),
        examples=[
            {'text': 'What is the weather like in Miami?'},
            {'text': 'What is the weather like in London?'},
        ],
    )
    with gr.Row():
        prompt = gr.Textbox(
            lines=1,
            show_label=False,
            placeholder='What is the weather like in New York City?',
        )
    generation = prompt.submit(
        stream_from_agent,
        inputs=[prompt, chatbot, past_messages],
        outputs=[prompt, chatbot, past_messages],
    )
    chatbot.example_select(select_data, None, [prompt])
    chatbot.retry(
        handle_retry, [chatbot, past_messages], [prompt, chatbot, past_messages]
    )
    chatbot.undo(undo, [chatbot, past_messages], [prompt, chatbot, past_messages])


if __name__ == '__main__':
    demo.launch()

UI 흐름을 간단히 짚어볼게요. stream_from_agent는 run_stream으로 에이전트를 실행하면서, 메시지의 parts를 순회해 도구 호출(ToolCallPart)이면 "🛠️ Using ..." 배지와 함께 파라미터를, 도구 반환(ToolReturnPart)이면 그 결과를 채팅에 붙여 넣어요. 최종 응답은 stream_text()로 점진적으로 이어 붙여요. 이때 message_history=past_messages로 이전 대화를 넘겨주기 때문에, 다중 턴 대화가 가능해요.

더 알아보기 (Learn more)