AI Agent

Instructor 라이브러리로 구조화 출력 실전 2026 — LLM에서 신뢰할 수 있는 JSON을 뽑는 법

cell-devlog 2026. 5. 29. 11:35
반응형

LLM에게 JSON으로 응답하라고 프롬프트를 쓰면 됩니다. 대부분 잘 됩니다. 문제는 나머지 5~10%입니다. Markdown 펜스가 붙거나, 필드가 빠지거나, 타입이 틀리거나, 음수가 와야 하는데 양수가 옵니다. Instructor는 그 나머지 5~10%를 자동으로 잡아냅니다.


핵심 요약 → Instructor = Pydantic 모델 기반 LLM 구조화 출력 라이브러리 (월 300만 다운로드, GitHub 11k★) → 핵심 기능: 스키마 강제 + 자동 검증 + 검증 실패 시 자동 재시도 → from_provider("provider/model") — 15개+ 프로바이더를 동일 인터페이스로 → 지원: OpenAI·Claude·Gemini·Ollama·DeepSeek·Groq 등 → 검증 실패 시 에러 메시지를 LLM에게 피드백 → 자동 수정 요청 → 스트리밍 부분 객체(Partial), 리스트, 멀티모달 이미지·PDF 입력 지원 → 에이전트보다 단순 추출이 필요할 때 PydanticAI보다 가볍고 빠름


왜 프롬프트만으로 부족한가

# 순수 프롬프트 방식 — 실전에서 자주 겪는 문제들

import anthropic
import json

client = anthropic.Anthropic()

response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=512,
    messages=[{
        "role": "user",
        "content": "사용자 정보를 JSON으로 추출해줘: '홍길동, 32세, 개발자'"
    }]
)

raw = response.content[0].text
# 실제로 올 수 있는 응답들:

# ✅ 정상
# {"name": "홍길동", "age": 32, "job": "개발자"}

# ❌ Markdown 펜스 포함
# ```json
# {"name": "홍길동", "age": 32, "job": "개발자"}
# ```

# ❌ 자연어 포함
# "네, JSON으로 추출했습니다: {"name": "홍길동"...}"

# ❌ 필드명 불일치
# {"이름": "홍길동", "나이": 32}

# ❌ 타입 오류
# {"name": "홍길동", "age": "32"}  ← str이 와야 할 곳에 int

# 결과: 파싱 코드가 에러 처리 코드로 가득 차게 됨
try:
    data = json.loads(raw.strip().strip('```json').strip('```'))
except:
    # 재시도 로직 직접 작성해야 함...
    pass

1. 설치 및 기본 설정

# 기본 설치
pip install instructor

# 프로바이더별 추가 의존성
pip install instructor[anthropic]   # Claude
pip install instructor[google-genai] # Gemini
# OpenAI는 기본 포함
import instructor
from pydantic import BaseModel

# ── 방법 1: from_provider (권장 — 2026년 신규 인터페이스) ──

# 동일한 인터페이스로 어떤 프로바이더든 전환 가능
client_openai    = instructor.from_provider("openai/gpt-5.5")
client_claude    = instructor.from_provider("anthropic/claude-sonnet-4-6")
client_gemini    = instructor.from_provider("google/gemini-3.5-flash")
client_local     = instructor.from_provider("ollama/qwen3-coder")  # 로컬
client_deepseek  = instructor.from_provider("deepseek/deepseek-v4-flash")

# API 키 직접 전달도 가능
client_claude = instructor.from_provider(
    "anthropic/claude-sonnet-4-6",
    api_key="sk-ant-..."  # 환경 변수 대신
)

# ── 방법 2: from_openai / from_anthropic (기존 방식도 유효) ──
import anthropic

raw_client = anthropic.Anthropic()
client_claude = instructor.from_anthropic(raw_client)

2. 핵심 패턴 — 모델 정의 + 추출

from pydantic import BaseModel, Field
from typing import Optional
import instructor

client = instructor.from_provider("anthropic/claude-sonnet-4-6")


# ── 기본 추출 ──

class UserProfile(BaseModel):
    name: str
    age: int
    occupation: str
    email: Optional[str] = None

user = client.chat.completions.create(
    response_model=UserProfile,
    messages=[{
        "role": "user",
        "content": "홍길동은 32살이고 서울에서 백엔드 개발자로 일하고 있어요. 연락처는 hong@example.com"
    }]
)

print(user.name)        # "홍길동"
print(user.age)         # 32  (int, 타입 보장)
print(user.occupation)  # "백엔드 개발자"
print(user.email)       # "hong@example.com"
# → JSON 파싱, 타입 변환, 에러 처리 전혀 없음


# ── 중첩 모델 ──

class Address(BaseModel):
    city: str
    country: str

class Company(BaseModel):
    name: str
    founded: int
    headquarters: Address      # 중첩 모델
    ceo: Optional[str] = None

company = client.chat.completions.create(
    response_model=Company,
    messages=[{
        "role": "user",
        "content": "Anthropic은 2021년 설립됐고 미국 샌프란시스코에 위치하며 Dario Amodei가 CEO입니다."
    }]
)

print(company.name)                    # "Anthropic"
print(company.headquarters.city)       # "San Francisco"  (중첩 접근)
print(company.founded)                 # 2021

3. 자동 재시도 — Instructor의 핵심 가치

from pydantic import BaseModel, field_validator, model_validator
from typing import Literal
import instructor

client = instructor.from_provider("anthropic/claude-sonnet-4-6")


# ── 커스텀 검증 + 자동 재시도 ──

class SentimentAnalysis(BaseModel):
    sentiment: Literal["positive", "negative", "neutral"]
    confidence: float = Field(ge=0.0, le=1.0)  # 0~1 범위 강제
    summary: str = Field(min_length=10)         # 최소 10자

    @field_validator("confidence")
    @classmethod
    def validate_confidence(cls, v):
        if v < 0 or v > 1:
            raise ValueError("신뢰도는 0과 1 사이여야 합니다")
        return round(v, 2)

    @model_validator(mode="after")
    def validate_consistency(self):
        # 비즈니스 로직 검증: positive인데 신뢰도가 0.3 미만이면 의심
        if self.sentiment == "positive" and self.confidence < 0.3:
            raise ValueError(
                "positive 감정인데 신뢰도가 너무 낮습니다. 재분석이 필요합니다."
            )
        return self


# max_retries: 검증 실패 시 최대 재시도 횟수
# → 실패 시 에러 메시지를 LLM에게 피드백해서 수정 요청
result = client.chat.completions.create(
    response_model=SentimentAnalysis,
    max_retries=3,   # ← 이게 핵심
    messages=[{
        "role": "user",
        "content": "이 리뷰 감정 분석해줘: '이 제품 정말 최악이에요. 돈 낭비입니다.'"
    }]
)

print(result.sentiment)    # "negative"
print(result.confidence)   # 0.95
print(result.summary)      # "제품에 대한 강한 부정적 평가..."

# 재시도 흐름:
# 1회: LLM이 confidence=1.5 반환 → field_validator 실패
# → Instructor가 에러 메시지("신뢰도는 0과 1 사이여야 합니다")를
#   다음 메시지로 LLM에게 피드백
# 2회: LLM이 올바른 값 반환 → 성공

4. Enum + Literal — 허용 값 강제

from enum import Enum
from pydantic import BaseModel
from typing import Literal

# ── Enum 방식 ──

class Priority(str, Enum):
    LOW    = "low"
    MEDIUM = "medium"
    HIGH   = "high"
    URGENT = "urgent"

class TaskClassification(BaseModel):
    category: Literal["bug", "feature", "docs", "refactor", "test"]
    priority: Priority
    estimated_hours: float = Field(gt=0, le=80)
    requires_review: bool

task = client.chat.completions.create(
    response_model=TaskClassification,
    messages=[{
        "role": "user",
        "content": "로그인 페이지에서 비밀번호 재설정이 안 되는 버그. 중요한 기능이라 빨리 고쳐야 함."
    }]
)

print(task.category)          # "bug"
print(task.priority)          # Priority.HIGH  (Enum 타입 보장)
print(task.priority.value)    # "high"
print(task.requires_review)   # True
# → "긴급" "급함" 같은 애매한 표현이 아닌 정확한 Enum 값으로 수신

5. 리스트 추출 + 스트리밍

from instructor import Partial
from typing import Iterable

# ── 리스트 일괄 추출 ──

class Product(BaseModel):
    name: str
    price: float
    category: str

# Iterable[Product] → 여러 객체를 한 번에 추출
products = client.chat.completions.create(
    response_model=Iterable[Product],
    messages=[{
        "role": "user",
        "content": """다음 상품 목록에서 정보를 추출해:
- 맥북 프로 14인치: 299만원, 노트북
- 에어팟 프로: 32만원, 이어폰
- 아이패드 에어: 89만원, 태블릿"""
    }]
)

for product in products:
    print(f"{product.name}: {product.price}원 ({product.category})")


# ── 스트리밍 부분 객체 (Partial) ──
# 긴 응답을 기다리지 않고 필드가 채워지는 대로 처리

class ResearchReport(BaseModel):
    title: str
    executive_summary: str
    key_findings: list[str]
    conclusion: str

for partial_report in client.chat.completions.create(
    response_model=Partial[ResearchReport],   # Partial 래핑
    stream=True,
    messages=[{
        "role": "user",
        "content": "AI 에이전트 시장 현황 리포트 작성해줘"
    }]
):
    # 필드가 채워지는 순간순간 처리 가능
    if partial_report.title:
        print(f"\r제목: {partial_report.title}", end="")
    if partial_report.executive_summary:
        print(f"\n요약: {partial_report.executive_summary[:50]}...", end="")

6. 멀티모달 — 이미지·PDF 구조화 추출

from instructor.processing.multimodal import Image, PDF

# ── 이미지에서 구조화 데이터 추출 ──

class InvoiceData(BaseModel):
    vendor_name: str
    invoice_number: str
    total_amount: float
    currency: str
    issue_date: str
    items: list[dict]

# 이미지 로드 방법 4가지
invoice = client.chat.completions.create(
    response_model=InvoiceData,
    messages=[{
        "role": "user",
        "content": [
            "이 인보이스에서 정보를 추출해줘",
            Image.from_url("https://example.com/invoice.jpg"),      # URL
            # Image.from_path("./invoice.png"),                     # 로컬 파일
            # Image.from_base64("base64_string_here"),              # Base64
            # Image.autodetect("url_or_path_or_base64"),            # 자동 감지
        ]
    }],
    max_retries=3
)

print(invoice.vendor_name)    # "ABC Corp"
print(invoice.total_amount)   # 1500000.0
print(invoice.currency)       # "KRW"


# ── PDF에서 구조화 데이터 추출 ──

class ContractSummary(BaseModel):
    parties: list[str]
    effective_date: str
    termination_clause: str
    payment_terms: str
    key_obligations: list[str]

contract = client.chat.completions.create(
    response_model=ContractSummary,
    messages=[{
        "role": "user",
        "content": [
            "이 계약서의 핵심 내용을 추출해줘",
            PDF.from_path("./contract.pdf")
        ]
    }]
)

7. 프로바이더 전환 패턴 — 동일 코드로 모델 교체

# 프로바이더 추상화 — 비용·성능 라우팅

import instructor
from pydantic import BaseModel

class ExtractedEntity(BaseModel):
    entities: list[str]
    entity_types: list[str]
    sentiment: str

def extract_entities(
    text: str,
    provider: str = "anthropic/claude-sonnet-4-6"  # 기본값
) -> ExtractedEntity:
    """
    provider만 바꾸면 동일 로직으로 다른 모델 사용
    """
    client = instructor.from_provider(provider)
    return client.chat.completions.create(
        response_model=ExtractedEntity,
        max_retries=3,
        messages=[{
            "role": "user",
            "content": f"다음 텍스트에서 개체명을 추출해줘: {text}"
        }]
    )

# 개발: 저렴한 모델로 빠르게 테스트
result = extract_entities(text, provider="openai/gpt-5.5-instant")

# 프로덕션: 정확한 모델
result = extract_entities(text, provider="anthropic/claude-sonnet-4-6")

# 비용 절감: 로컬 모델
result = extract_entities(text, provider="ollama/qwen3-coder")

# 배치 처리: 가장 싼 API
result = extract_entities(text, provider="deepseek/deepseek-v4-flash")

8. 에이전트 파이프라인에서의 실전 활용

# Instructor를 에이전트 파이프라인의 구조화 레이어로 활용

import instructor
from pydantic import BaseModel, Field
from typing import Optional

client = instructor.from_provider("anthropic/claude-sonnet-4-6")


# ── 에이전트 분석 결과 구조화 ──

class CodeAnalysisResult(BaseModel):
    """코드 분석 에이전트 출력 스키마"""
    has_bugs: bool
    bugs: list[dict] = Field(default_factory=list)
    security_issues: list[str] = Field(default_factory=list)
    complexity_score: float = Field(ge=1.0, le=10.0)
    refactor_suggestions: list[str] = Field(default_factory=list)
    estimated_fix_time_hours: Optional[float] = None

    @field_validator("bugs")
    @classmethod
    def validate_bug_schema(cls, bugs):
        for bug in bugs:
            if "line" not in bug or "description" not in bug:
                raise ValueError("버그 항목에 line, description 필드가 필요합니다")
        return bugs


def analyze_code(code: str) -> CodeAnalysisResult:
    return client.chat.completions.create(
        response_model=CodeAnalysisResult,
        max_retries=3,
        messages=[{
            "role": "user",
            "content": f"다음 코드를 분석해줘:\n\n```python\n{code}\n```"
        }]
    )


# ── Plan-and-Execute와 연동 ──

class PlanStep(BaseModel):
    step_id: int
    description: str
    tool: Literal["read_file", "write_file", "run_command", "search"]
    estimated_minutes: float = Field(gt=0)
    dependencies: list[int] = Field(default_factory=list)

class ExecutionPlan(BaseModel):
    goal: str
    total_steps: int
    steps: list[PlanStep]
    risk_level: Literal["low", "medium", "high"]

def create_plan(goal: str) -> ExecutionPlan:
    """Plan-and-Execute 패턴의 Planner를 Instructor로 구조화"""
    return client.chat.completions.create(
        response_model=ExecutionPlan,
        max_retries=3,
        messages=[{
            "role": "user",
            "content": f"다음 목표를 달성하기 위한 실행 계획을 수립해줘: {goal}"
        }]
    )

plan = create_plan("레거시 Flask 앱을 FastAPI로 마이그레이션")
print(f"총 {plan.total_steps}단계, 위험도: {plan.risk_level}")
for step in plan.steps:
    print(f"  {step.step_id}. {step.description} ({step.tool})")

9. Instructor vs 네이티브 Structured Outputs 비교

# 언제 Instructor를 쓰고 언제 네이티브 API를 쓰나

네이티브 Structured Outputs (OpenAI JSON mode 등):
  장점: 추가 라이브러리 없음, 모델이 직접 스키마 강제
  단점: 프로바이더마다 문법 다름, 재시도 로직 직접 구현 필요
  적합: 단일 프로바이더 사용, 간단한 스키마

Instructor:
  장점: 자동 재시도, 프로바이더 통합 인터페이스, 복잡한 검증 로직
  단점: 추가 의존성, 약간의 레이턴시 오버헤드(재시도 시)
  적합: 멀티 프로바이더 전환, 비즈니스 로직 검증, 프로덕션 안정성

PydanticAI:
  Instructor의 상위 호환 — 에이전트·툴·옵저버빌리티 포함
  적합: 복잡한 에이전트 워크플로 (단순 추출이 아닌 경우)
  관계: Instructor의 Pydantic 모델을 PydanticAI에서 그대로 재사용 가능

결론

Instructor가 빛나는 케이스

  • LLM 출력을 downstream 시스템(DB, API)에 바로 넣어야 할 때
  • 비즈니스 로직 검증이 있는 경우 (범위, 패턴, 일관성 체크)
  • 멀티 프로바이더 전환이 필요한 프로젝트
  • 이미지·PDF → 구조화 데이터 추출 파이프라인
  • 에이전트 플래너 출력 스키마 강제

핵심 패턴 3가지

  • from_provider() + response_model= → 5줄로 구조화 추출
  • max_retries=3 → 검증 실패 자동 재시도
  • @field_validator → 비즈니스 로직을 스키마에 내장

Instructor 없이도 되는 경우

  • 단순 요약·번역·창작 (구조화 출력 불필요)
  • 단일 프로바이더 + 간단한 JSON (네이티브 Structured Outputs 충분)
  • 복잡한 에이전트 워크플로 → PydanticAI로 직행

관련 글

반응형