PydanticAI 실전 가이드 — FastAPI 방식으로 만드는 타입 안전한 AI 에이전트

설치
# OpenAI 또는 Anthropic 프로바이더 선택
pip install pydantic-ai

# 사용할 모델에 따라 SDK 별도 설치
pip install openai          # GPT-5.4 등
pip install anthropic        # Claude Sonnet 4.6 등
pip install google-generativeai  # Gemini

첫 번째 에이전트 (Claude Sonnet 4.6 사용)
from pydantic_ai import Agent

agent = Agent(
    model="claude-sonnet-4-6",
    system_prompt="당신은 Python 전문가입니다. 간결하고 정확하게 답하세요."
)

result = agent.run_sync("PydanticAI와 LangChain의 차이를 세 줄로 요약해줘")
print(result.data)
print(f"토큰 비용: {result.usage()}")

타입 안전 Tool 등록 예시
from pydantic_ai import Agent, RunContext
from pydantic import BaseModel
from typing import Optional
import httpx

class WeatherResult(BaseModel):
    city: str
    temp_celsius: float
    condition: str
    humidity: Optional[int] = None

agent = Agent(
    model="claude-sonnet-4-6",
    system_prompt="날씨 정보를 조회해 사용자에게 알려주세요.",
    result_type=str  # 최종 응답은 문자열
)

@agent.tool
async def get_weather(ctx: RunContext, city: str) -> WeatherResult:
    """도시명을 받아 현재 날씨를 반환합니다."""
    # 실제 API 호출 (예시)
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            f"https://api.weatherapi.com/v1/current.json",
            params={"key": ctx.deps.weather_api_key, "q": city}
        )
    data = resp.json()
    return WeatherResult(
        city=data["location"]["name"],
        temp_celsius=data["current"]["temp_c"],
        condition=data["current"]["condition"]["text"],
        humidity=data["current"]["humidity"]
    )

# LLM이 city에 int를 넘기려 하면 Pydantic이 즉시 오류를 잡음

구조화 출력 — Pydantic 모델을 result_type으로 지정
from pydantic import BaseModel, Field
from pydantic_ai import Agent
from typing import List

class TechAnalysis(BaseModel):
    summary: str = Field(description="한 줄 요약")
    pros: List[str] = Field(description="장점 목록 (3~5개)")
    cons: List[str] = Field(description="단점 목록 (2~4개)")
    recommendation: str = Field(description="누구에게 추천하는지")
    confidence: float = Field(ge=0.0, le=1.0, description="분석 신뢰도 0~1")

agent = Agent(
    model="claude-sonnet-4-6",
    result_type=TechAnalysis,  # 응답을 TechAnalysis 스키마로 강제
    system_prompt="기술 도구를 분석해 구조화된 평가를 제공합니다."
)

result = agent.run_sync("PydanticAI 프레임워크를 분석해줘")
analysis: TechAnalysis = result.data  # IDE에서 타입 자동완성 동작
print(f"요약: {analysis.summary}")
print(f"신뢰도: {analysis.confidence:.0%}")

Logfire 통합 설정
pip install logfire

# main.py 또는 에이전트 초기화 코드 최상단에 추가
import logfire
logfire.configure()  # LOGFIRE_TOKEN 환경변수에서 자동 읽음
logfire.instrument_pydantic_ai()  # PydanticAI 에이전트 전체 자동 계측

# 이후 agent.run_sync() 호출 시 자동으로 추적 시작
# Logfire 대시보드: https://logfire.pydantic.dev

패턴 1 — FastAPI 엔드포인트에 에이전트 연결
from fastapi import FastAPI, Depends
from pydantic import BaseModel
from pydantic_ai import Agent
from dataclasses import dataclass

app = FastAPI()

@dataclass
class AppDeps:
    db_session: str  # 실제 DB 세션 타입으로 교체
    user_id: str

agent = Agent(
    model="claude-sonnet-4-6",
    deps_type=AppDeps,
    result_type=str
)

class QueryRequest(BaseModel):
    question: str
    user_id: str

@app.post("/ask")
async def ask(request: QueryRequest):
    deps = AppDeps(db_session="db_conn", user_id=request.user_id)
    result = await agent.run(request.question, deps=deps)
    return {"answer": result.data, "tokens": result.usage().total_tokens}

패턴 2 — 스트리밍 응답 (SSE)
from fastapi.responses import StreamingResponse
import asyncio

@app.post("/ask/stream")
async def ask_stream(request: QueryRequest):
    async def generate():
        async with agent.run_stream(request.question) as stream:
            async for chunk in stream.stream_text():
                yield f"data: {chunk}\n\n"
        yield "data: [DONE]\n\n"
    return StreamingResponse(generate(), media_type="text/event-stream")

패턴 3 — 재시도 + 에러 핸들링
from pydantic_ai import Agent
from pydantic_ai.exceptions import UnexpectedModelBehavior, ModelRetry

agent = Agent(
    model="claude-sonnet-4-6",
    retries=3  # 도구 실패 시 최대 3회 재시도
)

@agent.tool(retries=2)  # 이 Tool만 2회 재시도
async def risky_api_call(ctx, query: str) -> str:
    try:
        # 외부 API 호출
        result = await call_external_api(query)
        return result
    except TimeoutError:
        raise ModelRetry("API 타임아웃. 재시도합니다.")  # 에이전트가 자동 재시도

try:
    result = await agent.run("질문")
except UnexpectedModelBehavior as e:
    # 재시도 소진 후에도 실패 시
    logger.error(f"에이전트 실패: {e}")