에이전트

에이전트 (Agents)

에이전트는 Pydantic AI에서 LLM과 상호작용하는 일차적 인터페이스예요. 어떤 경우엔 단일 에이전트가 애플리케이션 전체를 제어하지만, 여러 에이전트가 상호작용해 더 복잡한 워크플로를 구현할 수도 있어요.

출처: 공식문서 — Agents

소개

Agent 클래스는 전체 API 문서가 있지만, 개념적으로 에이전트는 다음의 컨테이너로 볼 수 있어요:

컴포넌트	설명
지시사항(Instructions)	개발자가 LLM을 위해 작성한 지시 집합
함수 도구와 toolsets	응답 생성 중 정보를 얻기 위해 LLM이 호출할 수 있는 함수
구조화된 출력 타입	런 끝에 LLM이 반환해야 하는 구조화된 데이터 타입(지정한 경우)
의존성 타입 제약	동적 지시 함수·도구·출력 함수가 실행될 때 쓰는 의존성
LLM 모델	에이전트와 연관된 선택적 기본 LLM 모델. 에이전트 실행 시에도 지정 가능
모델 설정	요청을 미세 조정하는 선택적 기본 모델 설정. 실행 시에도 지정 가능
Capabilities	에이전트 동작을 확장하는 도구·훅·지시·모델 설정의 재사용 번들

이 각각은 개별로 설정할 수 있지만, capabilities로 관련 동작을 컴포즈·공유·설정 파일에서 로드하기 쉬운 재사용 단위로 묶을 수 있어요.

타이핑 측면에서 에이전트는 의존성·출력 타입에 대해 제네릭이에요. 예를 들어 Foobar 타입 의존성이 필요하고 list[str] 타입 출력을 만드는 에이전트는 Agent[Foobar, list[str]] 타입이에요. 실제로는 이런 걸 신경 쓸 필요가 없어요. 그냥 IDE가 올바른 타입을 쓰고 있는지 알려줄 뿐이고, 정적 타입 검사를 쓰기로 한다면 Pydantic AI와 잘 작동해요.

룰렛 휠을 시뮬레이션하는 장난감 예제:

from pydantic_ai import Agent, RunContext

roulette_agent = Agent(  # (1)
  'openai:gpt-5.2',
  deps_type=int,
  output_type=bool,
  system_prompt=(
      'Use the `roulette_wheel` function to see if the '
      'customer has won based on the number they provide.'
  ),
)


@roulette_agent.tool
async def roulette_wheel(ctx: RunContext[int], square: int) -> str:  # (2)
  """check if the square is a winner"""
  return 'winner' if square == ctx.deps else 'loser'


# Run the agent
success_number = 18  # (3)
result = roulette_agent.run_sync('Put my money on square eighteen', deps=success_number)
print(result.output)  # (4)
#> True

result = roulette_agent.run_sync('I bet five is the winner', deps=success_number)
print(result.output)
#> False

정수 의존성을 기대하고 불리언 출력을 만드는 에이전트를 만들어요. 이 에이전트는 Agent[int, bool] 타입을 가져요.

스퀘어가 승자인지 확인하는 도구를 정의해요. 여기서 RunContext는 의존성 타입 int로 파라미터화돼요. 의존성 타입을 잘못 쓰면 타이핑 오류를 받아요.

실제로는 random.randint(0, 36) 같은 난수를 쓰고 싶을 거예요.

result.output은 스퀘어가 승자인지 나타내는 불리언이에요. Pydantic이 출력 검증을 수행하고, 에이전트의 output_type 제네릭 파라미터에서 타입이 도출되므로 bool로 타이핑돼요.

에이전트는 FastAPI 앱처럼 재사용을 위해 설계됐어요 — 에이전트 하나를 인스턴스화해 작은 FastAPI 앱이나 APIRouter처럼 앱 전역에서 쓸 수도, 원하는 만큼 동적으로 만들 수도 있어요. 둘 다 유효하고 지원되는 사용 방식이에요.

에이전트 실행하기

에이전트를 실행하는 다섯 가지 방법:

1. agent.run() — 완료된 응답을 담은 RunResult를 반환하는 비동기 함수.
2. agent.run_sync() — 완료된 응답을 담은 RunResult를 반환하는 평범한 동기 함수(내부적으로 loop.run_until_complete(self.run())를 호출할 뿐).
3. agent.run_stream() — 텍스트와 구조화된 출력을 비동기 이터러블로 스트리밍하는 메서드를 담은 StreamedRunResult를 반환하는 비동기 컨텍스트 매니저. agent.run_stream_sync()은 같은 메서드들의 동기 버전을 담은 StreamedRunResultSync를 반환하는 동기 변형.
4. agent.run_stream_events() — 최종 런 결과를 담은 AgentRunResultEvent로 끝나는 AgentStreamEvent들의 비동기 이터레이터를 내놓는 비동기 컨텍스트 매니저.
5. agent.iter() — 에이전트의 기저 Graph 노드들의 비동기 이터러블인 AgentRun을 반환하는 컨텍스트 매니저.

처음 네 가지를 보여주는 간단한 예제:

from pydantic_ai import Agent, AgentRunResultEvent, AgentStreamEvent

agent = Agent('openai:gpt-5.2')

result_sync = agent.run_sync('What is the capital of Italy?')
print(result_sync.output)
#> The capital of Italy is Rome.


async def main():
    result = await agent.run('What is the capital of France?')
    print(result.output)
    #> The capital of France is Paris.

    async with agent.run_stream('What is the capital of the UK?') as response:
        async for text in response.stream_text():
            print(text)
            #> The capital of
            #> The capital of the UK is
            #> The capital of the UK is London.

    collected: list[AgentStreamEvent | AgentRunResultEvent] = []
    async with agent.run_stream_events('What is the capital of Mexico?') as events:
        async for event in events:
            collected.append(event)
    print(collected)
    """
    [
        PartStartEvent(index=0, part=TextPart(content='The capital of ')),
        FinalResultEvent(tool_name=None, tool_call_id=None),
        PartDeltaEvent(index=0, delta=TextPartDelta(content_delta='Mexico is Mexico ')),
        PartDeltaEvent(index=0, delta=TextPartDelta(content_delta='City.')),
        PartEndEvent(
            index=0, part=TextPart(content='The capital of Mexico is Mexico City.')
        ),
        AgentRunResultEvent(
            result=AgentRunResult(output='The capital of Mexico is Mexico City.')
        ),
    ]
    """

(이 예제를 실행하려면 asyncio를 import 하고 asyncio.run(main())을 추가하면 돼요. 다른 변경은 필요 없어요.)

이전 런의 메시지를 넘겨 대화를 이어가거나 맥락을 줄 수도 있는데, 메시지와 채팅 기록에 설명돼 있어요.

스트리밍 이벤트와 최종 출력

위 예제처럼 run_stream()은 에이전트의 최종 출력이 들어오는 대로 스트리밍하기 쉽게 해줘요. 또한 선택적 event_stream_handler 인수를 받아, 최종 출력이 만들어지기 전에 런 중 무슨 일이 일어나는지 통찰을 얻을 수 있어요. 실시간 세션 중에는 같은 핸들러 스트림에 실시간 전용 RealtimeEvent 멤버도 담길 수 있어요.

아래 예제는 이벤트와 텍스트 출력을 스트리밍하는 법을 보여줘요. 구조화된 출력을 스트리밍할 수도 있어요.

참고 — run_stream()과 run_stream_sync() 메서드는 출력 타입과 일치하는 첫 번째 출력(텍스트, 출력 도구 호출, 지연 도구 호출일 수 있음)을 에이전트 런의 최종 출력으로 간주해요. 모델이 이 "최종" 출력 이후에 (추가) 도구 호출을 만들어도요.

이런 "매달린(dangling)" 도구 호출은 에이전트의 end_strategy가 'graceful'이나 'exhaustive'로 설정되지 않는 한 실행되지 않고, 그렇게 해도 그 결과는 에이전트 런이 이미 완료된 것으로 간주되므로 모델에 다시 보내지지 않아요. 요컨대 모델이 도구 호출과 텍스트를 모두 반환하고 에이전트 출력 타입이 str이라면, 기본 설정에서는 스트리밍 모드에서 도구 호출이 실행되지 않아요.

에이전트가 도구 호출을 수행할 때 항상 계속 실행하고 싶고, 모델의 스트리밍 응답과 에이전트의 도구 실행에서 모든 이벤트를 스트리밍하고 싶다면, 다음 섹션에 설명된 대로 agent.run_stream_events()나 agent.iter()를 대신 쓰세요.

import asyncio
from collections.abc import AsyncIterable
from datetime import date

from pydantic_ai import (
    Agent,
    AgentStreamEvent,
    FinalResultEvent,
    FunctionToolCallEvent,
    FunctionToolResultEvent,
    PartDeltaEvent,
    PartStartEvent,
    RunContext,
    TextPartDelta,
    ThinkingPartDelta,
    ToolCallPartDelta,
)

weather_agent = Agent(
    'openai:gpt-5.2',
    system_prompt='Providing a weather forecast at the locations the user provides.',
)


@weather_agent.tool
async def weather_forecast(
    ctx: RunContext,
    location: str,
    forecast_date: date,
) -> str:
    return f'The forecast in {location} on {forecast_date} is 24°C and sunny.'


output_messages: list[str] = []

async def handle_event(event: AgentStreamEvent):
    if isinstance(event, PartStartEvent):
        output_messages.append(f'[Request] Starting part {event.index}: {event.part!r}')
    elif isinstance(event, PartDeltaEvent):
        if isinstance(event.delta, TextPartDelta):
            output_messages.append(f'[Request] Part {event.index} text delta: {event.delta.content_delta!r}')
        elif isinstance(event.delta, ThinkingPartDelta):
            output_messages.append(f'[Request] Part {event.index} thinking delta: {event.delta.content_delta!r}')
        elif isinstance(event.delta, ToolCallPartDelta):
            output_messages.append(f'[Request] Part {event.index} args delta: {event.delta.args_delta}')
    elif isinstance(event, FunctionToolCallEvent):
        output_messages.append(
            f'[Tools] The LLM calls tool={event.part.tool_name!r} with args={event.part.args} (tool_call_id={event.part.tool_call_id!r})'
        )
    elif isinstance(event, FunctionToolResultEvent):
        output_messages.append(f'[Tools] Tool call {event.tool_call_id!r} returned => {event.part.content}')
    elif isinstance(event, FinalResultEvent):
        output_messages.append(f'[Result] The model starting producing a final result (tool_name={event.tool_name})')


async def event_stream_handler(
    ctx: RunContext,
    event_stream: AsyncIterable[AgentStreamEvent],
):
    async for event in event_stream:
        await handle_event(event)

async def main():
    user_prompt = 'What will the weather be like in Paris on Tuesday?'

    async with weather_agent.run_stream(user_prompt, event_stream_handler=event_stream_handler) as run:
        async for output in run.stream_text():
            output_messages.append(f'[Output] {output}')


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

    print(output_messages)
    """
    [
        "[Request] Starting part 0: ToolCallPart(tool_name='weather_forecast', tool_call_id='0001')",
        '[Request] Part 0 args delta: {"location":"Pa',
        '[Request] Part 0 args delta: ris","forecast_',
        '[Request] Part 0 args delta: date":"2030-01-',
        '[Request] Part 0 args delta: 01"}',
        "[Tools] The LLM calls tool=\\'weather_forecast\\' with args={\"location\":\"Paris\",\"forecast_date\":\"2030-01-01\"} (tool_call_id=\\'0001\\')",
        "[Tools] Tool call '0001' returned => The forecast in Paris on 2030-01-01 is 24°C and sunny.",
        "[Request] Starting part 0: TextPart(content='It will be ')",
        '[Result] The model starting producing a final result (tool_name=None)',
        '[Output] It will be ',
        '[Output] It will be warm and sunny ',
        '[Output] It will be warm and sunny in Paris on ',
        '[Output] It will be warm and sunny in Paris on Tuesday.',
    ]
    """

(이 예제는 완전해서 그대로 실행할 수 있어요)

모든 이벤트 스트리밍

agent.run_stream()처럼, agent.run()도 선택적 event_stream_handler 인수를 받아 모델의 스트리밍 응답과 에이전트의 도구 실행에서 모든 이벤트를 스트리밍하게 해줘요. run_stream()과 달리, 최종 결과일 것처럼 보이는 도구 호출보다 텍스트가 먼저 와도 에이전트 그래프를 항상 완료까지 실행해요. 실시간 세션 중에는 이벤트 스트림 핸들러가 실시간 전용 RealtimeEvent 멤버도 받을 수 있어요.

편의상 agent.run_stream_events() 메서드도 run(event_stream_handler=...)의 래퍼로 제공돼요. 최종 런 결과를 실은 AgentRunResultEvent로 끝나는 AgentStreamEvent들에 대한 비동기 이터레이터를 내놓는 비동기 컨텍스트 매니저예요.

참고 — run_stream_events()와 run(event_stream_handler=...)는 들어오는 대로 원시 이벤트를 반환하므로, PartStartEvent와 이어지는 PartDeltaEvent들에서 스트리밍된 텍스트와 구조화된 출력을 직접 조립해야 해요.

양쪽 장점을 모두 얻으려면, 약간의 추가 복잡성을 대가로, 다음 섹션에 설명된 agent.iter()를 쓸 수 있어요. 이는 에이전트 그래프를 반복하고 매 단계에서 이벤트와 출력을 모두 스트리밍하게 해줘요. 검증된 구조화된 출력을 사용하는 집중 예시는 구조화된 응답을 더 빠르게 보이게 하기를 보세요.

import asyncio

from pydantic_ai import AgentRunResultEvent

from run_stream_event_stream_handler import handle_event, output_messages, weather_agent


async def main():
    user_prompt = 'What will the weather be like in Paris on Tuesday?'

    async with weather_agent.run_stream_events(user_prompt) as events:
        async for event in events:
            if isinstance(event, AgentRunResultEvent):
                output_messages.append(f'[Final Output] {event.result.output}')
            else:
                await handle_event(event)

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

    print(output_messages)
    """
    [
        "[Request] Starting part 0: ToolCallPart(tool_name='weather_forecast', tool_call_id='0001')",
        '[Request] Part 0 args delta: {"location":"Pa',
        '[Request] Part 0 args delta: ris","forecast_',
        '[Request] Part 0 args delta: date":"2030-01-',
        '[Request] Part 0 args delta: 01"}',
        "[Tools] The LLM calls tool=\\'weather_forecast\\' with args={\"location\":\"Paris\",\"forecast_date\":\"2030-01-01\"} (tool_call_id=\\'0001\\')",
        "[Tools] Tool call '0001' returned => The forecast in Paris on 2030-01-01 is 24°C and sunny.",
        "[Request] Starting part 0: TextPart(content='It will be ')",
        '[Result] The model starting producing a final result (tool_name=None)',
        "[Request] Part 0 text delta: 'warm and sunny '",
        "[Request] Part 0 text delta: 'in Paris on '",
        "[Request] Part 0 text delta: 'Tuesday.'",
        '[Final Output] It will be warm and sunny in Paris on Tuesday.',
    ]
    """

(이 예제는 완전해서 그대로 실행할 수 있어요)

커스텀 이벤트

프레임워크 자체 이벤트와 함께, 도구나 agent.iter()를 구동하는 코드가 같은 스트림에 자체 CustomEvent를 방출할 수 있어요. 모델의 컨텍스트에 아무것도 추가하지 않고, 스트림을 소비하는 사람에게 오래 걸리는 작업의 진행 업데이트·중간 결과·상태 정보를 표면화하는 데 유용해요.

어떤 이벤트 타입을 쓰나요?

Pydantic AI는 두 계열의 사용자 정의 이벤트가 있어요. 같은 스트림을 타고 같은 방식으로 정의되지만, 어느 것을 정의할지는 방출하는 코드를 누가 소유하는지가 결정하고, 그 구분은 런타임에 강제돼요: 잘못된 계열을 방출하면 UserError가 발생해요.

	CustomEvent	CapabilityEvent
언제 쓰나	앱이 자체 스트림 소비자나 프론트엔드에 뭔가 말하고 싶을 때	capability가 다른 capability와 호스트 앱에 뭔가 말하고 싶을 때
어디서 방출	앱 도구, 출력 검증기, 훅, event_stream_handler, AgentRun.emit()	capability 훅 또는 capability가 제공하는 도구
이름	progress처럼 평평하고 프로세스 전역	workspace.file_read처럼 네임스페이스됨
프론트엔드 도달	예, AG-UI와 Vercel AI 어댑터로	아니오, 내부 신호. 프론트엔드가 필요하면 CustomEvent로 재발행
결정을 실을 수 있나	아니오	예, dispatch='immediate'로

capability를 작성한다면 CapabilityEvent를 정의하세요(Capability 이벤트 참조). 그것의 이벤트는 런의 나머지와의 계약 일부이고, 네임스페이스가 두 capability가 이름에서 충돌하지 않게 해줘요. 애플리케이션을 작성한다면 CustomEvent를 정의하세요. capability의 이벤트를 프론트엔드에 표면화하려면 @agent.on_event로 듣고 공개 페이로드를 실은 자체 CustomEvent를 방출하세요.

이벤트 정의하고 방출하기

이벤트를 CustomEvent의 dataclass 서브클래스로 정의하세요 — 필드가 페이로드이고, 소비자는 클래스에 대한 isinstance 검사를 쓸 수 있어요. RunContext를 받는 앱의 비동기 코드 어디서든 이벤트 인스턴스와 함께 ctx.emit()을 await하세요. agent.iter()를 구동하는 코드는 대신 AgentRun.emit()을 써요. 동기 도구는 이벤트를 방출할 수 없어요. 이벤트를 방출해야 한다면 비동기 도구를 쓰세요. 도구 호출 안에서 방출하면 이벤트의 tool_call_id와 tool_name이 자동으로 찍혀서 소비자가 그것을 발원 호출로 귀속시킬 수 있어요. 이벤트는 event_stream_handler, run_stream_events(), agent.iter() 스트리밍, AG-UI와 Vercel AI UI 어댑터에 닿아요.

from collections.abc import AsyncIterator
from dataclasses import dataclass

from pydantic_ai import Agent, CustomEvent, RunContext
from pydantic_ai.messages import ModelMessage, ToolReturnPart
from pydantic_ai.models.function import (
    AgentInfo,
    DeltaToolCall,
    DeltaToolCalls,
    FunctionModel,
)


@dataclass(kw_only=True)
class SyncProgressEvent(CustomEvent):
    done: int
    total: int


async def model_function(
    messages: list[ModelMessage], info: AgentInfo
) -> AsyncIterator[DeltaToolCalls | str]:
    if any(
        isinstance(part, ToolReturnPart)
        for message in messages
        for part in message.parts
    ):
        yield 'All 3 files synchronized.'
    else:
        yield {
            0: DeltaToolCall(
                name='sync_files', json_args='{"count": 3}', tool_call_id='sync'
            )
        }


agent = Agent(FunctionModel(stream_function=model_function))
progress: list[str] = []


@agent.on_event(SyncProgressEvent)
async def record_progress(ctx: RunContext, event: SyncProgressEvent) -> None:
    progress.append(
        f'{event.done}/{event.total} from {event.tool_name} ({event.tool_call_id})'
    )


@agent.tool
async def sync_files(ctx: RunContext, count: int) -> str:
    for i in range(1, count + 1):
        # Do some long-running work, emitting a progress event after each step.
        await ctx.emit(SyncProgressEvent(done=i, total=count))
    return f'Synchronized {count} files.'


async def main():
    await agent.run('Synchronize my files')
    print(progress)
    """
    [
        '1/3 from sync_files (sync)',
        '2/3 from sync_files (sync)',
        '3/3 from sync_files (sync)',
    ]
    """

(이 예제는 완전해서 그대로 실행할 수 있어요 — main을 실행하려면 asyncio.run(main())을 추가해야 해요)

런 이벤트의 어떤 소비자든 그것을 볼 수 있어요: 위의 @agent.on_event 리스너, 이벤트 훅, event_stream_handler=, run_stream_events(), agent.iter() 스트리밍, AG-UI와 Vercel AI 어댑터. 페이로드 필드는 어떤 객체든 담을 수 있지만, durable execution과 UI 어댑터를 통과하려면 pydantic으로 직렬화 가능해야 해요.

페이로드는 봉투 자신이 필요한 필드 이름을 쓸 수 없어요: data, tool_call_id, tool_name, event_kind는 클래스가 정의될 때 거부되므로, 충돌할 필드에는 다른 이름(payload, call_id)을 고르세요.

방출은 런이 진행 중일 때만, 그리고 방출 코드가 소유한 계열에서만 작동하므로, 각각은 UserError를 발생시켜요: capability에서 CustomEvent 방출, 앱 코드에서 CapabilityEvent 방출, 런이 끝난 후 AgentRun.emit() 호출. AgentRun.emit()으로 방출된 이벤트는 모든 이벤트와 출력 스트리밍에 보이듯 node.stream(run.ctx)로 런 노드를 스트리밍하는 소비자에게 닿아요. 맨 async for node in run은 이벤트 스트림을 소비하지 않아서 아무것도 표면화되지 않아요.

이벤트는 방출되자마자 스트림 소비자에게 전달되므로, 진행 이벤트는 방출하는 도구가 아직 실행 중일 때 표면화돼요(반환 시가 아니라). 동시에 실행되는 도구의 이벤트는 방출 순서로 인터리브돼요(최선 노력 순서).

필드를 공유하는 이벤트는 베이스를 공유할 수 있어요. 베이스에 자체 @dataclass 데코레이터를 주고 — 장식되지 않은 것은 필드를 추가하지 않는데, 이는 페이로드에서 조용히 빠진 채 표면화되도록 두지 않고 거부돼요 — abstract=True로 표시해 이벤트 레지스트리 밖에 유지하고 스스로 방출될 수 없게 하세요:

from dataclasses import dataclass

from pydantic_ai import CustomEvent


@dataclass(kw_only=True)
class AppEvent(CustomEvent, abstract=True):
    request_id: str


@dataclass(kw_only=True)
class ReindexProgressEvent(AppEvent):
    done: int
    total: int

CapabilityEvent 베이스도 같은 방식으로 동작하고, 베이스는 계열의 namespace=를 두기 자연스러운 자리예요.

커스텀 이벤트 이름은 클래스 이름에서 Event를 제거하고 나머지를 snake case로 변환해 파생돼요. 그래서 SyncProgressEvent는 sync_progress를 써요. 클래스 인수로 이름을 오버라이드해요, 예: class SyncProgressEvent(CustomEvent, name='sync_status'). 이름은 클래스가 정의될 때 등록되고 프로세스 내에서 고유해야 해요. 같은 클래스 정의를 다시 실행하면(노트북 셀 재실행처럼) 등록을 교체해요.

이름은 이벤트의 위치 식별자일 뿐 아니라 와이어 식별자예요. 직렬화된 이벤트가 나르는 것이므로, 클래스를 리네임하면 태그도 함께 리네임돼요. 리네임은 이벤트가 방출한 프로세스를 오래 사는 곳 — durable execution 기록과 캐시, 영속 이벤트 로그, 이름에 매칭하는 프론트엔드 — 에서 호환성 브레이크예요. 클래스를 자유롭게 리네임하고 싶을 때 태그를 고정하려면 명시적 name=을 넘기세요.

태그가 당신이 제어하지 않는 것과 일치해야 할 때도 이름을 명시하세요. UI 어댑터가 와이어에 두는 식별자예요(AG-UI 이벤트의 name, Vercel AI 청크의 data-{name} 타입), 그리고 파생은 항상 snake case만 만들어요: 이미 data-indexProgress나 점 구분 ui.progress를 기대하는 프론트엔드는 그에 맞게 클래스를 리네임하는 대신 name='indexProgress'나 name='ui.progress'가 필요해요.

이벤트는 AgentStreamEvent 직렬화를 통해 원래 클래스로 왕복해요. 클래스가 등록되기 전에 이벤트가 역직렬화되면 UnknownCustomEvent가 되고, 페이로드는 data에 보존되며 UserWarning이 방출돼요. 이벤트를 역직렬화하는 어댑터를 만들기 전에 이벤트를 정의하는 모듈을 import하세요. 각 pydantic TypeAdapter는 생성될 때 등록된 이벤트 클래스를 포착해요. 등록된 이벤트의 페이로드 스키마는 메시지 타입과 같은 호환성 기대를 따라요: 로컬 클래스에 더 이상 검증되지 않는 페이로드는 조용히 저하되는 대신 크게 실패하므로, 이벤트를 정의하는 모듈의 호환 버전을 직렬화·역직렬화 양쪽에 유지하세요.

이벤트 이름은 앱 전역 레지스트리를 공유하고, 이미 등록된 이름으로 두 번째 클래스를 정의하면 즉시 오류가 나요. 커스텀 이벤트는 앱에 속하므로, 에이전트 런에 이벤트를 방출하는 라이브러리는 네임스페이스된 capability 이벤트를 capability에 정의해야 해요. capability 밖에서 런에 닿는 라이브러리 — 사용자가 등록하도록 건네는 맨 도구 — 만이 앱 수준 이벤트를 방출할 수 있고, 그때도 앱 자체 이벤트 이름과 충돌하지 않게 점 구분 접두사(name='mylib.progress') 아래 등록해야 해요.

UI 어댑터는 CustomEvent.to_payload()를 호출해 프론트엔드 페이로드를 얻는데, 기본은 이벤트의 자체 필드예요. UI가 다른 페이로드를 받아야 할 때 그걸 오버라이드하세요.

커스텀 이벤트는 기본으로 프론트엔드로 전달돼요. 방출하는 앱이 그 프론트엔드를 서빙하는 앱이기도 하기 때문이에요. 서버측 소비자 — 메트릭, 감사 로그, 당신 자신의 event_stream_handler — 만을 위한 이벤트는 ui=False로 옵트아웃해요. 그러면 모든 인프로세스 소비자에 닿는 반면 AG-UI와 Vercel AI 어댑터는 건너뛰어요:

from dataclasses import dataclass

from pydantic_ai import CustomEvent


@dataclass(kw_only=True)
class IndexProgressEvent(CustomEvent, ui=False):
    done: int
    total: int

서브클래스는 설정을 상속하고, 검사가 프로토콜별 핸들러보다 앞서 일어나므로 다른 프로토콜의 어댑터도 지켜요. 아무것도 보내지 않는 대신 다른 페이로드를 보내려면 to_payload()를 오버라이드하세요 — 그게 None을 반환하면 null 페이로드의 이벤트가 보내지는데, 그게 이름만의 신호를 보내는 방법이에요.

플래그는 와이어가 아니라 클래스에 살아요. 그래서 정의 모듈이 import되지 않은 곳에서 역직렬화된 이벤트는 앱이 선언한 것을 ui가 말하지 않는 UnknownCustomEvent로 도착해요. 그것들도 전달되지 않으므로, 프로세스 경계를 넘는 이벤트는 클래스가 옵트아웃한 페이로드를 새지 않아요. 이벤트가 다른 프로세스에서 프론트엔드에 닿는다면 — durable execution 워크플로, 큐, 웹소켓 팬아웃, 요청 없는 이벤트 인코딩처럼 — 그것들을 정의하는 모듈을 거기서 import하세요. 그렇지 않으면 커스텀 이벤트 중 아무것도 프론트엔드에 닿지 않아요.

에이전트의 그래프 반복하기

후드 아래에서 Pydantic AI의 각 Agent는 실행 흐름을 관리하기 위해 pydantic-graph를 써요. pydantic-graph는 Python에서 유한 상태 머신을 만들고 실행하는 제네릭·타입 중심 라이브러리예요. 실제로 Pydantic AI에 의존하지 않아요 — GenAI와 무관한 워크플로에 단독으로 쓸 수 있어요 — 하지만 Pydantic AI는 에이전트 런에서 모델 요청과 모델 응답 처리를 오케스트레이션하기 위해 그것을 활용해요.

많은 시나리오에서 pydantic-graph를 신경 쓸 필요가 없어요. agent.run(...) 호출이 기저 그래프를 시작부터 끝까지 그냥 순회하니까요. 그러나 더 깊은 통찰이나 제어 — 예를 들어 특정 단계에서 자체 로직을 주입하는 것 — 이 필요할 때, Pydantic AI는 Agent.iter로 저수준 반복 프로세스를 노출해요. 이 메서드는 AgentRun을 반환하는데, 이걸 비동기 반복하거나 next 메서드로 노드별 수동 구동할 수 있어요. 에이전트 그래프가 End를 반환하면, 모든 단계의 상세 기록과 함께 최종 결과를 얻어요.

async for 반복

iter와 async for를 써서 에이전트가 실행하는 각 노드를 기록하는 예시:

from pydantic_ai import Agent

agent = Agent('openai:gpt-5.2')


async def main():
    nodes = []
    # Begin an AgentRun, which is an async-iterable over the nodes of the agent's graph
    async with agent.iter('What is the capital of France?') as agent_run:
        async for node in agent_run:
            # Each node represents a step in the agent's execution
            nodes.append(node)
    print(nodes)
    """
    [
        UserPromptNode(
            user_prompt='What is the capital of France?',
            instructions_functions=[],
            system_prompts=(),
            system_prompt_functions=[],
            system_prompt_dynamic_functions={},
        ),
        ModelRequestNode(
            request=ModelRequest(
                parts=[
                    UserPromptPart(
                        content='What is the capital of France?',
                        timestamp=datetime.datetime(...),
                    )
                ],
                timestamp=datetime.datetime(...),
                run_id='...',
                conversation_id='...',
            )
        ),
        CallToolsNode(
            model_response=ModelResponse(
                parts=[TextPart(content='The capital of France is Paris.')],
                usage=RequestUsage(
                    cost=Decimal('0.000196'), input_tokens=56, output_tokens=7
                ),
                model_name='gpt-5.2',
                timestamp=datetime.datetime(...),
                run_id='...',
                conversation_id='...',
            )
        ),
        End(data=FinalResult(output='The capital of France is Paris.')),
    ]
    """
    print(agent_run.result.output)
    #> The capital of France is Paris.

(이 예제를 실행하려면 asyncio를 import 하고 asyncio.run(main())을 추가하면 돼요. 다른 변경은 필요 없어요.)

• AgentRun은 흐름의 각 노드(BaseNode 또는 End)를 내놓는 비동기 이터레이터예요.
• End 노드가 반환되면 런이 끝나요.

.next(...) 수동 사용

AgentRun.next(...) 메서드에 실행할 다음 노드를 넘겨 반복을 수동으로 구동할 수도 있어요. 이렇게 하면 실행 전에 노드를 검사하거나 수정하고, 자체 로직으로 노드를 건너뛰며, next()의 오류를 더 쉽게 잡을 수 있어요:

from pydantic_ai import Agent
from pydantic_graph import End

agent = Agent('openai:gpt-5.2')


async def main():
  async with agent.iter('What is the capital of France?') as agent_run:
      node = agent_run.next_node  # (1)

      all_nodes = [node]

      # Drive the iteration manually:
      while not isinstance(node, End):  # (2)
          node = await agent_run.next(node)  # (3)
          all_nodes.append(node)  # (4)

      print(all_nodes)
      """
      [
          UserPromptNode(
              user_prompt='What is the capital of France?',
              instructions_functions=[],
              system_prompts=(),
              system_prompt_functions=[],
              system_prompt_dynamic_functions={},
          ),
          ModelRequestNode(
              request=ModelRequest(
                  parts=[
                      UserPromptPart(
                          content='What is the capital of France?',
                          timestamp=datetime.datetime(...),
                      )
                  ],
                  timestamp=datetime.datetime(...),
                  run_id='...',
                  conversation_id='...',
              )
          ),
          CallToolsNode(
              model_response=ModelResponse(
                  parts=[TextPart(content='The capital of France is Paris.')],
                  usage=RequestUsage(
                      cost=Decimal('0.000196'), input_tokens=56, output_tokens=7
                  ),
                  model_name='gpt-5.2',
                  timestamp=datetime.datetime(...),
                  run_id='...',
                  conversation_id='...',
              )
          ),
          End(data=FinalResult(output='The capital of France is Paris.')),
      ]
      """

에이전트 그래프에서 실행될 첫 노드를 잡는 것으로 시작해요.

End 노드가 만들어지면 에이전트 런이 끝나요. End 인스턴스는 next에 넘길 수 없어요.

await agent_run.next(node)를 호출하면 그 노드를 에이전트 그래프에서 실행하고, 런 기록을 갱신하며, 실행할 다음 노드를 반환해요.

필요하면 새 node를 여기서 검사하거나 변경할 수도 있어요.

(이 예제를 실행하려면 asyncio를 import 하고 asyncio.run(main())을 추가하면 돼요. 다른 변경은 필요 없어요.)

사용량과 최종 출력 접근

AgentRun 객체에서 agent_run.usage로 언제든 사용 통계(토큰, 요청 등)를 꺼낼 수 있어요. 이 프로퍼티는 사용 데이터를 담은 RunUsage 객체를 반환해요.

RunUsage.cost는 각 요청의 사용량을 genai-prices로 계산한 런 총비용의 최선 노력(USD) 추정치를 추가로 담아요. genai-prices에 가격 데이터가 없는 모델·공급자에 대한 요청은 합계에 기여하지 않아요. 설치 후 출시된 모델의 가격 책정 방법은 모델 가격 최신 유지를 보세요.

런이 끝나면 agent_run.result가 최종 출력(과 관련 메타데이터)을 담은 AgentRunResult 객체가 돼요.

모든 이벤트와 출력 스트리밍

async for 반복과 결합해 에이전트 런을 스트리밍하는 예시:

import asyncio
from dataclasses import dataclass
from datetime import date

from pydantic_ai import (
    Agent,
    FinalResultEvent,
    FunctionToolCallEvent,
    FunctionToolResultEvent,
    PartDeltaEvent,
    PartStartEvent,
    RunContext,
    TextPartDelta,
    ThinkingPartDelta,
    ToolCallPartDelta,
)


@dataclass
class WeatherService:
    async def get_forecast(self, location: str, forecast_date: date) -> str:
        # In real code: call weather API, DB queries, etc.
        return f'The forecast in {location} on {forecast_date} is 24°C and sunny.'

    async def get_historic_weather(self, location: str, forecast_date: date) -> str:
        # In real code: call a historical weather API or DB
        return f'The weather in {location} on {forecast_date} was 18°C and partly cloudy.'


weather_agent = Agent[WeatherService, str](
    'openai:gpt-5.2',
    deps_type=WeatherService,
    output_type=str,  # We'll produce a final answer as plain text
    system_prompt='Providing a weather forecast at the locations the user provides.',
)


@weather_agent.tool
async def weather_forecast(
    ctx: RunContext[WeatherService],
    location: str,
    forecast_date: date,
) -> str:
    if forecast_date >= date.today():
        return await ctx.deps.get_forecast(location, forecast_date)
    else:
        return await ctx.deps.get_historic_weather(location, forecast_date)


output_messages: list[str] = []


async def main():
    user_prompt = 'What will the weather be like in Paris on Tuesday?'

    # Begin a node-by-node, streaming iteration
    async with weather_agent.iter(user_prompt, deps=WeatherService()) as run:
        async for node in run:
            if Agent.is_user_prompt_node(node):
                # A user prompt node => The user has provided input
                output_messages.append(f'=== UserPromptNode: {node.user_prompt} ===')
            elif Agent.is_model_request_node(node):
                # A model request node => We can stream tokens from the model's request
                output_messages.append('=== ModelRequestNode: streaming partial request tokens ===')
                async with node.stream(run.ctx) as request_stream:
                    final_result_found = False
                    async for event in request_stream:
                        if isinstance(event, PartStartEvent):
                            output_messages.append(f'[Request] Starting part {event.index}: {event.part!r}')
                        elif isinstance(event, PartDeltaEvent):
                            if isinstance(event.delta, TextPartDelta):
                                output_messages.append(
                                    f'[Request] Part {event.index} text delta: {event.delta.content_delta!r}'
                                )
                            elif isinstance(event.delta, ThinkingPartDelta):
                                output_messages.append(
                                    f'[Request] Part {event.index} thinking delta: {event.delta.content_delta!r}'
                                )
                            elif isinstance(event.delta, ToolCallPartDelta):
                                output_messages.append(
                                    f'[Request] Part {event.index} args delta: {event.delta.args_delta}'
                                )
                        elif isinstance(event, FinalResultEvent):
                            output_messages.append(
                                f'[Result] The model started producing a final result (tool_name={event.tool_name})'
                            )
                            final_result_found = True
                            break

                    if final_result_found:
                        # Once the final result is found, we can call `AgentStream.stream_text()` to stream the text.
                        # A similar `AgentStream.stream_output()` method is available to stream structured output.
                        async for output in request_stream.stream_text():
                            output_messages.append(f'[Output] {output}')
            elif Agent.is_call_tools_node(node):
                # A handle-response node => The model returned some data, potentially calls a tool
                output_messages.append('=== CallToolsNode: streaming partial response & tool usage ===')
                async with node.stream(run.ctx) as handle_stream:
                    async for event in handle_stream:
                        if isinstance(event, FunctionToolCallEvent):
                            output_messages.append(
                                f'[Tools] The LLM calls tool={event.part.tool_name!r} with args={event.part.args} (tool_call_id={event.part.tool_call_id!r})'
                            )
                        elif isinstance(event, FunctionToolResultEvent):
                            output_messages.append(
                                f'[Tools] Tool call {event.tool_call_id!r} returned => {event.part.content}'
                            )
            elif Agent.is_end_node(node):
                # Once an End node is reached, the agent run is complete
                assert run.result is not None
                assert run.result.output == node.data.output
                output_messages.append(f'=== Final Agent Output: {run.result.output} ===')


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

    print(output_messages)
    """
    [
        '=== UserPromptNode: What will the weather be like in Paris on Tuesday? ===',
        '=== ModelRequestNode: streaming partial request tokens ===',
        "[Request] Starting part 0: ToolCallPart(tool_name='weather_forecast', tool_call_id='0001')",
        '[Request] Part 0 args delta: {"location":"Pa',
        '[Request] Part 0 args delta: ris","forecast_',
        '[Request] Part 0 args delta: date":"2030-01-',
        '[Request] Part 0 args delta: 01"}',
        '=== CallToolsNode: streaming partial response & tool usage ===',
        "[Tools] The LLM calls tool=\\'weather_forecast\\' with args={\"location\":\"Paris\",\"forecast_date\":\"2030-01-01\"} (tool_call_id=\\'0001\\')",
        "[Tools] Tool call '0001' returned => The forecast in Paris on 2030-01-01 is 24°C and sunny.",
        '=== ModelRequestNode: streaming partial request tokens ===',
        "[Request] Starting part 0: TextPart(content='It will be ')",
        '[Result] The model started producing a final result (tool_name=None)',
        '[Output] It will be ',
        '[Output] It will be warm and sunny ',
        '[Output] It will be warm and sunny in Paris on ',
        '[Output] It will be warm and sunny in Paris on Tuesday.',
        '=== CallToolsNode: streaming partial response & tool usage ===',
        '=== Final Agent Output: It will be warm and sunny in Paris on Tuesday. ===',
    ]
    """

(이 예제는 완전해서 그대로 실행할 수 있어요)

런 취소하기

진행 중인 런은 완전히 취소될 수 있어요 — 예를 들어 사용자가 "정지" 버튼을 눌렀을 때요. CancellationToken을 만들고 런에 넘긴 다음, 정지 핸들러에서 cancel()을 호출하세요. 취소는 완료된 메시지 기록과 사용량과 함께 RunCancelled를 발생시켜서 대화를 영속화하고 재개할 수 있게 해요:

import asyncio

from pydantic_ai import Agent, CancellationToken, RunCancelled

agent = Agent('test')
tool_started = asyncio.Event()


@agent.tool_plain
async def slow_lookup() -> str:
  tool_started.set()
  await asyncio.sleep(10)
  return 'result'


async def main():
  token = CancellationToken()
  run = asyncio.create_task(
      agent.run('Look something up', cancellation_token=token)
  )
  await tool_started.wait()
  token.cancel()  # (1)

  try:
      await run
  except RunCancelled as exc:
      messages = exc.all_messages()
      print(f'Cancelled after {len(messages)} messages')
      #> Cancelled after 3 messages
      await agent.run(message_history=messages)  # (2)

cancel()은 멱등이고 스레드 안전해요. 하나의 토큰이 여러 동시 런을 지배해 모두 취소할 수 있어요. 토큰은 일회용이에요. 일단 취소되면 취소된 채로 남고, 이미 취소된 토큰을 런에 넘기면 그 런이 시작되지 못하게 해요(또한 "취소가 런보다 앞지르는" 틈도 닫아요). 따라서 런이나 정지 제스처마다 새 토큰을 만들어야 해요. 세션에 걸쳐 하나를 재사용하면 첫 런 이후의 모든 런이 시작도 전에 취소돼요.

RunCancelled.all_messages()는 완료된 도구 결과를 포함해 취소 전에 완료된 모든 것을 담아요. 매달린 도구 호출은 내역이 재개될 때 자동 복구돼요.

UI 어댑터 사용자는 on_cancel 콜백으로 이 재개 가능한 내역을 영속화할 수 있어요.

(이 예제를 실행하려면 asyncio를 import 하고 asyncio.run(main())을 추가하면 돼요. 다른 변경은 필요 없어요.)

agent.run_sync()도 같은 토큰을 받아요. 다른 스레드에서 token.cancel()을 호출하는 게 블로킹된 동안 동기 런을 중단시키는 유일한 방법이에요.

어느 메커니즘, 어느 예외 — CancellationToken이 기본으로 잡아야 할 것이에요. 런 밖에서, 다른 스레드에서, run_sync()에 대해 작동하는 유일한 표면이고, 하나의 토큰이 여러 런을 동시에 지배할 수 있으니까요. 나머지는 토큰이 닿을 수 없는 곳을 위해 존재해요:

취소할 때 당신이 있는 곳	사용	런이 끝나는 방법
런 밖 (정지 버튼, 다른 스레드)	CancellationToken	RunCancelled
도구, event_stream_handler, capability 훅 안	RunContext.cancel()	RunCancelled
run_stream_events() 소비	내놓은 핸들의 AgentRunEvents.cancel()	RunCancelled
agent.iter()로 직접 그래프 구동	AgentRun.cancel()	RunCancelled
환경이 취소 (asyncio.timeout(), TaskGroup, 셧다운)	(아무것도 호출 안 함)	CancelledError

처음 네 개는 일급(first-party) 이에요: Pydantic AI가 런 자체를 멈추고 재개 가능한 기록을 실은 평범한 잡을 수 있는 예외 RunCancelled를 발생시켜요. 마지막은 외부예요: CancelledError가 변경 없이 계속 전파돼요 — 그래서 asyncio.timeout()은 여전히 TimeoutError를, TaskGroup은 여전히 해체를, Temporal은 여전히 워크플로를 Cancelled 로 끝내요 — [RunCancelled.from_cancellation()](https://pydantic.dev/docs/ai/api/pydantic-ai/exceptions/#pydantic_ai.exceptions.RunCancelled.from_cancellation)을 위한 같은 기록이 _붙어_ 있으면서요. Pydantic AI는 그 의미론을 깨지 않고는 외부 CancelledError를 RunCancelled`로 바꿀 수 없어요. 그래서 취소에 두 종류가 있는 거예요. 다음에서 다뤄요.

주변 환경이 런을 취소하면 — 예를 들어 asyncio.timeout(), TaskGroup, 앱 셧다운으로 — CancelledError는 그대로 남아요. RunCancelled.from_cancellation()이 붙은 런 상태를 제공해요:

import asyncio

from pydantic_ai import Agent, RunCancelled

agent = Agent('test')
tool_started = asyncio.Event()


@agent.tool_plain
async def slow_lookup() -> str:
  tool_started.set()
  await asyncio.sleep(10)
  return 'result'


async def main():
  task = asyncio.create_task(agent.run('Look something up'))
  await tool_started.wait()
  task.cancel()  # (1)

  try:
      await task
  except asyncio.CancelledError as exc:
      cancelled = RunCancelled.from_cancellation(exc)  # (2)
      assert cancelled is not None
      messages = cancelled.all_messages()
      print(f'Cancelled after {len(messages)} messages')
      #> Cancelled after 3 messages
      await agent.run(message_history=messages)  # (3)

이것은 주변 asyncio 환경이 부과한 취소를 보여줘요. 앱 정지 제스처에는 CancellationToken을 선호하세요.

외부 취소는 절대 변환되지 않아요: asyncio.timeout(), TaskGroup, Temporal 취소 의미론이 보존돼요. 런 상태는 원래 CancelledError에 타고 가요.

RunCancelled.all_messages()는 완료된 도구 결과를 포함해 취소 전에 완료된 모든 것을 담아요. 매달린 도구 호출은 내역이 재개될 때 자동 복구돼요.

(이 예제를 실행하려면 asyncio를 import 하고 asyncio.run(main())을 추가하면 돼요. 다른 변경은 필요 없어요.)

Python 3.10에서 asyncio는 await task 경계를 가로질러 CancelledError를 재생성하지만, __context__를 통해 붙은 런 상태를 실은 원래 예외를 체이닝해요. from_cancellation()이 그것을 순회해요. 체인은 취소된 태스크의 첫 await에만 붙으므로, 같은 태스크의 이후 await는 체인 없는 예외를 봐요. capture_run_messages()는 기록만 필요할 때의 폴백이에요.

run_stream_events()를 소비할 때, 내놓은 AgentRunEvents 핸들은 태스크 재처리가 필요 없는 일급 대안을 제공해요: AgentRunEvents.cancel()은 다른 태스크(예: UI의 "정지" 핸들러)에서 호출해도 안전하고 계속 반복 시 RunCancelled로 표면화돼요:

from pydantic_ai import Agent, RunCancelled

agent = Agent('test')


async def main():
  async with agent.run_stream_events('Write a long essay about Python') as events:
      try:
          async for _event in events:
              events.cancel()  # (1)
      except RunCancelled as exc:
          print(f'Cancelled after {len(exc.all_messages())} messages')
          #> Cancelled after 2 messages

멱등이고, 런이 끝나면 no-op이며, 첫 반복 전에 호출해 런이 아예 시작되지 않게 할 수 있어요.

(이 예제를 실행하려면 asyncio를 import 하고 asyncio.run(main())을 추가하면 돼요. 다른 변경은 필요 없어요.)

소비 태스크를 외부에서 취소하는 것도 여기서 작동해요: 백그라운드 런이 해체되고, 전파하는 CancelledError가 from_cancellation()을 위한 런 상태를 실으며, 핸들의 all_messages()와 usage는 계속 접근 가능해요.

도구, event_stream_handler, capability 훅에서 취소를 요청하려면 RunContext.cancel()을 호출하세요. 이것은 일급 취소를 요청하므로, 외부 CancelledError가 아니라 RunCancelled로 런이 끝나요. cancel() 자체는 정상 반환돼요 — 취소는 호출 코드의 다음 await에서 전달되고 도구의 반환 값은 폐기돼요 — 그래서 도구는 요청 후에도 여전히 정리를 실행할 수 있어요:

from pydantic_ai import Agent, RunCancelled, RunContext

agent = Agent('test')


@agent.tool
async def stop(ctx: RunContext) -> str:
    ctx.cancel()
    return 'discarded'  # cancel() returned; this value is never sent to the model


async def main():
    try:
        await agent.run('Stop now')
    except RunCancelled as exc:
        print(f'Cancelled after {len(exc.all_messages())} messages')
        #> Cancelled after 2 messages

취소는 협력적이에요 — Pydantic AI는 진행 중 작업의 취소를 요청하고, 취소 후 도착한 결과를 폐기하며, 소유한 리소스를 닫아요. 비동기 도구는 중단 지점에서 CancelledError를 받아요. 동기(def) 도구는 워커 스레드에서 실행되는데, Python이 안전하게 종료할 수 없어요. 실행 모드에 따라 취소가 워커를 기다리거나 백그라운드에서 끝나게 둘 수 있어요. 어느 쪽이든 결과는 폐기되지만 부수 효과는 롤백되지 않아요. 공급자측 모델 생성을 취소하는 것은 최선 노력이고 공급자에 달려 있어요.

취소가 어떤 방식으로 도착할지 제어하지 못할 수 있어요: 호출자가 정지 제스처를 위해 agent.run()을 태스크로 감싸는 반면, 도구 — 어쩌면 다른 라이브러리에서 온 — 는 내부적으로 ctx.cancel()을 호출해요. 각각을 자기 방식으로 처리하세요 — 일급 RunCancelled는 소비하되, 외부 CancelledError는 타임아웃과 태스크 그룹이 여전히 제대로 해체되도록 계속 전파시키고, 필요하면 그 상태를 먼저 포착하세요:

import asyncio

from pydantic_ai import Agent, RunCancelled, RunContext

agent = Agent('test')


@agent.tool
async def imported_tool(ctx: RunContext) -> str:
  ctx.cancel()  # (1)
  return 'discarded'


async def main():
  task = asyncio.create_task(agent.run('Go'))
  try:
      await task
  except RunCancelled as exc:  # (2)
      print(f'Cancelled after {len(exc.all_messages())} messages')
      #> Cancelled after 2 messages
  except asyncio.CancelledError as exc:  # (3)
      cancelled = RunCancelled.from_cancellation(exc)
      if cancelled is not None:
          ...  # persist cancelled.all_messages() before re-raising
      raise

여기서 도구가 일급으로 취소하므로 await task는 RunCancelled를 발생시켜요. 만약 정지 버튼이 대신 task.cancel()을 호출했다면 await task는 CancelledError를 발생시키고 두 번째 핸들러가 실행됐을 거예요.

일급 취소는 당신이 소비할 수 있는 RunCancelled예요: 런이 당신 자신의 코드가 요청해서 멈췄으므로 정상 반환해도 돼요.

외부 취소는 CancelledError로 남고, 정지 버튼의 task.cancel()은 타임아웃이나 TaskGroup 해체와 구별할 수 없어요 — 그래서 다시 발생시키고(삼키면 그 해체들을 깨뜨려요), 부분 상태를 먼저 포착하려고만 from_cancellation()에 닿아요. 아무것도 붙어 있지 않을 때(예: 이 런과 무관한 앱 셧다운) None을 반환해요.

(이 예제를 실행하려면 asyncio를 import 하고 asyncio.run(main())을 추가하면 돼요. 다른 변경은 필요 없어요.)

왜 두 예외 타입인가? — 취소는 두 곳에서 발원할 수 있고, 그중 하나만이 Pydantic AI가 이름을 지을 몫이에요:

• 당신의 앱이 전용 취소 메서드 중 하나로 런을 멈추기로 결정. Pydantic AI가 그 취소를 스스로 발행했으므로, asyncio가 해석하기 전에 소비하고 대신 RunCancelled를 발생시킬 수 있어요: 런이 평범하고 잡을 수 있는 앱 오류로 끝나요.
• asyncio 환경이 런이 우연히 얹혀 있는 태스크를 취소: asyncio.Task.cancel(), asyncio.timeout() 만료, 형제가 실패한 후 TaskGroup 해체, 서버 셧다운, durable execution 아래 워크플로 취소. 이 모두가 바로 그 CancelledError 신호를 전달하므로, Pydantic AI는 정지 버튼과 타임아웃을 구별할 수 없어요 — 그리고 예외 타입은 그것 위에 세워진 모든 것에 무게를 실어요: asyncio.timeout()은 TimeoutError만 만들고, TaskGroup은 태스크를 깨끗이 취소된 것으로만 취급하며, Temporal은 CancelledError 자체가 계속 전파돼야만 워크플로를 Cancelled 로 끝내요. 그 자리에 RunCancelled를 발생시키면 그것들 각각을 조용히 깨뜨려요. 그래서 런 상태는 대체하는 게 아니라, from_cancellation()을 위해 전파하는 CancelledError에 붙어 있어요.

취소는 종결이에요: capability 훅은 그것을 관찰하고 정리할 수 있지만 런을 성공으로 복구할 수는 없어요 — Python 3.11+에서는 사용자 코드가 전달된 취소를 흡수해도 성립하고, Python 3.10에서는 최선 노력이에요. 일급 취소와 외부 취소가 경주하면 외부 취소가 이겨요. Python 3.10에서는 그 경주를 구별할 수 없어서, 일급 취소가 대신 이겨요.

에이전트 그래프에 대한 세밀한 제어를 위해, agent.iter()가 반환한 핸들에서 AgentRun.cancel()을 호출하세요:

from pydantic_ai import Agent, RunCancelled

agent = Agent('test')


async def main():
  try:
      async with agent.iter('Write a long essay about Python') as agent_run:
          async for node in agent_run:
              if Agent.is_call_tools_node(node):
                  agent_run.cancel()  # (1)
  except RunCancelled as exc:
      print(f'Cancelled after {len(exc.all_messages())} messages')  # (2)
      #> Cancelled after 2 messages

AgentRun.cancel()은 다른 태스크에서 호출해도 안전하고 런이 끝나면 no-op이에요.

agent.iter() 블록 안에서 취소는 asyncio.CancelledError로 표면화되고, 컨텍스트가 나간 후에는 일급 취소가 분리된 상태 스냅샷과 함께 RunCancelled를 발생시켜요.

(이 예제를 실행하려면 asyncio를 import 하고 asyncio.run(main())을 추가하면 돼요. 다른 변경은 필요 없어요.)

취소 후의 메시지 기록

스트림이 생성 중간에 취소되면, 응답이 메시지 기록에 state='interrupted'로 기록돼요. 기록에는 취소 전에 받은 부분 콘텐츠가 포함돼요:

from pydantic_ai import Agent

agent = Agent('openai:gpt-5.2')


async def main():
  async with agent.run_stream('Tell me about Python') as result:
      async for text in result.stream_text(delta=True):
          break
      await result.cancel()

  messages = result.all_messages()  # (1)
  print(messages[-1].state)  # (2)
  #> interrupted

메시지 기록에는 취소 전에 받은 부분 콘텐츠와 함께 중단된 응답이 포함돼요.

중단된 응답 상태는 앱이 기록을 재사용하기 전에 부분 응답을 유지·검사·폐기할지 결정하게 해줘요.

(이 예제를 실행하려면 asyncio를 import 하고 asyncio.run(main())을 추가하면 돼요. 다른 변경은 필요 없어요.)

중단된 기록 재사용 — 중단된 기록은 다른 런에 직접 넘길 수 있어요. 다음 모델 요청 전에 Pydantic AI가 기록을 복구해요: 결과를 받지 못한 어떤 도구 호출 — 인수가 스트림 중간에 잘린 것 포함 — 은 인터럽트됐다고 모델에 알리는 합성 ToolReturnPart로 응답돼요.

취소된 스트림의 사용량 추적 — 취소 후 usage가 보고하는 토큰 사용량은 부분적이고 공급자 의존적이에요. Pydantic AI는 즉시 스트림에서 당기기를 멈추므로 최종 사용량 이벤트가 도착하지 않을 수 있고, 일부 공급자 SDK는 로컬 스트림이 닫힌 후에도 서버측 생성이 계속될 수 있어요. 비용에 민감한 회계에 취소된 스트림 사용량을 의존하지 마세요. OpenAI chat completions의 경우 openai_continuous_usage_stats가 각 청크와 함께 누적 사용량 데이터를 요청해 스트림 내 사용량 보고를 개선할 수 있지만, 취소된 스트림 사용량은 여전히 최선 노력이에요.

취소와 서브에이전트

취소는 런 범위예요: cancel()은 그 RunContext가 속한 런을 취소하고, CancellationToken은 그것이 붙은 런들을 취소해요. 이는 에이전트 위임 — await sub_agent.run(...)으로 다른 에이전트를 실행하는 도구 — 을 쓸 때 중요해요:

• 서브에이전트가 스스로 취소해도 부모를 취소하지 않아요 — 도구 본문 안에서 await될 때요. 서브에이전트(또는 그 도구 중 하나)가 ctx.cancel()을 호출하면 서브에이전트의 런이 취소돼요. 위임 도구는 RunCancelled를 보는데, 잡히지 않으면 부모에게 부모 런의 취소가 아니라 부모 모델이 반응할 수 있는 실패한 도구 반환 으로 표면화돼요. 이 격리는 도구 본문에 특정돼요: event_stream_handler, 출력 검증기, capability 훅에서 await된 서브에이전트는 부모의 태스크에서 직접 실행되므로, 그 cancel()은 부모 자신의 RunCancelled로 표면화돼요.
• 부모도 취소하려면 위임 도구에서 옵트인하세요 — RunCancelled를 잡아 부모 컨텍스트에서 ctx.cancel()을 호출하거나(또는 다른 오류 재발생) 하세요.
• 한 번에 전체 런 트리를 취소하려면 부모와 서브에이전트 간에 하나의 CancellationToken을 공유하세요 — 취소하면 전부 멈춰요. 이렇게(또는 외부 asyncio.CancelledError로) 취소된 부모는 인라인으로 await하고 있는 서브에이전트 런도 해체하는데, 같은 태스크에서 실행되기 때문이에요.

추가 구성 (Additional Configuration)

모델 가격 최신 유지하기

Pydantic AI는 출시 시 모델 가격을 번들해요. 설치 후 출시된 모델의 비용을 추정하려면 앱 시작 시 갱신된 가격을 다운로드하세요:

from pydantic_ai import prices

updater = prices.update_in_background()

try:
    ...  # run your app
finally:
    updater.stop()

가격 목록은 즉시 갱신된 후 백그라운드 스레드에서 매시간 갱신돼요. 실패한 다운로드는 가장 최근 가격을 계속 사용해요.

커스텀 URL이나 갱신 간격은 같은 백그라운드 태스크를 공유하는 genai_prices.UpdatePrices를 쓰세요.

사용량 한계 (Usage Limits)

Pydantic AI는 모델 런에서 사용량(토큰·요청·도구 호출·비용)을 제한하는 UsageLimits 구조를 제공해요.

run{_sync,_stream} 함수에 usage_limits 인수를 넘겨 이 설정을 적용할 수 있어요.

출력 토큰 수를 제한하는 예시:

from pydantic_ai import Agent, UsageLimitExceeded, UsageLimits

agent = Agent('anthropic:claude-sonnet-4-6')

result_sync = agent.run_sync(
    'What is the capital of Italy? Answer with just the city.',
    usage_limits=UsageLimits(output_tokens_limit=10),
)
print(result_sync.output)
#> Rome
print(result_sync.usage)
#> RunUsage(cost=Decimal('0.000201'), input_tokens=62, output_tokens=1, requests=1)

try:
    result_sync = agent.run_sync(
        'What is the capital of Italy? Answer with a paragraph.',
        usage_limits=UsageLimits(output_tokens_limit=10),
    )
except UsageLimitExceeded as e:
    print(e)
    """
    Exceeded the output_tokens_limit of 10 (output_tokens=32). Consider raising the limit, or see the docs on usage limits for budget-aware patterns: https://pydantic.dev/docs/ai/core-concepts/agent/#usage-limits
    """

요청 수 제한은 무한 루프나 과도한 도구 호출을 막는 데 유용해요:

from typing_extensions import TypedDict

from pydantic_ai import Agent, ModelRetry, UsageLimitExceeded, UsageLimits


class NeverOutputType(TypedDict):
  """
  Never ever coerce data to this type.
  """

  never_use_this: str


agent = Agent(
  'anthropic:claude-sonnet-4-6',
  retries={'tools': 3},
  output_type=NeverOutputType,
  system_prompt='Any time you get a response, call the `infinite_retry_tool` to produce another response.',
)


@agent.tool_plain(retries=5)  # (1)
def infinite_retry_tool() -> int:
  raise ModelRetry('Please try again.')


try:
  result_sync = agent.run_sync(
      'Begin infinite retry loop!', usage_limits=UsageLimits(request_limit=3)  # (2)
  )
except UsageLimitExceeded as e:
  print(e)
  """
  The next request would exceed the request_limit of 3. Consider raising the limit, or see the docs on usage limits for budget-aware patterns: https://pydantic.dev/docs/ai/core-concepts/agent/#usage-limits
  """

이 도구는 오류 전에 5번 재시도할 수 있어요. 루프에 갇힐 수 있는 도구를 시뮬레이션해요.

이 런은 3 요청 후 오류가 나서 무한 도구 호출을 막아요.

도구 호출 상한

단일 런 내 성공적인 도구 호출 수에 제한이 필요하다면 tool_calls_limit을 쓰세요:

from pydantic_ai import Agent
from pydantic_ai.exceptions import UsageLimitExceeded
from pydantic_ai.usage import UsageLimits

agent = Agent('anthropic:claude-sonnet-4-6')

@agent.tool_plain
def do_work() -> str:
    return 'ok'

try:
    # Allow at most one executed tool call in this run
    agent.run_sync('Please call the tool twice', usage_limits=UsageLimits(tool_calls_limit=1))
except UsageLimitExceeded as e:
    print(e)
    """
    The next tool call(s) would exceed the tool_calls_limit of 1 (tool_calls=2). Consider raising the limit, or see the docs on usage limits for budget-aware patterns: https://pydantic.dev/docs/ai/core-concepts/agent/#usage-limits
    """

참고

• 사용량 한계는 도구를 많이 등록했다면 특히 관련돼요. request_limit로 모델 턴 수를, tool_calls_limit으로 런 내 성공적인 도구 실행 수를 묶으세요.
• tool_calls_limit은 도구 호출을 실행하기 전에 검사돼요. 모델이 한계를 초과할 병렬 도구 호출을 반환하면 어떤 도구도 실행되지 않아요.

도구와 capabilities는 ctx.usage_limits에서 런의 한계를 읽을 수 있어요(지금까지 사용량을 위한 ctx.usage와 함께). 그래서 예산 인지 도구·capability는 한계의 중복 사본으로 설정되지 않고 남은 예산을 공개하거나 적응할 수 있어요. 그것은 런이 이미 집행하는 것을 반영하고 관례상 읽기 전용이에요.

요청당 입력 크기 제한

위 토큰 한계는 전체 런에 걸쳐 누적돼요. 단일 요청의 입력 크기(실제로 모델에 보내는 컨텍스트 창)를 상한하려면 per_request_input_tokens_limit을 쓰세요. 프롬프트 캐싱이 누적 입력을 비용의 나쁜 대리자로 만들 때 유용해요: 재전송된 캐시된 접두사는 저렴하지만, 단일 과대 컨텍스트가 모델 성능을 떨어뜨리고 캐시 미스 비용을 몰아요.

from pydantic_ai import Agent, UsageLimitExceeded, UsageLimits

agent = Agent('anthropic:claude-sonnet-4-6')

try:
    agent.run_sync(
        'What is the capital of Italy? Answer with just the city.',
        usage_limits=UsageLimits(per_request_input_tokens_limit=10),
    )
except UsageLimitExceeded as e:
    print(e)
    """
    Exceeded the per_request_input_tokens_limit of 10 (request_input_tokens=62). Consider raising the limit, or see the docs on usage limits for budget-aware patterns: https://pydantic.dev/docs/ai/core-concepts/agent/#usage-limits
    """

기본적으로 한계는 응답 후 공급자가 보고한 입력 토큰에 대해 검사되므로, 과대 요청은 여전히 보내지고 청구돼요(input_tokens_limit과 일치). 요청이 보내지기 전에 한계를 집행하려면 count_tokens_before_request=True를 설정해 토큰 계수 패스를 돌리세요.

런 비용 상한

토큰 한계는 지출의 대리자예요: 같은 토큰 수가 다른 모델에서 엄청나게 다른 비용이 들므로, 한 모델에 맞춘 한계는 다음에 틀려요. 런이 쓸 실제 달러를 묶으려면 RunUsage.cost(USD)를 상한하는 cost_limit을 쓰세요:

from decimal import Decimal

from pydantic_ai import Agent, UsageLimitExceeded, UsageLimits

agent = Agent('anthropic:claude-sonnet-4-6')

try:
    agent.run_sync(
        'What is the capital of Italy? Answer with just the city.',
        usage_limits=UsageLimits(cost_limit=Decimal('0.0001')),
    )
except UsageLimitExceeded as e:
    print(e)
    """
    Exceeded the `cost_limit` of 0.0001 (`usage.cost`=Decimal('0.000201')). Consider raising the limit, or see the docs on usage limits for budget-aware patterns: https://pydantic.dev/docs/ai/core-concepts/agent/#usage-limits
    """

output_tokens_limit처럼, 응답의 출력 비용은 도착해야 알 수 있으므로 각 응답 후에 검사돼요. count_tokens_before_request=True를 설정하면 계산된 입력 토큰도 가격을 매기고, 그 하한만으로 한계를 넘으면 요청을 사전에 거부해요.

참고 — 비용은 최선 노력이에요: genai-prices에 가격 데이터가 없는 모델·공급자(설치 후 출시된 모델을 가격 최신 유지로 갱신하지 않으면 포함)에 대해 None이에요. cost_limit이 있으면, 전혀 가격을 매길 수 없는 런은 조용히 제약 없이 두는 대신 CostNotFoundWarning을 방출해요. 예상치 못한 가격 책정 실패는 CostCalculationFailedWarning을 방출해요. cost_limit을 강한 청구 보장으로 의존하지 마세요 — request_limit이나 공급자 자신의 지출 제어와 짝지으세요.

모델 (런) 설정

Pydantic AI는 요청을 미세 조정하는 settings.ModelSettings 구조를 제공해요. 이 구조로 temperature, max_tokens, top_k, timeout 등 모델 동작에 영향을 주는 공통 파라미터를 구성할 수 있어요.

이 설정을 적용하는 세 가지 방법이 있는데 명확한 우선순위가 있어요:

1. 모델 수준 기본값 — settings 파라미터로 모델 인스턴스를 만들 때 설정. 그 모델의 기본값 기반 역할.
2. 에이전트 수준 기본값 — model_settings 인수로 Agent 초기화 시 설정. 모델 기본값과 병합되며 에이전트 설정이 우선.
3. 런타임 오버라이드 — model_settings 인수로 run{_sync,_stream} 함수에 전달. 최고 우선순위이고 결합된 에이전트·모델 기본값과 병합.

예를 들어 더 덜 무작위한 동작을 보장하려고 temperature를 0.0으로 설정하고 싶다면:

from pydantic_ai import Agent, ModelSettings
from pydantic_ai.models.openai import OpenAIChatModel

# 1. Model-level defaults
model = OpenAIChatModel(
    'gpt-5.2',
    settings=ModelSettings(temperature=0.8, max_tokens=500)  # Base defaults
)

# 2. Agent-level defaults (overrides model defaults by merging)
agent = Agent(model, model_settings=ModelSettings(temperature=0.5))

# 3. Run-time overrides (highest priority)
result_sync = agent.run_sync(
    'What is the capital of Italy?',
    model_settings=ModelSettings(temperature=0.0)  # Final temperature: 0.0
)
print(result_sync.output)
#> The capital of Italy is Rome.

최종 요청은 temperature=0.0(런타임), max_tokens=500(모델)을 사용해, 런타임이 우선하며 설정이 어떻게 병합되는지 보여줘요.

동적 모델 설정

에이전트 수준과 런 수준 model_settings는 둘 다 RunContext를 받아 ModelSettings를 반환하는 호출 가능한 것을 받아들여요. 그 호출 가능한 것은 매 모델 요청 전에 호출되므로, 설정이 단계마다 달라질 수 있어요. 지금까지 해결된 설정은 콜백 안에서 ctx.model_settings로 사용 가능해요.

설정은 계층으로 해결되고, 각각 이전 위에 병합돼요:

1. 모델 기본값 (model.settings)
2. 에이전트 수준 (Agent(model_settings=...))
3. capability 수준 (예: Thinking() — Capabilities 참조)
4. 런 수준 (agent.run(model_settings=...))

콜백 안에서 ctx.model_settings는 이전 모든 계층(위치 의존)의 병합 결과를 담아요. 예를 들어 에이전트 수준 콜백은 모델 기본값만 보고, 런 수준 콜백은 모델 기본값 + 에이전트 수준 + capability 수준 설정을 봐요. 이전 계층이 설정한 필드를 재설정하려면 명시적으로 설정하세요(예: {'temperature': None}).

from pydantic_ai import Agent, ModelSettings

agent = Agent(
    'test',
    model_settings=lambda ctx: ModelSettings(
        temperature=0.0 if ctx.run_step <= 1 else 0.7,
    ),
)

모델 설정 지원 — 모델 수준 설정은 모든 구체 모델 구현(OpenAI, Anthropic, Google 등)이 지원해요. FallbackModel과 WrapperModel 같은 래퍼 모델은 자체 설정이 없어요. 기저 모델의 설정을 써요.

런 메타데이터

런 메타데이터는 각 에이전트 실행을 컨텍스트 세부사항(예: 트레이스·로그를 필터링할 테넌트 ID)으로 태그하고, 완료 후 AgentRun.metadata, AgentRunResult.metadata, StreamedRunResult.metadata로 읽을 수 있게 해줘요. 해결된 메타데이터는 런 중 [RunContext](https://pydantic.dev/docs/ai/api/pydantic-ai/tools/#pydantic_ai.tools.RunContext)]에 붙고, 계측이 켜져 있으면 관찰성 도구를 위해 런 span 속성에 추가돼요.

Agent에서 메타데이터를 구성하거나 런에 넘기세요. 둘 다 정적 사전 또는 RunContext를 받는 호출 가능한 것을 받아요. 메타데이터는 런이 시작할 때 계산(호출 가능하면)되어 적용되고, 런이 성공적으로 끝난 후 재계산되므로, 런 종료 값을 포함할 수 있어요. 에이전트 수준 메타데이터와 런별 메타데이터는 병합되며, 런별 값이 에이전트 수준을 오버라이드해요.

from dataclasses import dataclass

from pydantic_ai import Agent


@dataclass
class Deps:
    tenant: str


agent = Agent[Deps](
    'openai:gpt-5.2',
    deps_type=Deps,
    metadata=lambda ctx: {'tenant': ctx.deps.tenant},  # agent-level metadata
)

result = agent.run_sync(
    'What is the capital of France?',
    deps=Deps(tenant='tenant-123'),
    metadata=lambda ctx: {'num_requests': ctx.usage.requests},  # per-run metadata
)
print(result.output)
#> The capital of France is Paris.
print(result.metadata)
#> {'tenant': 'tenant-123', 'num_requests': 1}

동시성 제한

max_concurrency 파라미터로 동시 에이전트 런 수를 제한할 수 있어요. 여러 에이전트 인스턴스를 병렬로 실행할 때 외부 리소스를 압도하지 않거나 속도 제한을 집행하고 싶을 때 유용해요.

import asyncio

from pydantic_ai import Agent, ConcurrencyLimit

# Simple limit: allow up to 10 concurrent runs
agent = Agent('openai:gpt-5', max_concurrency=10)


# With backpressure: limit concurrent runs and queue depth
agent_with_backpressure = Agent(
    'openai:gpt-5',
    max_concurrency=ConcurrencyLimit(max_running=10, max_queued=100),
)


async def main():
    # These will be rate-limited to 10 concurrent runs
    results = await asyncio.gather(
        *[agent.run(f'Question {i}') for i in range(20)]
    )
    print(len(results))
    #> 20

동시성 한계에 도달하면 agent.run()이나 agent.iter()의 추가 호출은 슬롯이 생길 때까지 기다려요. max_queued를 구성하고 큐가 차면 ConcurrencyLimitExceeded 예외가 발생해요.

계측이 켜져 있으면 대기 작업은 큐 깊이와 한계를 보여주는 속성이 있는 "waiting for concurrency" span으로 나타나요.

모델별 설정

모델 동작을 더 커스터마이즈하려면 선택한 모델과 연관된 GoogleModelSettings 같은 ModelSettings의 서브클래스를 쓸 수 있어요.

예를 들어:

from pydantic_ai import Agent, UnexpectedModelBehavior
from pydantic_ai.models.google import GoogleModelSettings

agent = Agent('google:gemini-3-flash-preview')

try:
  result = agent.run_sync(
      'Write a list of 5 very rude things that I might say to the universe after stubbing my toe in the dark:',
      model_settings=GoogleModelSettings(
          temperature=0.0,  # general model settings can also be specified
          gemini_safety_settings=[
              {
                  'category': 'HARM_CATEGORY_HARASSMENT',
                  'threshold': 'BLOCK_LOW_AND_ABOVE',
              },
              {
                  'category': 'HARM_CATEGORY_HATE_SPEECH',
                  'threshold': 'BLOCK_LOW_AND_ABOVE',
              },
          ],
      ),
  )
except UnexpectedModelBehavior as e:
  print(e)  # (1)
  """
  Content filter 'SAFETY' triggered, body:
  <safety settings details>
  """

이 오류는 안전 기준값이 초과됐기 때문에 발생해요.

런 vs 대화 (Runs vs. Conversations)

에이전트 런이 전체 대화를 나타낼 수도 있어요 — 단일 런에서 교환할 수 있는 메시지 수에 제한은 없어요. 하지만 대화는 별도 상호작용이나 API 호출 사이에 상태를 유지해야 한다면 여러 런으로 구성될 수도 있어요.

여러 런으로 구성된 대화 예시:

from pydantic_ai import Agent

agent = Agent('openai:gpt-5.2')

# First run
result1 = agent.run_sync('Who was Albert Einstein?')
print(result1.output)
#> Albert Einstein was a German-born theoretical physicist.

# Second run, passing previous messages
result2 = agent.run_sync(
  'What was his most famous equation?',
  message_history=result1.new_messages(),  # (1)
)
print(result2.output)
#> Albert Einstein's most famous equation is (E = mc^2).

대화를 이어가요. message_history가 없으면 모델은 "his"가 누구를 가리키는지 모를 거예요.

(이 예제는 완전해서 그대로 실행할 수 있어요)

디자인상 타입 안전 (Type safe by design)

Pydantic AI는 mypy와 pyright 같은 정적 타입 검사기와 잘 작동하도록 설계됐어요.

타이핑은 (어느 정도) 선택적이에요 — Pydantic AI는 쓰기로 하면 타입 검사가 최대한 유용하도록 설계했지만, 어디서나 항상 타입을 쓰지 않아도 돼요.

그렇긴 해도, Pydantic AI가 Pydantic을 쓰고 Pydantic이 스키마·검증의 정의로 타입 힌트를 쓰기 때문에, 일부 타입(구체적으로 도구 파라미터의 타입 힌트와 Agent의 output_type 인수)은 런타임에 사용돼요.

(라이브러리 개발자인) 우리는 타입 힌트가 돕기보다 헷갈리게 하면 실수한 거예요. 그런 걸 발견하면 무슨 점이 짜증나는지 설명하는 이슈를 만들어 주세요!

특히 에이전트는 그 의존성 타입과 반환하는 출력 타입 모두에 대해 제네릭이라, 타입 힌트로 올바른 타입을 쓰고 있는지 보장할 수 있어요.

타입 실수가 있는 다음 스크립트를 보세요:

from dataclasses import dataclass

from pydantic_ai import Agent, RunContext


@dataclass
class User:
  name: str


agent = Agent(
  'test',
  deps_type=User,  # (1)
  output_type=bool,
)


@agent.system_prompt
def add_user_name(ctx: RunContext[str]) -> str:  # (2)
  return f"The user's name is {ctx.deps}."


def foobar(x: bytes) -> None:
  pass


result = agent.run_sync('Does their name start with "A"?', deps=User('Anne'))
foobar(result.output)  # (3)

에이전트는 deps로 User 인스턴스를 기대하도록 정의돼요.

하지만 여기서 add_user_name은 의존성으로 User가 아니라 str을 받도록 정의돼 있어요.

에이전트가 bool을 반환하도록 정의됐으므로, foobar가 bytes를 기대해서 타입 오류가 발생해요.

mypy를 실행하면 다음 출력이 주어져요:

➤ uv run mypy type_mistakes.py
type_mistakes.py:18: error: Argument 1 to "system_prompt" of "Agent" has incompatible type "Callable[[RunContext[str]], str]"; expected "Callable[[RunContext[User]], str]"  [arg-type]
type_mistakes.py:28: error: Argument 1 to "foobar" has incompatible type "bool"; expected "bytes"  [arg-type]
Found 2 errors in 1 file (checked 1 source file)

pyright도 같은 문제를 식별할 거예요.

시스템 프롬프트 (System Prompts)

시스템 프롬프트는 단지 문자열이므로(또는 연결되는 문자열 시퀀스) 처음 보면 단순해 보일 수 있지만, 올바른 시스템 프롬프트를 만드는 것이 모델을 원하는 대로 행동하게 하는 핵심이에요.

팁 — 대부분의 사용 사례에서는 "시스템 프롬프트" 대신 instructions를 써야 해요.

뭘 하고 있는지 안다면, 그리고 시스템 프롬프트 메시지를 이후 완성 요청에서 LLM에 보내지는 메시지 기록에 유지하고 싶다면, system_prompt 인수/데코레이터로 그걸 달성할 수 있어요.

자세한 내용은 아래 지시사항 섹션을 보세요.

일반적으로 시스템 프롬프트는 두 범주로 나눠져요:

1. 정적 시스템 프롬프트: 코드 작성 시 알려져 있고, Agent 생성자의 system_prompt 파라미터로 정의할 수 있어요.
2. 동적 시스템 프롬프트: 런타임까지 알 수 없는 컨텍스트에 어떤 식으로든 의존하고, @agent.system_prompt로 장식된 함수로 정의해야 해요.

둘 다 단일 에이전트에 추가할 수 있고, 런타임에 정의된 순서대로 추가돼요.

두 유형을 모두 쓰는 예시:

from datetime import date

from pydantic_ai import Agent, RunContext

agent = Agent(
  'openai:gpt-5.2',
  deps_type=str,  # (1)
  system_prompt="Use the customer's name while replying to them.",  # (2)
)


@agent.system_prompt  # (3)
def add_the_users_name(ctx: RunContext[str]) -> str:
  return f"The user's name is {ctx.deps}."


@agent.system_prompt
def add_the_date() -> str:  # (4)
  return f'The date is {date.today()}.'


result = agent.run_sync('What is the date?', deps='Frank')
print(result.output)
#> Hello Frank, the date today is 2032-01-02.

에이전트가 문자열 의존성을 기대해요.

에이전트 생성 시 정의된 정적 시스템 프롬프트.

RunContext로 데코레이터를 통해 정의된 동적 시스템 프롬프트. 에이전트가 생성될 때가 아니라 run_sync 직후에 호출되므로, 그 런에 쓰인 의존성 같은 런타임 정보의 혜택을 받을 수 있어요.

또 다른 동적 시스템 프롬프트. 시스템 프롬프트는 RunContext 파라미터를 가질 필요가 없어요.

(이 예제는 완전해서 그대로 실행할 수 있어요)

지시사항 (Instructions)

지시사항은 시스템 프롬프트와 비슷해요. 주요 차이는 Agent.run과 유사한 메서드 호출에 명시적 message_history가 주어지면, 기록 안의 기존 메시지들의 지시사항 이 모델에 대한 요청에 포함되지 않고 — 현재 에이전트의 지시사항만 포함된다는 거예요.

이렇게 쓰세요:

• instructions — 모델에 대한 요청에 현재 에이전트의 시스템 프롬프트만 포함되길 원할 때.
• system_prompt — 모델에 대한 요청이 (다른 에이전트로 만든 것일 수 있는) 이전 요청에서 쓰인 시스템 프롬프트를 유지 하길 원할 때.

일반적으로 특별한 이유가 없으면 system_prompt 대신 instructions를 권장해요.

지시사항은 시스템 프롬프트처럼 다양한 시점에 지정될 수 있어요:

1. 정적 지시사항: 코드 작성 시 알려져 있고, Agent 생성자의 instructions 파라미터로 정의할 수 있어요.
2. 동적 지시사항: 런타임에만 사용 가능한 컨텍스트에 의존하고, @agent.instructions로 장식된 함수로 정의해야 해요. message_history가 있을 때 재사용될 수 있는 동적 시스템 프롬프트와 달리, 동적 지시사항은 항상 재평가돼요.
3. 런타임 지시사항: 특정 런을 위한 추가 지시사항으로, instructions 인수로 실행 메서드 중 하나에 넘길 수 있어요.

세 유형 모두 단일 에이전트에 추가할 수 있고, 런타임에 정의된 순서대로 추가돼요. 각 지시사항은 내부적으로 정적(instructions 파라미터의 리터럴 문자열) 또는 동적(@agent.instructions 함수, 런타임 지시사항, toolset 지시사항)으로 분류돼요. 정적 지시사항은 항상 동적 것보다 먼저 정렬돼요. 이 정렬은 프롬프트 캐싱을 지원하는 공급자(예: Anthropic과 Bedrock)가 안정된 정적 접두사를 캐시하면서 동적 지시사항을 캐시 경계 밖에 두게 해줘요.

정적 지시사항과 동적 지시사항을 모두 쓰는 예시:

from datetime import date

from pydantic_ai import Agent, RunContext

agent = Agent(
  'openai:gpt-5.2',
  deps_type=str,  # (1)
  instructions="Use the customer's name while replying to them.",  # (2)
)


@agent.instructions  # (3)
def add_the_users_name(ctx: RunContext[str]) -> str:
  return f"The user's name is {ctx.deps}."


@agent.instructions
def add_the_date() -> str:  # (4)
  return f'The date is {date.today()}.'


result = agent.run_sync('What is the date?', deps='Frank')
print(result.output)
#> Hello Frank, the date today is 2032-01-02.

에이전트가 문자열 의존성을 기대해요.

에이전트 생성 시 정의된 정적 지시사항.

RunContext로 데코레이터를 통해 정의된 동적 지시사항. 에이전트가 생성될 때가 아니라 run_sync 직후에 호출되므로, 그 런에 쓰인 의존성 같은 런타임 정보의 혜택을 받을 수 있어요.

또 다른 동적 지시사항. 지시사항은 RunContext 파라미터를 가질 필요가 없어요.

(이 예제는 완전해서 그대로 실행할 수 있어요)

빈 문자열을 반환하면 지시사항 메시지가 추가되지 않는다는 점에 주의하세요.

지시사항은 capabilities의 get_instructions(), toolsets의 get_instructions(), 또는 에이전트 의존성에 대해 렌더링된 템플릿 문자열에서 올 수도 있어요.

지시사항 파트 (Instruction parts)

각 소스는 자체 지시사항 파트에 기여해요. 파트는 빈 줄로 구분된 하나의 문자열로 모델에 보내지고, ModelRequestParameters.instruction_parts에서 InstructionPart로 개별적으로도 사용 가능해요.

당신이 name을 선언하고, 프레임워크가 id를 발행해요. 당신이 소유하는 것에 상대적으로 파트 이름을 지으세요 — 'limits'지, 'toolset:weather:limits'가 아니에요 — 그리고 기여하는 소스가 나머지를 제공하므로, 자신의 정체성을 반복할 일이 없고 다른 소스의 키를 주장할 수 없어요. 돌려받는 InstructionId는 파트를 기여한 source와 name(있으면)을 짝짓고, 그 세그먼트들을 :로 이은 것으로 렌더링돼요:

str(part.id)	주소 지정
'agent'	에이전트 자신의 instructions
'toolset:<toolset id>'	id가 있는 toolset이 기여한 모든 것
'capability:<capability id>'	id가 있는 capability가 기여한 모든 것
'agent:<name>'	에이전트가 이름 지은 하나의 파트
'toolset:<toolset id>:<name>'	그 toolset이 이름 지은 하나의 파트
'capability:<capability id>:<name>'	그 capability가 이름 지은 하나의 파트

파트가 무엇이냐에 따라 이름을 선언하는 두 가지 방법:

• 함수는 등록된 곳에서 이름 지어져요: @agent.instructions(name=...), @capability.instructions(name=...).
• 리터럴 텍스트는 지시사항이 받아들여지는 어디든 InstructionPart를 넘겨 자기 이름을 텍스트 자신에 실어요 — Agent(instructions=...), Capability(instructions=...), FunctionToolset(instructions=...), 또는 capability나 toolset의 get_instructions() 구현. 파트는 또한 자신이 dynamic으로 여겨지는지 결정하는데, 그것이 파트를 캐시 가능한 접두사 밖에 두는 요인이고(프롬프트 캐싱 참조), 파트는 항상 통째로 유지되며 이웃과 병합되지 않아요.

파트의 id는 런에 걸쳐 안정적이므로, 지시사항 구성을 다른 곳에 저장하는 앱(예: 사용자가 MCP 서버가 기여하는 지시사항을 편집하는 UI)은 파트의 위치나 문구 대신 그 id에 구성을 키잉할 수 있는데, 둘 다 에이전트가 진화함에 따라 바뀌니까요.

소스 키는 다른 모든 것이 그 위에 세워지는 것이므로 의미를 영구히 유지해요: 나중에 더 많은 소스에 id를 주는 것은 키를 추가할 뿐, 기존 키가 주소 지정하는 것을 결코 바꾸지 않아요.

capability id, toolset id, 지시사항 이름은 :를 포함할 수 없어요. 그 문자는 이 세그먼트 사이의 구분자로 예약돼 있기 때문이에요. 이름 'agent'도 예약돼 있어요. 단독으로 에이전트 자신의 지시사항의 키이기 때문이에요.

from pydantic_ai import Agent, RunContext
from pydantic_ai.capabilities import Capability
from pydantic_ai.models.test import TestModel

model = TestModel()
agent = Agent(
    model,
    instructions='Be concise.',
    capabilities=[Capability(instructions='Cite your sources.', id='research')],
)


@agent.instructions(name='local_time')
def local_time() -> str:
    return 'The time is 10:00.'


@agent.instructions
def user_name(ctx: RunContext[None]) -> str:
    return 'The user is Frank.'


agent.run_sync('What is the capital of Italy?')

parts = model.last_model_request_parameters.instruction_parts or []
print([(part.name, str(part.id) if part.id is not None else None, part.content) for part in parts])
"""
[
    (None, 'agent', 'Be concise.'),
    (None, 'capability:research', 'Cite your sources.'),
    ('local_time', 'agent:local_time', 'The time is 10:00.'),
    (None, None, 'The user is Frank.'),
]
"""

(이 예제는 완전해서 그대로 실행할 수 있어요)

id에 구성을 키잉하기 전에 알 두 가지 결과:

• 키는 그 아래 모든 것을 덮어요. 소스가 여러 파트를 기여하고 그중 어느 것도 이름 지어지지 않았다면, 모두 소스 키를 실어요. 그래서 그 키의 텍스트를 교체하면 계산된 파트를 포함해 모두 교체돼요. 그것이 "내가 이 capability가 모델에 말하는 것을 제어한다"의 정직한 의미지만, 소스가 이름 지지 않은 파트는 하나씩 주소 지정할 수 없다는 뜻이에요. 'agent'는 의도된 예외예요: 그 에이전트가 지어진 리터럴 지시사항만 덮으므로, 기본 프롬프트를 인수받아도 날짜나 사용자 이름을 주입하는 @agent.instructions 함수를 조용히 삼키지 않아요.
• 어떤 파트는 전혀 주소 지정될 수 없어요. 그것들은 다른 것처럼 프롬프트에 참여하지만 아무것도 키잉하지 않아요: name 없는 지시사항 함수(함수의 자체 이름은 고유하지 않고, 람다나 템플릿 문자열은 이름이 없어요), Agent(instructions=...)에 넘긴 호출 가능한 것, 특정 런의 런타임 지시사항으로 넘긴 것, id 없는 toolset·capability의 것. 자체 호출 가능한 것을 오버라이드 가능하게 해야 한다면, 생성자에 넘기는 대신 @agent.instructions(name=...)이나 @capability.instructions(name=...)으로 등록하세요.
• 소스에 id가 없는 파트에 이름을 주면 그 id는 None으로 남아요. 이름을 한정할 소스 키가 없기 때문이에요. 이름은 여전히 파트와 함께 여행해서, 작성자가 무엇이라 불렀는지 볼 수 있지만 아무것도 주소 지정하지 않아요.

반성과 자기 교정 (Reflection and self-correction)

함수 도구 파라미터 검증과 구조화된 출력 검증 양쪽의 검증 오류를 재시도 요청과 함께 모델에 다시 전달할 수 있어요.

도구나 출력 함수 안에서 ModelRetry를 올려 모델이 응답 생성을 재시도해야 한다고 알릴 수도 있어요.

이것은 런 중 재시도할 수 있는 여러 계층 중 하나로, 각각 자체 예산이 있어요.

• 기본 재시도 횟수는 1이지만, retries나 AgentRetries로 전체 에이전트, 특정 도구, 출력에 대해 바꿀 수 있어요. 에이전트 재시도 예산의 도구·출력 양쪽 모두 agent.run(retries={'tools': ..., 'output': ...})(그리고 친구들)로 오버라이드할 수 있고, 런 블록은 agent.override()로요. 이 호출 지점에서 맨 int는 생성 때처럼 두 예산을 모두 오버라이드해요. retries={'tools': ...} 같은 dict를 넘겨 하나만 오버라이드하세요. 도구 재시도 기본값과 그 런별 오버라이드는 함수 도구·출력 도구·MCP 도구에 적용돼요.
• 도구, 출력 검증기, 출력 함수 안에서 ctx.retry로 현재 재시도 횟수에 접근할 수 있어요.

출력 재시도가 어떻게 집행되나

Pydantic AI는 모델이 최종 출력을 반환하는 방법에 따라 출력 재시도 예산을 다르게 집행해요:

• 텍스트 출력 경로 (output_type=str, 텍스트 전용 출력, 비었거나 쓸 수 없는 모델 응답): 단일 전역 예산이 런 전체에 공유돼요. 무효한 응답마다 예산 한 단위를 소비하고, 소진되면 런이 메시지 'Exceeded maximum output retries (N)'와 함께 UnexpectedModelBehavior를 발생시켜요.
• 도구 출력 경로 (output_type=ToolOutput(...), 구조화된 출력): 출력 재시도 예산은 기본 도구별 한계 예요. 도구별 오버라이드는 ToolOutput(max_retries=N)로 도구 출력을 보세요.

출력 검증기 안에서 예산이 어떻게 나타나는가 — 각 경로에서 ctx.max_retries와 ctx.retry가 무엇을 반영하는지 포함 — 는 출력 검증기 섹션을 보세요.

도구 재시도는 도구별로 추적돼요 — 도구별 카운터 모델과 세 구성 수준은 도구 실행, 재시도, 실패를 보세요.

예시:

from pydantic import BaseModel

from pydantic_ai import Agent, RunContext, ModelRetry

from fake_database import DatabaseConn


class ChatResult(BaseModel):
    user_id: int
    message: str


agent = Agent(
    'openai:gpt-5.2',
    deps_type=DatabaseConn,
    output_type=ChatResult,
)


@agent.tool(retries=2)
def get_user_by_name(ctx: RunContext[DatabaseConn], name: str) -> int:
    """Get a user's ID from their full name."""
    print(name)
    #> John
    #> John Doe
    user_id = ctx.deps.users.get(name=name)
    if user_id is None:
        raise ModelRetry(
            f'No user found with name {name!r}, remember to provide their full name'
        )
    return user_id


result = agent.run_sync(
    'Send a message to John Doe asking for coffee next week', deps=DatabaseConn()
)
print(result.output)
"""
user_id=123 message='Hello John, would you be free for coffee sometime next week? Let me know what works for you!'
"""

디버깅과 모니터링 (Debugging and Monitoring)

에이전트는 전통적인 소프트웨어와 다른 관찰성 접근이 필요해요. 전통적 웹 엔드포인트나 데이터 파이프라인은 코드를 읽어 행동을 대부분 예측할 수 있어요. 에이전트는 그게 훨씬 어려워요. 모델의 결정은 확률적이고, 그 확률성은 에이전트가 추론·도구 호출·결과 관찰·다시 추론하는 에이전트 루프를 통해 증폭돼요. 실제로 무슨 일이 있었는지 봐야 해요.

이는 개발 중(이해하고 반복하기 위해)과 프로덕션(문제 디버그·행동 모니터링) 모두에서, 나중에 검토할 수 있는 형태로 무슨 일이 일어나는지 기록하도록 앱을 설정하는 것을 뜻해요. 인체공학도 중요해요: 일어난 모든 것의 평문 덤프는 개발 중에도 에이전트 행동을 검토하는 실용적인 방법이 아니에요. 각 결정과 도구 호출을 대화형으로 단계별로 살펴볼 수 있는 도구를 원해요.

Pydantic AI 워크플로를 염두에 두고 설계된 Pydantic Logfire를 권장해요.

Logfire로 추적

import logfire

logfire.configure()
logfire.instrument_pydantic_ai()

Logfire 계측을 켜면 모든 에이전트 런이 다음을 보여주는 상세 트레이스를 만들어요:

• 모델과 주고받은 메시지 (시스템, 사용자, 어시스턴트)
• 인수와 반환 값을 포함한 도구 호출
• 요청별·누적 토큰 사용량
• 각 연산의 지연
• 전체 컨텍스트가 있는 오류

이 가시성은 다음에 매우 귀중해요:

• 에이전트가 특정 결정을 내린 이유 이해
• 예상치 못한 행동 디버깅
• 성능·비용 최적화
• 프로덕션 배포 모니터링

Evals로 체계적 테스트

런타임 디버깅 너머의 에이전트 행동 체계적 평가를 위해 Pydantic Evals가 AI 시스템 테스트를 위한 코드 우선 프레임워크를 제공해요:

from pydantic_evals import Case, Dataset

dataset = Dataset(
    name='agent_eval',
    cases=[
        Case(name='capital_question', inputs='What is the capital of France?', expected_output='Paris'),
    ]
)
report = dataset.evaluate_sync(my_agent_function)

Evals로 테스트 케이스를 정의하고, 에이전트에 대해 실행하며, 결과를 채점할 수 있어요. Logfire와 결합하면 평가 결과가 웹 UI에 나타나 런 간 시각화·비교가 가능해요. 설정은 Logfire 통합 가이드를 보세요.

다른 백엔드 사용

Pydantic AI의 계측은 OpenTelemetry 위에 지어져 있어서, 트레이스를 호환 백엔드 어디로든 보낼 수 있어요. 편의상 Logfire SDK를 써도 다른 백엔드로 데이터를 보내도록 구성할 수 있어요. 설정은 대체 백엔드를 보세요.

전체 Logfire 통합 가이드 →

모델 오류 (Model errors)

모델이 예상치 못하게 행동하면(예: 재시도 한계를 초과하거나 API가 503을 반환), 에이전트 런은 UnexpectedModelBehavior를 발생시켜요.

이런 경우 capture_run_messages로 런 중 주고받은 메시지에 접근해 문제를 진단할 수 있어요.

실패가 아니라 취소된 런에는 RunCancelled와 RunCancelled.from_cancellation()이 런의 기록을 직접 실어요 — 런 취소하기 참조.

from pydantic_ai import Agent, ModelRetry, UnexpectedModelBehavior, capture_run_messages

agent = Agent('openai:gpt-5.2')


@agent.tool_plain
def calc_volume(size: int) -> int:  # (1)
  if size == 42:
      return size**3
  else:
      raise ModelRetry('Please try again.')


with capture_run_messages() as messages:  # (2)
  try:
      result = agent.run_sync('Please get me the volume of a box with size 6.')
  except UnexpectedModelBehavior as e:
      print('An error occurred:', e)
      """
      An error occurred:
      Tool 'calc_volume' exceeded max retries count of 1. Consider raising the retry limit, or see the docs on tool retries: https://pydantic.dev/docs/ai/tools-toolsets/tools-advanced/#tool-retries
      """
      print('cause:', repr(e.__cause__))
      #> cause: ModelRetry('Please try again.')
      print('messages:', messages)
      """
      messages:
      [
          ModelRequest(
              parts=[
                  UserPromptPart(
                      content='Please get me the volume of a box with size 6.',
                      timestamp=datetime.datetime(...),
                  )
              ],
              timestamp=datetime.datetime(...),
              run_id='...',
              conversation_id='...',
          ),
          ModelResponse(
              parts=[
                  ToolCallPart(
                      tool_name='calc_volume',
                      args={'size': 6},
                      tool_call_id='pyd_ai_tool_call_id',
                  )
              ],
              usage=RequestUsage(
                  cost=Decimal('0.0001645'), input_tokens=62, output_tokens=4
              ),
              model_name='gpt-5.2',
              timestamp=datetime.datetime(...),
              run_id='...',
              conversation_id='...',
          ),
          ModelRequest(
              parts=[
                  RetryPromptPart(
                      content='Please try again.',
                      tool_name='calc_volume',
                      tool_call_id='pyd_ai_tool_call_id',
                      timestamp=datetime.datetime(...),
                  )
              ],
              timestamp=datetime.datetime(...),
              run_id='...',
              conversation_id='...',
          ),
          ModelResponse(
              parts=[
                  ToolCallPart(
                      tool_name='calc_volume',
                      args={'size': 6},
                      tool_call_id='pyd_ai_tool_call_id',
                  )
              ],
              usage=RequestUsage(
                  cost=Decimal('0.000238'), input_tokens=72, output_tokens=8
              ),
              model_name='gpt-5.2',
              timestamp=datetime.datetime(...),
              run_id='...',
              conversation_id='...',
          ),
          ModelRequest(
              parts=[],
              timestamp=datetime.datetime(...),
              run_id='...',
              conversation_id='...',
              state='interrupted',
          ),
      ]
      """
  else:
      print(result.output)

이 경우 ModelRetry를 반복해 올릴 도구를 정의해요.

capture_run_messages는 런 중 주고받은 메시지를 포착하는 데 쓰여요.

(이 예제는 완전해서 그대로 실행할 수 있어요)

런이 스트리밍 중 예외, 도구 안 예외, 외부 취소로 중단되면, Pydantic AI는 가능한 곳에서 부분 상태를 여전히 포착해요. 부분 ModelResponse와 ModelRequest 메시지는 state='interrupted'를 가져서 영속 계층과 UI가 완전한 메시지와 구별할 수 있어요.

모델 응답의 경우 중단된 메시지는 중단 전에 스트리밍된 응답 파트를 담아요. 모델 요청의 경우 중단된 메시지는 도구 실행이 멈추기 전에 완료된 도구 결과를 담아요 — 아무것도 없으면 요청은 여전히 기록되고 파트 없이, 응답의 도구 호출이 포기된 지점을 표시해요. 포착된 메시지는 정확히 일어난 일을 반영해요 — 절반 만들어진 도구 호출 파트는 포착 시점에 합성 도구 결과로 바뀌지 않아요. 중단된 내역이 다시 런에 넘겨지면, 다음 모델 요청 전에 자동 복구돼요.

이 예시에서 get_volume이 get_mass가 발생시키기 전에 완료되므로, 중단된 요청은 완료된 get_volume 반환을 담아요:

from pydantic_ai import Agent, ModelRequest, capture_run_messages
from pydantic_ai.messages import (
    ModelMessage,
    ModelResponse,
    ToolCallPart,
    ToolReturnPart,
)
from pydantic_ai.models.function import AgentInfo, FunctionModel


def call_tools(_messages: list[ModelMessage], _info: AgentInfo) -> ModelResponse:
    return ModelResponse(
        parts=[
            ToolCallPart(tool_name='get_volume', args={'size': 6}, tool_call_id='volume_call'),
            ToolCallPart(tool_name='get_mass', args={'size': 6}, tool_call_id='mass_call'),
        ]
    )


agent = Agent(FunctionModel(function=call_tools))


@agent.tool_plain(sequential=True)
def get_volume(size: int) -> int:
    return size**3


@agent.tool_plain(sequential=True)
def get_mass(size: int) -> int:
    raise RuntimeError('missing density')


with capture_run_messages() as messages:
    try:
        agent.run_sync('Calculate volume and mass.')
    except RuntimeError as exc:
        print(f'Run failed: {exc}')
        #> Run failed: missing density

interrupted_request = next(
    message for message in messages if isinstance(message, ModelRequest) and message.state == 'interrupted'
)
assert any(
    isinstance(part, ToolReturnPart) and part.tool_name == 'get_volume' and part.content == 216
    for part in interrupted_request.parts
)

참고 — 하나의 capture_run_messages 컨텍스트 안에서 run, run_sync, run_stream을 두 번 이상 호출하면, messages는 첫 호출 동안만 주고받은 메시지를 나타내요.

capture_run_messages 컨텍스트는 중첩될 수 있어요: 각 컨텍스트는 그것이 가장 안쪽 활성 컨텍스트인 런을 포착해요. 중첩 컨텍스트 안에서 시작된 런은 바깥 컨테크스트가 아니라 그 중첩 컨텍스트가 포착해요. 이는 중첩 에이전트 런(예: 다른 에이전트를 호출하는 도구 안)을 자체 capture_run_messages로 감싸 내부 런의 메시지를 독립적으로 검사할 수 있음을 뜻해요.

에이전트 스펙 (Agent Specs)

에이전트는 agent specs로 YAML이나 JSON에서 선언적으로도 정의할 수 있어요. 이는 에이전트 구성을 애플리케이션 코드에서 분리해요:

model: anthropic:claude-opus-4-6
instructions: You are a helpful assistant.
capabilities:
  - WebSearch
  - Thinking:
      effort: high

from pydantic_ai import Agent

agent = Agent.from_file('agent.yaml')

더 알아보기 (Learn more)

• 함수 도구 (Function Tools)
• 고급 도구 기능 (Advanced Tool Features)
• 출력 (Output)
• 의존성 (Dependencies)
• 메시지와 채팅 기록 (Messages and chat history)
• 타임아웃 (Timeouts)
• 재시도 (Retries)
• 훅 (Hooks)