Pydantic AI: Python AI 에이전트에서 타입 안전한 LLM 도구 호출 구현

# 타입 검증 없는 기존 방식 — 런타임에서야 오류 발각
def process_tool_call(args: dict):
    amount = args["amount"]   # LLM이 "22.5"(str)를 넘겼지만 통과
    return amount * 1.1        # TypeError: can't multiply str by float
 
# Pydantic AI 방식 — 인자 타입이 보장되고, 오류 시 LLM에게 재시도 요청
@agent.tool
def process_amount(ctx: RunContext[str], amount: float) -> float:
    return amount * 1.1        # Pydantic이 검증 완료, float 보장

import os
from pydantic_ai import Agent, RunContext
from pydantic import BaseModel
 
MODEL = os.getenv("LLM_MODEL", "anthropic:claude-sonnet-4-20250514")
 
class WeatherResult(BaseModel):
    city: str
    temperature: float
    condition: str
 
agent = Agent(
    MODEL,
    deps_type=str,              # 주입할 의존성 타입 (여기선 API 키)
    output_type=WeatherResult,  # LLM 출력의 검증 타입
    retries=2,                  # 검증 실패 시 최대 재시도 횟수
)
 
@agent.tool_plain
def get_time() -> str:
    """현재 UTC 시각을 반환합니다."""
    from datetime import datetime, timezone
    return datetime.now(timezone.utc).isoformat()
 
@agent.tool
def get_weather(ctx: RunContext[str], city: str) -> dict:
    """특정 도시의 날씨를 조회합니다.
 
    city: 조회할 도시명 (한글 가능)
    """
    api_key = ctx.deps  # 타입 안전하게 주입된 의존성
    return {"city": city, "temperature": 22.5, "condition": "맑음"}
 
result = agent.run_sync("서울 날씨 알려줘", deps="my-api-key")
print(result.output.temperature)  # float임이 타입 시스템에서 보장됨

from pydantic import BaseModel, Field
from typing import Optional
 
class SearchParams(BaseModel):
    query: str = Field(description="검색어")
    max_results: int = Field(default=10, ge=1, le=100, description="최대 결과 수")
    language: Optional[str] = Field(default=None, description="언어 코드 (예: ko, en)")
 
@agent.tool
def search_docs(ctx: RunContext[str], params: SearchParams) -> list[str]:
    """문서를 검색합니다."""
    ...

import asyncio
import os
from pydantic import BaseModel, field_validator  # Pydantic v2 문법
from pydantic_ai import Agent
from typing import Optional
 
MODEL = os.getenv("LLM_MODEL", "openai:gpt-4.1")
 
class InvoiceData(BaseModel):
    vendor: str
    amount: float
    date: str
    items: list[str]
    currency: Optional[str] = "KRW"
 
    @field_validator('amount', mode='before')  # Pydantic v2: @field_validator
    @classmethod
    def amount_must_be_positive(cls, v):
        v = float(v)
        if v <= 0:
            raise ValueError('금액은 양수여야 합니다')
        return v
 
agent = Agent(MODEL, output_type=InvoiceData, retries=2)
 
async def extract_invoice(raw_text: str) -> InvoiceData:
    result = await agent.run(f"다음 영수증에서 정보를 추출해줘:\n{raw_text}")
    return result.output  # InvoiceData 인스턴스, 타입 보장됨
 
async def main():
    raw_text = """
    거래처: (주)테크솔루션
    결제금액: 550,000원
    결제일: 2026-05-15
    항목: 클라우드 서버비용, 모니터링 도구 라이선스
    """
    invoice = await extract_invoice(raw_text)
 
    print(f"거래처: {invoice.vendor}")           # str
    print(f"금액: {invoice.amount:,.0f}원")       # float
    print(f"항목: {', '.join(invoice.items)}")    # list[str]
 
    # DB 인서트 전 타입 오류 걱정 없음 (db는 실제 DB 클라이언트)
    await db.insert("invoices", invoice.model_dump())
 
asyncio.run(main())

import os
from dataclasses import dataclass
from pydantic_ai import Agent, RunContext
 
MODEL = os.getenv("LLM_MODEL", "anthropic:claude-sonnet-4-20250514")
 
# AsyncDatabase, ExternalAPIClient는 실제 프로젝트에서 정의하는 클라이언트입니다.
# 예: asyncpg.Connection, httpx.AsyncClient 등
@dataclass
class SupportDeps:
    db: "AsyncDatabase"
    user_id: int
    api_client: "ExternalAPIClient"
 
agent = Agent(
    MODEL,
    deps_type=SupportDeps,
    system_prompt="고객 지원 에이전트입니다. 정확한 정보만 제공합니다.",
    retries=2,
)
 
@agent.tool
async def get_order_status(ctx: RunContext[SupportDeps], order_id: str) -> str:
    """주문 상태를 조회합니다.
 
    order_id: 조회할 주문 ID (예: ORD-2026-001234)
    """
    order = await ctx.deps.db.fetch_order(ctx.deps.user_id, order_id)
    if not order:
        return "해당 주문을 찾을 수 없습니다."
    return f"주문 {order_id} 상태: {order.status} (예상 도착: {order.estimated_arrival})"
 
@agent.tool
async def get_product_info(ctx: RunContext[SupportDeps], product_id: str) -> dict:
    """상품 정보를 조회합니다.
 
    product_id: 조회할 상품 ID
    """
    return await ctx.deps.api_client.fetch_product(product_id)
 
# 실제 사용
async def handle_request(user_id: int, message: str):
    deps = SupportDeps(db=real_db, user_id=user_id, api_client=real_client)
    result = await agent.run(message, deps=deps)
    return result.output
 
# 테스트 시 — mock 객체로 교체하면 됩니다
async def test_order_status():
    test_deps = SupportDeps(db=mock_db, user_id=99999, api_client=mock_client)
    result = await agent.run("주문 ORD-2026-001234 상태 알려줘", deps=test_deps)
    assert "상태" in result.output

import os
from fastapi import FastAPI, Depends
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from pydantic_ai import Agent
 
MODEL = os.getenv("LLM_MODEL", "anthropic:claude-sonnet-4-20250514")
app = FastAPI()
 
class ChatRequest(BaseModel):
    message: str
    user_id: int
 
agent = Agent(MODEL, deps_type=SupportDeps, retries=2)
 
# get_db(), get_client()는 실제 프로젝트에서 정의하는 팩토리 함수입니다.
def get_deps(request: ChatRequest) -> SupportDeps:
    return SupportDeps(
        db=get_db(),
        user_id=request.user_id,
        api_client=get_client(),
    )
 
@app.post("/chat/stream")
async def chat_stream(
    request: ChatRequest,
    deps: SupportDeps = Depends(get_deps),
):
    async def generate():
        async with agent.run_stream(request.message, deps=deps) as response:
            async for chunk in response.stream_text():
                yield f"data: {chunk}\n\n"
        yield "data: [DONE]\n\n"
 
    return StreamingResponse(generate(), media_type="text/event-stream")
 
@app.post("/chat")
async def chat(
    request: ChatRequest,
    deps: SupportDeps = Depends(get_deps),
):
    result = await agent.run(request.message, deps=deps)
    return {"response": result.output}

import os
from pydantic_ai import Agent, RunContext
 
FAST_MODEL = os.getenv("FAST_LLM_MODEL", "openai:gpt-4.1")
MAIN_MODEL = os.getenv("LLM_MODEL", "anthropic:claude-sonnet-4-20250514")
 
# 서브 에이전트들 — API 키를 deps로 전달하는 경우 str 타입 사용
code_analyzer = Agent(FAST_MODEL, deps_type=str, output_type=str)
security_reviewer = Agent(MAIN_MODEL, deps_type=str, output_type=str)
 
coordinator = Agent(
    MAIN_MODEL,
    deps_type=str,
    system_prompt="코드 리뷰를 총괄합니다. 분석, 보안 검토 결과를 종합해 최종 리뷰를 작성합니다.",
    retries=2,
)
 
@coordinator.tool
async def analyze_code(ctx: RunContext[str], code: str) -> str:
    """코드 품질과 구조를 분석합니다.
 
    code: 분석할 소스 코드
    """
    result = await code_analyzer.run(
        f"다음 코드를 분석해줘:\n{code}",
        deps=ctx.deps,  # API 키를 서브 에이전트에 그대로 전달
    )
    return result.output
 
@coordinator.tool
async def review_security(ctx: RunContext[str], code: str) -> str:
    """보안 취약점을 검토합니다.
 
    code: 검토할 소스 코드
    """
    result = await security_reviewer.run(
        f"보안 관점에서 리뷰해줘:\n{code}",
        deps=ctx.deps,
    )
    return result.output
 
async def review_code(user_code: str, api_key: str) -> str:
    result = await coordinator.run(
        f"다음 코드를 리뷰해줘:\n{user_code}",
        deps=api_key,
    )
    return result.output