PydanticAI 완전가이드 2026 — FastAPI 철학의 에이전트 프레임워크

FastAPI가 웹 개발에 가져온 "타입 안전 + DI + 자동 문서화"의 느낌을 LLM 에이전트에 그대로 가져오려는 시도입니다. PydanticAI는 그것이 무엇인지 보여줍니다.

---

핵심 요약 → PydanticAI = Pydantic 팀이 만든 에이전트 프레임워크, v1.0 2025년 9월 → v1.70+ 2026년 3월 → 핵심 철학: "FastAPI feeling to GenAI" — 타입 안전, DI, IDE 자동완성 → 4대 핵심: 타입 안전 출력 + 의존성 주입(RunContext) + Pydantic Evals + Logfire → 75+ 프로바이더 지원: OpenAI·Anthropic·Gemini·Ollama·DeepSeek·Grok 등 → MCP 네이티브 지원 — MCPServerStdio로 외부 툴 직접 연결 → LangGraph와의 차이: 단순 에이전트에 더 간결, 복잡 그래프는 LangGraph → Instructor와의 차이: Instructor는 구조화 추출 특화, PydanticAI는 풀 에이전트 프레임워크

---

왜 PydanticAI인가

Pydantic 팀은 Pydantic 검증이 OpenAI SDK, Google ADK, Anthropic SDK, LangChain, LlamaIndex, CrewAI 등 사실상 모든 Python AI 프레임워크의 검증 레이어로 쓰이고 있음을 발견했습니다. 그러나 정작 에이전트를 직접 만들 때 FastAPI 같은 느낌을 주는 프레임워크가 없었고, PydanticAI를 만든 이유가 바로 그것입니다.

# PydanticAI vs 기존 접근법

순수 Anthropic SDK:
  client.messages.create() → raw text
  → JSON 파싱, 타입 변환, 재시도 직접 구현
  → 툴 호출 루프 직접 구현
  → DI 없음 → 전역 변수 또는 클로저 남발

LangChain:
  유연하지만 추상화 레이어가 많아 디버깅 어려움
  타입 정보 약함 → IDE 자동완성 미흡

PydanticAI:
  @agent.tool 데코레이터 → 타입 안전 툴 등록
  deps_type + RunContext → DI 패턴 (FastAPI의 Depends와 동일 개념)
  output_type=MyModel → Pydantic 검증 자동
  IDE가 에러를 런타임이 아닌 작성 시점에 잡아냄

---

1. 설치 + 기본 에이전트

# 설치
# pip install pydantic-ai
# pip install pydantic-ai[logfire]  # Logfire 포함

from pydantic import BaseModel
from pydantic_ai import Agent

# ── 가장 단순한 에이전트 ──

# 프로바이더/모델 문자열 형식: "provider:model-name"
agent = Agent(
    "anthropic:claude-sonnet-4-6",      # 또는 "openai:gpt-5.5"
    system_prompt="당신은 도움이 되는 어시스턴트입니다.",
)

result = agent.run_sync("파이썬에서 리스트 컴프리헨션 예시 알려줘")
print(result.output)  # → str


# ── 구조화 출력 ──

class CodeReview(BaseModel):
    has_bugs: bool
    severity: str           # "low" | "medium" | "high"
    issues: list[str]
    suggestions: list[str]
    score: int              # 1~10

review_agent = Agent(
    "anthropic:claude-sonnet-4-6",
    output_type=CodeReview,   # ← Pydantic 모델 지정
    system_prompt="당신은 시니어 코드 리뷰어입니다.",
)

result = review_agent.run_sync(
    "def divide(a, b): return a / b"
)

review = result.output   # → CodeReview 타입 보장
print(review.has_bugs)   # True
print(review.issues)     # ["0으로 나누기 예외 처리 없음"]
print(review.score)      # 4  (int 타입 보장)
# → JSON 파싱·타입 변환 코드 없음

---

2. 의존성 주입 — PydanticAI의 핵심

PydanticAI의 의존성 주입은 deps_type과 RunContext를 통해 에이전트 정의와 런타임 의존성(DB 연결, API 클라이언트, 사용자 세션)을 분리합니다. FastAPI의 Depends()와 동일한 패턴으로, 툴이 자동으로 올바른 컨텍스트를 받아 타이트 커플링 없이 테스트가 쉬워집니다.

from dataclasses import dataclass
from pydantic import BaseModel
from pydantic_ai import Agent, RunContext
import httpx

# ── 의존성 타입 정의 ──

@dataclass
class AppDependencies:
    """런타임 의존성 — DB 연결, API 클라이언트 등"""
    db_pool:    object         # 실제로는 asyncpg.Pool 등
    http_client: httpx.AsyncClient
    user_id:    int
    api_key:    str

# ── 구조화 출력 모델 ──

class CustomerReport(BaseModel):
    customer_name: str
    account_balance: float
    risk_level: str            # "low" | "medium" | "high"
    recommendation: str

# ── 에이전트 정의 ──

support_agent = Agent(
    "anthropic:claude-sonnet-4-6",
    deps_type=AppDependencies,      # ← 의존성 타입 선언
    output_type=CustomerReport,     # ← 출력 타입 선언
    system_prompt="당신은 은행 고객 지원 에이전트입니다.",
)

# ── 동적 시스템 프롬프트 (deps 접근 가능) ──

@support_agent.instructions
async def add_customer_context(ctx: RunContext[AppDependencies]) -> str:
    """deps를 활용해 런타임 컨텍스트 동적 주입"""
    customer = await ctx.deps.db_pool.fetchrow(
        "SELECT name, tier FROM customers WHERE id = $1",
        ctx.deps.user_id
    )
    return f"고객 이름: {customer['name']}, 등급: {customer['tier']}"


# ── 툴 등록 (deps 자동 수신) ──

@support_agent.tool
async def get_account_balance(
    ctx: RunContext[AppDependencies],  # deps 자동 주입
    include_pending: bool = False      # LLM이 채우는 파라미터
) -> float:
    """고객의 현재 계좌 잔액을 반환합니다.

    Args:
        include_pending: 보류 중인 거래 포함 여부
    """
    # docstring → LLM에게 툴 설명으로 전달
    # include_pending → LLM이 결정해서 채움
    query = """
        SELECT balance + CASE WHEN $2 THEN pending_amount ELSE 0 END
        FROM accounts WHERE user_id = $1
    """
    balance = await ctx.deps.db_pool.fetchval(
        query, ctx.deps.user_id, include_pending
    )
    return float(balance)


@support_agent.tool
async def check_recent_transactions(
    ctx: RunContext[AppDependencies],
    days: int = 7
) -> list[dict]:
    """최근 거래 내역을 조회합니다.

    Args:
        days: 조회 기간 (일), 기본값 7일
    """
    rows = await ctx.deps.db_pool.fetch(
        "SELECT * FROM transactions WHERE user_id = $1 AND date > NOW() - $2 * INTERVAL '1 day'",
        ctx.deps.user_id, days
    )
    return [dict(row) for row in rows]


# ── 실행 ──

async def handle_support_request(user_id: int, message: str):
    async with httpx.AsyncClient() as client:
        deps = AppDependencies(
            db_pool=db_pool,        # 앱 레벨에서 생성된 풀
            http_client=client,
            user_id=user_id,
            api_key=API_KEY
        )

        result = await support_agent.run(
            message,
            deps=deps               # ← 실행 시 의존성 주입
        )

        return result.output        # CustomerReport 타입 보장

---

3. 툴 vs 툴_플레인 — deps 수신 여부

from pydantic_ai import Agent, RunContext
import random

agent = Agent("anthropic:claude-sonnet-4-6", deps_type=str)

# @agent.tool — deps(RunContext) 접근 가능
@agent.tool
async def greet_user(ctx: RunContext[str]) -> str:
    """사용자 이름으로 인사합니다."""
    return f"안녕하세요, {ctx.deps}님!"   # deps = 사용자 이름

# @agent.tool_plain — deps 불필요, 순수 함수
@agent.tool_plain
def roll_dice() -> int:
    """1~6 사이의 주사위를 굴립니다."""
    return random.randint(1, 6)

# @agent.tool_plain — 파라미터도 없는 단순 툴
@agent.tool_plain
def get_current_time() -> str:
    """현재 시각을 반환합니다."""
    from datetime import datetime
    return datetime.now().strftime("%H:%M:%S")


# 실행
result = agent.run_sync(
    "주사위 굴려서 내 이름이랑 같이 알려줘",
    deps="홍길동"      # ← deps 주입
)
print(result.output)
# → "안녕하세요 홍길동님! 주사위를 굴렸더니 5가 나왔습니다."

---

4. 멀티 에이전트 — 에이전트가 에이전트를 호출

from dataclasses import dataclass
from pydantic import BaseModel
from pydantic_ai import Agent, RunContext

@dataclass
class SharedDeps:
    http_client: object
    api_key: str


# ── 전문화된 서브 에이전트들 ──

flight_agent = Agent(
    "anthropic:claude-sonnet-4-6",
    deps_type=SharedDeps,
    output_type=str,
    system_prompt="항공편 검색 전문 에이전트입니다.",
)

seat_agent = Agent(
    "openai:gpt-5.5",          # 다른 모델 사용 가능
    deps_type=SharedDeps,
    output_type=str,
    system_prompt="좌석 선호도 분석 전문 에이전트입니다.",
)

# ── 오케스트레이터 에이전트 ──

orchestrator = Agent(
    "anthropic:claude-opus-4-7",    # 강력한 모델로 오케스트레이션
    deps_type=SharedDeps,
    system_prompt="여행 예약을 돕는 에이전트입니다.",
)

@orchestrator.tool
async def search_flights(
    ctx: RunContext[SharedDeps],
    origin: str,
    destination: str,
    date: str
) -> str:
    """항공편을 검색합니다."""
    # flight_agent를 서브 에이전트로 호출
    result = await flight_agent.run(
        f"{origin} → {destination}, {date} 항공편 검색",
        deps=ctx.deps    # 동일 deps 전달
    )
    return result.output


@orchestrator.tool
async def analyze_seat_preference(
    ctx: RunContext[SharedDeps],
    passenger_history: str
) -> str:
    """탑승객의 좌석 선호도를 분석합니다."""
    result = await seat_agent.run(
        f"탑승 이력 분석: {passenger_history}",
        deps=ctx.deps
    )
    return result.output


# 실행
async def book_trip():
    deps = SharedDeps(http_client=..., api_key="...")
    result = await orchestrator.run(
        "서울 → 도쿄, 다음 주 토요일, 창문 자리 선호",
        deps=deps
    )
    print(result.output)

---

5. MCP 네이티브 통합

from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStdio, MCPServerHTTP

# ── stdio MCP 서버 연결 ──

agent_with_mcp = Agent(
    "anthropic:claude-sonnet-4-6",
    toolsets=[
        # 로컬 MCP 서버
        MCPServerStdio("python", args=["my_mcp_server.py"]),

        # 원격 HTTP MCP 서버
        MCPServerHTTP("https://my-mcp-server.example.com/mcp"),
    ]
)

# async context manager로 MCP 서버 수명 관리
async def run_with_mcp():
    async with agent_with_mcp:    # MCP 서버 시작
        result = await agent_with_mcp.run(
            "내부 DB에서 이번 달 이상 거래 조회해줘"
        )
        print(result.output)
    # context 종료 시 MCP 서버 자동 정리


# ── deps와 MCP 함께 사용 ──

from dataclasses import dataclass
from pydantic_ai import RunContext
from pydantic_ai.mcp import CallToolFunc, MCPServerStdio, ToolResult
from typing import Any

@dataclass
class Deps:
    user_id: int

async def inject_user_context(
    ctx: RunContext[Deps],
    call_tool: CallToolFunc,
    name: str,
    tool_args: dict[str, Any],
) -> ToolResult:
    """MCP 툴 호출 시 user_id를 메타데이터로 주입"""
    return await call_tool(name, tool_args, {"user_id": ctx.deps.user_id})

server = MCPServerStdio(
    "python",
    args=["mcp_server.py"],
    process_tool_call=inject_user_context   # 툴 호출 전처리
)

---

6. 스트리밍 + 구조화 출력

from pydantic import BaseModel
from pydantic_ai import Agent

class AnalysisReport(BaseModel):
    title: str
    summary: str
    key_points: list[str]
    conclusion: str

agent = Agent(
    "anthropic:claude-sonnet-4-6",
    output_type=AnalysisReport
)

# ── 구조화 출력 스트리밍 ──

async def stream_report():
    async with agent.run_stream("AI 에이전트 시장 분석해줘") as stream:
        # 필드가 채워지는 대로 즉시 접근 가능
        async for partial in stream.stream_structured():
            if partial.title:
                print(f"\r제목: {partial.title}", end="")
            if partial.summary:
                print(f"\n요약: {partial.summary[:80]}...", end="")

        # 최종 완성 객체
        final: AnalysisReport = await stream.get_output()
        print(f"\n결론: {final.conclusion}")
        print(f"핵심 포인트: {len(final.key_points)}개")

---

7. 테스트 — TestModel + 실제 API 없이 검증

# PydanticAI의 핵심 장점: LLM 없이 에이전트 로직 테스트 가능

import pytest
from pydantic_ai import Agent, RunContext
from pydantic_ai.models.test import TestModel
from pydantic import BaseModel
from dataclasses import dataclass

@dataclass
class TestDeps:
    user_name: str
    balance: float

class AccountInfo(BaseModel):
    greeting: str
    balance_formatted: str
    is_vip: bool

agent = Agent(
    "anthropic:claude-sonnet-4-6",
    deps_type=TestDeps,
    output_type=AccountInfo,
)

@agent.tool
def get_balance(ctx: RunContext[TestDeps]) -> float:
    return ctx.deps.balance


# ── 테스트 ──

@pytest.mark.asyncio
async def test_agent_without_api():
    """실제 LLM 호출 없이 에이전트 로직 검증"""
    with agent.override(model=TestModel()):
        # TestModel: 실제 LLM 대신 사용
        # 툴은 실제로 실행됨 — 로직 검증 가능
        result = await agent.run(
            "내 계좌 정보 보여줘",
            deps=TestDeps(user_name="홍길동", balance=5_000_000.0)
        )

        # 툴이 호출됐는지 확인
        tool_calls = [m for m in result.all_messages()
                      if hasattr(m, "parts")]
        assert len(tool_calls) > 0

        # TestModel은 output_type의 기본값 반환
        assert isinstance(result.output, AccountInfo)


@pytest.mark.asyncio
async def test_balance_tool():
    """툴 로직만 단독 테스트"""
    @dataclass
    class MockCtx:
        deps: TestDeps

    ctx = MockCtx(deps=TestDeps(user_name="test", balance=1234.56))
    balance = get_balance(ctx)
    assert balance == 1234.56

---

8. Logfire 통합 — 1줄 옵저버빌리티

import logfire
from pydantic_ai import Agent

# ── 1줄로 전체 에이전트 추적 활성화 ──

logfire.configure()
logfire.instrument_pydantic_ai()   # 이게 전부

# 이후 모든 agent.run() 호출에서 자동으로:
# - 모델 이름, 토큰 사용량
# - 각 툴 호출 입력/출력
# - 검증 결과
# - 전체 레이턴시
# → Logfire 대시보드 또는 OTel 호환 백엔드로 전송

agent = Agent("anthropic:claude-sonnet-4-6")
result = agent.run_sync("코드 리뷰해줘")

# Logfire 외 OTel 백엔드 사용 시 (Jaeger, Grafana 등)
# LOGFIRE_SEND_TO_LOGFIRE=false
# OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317
# → 위 환경 변수만 설정하면 기존 OTel 파이프라인으로 전송

---

9. 모델 전환 — 프로바이더 교체가 한 줄

from pydantic_ai import Agent

# 동일 코드로 모델만 교체
agents = {
    "production":  Agent("anthropic:claude-sonnet-4-6"),
    "premium":     Agent("anthropic:claude-opus-4-7"),
    "fast":        Agent("google:gemini-3.5-flash"),
    "cheap":       Agent("deepseek:deepseek-v4-flash"),
    "local":       Agent("ollama:qwen3-coder:30b-a3b"),
}

# 환경 변수 기반 모델 선택
import os
model = os.getenv("LLM_MODEL", "anthropic:claude-sonnet-4-6")
agent = Agent(model)

# 런타임 모델 교체 (테스트·A-B 비교)
from pydantic_ai.models.test import TestModel

with agent.override(model=TestModel()):
    result = agent.run_sync("테스트 쿼리")   # API 비용 $0

---

10. PydanticAI vs 다른 프레임워크

# 선택 기준

PydanticAI 선택:
  ✅ FastAPI 스타일이 익숙한 Python 팀
  ✅ 타입 안전이 최우선 (IDE 자동완성, 정적 분석)
  ✅ 단순~중간 복잡도 에이전트 (툴 + DI + 구조화 출력)
  ✅ Logfire 통합 옵저버빌리티 원할 때
  ✅ 테스트 가능성이 중요한 프로덕션 코드

LangGraph 선택:
  ✅ 복잡한 그래프 기반 워크플로 (조건부 라우팅, 상태 머신)
  ✅ 시각적 그래프 디버깅 필요
  ✅ LangSmith 생태계 이미 사용 중

CrewAI 선택:
  ✅ 역할 기반 멀티 에이전트 팀 시뮬레이션
  ✅ 비개발자도 에이전트 구성 가능한 UX 필요

Instructor 선택:
  ✅ 에이전트 필요 없고 구조화 추출만 필요
  ✅ 기존 코드에 최소 변경으로 Pydantic 검증 추가

Google ADK 선택:
  ✅ GCP 생태계에 올인
  ✅ Gemini 모델 네이티브 통합 최우선

---

결론

✅ PydanticAI를 선택해야 하는 이유

• deps_type + RunContext = FastAPI의 DI를 에이전트에 그대로 — 코드가 테스트 가능해짐
• output_type=MyModel = LLM 출력이 항상 타입 보장 — 파싱 에러 없음
• logfire.instrument_pydantic_ai() 한 줄 = 전체 에이전트 추적
• TestModel = LLM API 없이 에이전트 로직 테스트 → CI/CD 통합 가능

✅ 2026년 프로덕션 에이전트의 기준

• "타입 에러를 런타임이 아닌 작성 시점에 잡는다"
• PydanticAI는 그 기준을 LLM 에이전트에도 적용한 첫 번째 프레임워크

❌ 주의사항

• 복잡한 그래프 워크플로는 PydanticAI Graph 사용 or LangGraph 병행 검토
• v1.70+ 기준 — 아직 빠른 업데이트 중, API 변경 가능성 있음
• deps_type이 없는 간단한 태스크엔 Instructor가 더 가벼운 선택