반응형
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로 직행
관련 글
반응형
'AI Agent' 카테고리의 다른 글
| PydanticAI 완전가이드 2026 — FastAPI 철학의 에이전트 프레임워크 (0) | 2026.05.29 |
|---|---|
| OpenTelemetry로 LLM 에이전트 추적 — 스팬 계측, 토큰 비용 추적, 프로덕션 디버깅 (0) | 2026.05.29 |
| 멀티에이전트 시스템: 오케스트레이터-워커 병렬 에이전트 패턴 — N개 서브태스크 동시 실행, 비용·레이턴시 트레이드오프 계산 (0) | 2026.05.29 |
| Plan-and-Execute 에이전트 패턴 — 계획과 실행을 분리하면 비용이 절반이 된다 (0) | 2026.05.29 |
| AI 에이전트 배포 의사결정 매트릭스 2026 — SaaS·자체호스팅·하이브리드, 어떤 것을 선택해야 하나 (0) | 2026.05.28 |