AI Sparkup

최신 AI 쉽게 깊게 따라잡기⚡

Pydantic AI 튜토리얼 – 구조화 출력·툴·의존성 주입으로 AI 에이전트 구축하기

pydantic-ai로 Python에서 프로덕션급 AI 에이전트를 만드는 방법을 단계별로 익힌다. 기본 에이전트 실행부터 구조화 출력, 커스텀 툴, 런타임 의존성 주입, 그리고 웹 검색·추론 내장 기능까지 순서대로 다룬다.

전제 조건

  • Python 3.9 이상
  • Pydantic BaseModel과 타입 힌트에 대한 기본 이해
  • 지원 제공자의 API 키 (이 튜토리얼은 openai:gpt-4o-mini 사용)

1단계: 설치 및 기본 실행

가상 환경을 생성하고 패키지를 설치한다.

python -m venv venv
source venv/bin/activate
pip install pydantic-ai
export OPENAI_API_KEY="YOUR-API-KEY-HERE"

가장 단순한 에이전트를 만들어 동작을 확인한다.

from pydantic_ai import Agent

agent = Agent(
    "openai:gpt-4o-mini",
    instructions="You are a concise assistant. Answer in one or two sentences.",
)

result = agent.run_sync("What is a large language model?")
print(result.output)
  • 모델 문자열은 "provider:model-name" 형식이다. 프리픽스만 바꾸면 Anthropic·Google 등 다른 제공자로 교체 가능하다.
  • 비동기 환경에서는 await agent.run(...) 을 사용한다.

2단계: 구조화 출력 (Structured Output)

output_type에 Pydantic 모델을 지정하면 LLM 응답을 검증된 객체로 받는다. 검증 실패 시 프레임워크가 자동 재시도한다.

from pydantic import BaseModel, Field
from pydantic_ai import Agent

class JobPosting(BaseModel):
    job_title: str
    company_name: str
    required_skills: list[str] = Field(description="Explicitly stated technical skills")
    seniority_level: str = Field(description="e.g. Junior, Mid-level, Senior")
    is_remote: bool

agent = Agent(
    "openai:gpt-4o-mini",
    output_type=JobPosting,
    instructions="Extract structured job posting information. Only include what is explicitly stated.",
)

result = agent.run_sync("""
We are hiring a Senior Python Engineer at CoolData Inc.
The role is fully remote. Required: FastAPI, PostgreSQL, Docker.
""")

posting = result.output
print(posting.job_title, posting.seniority_level, posting.is_remote)
# Senior Python Engineer Senior True
print(posting.model_dump())

Field(description=...) 어노테이션은 LLM에게 필드 의도를 전달해 검증 재시도를 줄인다.

3단계: 함수 툴 등록

@agent.tool_plain 으로 Python 함수를 툴로 등록한다. LLM이 독스트링과 타입 힌트를 읽어 언제 호출할지 결정하므로 독스트링은 반드시 명확하게 작성한다.

import json
from pydantic import BaseModel
from pydantic_ai import Agent

NUTRITION_DB = {
    "chicken breast": {"calories": 165, "protein_g": 31, "carbs_g": 0, "fat_g": 3.6},
    "brown rice":     {"calories": 216, "protein_g": 5,  "carbs_g": 45, "fat_g": 1.8},
    "broccoli":       {"calories": 55,  "protein_g": 3.7,"carbs_g": 11, "fat_g": 0.6},
}

class MealSummary(BaseModel):
    total_calories: int
    total_protein_g: float
    total_carbs_g: float
    total_fat_g: float
    health_verdict: str
    recommendation: str

agent = Agent(
    "openai:gpt-4o-mini",
    output_type=MealSummary,
    instructions="Use tools to look up ingredient data, compute totals, and give a verdict.",
)

@agent.tool_plain
def get_ingredient_nutrition(ingredient: str) -> str:
    """Look up calories, protein, carbs, and fat per 100g for a single ingredient."""
    data = NUTRITION_DB.get(ingredient.lower().strip())
    if data:
        return json.dumps({"ingredient": ingredient, **data})
    return f"Not found. Available: {', '.join(NUTRITION_DB)}"

result = agent.run_sync(
    "Analyse: 200g chicken breast, 150g brown rice, 100g broccoli."
)
print(result.output.model_dump())

에이전트는 재료당 툴을 한 번씩 호출해 영양소를 합산한 뒤 MealSummary를 반환한다.

4단계: 의존성 주입 (Dependency Injection)

프로덕션에서는 DB 연결·API 클라이언트를 전역 상태로 두지 않고 RunContext를 통해 주입한다. @agent.tool(tool_plain 아님)을 사용하면 ctx: RunContext[T]가 첫 번째 인자로 전달된다.

from dataclasses import dataclass
from pydantic_ai import Agent, RunContext

@dataclass
class NutritionService:
    database: dict

    def lookup(self, ingredient: str) -> dict | None:
        return self.database.get(ingredient.lower().strip())

    def all_ingredients(self) -> list[str]:
        return list(self.database.keys())

agent = Agent(
    "openai:gpt-4o-mini",
    output_type=MealSummary,
    deps_type=NutritionService,
    instructions="Use tools to compute meal totals and provide a verdict.",
)

@agent.tool
def get_ingredient_nutrition(ctx: RunContext[NutritionService], ingredient: str) -> str:
    """Look up nutritional info (per 100g) for a single ingredient."""
    data = ctx.deps.lookup(ingredient)
    if data:
        return json.dumps({"ingredient": ingredient, **data})
    return f"Not found. Available: {', '.join(ctx.deps.all_ingredients())}"

service = NutritionService(database=NUTRITION_DB)
result = agent.run_sync("Analyse: 150g chicken breast, 200g brown rice.", deps=service)

테스트 시에는 mock으로 교체해 에이전트 정의를 수정하지 않아도 된다.

mock_service = NutritionService(database={"test item": {"calories": 100, "protein_g": 10, "carbs_g": 10, "fat_g": 5}})
with agent.override(deps=mock_service):
    result = agent.run_sync("Analyse 100g test item.", deps=mock_service)
    assert result.output.total_calories == 100

5단계: 내장 기능 (WebSearch·Thinking)

capabilities 인자로 웹 검색과 추론 모드를 추가한다.

from pydantic_ai import Agent
from pydantic_ai.capabilities import WebSearch, Thinking

# 웹 검색만
agent = Agent("openai:gpt-4o-mini", capabilities=[WebSearch()])
result = agent.run_sync("What is the current price of gold?")

# 추론 + 웹 검색 조합
agent = Agent(
    "openai:gpt-4o-mini",
    instructions="You are a research assistant.",
    capabilities=[Thinking(effort="high"), WebSearch()],
)
result = agent.run_sync("What were the biggest AI breakthroughs this month?")

Thinkingeffort 수준(“low”, “medium”, “high”)은 제공자의 네이티브 추론 설정에 매핑된다.

다음 단계

  • MCP 서버 통합 및 고급 툴셋 탐색
  • Logfire와 연동해 LLM 호출·툴 실행·검증 재시도 전 구간 관찰
  • Pydantic AI 공식 문서에서 스트리밍·멀티 에이전트 패턴 참고

참고 자료



AI Sparkup 구독하기

최신 게시물 요약과 더 심층적인 정보를 이메일로 받아 보세요! (무료)