PydanticAI 완전 가이드 — LLM 출력을 문자열 말고 타입으로 받는 법

response.text 파싱하다가 KeyError 터진 적 있으시죠. LLM이 JSON 대신 markdown 코드블록으로 감싸서 줬을 때요. PydanticAI는 LLM 출력을 Pydantic 모델로 강제합니다. 파싱 없이 .output.field로 바로 접근합니다.

[핵심 요약]
→ PydanticAI: FastAPI 팀(Pydantic 팀)이 만든 Python 에이전트 프레임워크
→ 핵심: output_type으로 LLM 출력을 Pydantic 모델로 강제 검증
→ 지원 모델: Claude, GPT, Gemini, Ollama, Groq, Bedrock 등 단일 인터페이스
→ 타입 안전: 에이전트 입출력 전 구간 mypy/pyright 타입 체크
→ 의존성 주입: DB 연결, API 클라이언트를 전역 변수 없이 타입 안전하게 주입
→ Logfire 통합: 에이전트 실행 전 구간 자동 트레이싱
→ v1.x 안정 API: 2025년 말 GA, 2026년 4월 기준 주간 릴리즈 유지

---

LLM 출력 파싱의 현실

기존 방식:
→ response = client.messages.create(...)
→ text = response.content[0].text
→ json.loads(text)  # JSON 파싱 시도
→ KeyError: 'result'  # 모델이 다른 키 이름으로 줬음
→ json.JSONDecodeError  # ```json 코드블록으로 감싸서 줬음
→ 타입 힌트 없음 → IDE 자동완성 없음 → 런타임에서 오류 발견

PydanticAI:
→ output_type=MyModel 지정
→ result = agent.run_sync("쿼리")
→ result.output.field  # 타입 안전, 자동완성, 검증 완료
→ 검증 실패 시 모델에게 자동 retry 요청
→ mypy/pyright 전 구간 타입 체크

---

실전 1 — 설치 및 기본 구조화 출력

# 전체 프로바이더 설치
pip install "pydantic-ai[all]"

# 또는 필요한 것만
pip install pydantic-ai
pip install anthropic  # Claude 사용 시

# 1. 가장 단순한 구조화 출력
from pydantic import BaseModel, Field
from pydantic_ai import Agent

# 출력 스키마 정의
class MovieReview(BaseModel):
    title: str
    rating: float = Field(ge=0, le=10, description="0~10점 평점")
    summary: str = Field(description="한 줄 요약")
    recommended: bool

# 에이전트 생성 — output_type으로 스키마 강제
agent = Agent(
    'anthropic:claude-sonnet-4-6',
    output_type=MovieReview,
    instructions="영화 리뷰를 분석해서 구조화된 정보를 반환해.",
)

result = agent.run_sync("인터스텔라는 정말 훌륭한 SF 영화야. 시각 효과가 압도적이고 스토리도 탄탄해.")

# ✅ 파싱 없이 바로 타입 안전하게 접근
print(result.output.title)        # 인터스텔라
print(result.output.rating)       # 9.2 (float, 검증됨)
print(result.output.recommended)  # True (bool, 검증됨)

[output_type 동작 방식]
→ Pydantic 모델 → JSON Schema 자동 변환 → LLM 프롬프트에 주입
→ LLM이 스키마 벗어난 값 반환 시 → 자동 retry (기본 1회)
→ ge=0, le=10 같은 validator도 LLM 프롬프트 설명으로 포함
→ Field(description=...) → 모델이 보는 필드 설명으로 사용
→ result.output은 Agent 제네릭 타입으로 IDE 자동완성 지원

---

실전 2 — Tool 정의 + 의존성 주입

툴은 LLM이 실행 중 호출하는 함수입니다. 의존성 주입으로 DB 연결이나 API 클라이언트를 전역 변수 없이 안전하게 넘깁니다.

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

# 의존성 타입 정의
@dataclass
class SearchDeps:
    api_key: str
    max_results: int = 5

# 출력 스키마
class SearchResult(BaseModel):
    query: str
    results: list[str] = Field(description="검색 결과 목록")
    total_count: int
    confidence: float = Field(ge=0, le=1)

# 에이전트 — 의존성 타입 지정
agent = Agent(
    'anthropic:claude-sonnet-4-6',
    deps_type=SearchDeps,
    output_type=SearchResult,
    instructions="주어진 쿼리로 검색하고 결과를 구조화해서 반환해.",
)

# 툴 정의 — ctx로 의존성 접근
@agent.tool
async def search_web(ctx: RunContext[SearchDeps], query: str) -> list[str]:
    """웹에서 쿼리를 검색합니다."""
    # ctx.deps로 주입된 의존성 접근
    api_key = ctx.deps.api_key
    max_results = ctx.deps.max_results

    # 실제 검색 API 호출
    results = await call_search_api(query, api_key, max_results)
    return results

@agent.tool
async def filter_results(
    ctx: RunContext[SearchDeps],
    results: list[str],
    keyword: str
) -> list[str]:
    """결과를 키워드로 필터링합니다."""
    return [r for r in results if keyword.lower() in r.lower()]

# 실행 — 의존성 주입
async def main():
    deps = SearchDeps(api_key="your-api-key", max_results=10)

    result = await agent.run(
        "PydanticAI 프레임워크 최신 뉴스 검색해줘",
        deps=deps,
    )

    print(result.output.query)        # 검색 쿼리
    print(result.output.results)      # 검색 결과 목록
    print(result.output.confidence)   # 신뢰도 (0~1, 검증됨)

asyncio.run(main())

[의존성 주입 핵심]
→ deps_type: 에이전트 수준에서 의존성 타입 선언
→ RunContext[DepsType]: 툴 함수에서 타입 안전하게 deps 접근
→ 전역 변수 없이 DB 연결, API 클라이언트 공유
→ 테스트 시 mock 의존성 주입 가능
→ 여러 툴이 동일한 deps 인스턴스 공유

---

실전 3 — 멀티 모델 + 재시도 + 스트리밍

from pydantic import BaseModel, field_validator
from pydantic_ai import Agent, ModelRetry
from typing import Literal

class CodeReview(BaseModel):
    summary: str
    issues: list[str]
    severity: Literal["low", "medium", "high", "critical"]
    score: int = Field(ge=0, le=10)
    suggested_fix: str | None = None

    # 커스텀 validator — LLM 출력 후 검증
    @field_validator('issues')
    @classmethod
    def issues_not_empty(cls, v):
        if len(v) == 0:
            raise ValueError("이슈 목록은 비어있을 수 없음")
        return v

reviewer = Agent(
    'anthropic:claude-sonnet-4-6',
    output_type=CodeReview,
    instructions="코드를 리뷰하고 이슈를 구체적으로 나열해.",
    # 검증 실패 시 최대 3회 재시도
    output_retries=3,
)

# 출력 validator — 재시도 트리거
@reviewer.output_validator
async def validate_review(ctx, review: CodeReview) -> CodeReview:
    if review.severity == "critical" and review.score > 3:
        # critical인데 점수가 너무 높으면 재시도 요청
        raise ModelRetry("critical 심각도면 점수가 3 이하여야 합니다. 다시 검토해주세요.")
    return review

# 스트리밍 출력
async def stream_review(code: str):
    async with reviewer.run_stream(f"이 코드 리뷰해줘:\n{code}") as result:
        # 구조화 출력도 스트리밍 가능
        async for chunk in result.stream_text():
            print(chunk, end="", flush=True)

        # 스트리밍 완료 후 최종 구조화 출력 접근
        final = await result.get_output()
        print(f"\n최종 점수: {final.score}")
        print(f"심각도: {final.severity}")

# 모델 전환 — 문자열 하나만 바꾸면 됨
agents = {
    "claude": Agent('anthropic:claude-sonnet-4-6', output_type=CodeReview),
    "gpt":    Agent('openai:gpt-4.1', output_type=CodeReview),
    "gemini": Agent('google-gla:gemini-3-flash', output_type=CodeReview),
    "local":  Agent('ollama:llama3.3', output_type=CodeReview),
}

[재시도 + 검증 흐름]
→ LLM 출력 → Pydantic 검증 → 실패 시 에러 메시지 LLM에 전달 → 재시도
→ field_validator: 데이터 수준 검증 (값 범위, 형식)
→ output_validator: 비즈니스 로직 수준 검증 (조합 규칙)
→ ModelRetry: 재시도 이유를 LLM에 명시적으로 전달
→ output_retries: 전체 재시도 횟수 제한

---

실전 4 — FastAPI 통합 + Logfire 트레이싱

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from pydantic_ai import Agent
import logfire

# Logfire 트레이싱 설정
logfire.configure(token="your-logfire-token")
logfire.instrument_pydantic_ai()  # 모든 에이전트 자동 트레이싱

app = FastAPI()

class AnalysisRequest(BaseModel):
    text: str
    language: str = "ko"

class SentimentResult(BaseModel):
    sentiment: str  # positive/negative/neutral
    score: float = Field(ge=-1, le=1)
    key_phrases: list[str]
    language_detected: str

analyzer = Agent(
    'anthropic:claude-sonnet-4-6',
    output_type=SentimentResult,
    instructions="텍스트의 감정을 분석해서 구조화된 결과를 반환해.",
)

@app.post("/analyze")
async def analyze_sentiment(request: AnalysisRequest):
    try:
        result = await analyzer.run(
            f"언어: {request.language}\n텍스트: {request.text}"
        )
        return result.output  # Pydantic 모델 → 자동 JSON 직렬화

    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

# 사용량 추적
@app.get("/usage")
async def get_usage():
    result = await analyzer.run("테스트")
    return {
        "input_tokens": result.usage().input_tokens,
        "output_tokens": result.usage().output_tokens,
        "total_requests": result.usage().requests,
    }

[Logfire 트레이싱으로 보이는 것들]
→ 에이전트 실행 전 구간 스팬 (시작~종료)
→ 툴 호출별 입력/출력/소요시간
→ 모델 요청/응답 토큰 수
→ 검증 실패 및 재시도 이벤트
→ 의존성 주입 흐름
→ Logfire 대시보드에서 실시간 조회

---

마무리

✅ PydanticAI 써야 하는 경우
→ LLM 출력을 파싱하다가 런타임 오류 자주 터지는 팀
→ FastAPI + Pydantic 이미 쓰는 Python 팀 (패턴 동일)
→ 타입 안전성 + mypy/pyright 체크가 필요한 프로덕션 환경
→ 여러 LLM 프로바이더를 전환하며 쓰는 경우
→ 비즈니스 규칙 기반 출력 검증이 필요한 경우

❌ PydanticAI가 안 맞는 경우
→ TypeScript/JavaScript 팀 (Python 전용)
→ 단순 챗봇 — 구조화 출력 필요 없는 경우
→ 복잡한 멀티에이전트 그래프 (LangGraph가 더 적합)
→ Bedrock에서 Claude 쓰면서 구조화 출력 스트리밍 필요한 경우
   (Bedrock+Claude 조합은 구조화 출력 스트리밍 미지원)

---

관련 글

https://cell-devlog.tistory.com/22

LLM 출력 파싱 실패를 없애는 법 — Pydantic으로 JSON 검증 완전 정리