Multi-agent Applications

Pydantic AI로 애플리케이션을 만들 때의 복잡도는 대략 다섯 단계로 나눌 수 있어요.

단일 에이전트 워크플로 — 대부분의 pydantic_ai 문서가 다루는 내용이에요.
에이전트 위임 (Agent delegation) — 에이전트가 도구를 통해 다른 에이전트를 쓰는 것. 손으로 연결하거나 Harness의 SubAgents 기능을 써요.
프로그램적 에이전트 핸드오프 (Programmatic agent hand-off) — 한 에이전트가 실행된 뒤, 애플리케이션 코드가 다른 에이전트를 호출하는 것.
그래프 기반 제어 흐름 (Graph based control flow) — 가장 복잡한 경우, 그래프 기반 상태 머신으로 여러 에이전트의 실행을 제어해요.
딥 에이전트 (Deep Agents) — 계획, 파일 연산, 작업 위임, 샌드박스 코드 실행을 갖춘 자율 에이전트.

물론 한 애플리케이션에서 여러 전략을 결합할 수도 있어요.

출처: 공식문서

에이전트 위임 (Agent delegation)

"에이전트 위임"은 에이전트가 다른 에이전트에게 작업을 넘기고, 위임받은 에이전트(도구 안에서 호출된 에이전트)가 끝나면 다시 제어권을 되찾는 시나리오를 말해요. 제어권을 첫 번째 에이전트로 돌아오지 않고 완전히 다른 에이전트에 넘기고 싶다면 출력 함수를 쓸 수 있어요.

SubAgents가 이걸 대신 해줘요

Pydantic AI Harness에는 위임을 위한 가장 편리한 방식인 SubAgents 기능이 들어 있어요. 이름 붙인 에이전트 목록을 넘기면, 단일 delegate_task(agent_name, task) 도구를 노출하고, 의존성을 전달하며, 사용량 한도를 스레딩하고, 사용 가능한 위임자 목록을 캐시 가능한 프리픽스에 유지되도록 시스템 프롬프트에 정적 지시문으로 나열해요(see prompt caching). 각 위임은 고유한 메시지 히스토리를 가진 자체 실행으로 진행되므로, 위임자는 부모 대화를 보지 못해요.

SubAgents가 주지 못하는 제어가 필요할 때 — 특수한 도구 스키마, 위임자별 인자 검증, 부모의 메시지 히스토리 전달 등 — 아래처럼 위임 도구를 손으로 작성해요.

에이전트는 상태가 없고(stateless) 전역으로 설계되었기 때문에, 에이전트 자체를 에이전트 의존성에 넣을 필요는 없어요.

보통은 위임 에이전트 실행의 usage 키워드 인자에 ctx.usage를 넘겨서, 그 실행 안의 사용량이 부모 에이전트 실행의 총 사용량에 포함되게 하고 싶을 거예요.

취소(Cancellation)는 실행 범위별로 적용돼요. 위임 에이전트가 스스로 취소하면 부모를 취소하는 대신 실패한 도구 반환으로 부모에게 표면화되고, 공유된 CancellationToken은 실행 트리 전체를 한 번에 취소해요.

여러 모델

에이전트 위임은 각 에이전트마다 같은 모델을 쓸 필요가 없어요. 한 실행 안에서 다른 모델을 선택하면, 최종 result.usage에는 계산 가능한 응답별 비용이 여전히 누적돼요. 다만 모델마다 가격이 다를 수 있어서, 화폐 비용은 집계된 토큰 수로 재구성할 수 없어요. 예상치 못한 비용이나 통제 불능의 도구 루프를 피하려면 UsageLimits — cost_limit, request_limit, total_tokens_limit, tool_calls_limit 포함 — 를 쓰면 돼요.

agent_delegation_simple.py

from pydantic_ai import Agent, RunContext, UsageLimits

joke_selection_agent = Agent(  # (1)
  'openai:gpt-5.2',
  name='joke_selection_agent',  # (2)
  instructions=(
      'Use the `joke_factory` to generate some jokes, then choose the best. '
      'You must return just a single joke.'
  ),
)
joke_generation_agent = Agent(  # (3)
  'google:gemini-3-flash-preview', name='joke_generation_agent', output_type=list[str]
)


@joke_selection_agent.tool
async def joke_factory(ctx: RunContext, count: int) -> list[str]:
  r = await joke_generation_agent.run(  # (4)
      f'Please generate {count} jokes.',
      usage=ctx.usage,  # (5)
  )
  return r.output  # (6)


result = joke_selection_agent.run_sync(
  'Tell me a joke.',
  usage_limits=UsageLimits(request_limit=5, total_tokens_limit=500),
)
print(result.output)
#> Did you hear about the toothpaste scandal? They called it Colgate.
print(result.usage)
"""
RunUsage(
  cost=Decimal('0.00051200'),
  input_tokens=165,
  output_tokens=24,
  requests=3,
  tool_calls=1,
)
"""

"부모"이자 제어하는 에이전트예요.
name을 넘기는 것은 선택이지만 둘 이상의 에이전트를 실행할 때 권장돼요. 각 에이전트의 실행 스팬을 라벨링해서, 둘 다 이름을 지으면 Logfire에서 부모와 위임자를 구분할 수 있어요. 생략하면 에이전트가 할당된 변수에서 이름이 추론되고, 그럴 수 없을 때(예: 에이전트를 리스트나 dict에 보관) 'agent'로 폴백해요.
부모 에이전트의 도구 안에서 호출되는 "위임" 에이전트예요.
부모 에이전트의 도구 안에서 위임 에이전트를 호출해요.
부모 에이전트의 사용량을 위임 에이전트에 넘겨서, 최종 result.usage가 두 에이전트의 사용량을 모두 포함하게 해요.
함수가 list[str]를 반환하고 joke_generation_agent의 output_type도 list[str]이므로, 도구에서 그냥 r.output을 반환하면 돼요.

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

async def 함수에서 위임하세요. 동기 함수가 아니라요

위의 joke_factory는 async def이고 await joke_generation_agent.run(...)을 쓰는 걸 눈여겨보세요. 이건 스타일이 아니라 필수예요. run_sync()와 run_stream_sync()는 도구, 출력 함수, 또는 에이전트 실행 중 호출되는 다른 함수 안에서 사용할 수 없고, 거기서 UserError를 던져요.

부모 에이전트는 위 예시처럼 run_sync()으로 시작할 수 있어요. 위임 함수만 async def이면 돼요. 블로킹 작업도 필요하다면, async def로 유지하고 그 부분만 asyncio.to_thread()로 밀어 넣으세요.

Temporal 워크플로 안의 위임

Temporal 액티비티에서 실행되는 도구는 실행 컨텍스트의 복사본을 받으므로, usage=ctx.usage가 위임자의 사용량을 부모 실행으로 되돌려주지 못해요. Agent Run Context and Dependencies를 참고하세요.

이 예시의 제어 흐름은 꽤 단순해서 다음과 같이 요약할 수 있어요.

graph TD
  START --> joke_selection_agent
  joke_selection_agent --> joke_factory["joke_factory (tool)"]
  joke_factory --> joke_generation_agent
  joke_generation_agent --> joke_factory
  joke_factory --> joke_selection_agent
  joke_selection_agent --> END

에이전트 위임과 의존성

일반적으로 위임 에이전트는 호출하는 에이전트와 같은 의존성을 가지거나, 호출 에이전트 의존성의 부분집합인 의존성을 가져야 해요.

의존성 초기화

위에서는 "일반적으로"라고 했어요. 도구 호출 안에서 의존성을 초기화하고, 따라서 부모에 없는 상호 의존성을 위임 에이전트에서 사용하는 것을 막는 건 없으니까요. 다만 이는 부모 에이전트의 연결 등을 재사용하는 것보다 훨씬 느릴 수 있어 자주 피해야 해요.

agent_delegation_deps.py

from dataclasses import dataclass

import httpx

from pydantic_ai import Agent, RunContext


@dataclass
class ClientAndKey:  # (1)
  http_client: httpx.AsyncClient
  api_key: str


joke_selection_agent = Agent(
  'openai:gpt-5.2',
  name='joke_selection_agent',
  deps_type=ClientAndKey,  # (2)
  instructions=(
      'Use the `joke_factory` tool to generate some jokes on the given subject, '
      'then choose the best. You must return just a single joke.'
  ),
)
joke_generation_agent = Agent(
  'google:gemini-3-flash-preview',
  name='joke_generation_agent',
  deps_type=ClientAndKey,  # (4)
  output_type=list[str],
  instructions=(
      'Use the "get_jokes" tool to get some jokes on the given subject, '
      'then extract each joke into a list.'
  ),
)


@joke_selection_agent.tool
async def joke_factory(ctx: RunContext[ClientAndKey], count: int) -> list[str]:
  r = await joke_generation_agent.run(
      f'Please generate {count} jokes.',
      deps=ctx.deps,  # (3)
      usage=ctx.usage,
  )
  return r.output


@joke_generation_agent.tool  # (5)
async def get_jokes(ctx: RunContext[ClientAndKey], count: int) -> str:
  response = await ctx.deps.http_client.get(
      'https://example.com',
      params={'count': count},
      headers={'Authorization': f'Bearer {ctx.deps.api_key}'},
  )
  response.raise_for_status()
  return response.text


async def main():
  async with httpx.AsyncClient() as client:
      deps = ClientAndKey(client, 'foobar')
      result = await joke_selection_agent.run('Tell me a joke.', deps=deps)
      print(result.output)
      #> Did you hear about the toothpaste scandal? They called it Colgate.
      print(result.usage)  # (6)
      """
      RunUsage(
          cost=Decimal('0.00056350'),
          input_tokens=220,
          output_tokens=32,
          requests=4,
          tool_calls=2,
      )
      """

클라이언트와 API 키 의존성을 담을 dataclass를 정의해요.
호출 에이전트(joke_selection_agent)의 deps_type을 설정해요.
도구 호출 안에서 위임 에이전트의 실행 메서드에 의존성을 넘겨요.
위임 에이전트(joke_generation_agent)의 deps_type도 설정해요.
의존성을 사용해 HTTP 요청을 하는 도구를 위임 에이전트에 정의해요.
이제 사용량에 요청 4개가 포함돼요 — 호출 에이전트에서 2개, 위임 에이전트에서 2개.

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

이 예시는 꽤 단순한 에이전트 위임이어도 복잡한 제어 흐름을 만들 수 있음을 보여줘요.

graph TD
  START --> joke_selection_agent
  joke_selection_agent --> joke_factory["joke_factory (tool)"]
  joke_factory --> joke_generation_agent
  joke_generation_agent --> get_jokes["get_jokes (tool)"]
  get_jokes --> http_request["HTTP request"]
  http_request --> get_jokes
  get_jokes --> joke_generation_agent
  joke_generation_agent --> joke_factory
  joke_factory --> joke_selection_agent
  joke_selection_agent --> END

프로그램적 에이전트 핸드오프 (Programmatic agent hand-off)

"프로그램적 에이전트 핸드오프"는 여러 에이전트를 연속으로 호출하되, 다음에 어떤 에이전트를 호출할지를 애플리케이션 코드와/또는 루프 안의 사람이 결정하는 시나리오를 말해요.

여기서 에이전트들은 같은 deps를 쓸 필요가 없어요.

에이전트 간 메시지 히스토리

다른 에이전트에 이전 대화를 컨텍스트로 주려면, 실행 메서드에 message_history를 넘겨요. 지시문, 시스템 프롬프트, 도구 컨텍스트에 대한 자세한 내용은 Sharing messages between agents를 참고하세요.

여기선 연속으로 사용되는 두 에이전트를 보여줘요. 첫 번째는 항공편을 찾고, 두 번째는 사용자의 좌석 선호도를 추출해요.

programmatic_handoff.py

from typing import Literal

from pydantic import BaseModel, Field
from rich.prompt import Prompt

from pydantic_ai import Agent, ModelMessage, RunContext, RunUsage, UsageLimits


class FlightDetails(BaseModel):
  flight_number: str


class Failed(BaseModel):
  """Unable to find a satisfactory choice."""


flight_search_agent = Agent[object, FlightDetails | Failed](  # (1)
  'openai:gpt-5.2',
  name='flight_search_agent',
  output_type=FlightDetails | Failed,  # type: ignore
  instructions=(
      'Use the "flight_search" tool to find a flight '
      'from the given origin to the given destination.'
  ),
)


@flight_search_agent.tool  # (2)
async def flight_search(
  ctx: RunContext, origin: str, destination: str
) -> FlightDetails | None:
  # in reality, this would call a flight search API or
  # use a browser to scrape a flight search website
  return FlightDetails(flight_number='AK456')


usage_limits = UsageLimits(request_limit=15)  # (3)


async def find_flight(usage: RunUsage) -> FlightDetails | None:  # (4)
  message_history: list[ModelMessage] | None = None
  for _ in range(3):
      prompt = Prompt.ask(
          'Where would you like to fly from and to?',
      )
      result = await flight_search_agent.run(
          prompt,
          message_history=message_history,
          usage=usage,
          usage_limits=usage_limits,
      )
      if isinstance(result.output, FlightDetails):
          return result.output
      else:
          message_history = result.all_messages(
              output_tool_return_content='Please try again.'
          )


class SeatPreference(BaseModel):
  row: int = Field(ge=1, le=30)
  seat: Literal['A', 'B', 'C', 'D', 'E', 'F']


# This agent is responsible for extracting the user's seat selection
seat_preference_agent = Agent[object, SeatPreference | Failed](  # (5)
  'openai:gpt-5.2',
  name='seat_preference_agent',
  output_type=SeatPreference | Failed,  # type: ignore
  instructions=(
      "Extract the user's seat preference. "
      'Seats A and F are window seats. '
      'Row 1 is the front row and has extra leg room. '
      'Rows 14, and 20 also have extra leg room. '
  ),
)


async def find_seat(usage: RunUsage) -> SeatPreference:  # (6)
  message_history: list[ModelMessage] | None = None
  while True:
      answer = Prompt.ask('What seat would you like?')

      result = await seat_preference_agent.run(
          answer,
          message_history=message_history,
          usage=usage,
          usage_limits=usage_limits,
      )
      if isinstance(result.output, SeatPreference):
          return result.output
      else:
          print('Could not understand seat preference. Please try again.')
          message_history = result.all_messages()


async def main():  # (7)
  usage: RunUsage = RunUsage()

  opt_flight_details = await find_flight(usage)
  if opt_flight_details is not None:
      print(f'Flight found: {opt_flight_details.flight_number}')
      #> Flight found: AK456
      seat_preference = await find_seat(usage)
      print(f'Seat preference: {seat_preference}')
      #> Seat preference: row=1 seat='A'

첫 번째 에이전트(항공편 찾기)를 정의해요. PEP-747이 적용되기 전까지는 명시적 타입 어노테이션을 사용해요(see structured output). 출력 타입으로 유니온을 써서, 모델이 만족스러운 선택을 찾지 못했을 때 이를 알릴 수 있게 해요. 내부적으로 유니온의 각 멤버는 별도의 도구로 등록돼요.
항공편을 찾는 도구를 에이전트에 정의해요. 이 단순한 경우엔 도구 없이 에이전트를 구조화 데이터 반환으로 정의하고 항공편을 검색할 수도 있지만, 더 복잡한 시나리오에선 도구가 필요해요.
전체 앱의 사용량 한도를 정의해요.
항공편을 찾는 함수를 정의해요. 사용자에게 선호도를 묻고 에이전트를 호출해 항공편을 찾아요.
위의 flight_search_agent처럼, 명시적 타입 어노테이션으로 에이전트를 정의해요.
사용자의 좌석 선호도를 찾는 함수를 정의해요. 사용자에게 좌석 선호도를 묻고, 에이전트를 호출해 좌석 선호도를 추출해요.
이제 각 에이전트를 실행하는 로직을 별도 함수에 넣었으므로, 메인 앱은 아주 단순해져요.

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

이 예시의 제어 흐름은 다음과 같이 요약할 수 있어요.

graph TB
  START --> ask_user_flight["ask user for flight"]

  subgraph find_flight
    flight_search_agent --> ask_user_flight
    ask_user_flight --> flight_search_agent
  end

  flight_search_agent --> ask_user_seat["ask user for seat"]
  flight_search_agent --> END

  subgraph find_seat
    seat_preference_agent --> ask_user_seat
    ask_user_seat --> seat_preference_agent
  end

  seat_preference_agent --> END

이 핸드오프 패턴의 핵심

이 예시의 흐름을 정리하면 이래요. find_flight는 루프 안에서 사용자에게 출발·도착지를 물어 flight_search_agent를 호출해요. 결과가 FlightDetails면 성공이므로 반환하고, 그렇지 않으면 Failed(또는 추출 실패)를 받았으므로 result.all_messages(output_tool_return_content='Please try again.')로 메시지 히스토리를 갱신하고 다시 시도해요. output_tool_return_content는 위임받은 도구의 출력을 "다시 시도하세요" 문구로 바꿔 모델에게 제공해요.

두 에이전트(flight_search_agent, seat_preference_agent)는 같은 RunUsage 객체(usage)를 공유해요. 이를 main()에 전달해서 각 실행이 같은 누적 사용량을 공유하게 하고, usage_limits는 앱 전체에 일관된 request_limit=15를 적용해요.

message_history를 다시 넘기는 것이 "핸드오프"의 핵심이에요. 두 번째 에이전트(좌석)는 첫 번째 에이전트(항공편)와 다른 대화를 시작하지만, 각자 실패했을 때는 자기 메시지 히스토리로 재시도해요.

Pydantic Graphs

그래프를 언제, 어떻게 써야 하는지에 대한 내용은 graph 문서를 참고하세요.

딥 에이전트 (Deep Agents)

딥 에이전트는 여러 아키텍처 패턴과 기능을 결합해 복잡한 다단계 작업을 안정적으로 처리하는 자율 에이전트예요. 이러한 패턴은 Pydantic AI의 내장 기능과 (서드파티) toolset으로 구현할 수 있어요.

계획 및 진행 추적 (Planning and progress tracking) — 에이전트가 복잡한 작업을 단계로 나누고 진행 상황을 추적해서, 사용자가 에이전트가 무엇을 하고 있는지 볼 수 있게 해요. Task Management toolsets를 참고하세요.
파일 시스템 연산 (File system operations) — 인메모리 저장소, 실제 파일 시스템, 샌드박스 컨테이너를 아우르는 적절한 추상화 계층으로 파일을 읽고·쓰고·편집해요. File Operations toolsets를 참고하세요.
작업 위임 (Task delegation) — 재귀적 위임 문제를 방지하기 위해 격리된 컨텍스트로 특정 작업을 위한 전문화된 서브에이전트를 생성해요. 위의 Agent Delegation을 참고하세요.
샌드박스 코드 실행 (Sandboxed code execution) — 사고를 방지하기 위해 AI가 생성한 코드를 격리된 환경(보통 Docker 컨테이너)에서 실행해요. Code Execution toolsets를 참고하세요.
컨텍스트 관리 (Context management) — 토큰 한도를 넘을 긴 세션을 처리하기 위해 대화를 자동으로 요약해요. Processing Message History를 참고하세요.
Human-in-the-loop (사람 개입) — 코드 실행이나 파일 삭제 같은 위험한 연산에 대한 승인 워크플로예요. Requiring Tool Approval를 참고하세요.
지속 실행 (Durable execution) — 일시적인 API 실패, 애플리케이션 오류, 재시작을 넘어 에이전트 상태를 보존해요. Durable Execution을 참고하세요.

또한 커뮤니티는 이런 개념을 더 의견이 담긴 방식으로 묶는 패키지를 관리하고 있어요.

Vstorm의 pydantic-deep

멀티 에이전트 시스템 관찰하기 (Observing Multi-Agent Systems)

멀티 에이전트 시스템은 복잡해서 디버깅이 어려울 수 있어요. 여러 에이전트가 상호작용하면 실행 흐름을 이해하는 것이 필수적이 돼요.

에이전트 위임 추적 (Tracing Agent Delegation)

Logfire로 여러 에이전트를 오가는 전체 흐름을 추적할 수 있어요.

import logfire

logfire.configure()
logfire.instrument_pydantic_ai()

# Your multi-agent code here...

Logfire는 다음을 보여줘요.

요청의 어느 부분을 어떤 에이전트가 처리했는지
위임 결정 — 한 에이전트가 언제, 왜 다른 에이전트를 호출했는지
에이전트별로 세분화된 엔드투엔드 지연시간
에이전트별 토큰 사용량과 비용
에이전트 실행을 무엇이 촉발했는지 — HTTP 요청, 예약 작업, 사용자 액션이 그것을 시작했는지
도구 호출 안에서 일어난 일 — 도구가 실행하는 DB 쿼리, HTTP 요청, 파일 연산, 기타 계측된 코드

이것은 복잡한 에이전트 워크플로를 이해하고 최적화하는 데 필수적이에요. 멀티 에이전트 시스템에서 뭔가 잘못되면, 정확히 어느 에이전트가 어떤 작업을 시도하다 실패했는지, 문제가 에이전트의 추론에 있었는지 호출한 백엔드 시스템에 있었는지를 볼 수 있어요.

풀스택 가시성 (Full-Stack Visibility)

Pydantic AI 애플리케이션에 TypeScript 프론트엔드, API 게이트웨이, 다른 언어의 서비스가 포함되어 있다면 Logfire가 그것들도 추적해요. Logfire는 Python, JavaScript/TypeScript, Rust용 SDK를 제공하고, OpenTelemetry로 계측된 어떤 애플리케이션과도 호환돼요. 전체 스택의 트레이스를 통합 뷰에서 볼 수 있어요. 다른 언어에서 표준 OpenTelemetry로 데이터를 보내는 방법은 alternative clients guide를 참고하세요.

Pydantic AI의 계측은 OpenTelemetry를 기반으로 하므로, OTel 호환 백엔드를 어떤 것이든 사용할 수 있어요. 자세한 내용은 Logfire integration guide를 참고하세요.

예시

Pydantic AI에서 멀티 에이전트 패턴을 어떻게 쓰는지 보여주는 예시예요.

Flight booking

Multi-agent Applications