Pydantic AI: Python으로 AI를 다루는 방법

Pydantic AI: Python으로 AI를 다루는 방법

Pydantic AI는 Python을 위한 AI SDK예요. "How Python does AI"라는 한 줄이 전부를 말하는데, 에이전트, 실시간 음성, 이미지 생성, 임베딩까지 — 모든 모델과 모든 인터페이스가 타입으로 끝에서 끝까지 연결돼 있어요. 이 페이지는 Pydantic AI가 무엇이고, 어디서 어떻게 쓰는지, 그리고 왜 선택해야 하는지를 넓게 훑어보는 개요예요.

출처: 공식문서

Pydantic AI란?

Pydantic AI는 타입 세이프하고 확장 가능한 에이전트 루프를 제공하는 Python AI SDK예요. 모델은 문자열 하나로 어떤 모델이든 바꿔 끼울 수 있고, 같은 에이전트가 웹 프론트엔드 뒤에서든 터미널에서든 음성 통화에서든 지속형 백그라운드 큐에서든, 아니면 그냥 run()을 호출하는 평범한 객체로든 어디서든 돌아요. 이미지 생성임베딩도 같은 상자 안에 들어 있어요.

복잡하고 오래 가는 작업에 필요한 것들은 Pydantic AI Harnesscapability로 딸깍 끼워주는데, 메모리, 서브에이전트, 컨텍스트 관리부터 완전한 코딩 에이전트까지 갖춰져 있어요.

무엇을 만들고 있나요?

간단한 타입이 있는 데이터 추출부터 복잡하고 오래 걸리는 멀티에이전트 협업까지, Pydantic AI와 Harness가 전부 커버해요.

데이터 추출 — 에이전트에 output type도구를 주면, 매 실행마다 검증되고 타입이 있는 결과가 나와요.

from typing import Literal

from pydantic import BaseModel, Field

from pydantic_ai import Agent, RunContext


class Sentiment(BaseModel):
    label: Literal['positive', 'negative', 'neutral']
    score: float = Field(ge=-1, le=1)


agent = Agent('openai:gpt-5.6-sol', output_type=Sentiment)


@agent.tool
def recent_reviews(ctx: RunContext[None], product: str) -> list[str]:
    """Fetch recent review snippets for a product."""
    return ['The new release fixed everything I complained about!']


result = agent.run_sync('How are people feeling about the Extract app?')
print(result.output)
#> label='positive' score=0.9

@agent.tool 함수는 RunContext를 통해 의존성을 받아요. 나머지 시그니처와 docstring이 도구 스키마가 되고, 인자는 코드 실행 전에 검증되며, 실행은 반드시 Sentiment를 반환하도록 보장돼요. 그래서 IDE·타입 체커·LLM이 반환 타입에 대해 모두 같은 결론에 도달해요.

지속형 워크플로(durable execution) — 같은 에이전트에 TemporalDurability를 붙이면 Temporal 워크플로 안에서 돌 수 있어요. 매 모델·도구 호출이 지속형 액티비티가 되어, 백그라운드 큐에서 돌다가 재시작·실패·긴 대기를 견뎌요.

실시간 음성 — 같은 에이전트를 실시간 음성 세션에 올릴 수 있어요.

import asyncio

from pydantic_ai import Agent
from pydantic_ai.capabilities import MCP

agent = Agent(
    instructions='You are a helpful voice assistant.',
    capabilities=[MCP('https://internal.example.com/mcp')],  # capabilities work in voice too
)

@agent.tool_plain
def order_status(order_id: str) -> str:
    """Look up the status of an order."""
    return f'Order {order_id}: shipped, arriving Thursday.'

async with agent.realtime('openai:gpt-realtime-2.1').session() as session:
    microphone = asyncio.create_task(session.send_audio(microphone_chunks()))  # your microphone → the model
    speaker = asyncio.create_task(play_audio(session.stream_audio()))  # model audio → your speaker
    async for part in session.stream_transcripts():
        print(f'{part.speaker}: {part.transcript}')

이미지 생성 — 에이전트 실행 없이 전용 이미지 모델로 바로 생성할 수도 있어요.

from pathlib import Path

from pydantic_ai import ImageGenerator

generator = ImageGenerator('openai:gpt-image-2')
result = generator.generate_sync('A minimalist logo for a coffee shop called Extract.')
Path('logo.png').write_bytes(result.image.data)

아직 API 키가 없나요? 그 어떤 것도 시도하는 데 프로바이더 API 키는 필요 없어요. 내장된 'test' 모델(Agent('test'))을 쓰면 LLM을 호출하지 않고 완전히 오프라인으로 돌아서, 에이전트·도구·출력을 먼저 시험해볼 수 있어요. 진짜 모델을 쓸 준비가 되면 모델 및 프로바이더에서 프로바이더와 API 키를 정하면 돼요.

왜 Pydantic AI인가

  • 어떤 모델이든 하나의 Python API. 거의 모든 모델·프로바이더(OpenAI, Anthropic, Google, Bedrock, Azure AI Foundry, Groq, Mistral, xAI, Ollama 등 수십 개)를 문자열 하나로, 또는 Pydantic AI Gateway로 바꿔 끼워요. 하나의 키로 전부 쓰고 페일오버·비용 모니터링도 내장돼 있어요. 어떤 간판 기능도 특정 벤더에 묶여 있지 않아요.
  • 끝에서 끝까지 타입. 구조화 출력, 타입 있는 의존성 주입, 타입 있는 도구 덕분에 IDE·타입 체커·코딩 에이전트가 모두 에이전트가 뭘 반환하는지 알아요. 오류의 한 종류를 런타임에서 작성 시점으로 옮겨줘요.
  • 측정이지 감이 아니다. OpenTelemetry 네이티브 계측이라 어떤 OTel 백엔드와도 동작해요. 한 줄이면 Pydantic Logfire로 실시간 디버깅·트레이싱·비용 추적을 켜요.
  • 조립 가능한 배터리. 하나의 primitive인 capability가 도구·지시·훅·모델 설정을 재사용 가능한 단위로 묶어요. MCP, 웹 검색 같은 핵심 기능부터 CoderResearcher 같은 완전한 에이전트까지 capability 조합일 뿐이에요. 코드 없이 YAML/JSON agent 스펙으로도 가능해요.
  • 모든 인터페이스. 하나의 에이전트 정의가 CLI, 내장 웹 채팅, 실시간 음성으로 동작해요.
  • 지속형 실행. Temporal, DBOS, Prefect, Restate에서 퍼스트파티·공동 유지 관리 방식의 지속형 실행을 쓸 수 있어요. 에이전트가 재시작을 견디고 며칠을 돌 수 있어요.

하나로 묶기: 은행 고객지원 에이전트

여러 기능이 함께 동작하는 예시로, 의존성 주입(데이터베이스 연결을 지시·도구에 전달), 함수 도구의 호출, 매 실행마다 검증되는 구조화 출력, 고객 컨텍스트를 묶은 재사용 capability, 그리고 대화가 필요할 때만 모델이 불러오는 on-demand capability를 살펴볼 수 있어요.

from dataclasses import dataclass

from pydantic import BaseModel, Field
from pydantic_ai import Agent, Capability, RunContext

from bank_database import DatabaseConn


@dataclass
class SupportDependencies:  # (1)
  customer_id: int
  db: DatabaseConn  # (2)


class SupportOutput(BaseModel):  # (3)
  support_advice: str = Field(description='Advice returned to the customer')
  block_card: bool = Field(description="Whether to block the customer's card")
  risk: int = Field(description='Risk level of query', ge=0, le=10)


customer_context = Capability[SupportDependencies](  # (4)
  id='customer-context',
  description="Who the customer is and what's on their account.",
)


@customer_context.instructions  # (5)
async def add_customer_name(ctx: RunContext[SupportDependencies]) -> str:
  customer_name = await ctx.deps.db.customer_name(id=ctx.deps.customer_id)
  return f"The customer's name is {customer_name!r}"


@customer_context.tool  # (6)
async def customer_balance(
  ctx: RunContext[SupportDependencies], include_pending: bool
) -> float:
  """Returns the customer's current account balance."""  # (7)
  return await ctx.deps.db.customer_balance(
      id=ctx.deps.customer_id,
      include_pending=include_pending,
  )


refunds = Capability[SupportDependencies](  # (8)
  id='refunds',
  description='Refund eligibility and refund status.',
  defer_loading=True,
)


@refunds.tool
async def refund_status(ctx: RunContext[SupportDependencies]) -> str:
  """Look up the refund status for the customer's most recent charge."""
  return await ctx.deps.db.refund_status(id=ctx.deps.customer_id)


support_agent = Agent(  # (9)
  'openai:gpt-5.6-sol',  # (10)
  deps_type=SupportDependencies,
  output_type=SupportOutput,  # (11)
  instructions=(
      'You are a support agent in our bank, give the '
      'customer support and judge the risk level of their query.'
  ),
  capabilities=[customer_context, refunds],  # (12)
)


...  # (13)


async def main():
  deps = SupportDependencies(customer_id=123, db=DatabaseConn())
  result = await support_agent.run('What is my balance?', deps=deps)  # (14)
  print(result.output)
  """
  support_advice='Hello John, your current account balance, including pending transactions, is $123.45.' block_card=False risk=1
  """

  result = await support_agent.run('I just lost my card!', deps=deps)
  print(result.output)
  """
  support_advice="I'm sorry to hear that, John. We are temporarily blocking your card to prevent unauthorized transactions." block_card=True risk=8
  """

  result = await support_agent.run(  # (15)
      'Was I refunded for the duplicate charge on my last statement?', deps=deps
  )
  print(result.output)
  """
  support_advice='Good news, John: the duplicate charge on your last statement was refunded on 2026-05-01.' block_card=False risk=1
  """

핵심 포인트를 짚어볼게요. SupportDependencies dataclass가 지시·도구 함수가 실행될 때 필요한 데이터·연결·로직을 모델에 전달해요. SupportOutput은 Pydantic이 JSON Schema를 만들어 LLM에 어떻게 반환할지 알려주고, 실행 끝에 데이터를 검증해요. customer_context는 관련 지시와 도구를 하나의 재사용 단위로 묶은 capability라서, 다른 에이전트의 capabilities 목록에 그대로 넣을 수 있어요. refundsdefer_loading=True라 온디맨드 capability로 동작해요 — 프롬프트에 한 줄 카탈로그 항목으로만 보이고, 모델이 관련 있다고 판단해 load_capability 도구로 불러올 때까지 그 도구는 숨겨져 있어요. (이 코드는 brevity를 위해 일부 생략됐고 완전한 bank_support.py여기에서 볼 수 있어요.)

Pydantic Logfire 계측

Pydantic AI는 OpenTelemetry 네이티브라 Instrumentation capability가 매 모델·도구 호출에 표준 OTel 스팬을 만들어요. 가장 쉬운 설정은 logfire SDK인데, 이건 순수 OpenTelemetry라 Pydantic Logfire나 다른 백엔드를 모두 가리킬 수 있어요.

...
from pydantic_ai import Agent, RunContext

from bank_database import DatabaseConn

import logfire

logfire.configure()  # (1)
logfire.instrument_pydantic_ai()  # (2)
logfire.instrument_sqlite3()  # (3)

...

llms.txt

Pydantic AI 문서는 llms.txt 형식으로도 제공돼요. 두 가지 형식이 있어요: llms.txt(프로젝트 요약 + 문서 섹션 링크)와 llms-full.txt(모든 링크 내용 포함, 일부 LLM에겐 너무 클 수 있음). 현재는 IDE·코딩 에이전트가 자동으로 쓰진 않지만, 링크나 전체 텍스트를 직접 주면 활용해요.

다음 단계

더 알아보기 (Learn more)