Instructor: LLM이 100% 유효한 JSON을 출력하도록 강제하는 Python 라이브러리 —— 2026 가이드

# Instructor 설치
pip install instructor

# 선호하는 LLM 클라이언트 설치(OpenAI 예시)
pip install openai

import instructor
from openai import OpenAI
from pydantic import BaseModel

# OpenAI 클라이언트를 Instructor로 패치
client = instructor.from_openai(OpenAI())

# 출력 스키마를 Pydantic 모델로 정의
class UserProfile(BaseModel):
    name: str
    age: int
    email: str
    interests: list[str]

# 자연어에서 구조화된 데이터 추출
def extract_profile(user_description: str) -> UserProfile:
    return client.chat.completions.create(
        model="gpt-4o",
        response_model=UserProfile,
        messages=[
            {
                "role": "user",
                "content": f"이 설명에서 사용자 프로필을 추출: {user_description}"
            }
        ]
    )

# 사용법
profile = extract_profile(
    "Sarah는 시애틀 출신의 28세 소프트웨어 엔지니어입니다. "
    "그녀는 하이킹, 사진 촬영, 공상과학 소설 읽기를 좋아합니다. "
    "이메일은 sarah.chen@example.com입니다"
)

print(profile)
# UserProfile(name='Sarah', age=28, email='sarah.chen@example.com', 
#             interests=['hiking', 'photography', 'reading sci-fi novels'])

# 타입이 지정된 필드에 직접 접근
print(f"이름: {profile.name}, 나이: {profile.age}")
print(f"이메일 유효: {'@' in profile.email}")

from pydantic import BaseModel, Field, field_validator

class ValidatedProduct(BaseModel):
    name: str = Field(description="제품 이름, 최대 50자")
    price: float = Field(description="USD 가격, 양수여야 함")
    category: str = Field(description="다음 중 하나: electronics, clothing, food, books")
    
    @field_validator('category')
    @classmethod
    def validate_category(cls, v):
        allowed = {'electronics', 'clothing', 'food', 'books'}
        if v.lower() not in allowed:
            raise ValueError(f"카테고리는 다음 중 하나여야 함: {allowed}")
        return v.lower()
    
    @field_validator('price')
    @classmethod
    def validate_price(cls, v):
        if v <= 0:
            raise ValueError("가격은 양수여야 함")
        return round(v, 2)

# 검증 실패 시 Instructor가 자동 재시도
def parse_product(description: str) -> ValidatedProduct:
    return client.chat.completions.create(
        model="gpt-4o",
        response_model=ValidatedProduct,
        max_retries=3,  # 피드백과 함께 최대 3회 재시도
        messages=[
            {"role": "user", "content": f"이 제품을 파싱: {description}"}
        ]
    )

# 첫 시도에 문제가 있어도 정상 작동
product = parse_product(
    "무선 블루투스 노이즈 캔슬링 헤드폰, $79.99. "
    "전자제품 카테고리."
)
print(product)
# ValidatedProduct(name='무선 블루투스 헤드폰', 
#                  price=79.99, category='electronics')

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

class Address(BaseModel):
    street: str
    city: str
    state: str = Field(description="2글자 주 코드")
    zip_code: str
    country: str = "US"

class OrderItem(BaseModel):
    product_name: str
    quantity: int = Field(ge=1, description="1 이상이어야 함")
    unit_price: float = Field(gt=0)
    
    @property
    def total(self) -> float:
        return self.quantity * self.unit_price

class CustomerOrder(BaseModel):
    customer_name: str
    customer_email: str
    shipping_address: Address
    billing_address: Optional[Address] = None
    items: List[OrderItem]
    order_notes: Optional[str] = None
    
    @property
    def grand_total(self) -> float:
        return sum(item.total for item in self.items)

def extract_order(email_text: str) -> CustomerOrder:
    return client.chat.completions.create(
        model="gpt-4o",
        response_model=CustomerOrder,
        messages=[
            {"role": "user", "content": f"이메일에서 주문 추출:\n\n{email_text}"}
        ]
    )

order = extract_order("""
안녕하세요, 주문을 하고 싶습니다.

고객: John Smith (john.smith@email.com)
배송지: 123 Oak Street, San Francisco, CA 94102

상품:
- MacBook Pro M3, 수량 1, $1999
- USB-C 허브, 수량 2, 각 $49

노트북은 선물 포장해 주세요.
""")

print(f"고객: {order.customer_name}")
print(f"배송 도시: {order.shipping_address.city}")
print(f"주문 총액: ${order.grand_total:.2f}")

# --- Anthropic Claude ---
import anthropic
import instructor

anthropic_client = instructor.from_anthropic(anthropic.Anthropic())

result = anthropic_client.messages.create(
    model="claude-sonnet-4-20250514",
    response_model=UserProfile,
    max_tokens=1024,
    messages=[{"role": "user", "content": "추출: Mike 35세, 테니스 좋아함"}]
)
print(result)

# --- Google Gemini ---
import google.generativeai as genai
import instructor

gemini_client = instructor.from_gemini(
    genai.GenerativeModel("gemini-2.5-pro")
)

result = gemini_client.generate_content(
    response_model=UserProfile,
    contents=["추출: Lisa 29세, 요리와 요가 좋아함"]
)
print(result)

# --- Cohere ---
import cohere
import instructor

cohere_client = instructor.from_cohere(cohere.Client())

result = cohere_client.chat(
    response_model=UserProfile,
    message="추출: David 42세, 골프와 낚시 좋아함"
)
print(result)

import asyncio
import instructor
from openai import AsyncOpenAI
from pydantic import BaseModel

# 배치 처리를 위한 비동기 클라이언트 사용
async_client = instructor.from_openai(AsyncOpenAI())

class SentimentResult(BaseModel):
    text: str
    sentiment: str  # "positive", "negative", "neutral"
    confidence: float
    key_phrases: list[str]

async def analyze_single(text: str) -> SentimentResult:
    return await async_client.chat.completions.create(
        model="gpt-4o-mini",
        response_model=SentimentResult,
        messages=[
            {"role": "user", "content": f"감정 분석: {text}"}
        ]
    )

async def analyze_batch(texts: list[str]) -> list[SentimentResult]:
    """여러 텍스트를 동시에 처리."""
    tasks = [analyze_single(text) for text in texts]
    results = await asyncio.gather(*tasks)
    return results

# 100개 리뷰를 동시에 처리
texts = [
    "이 제품은 기대를 뛰어넘었어요!",
    "품질이 너무 안 좋아서 하루 만에 고장났어요.",
    "그냥 그래요. 특별한 건 없지만 쓸 만해요.",
    # ... 97개 더 추가
]

results = asyncio.run(analyze_batch(texts))
positive = sum(1 for r in results if r.sentiment == "positive")
print(f"긍정: {positive}/{len(results)}")

from typing import Iterable
from pydantic import BaseModel

class PartialArticle(BaseModel):
    title: str
    sections: list[str]
    key_points: list[str]

# 생성되는 대로 구조화된 데이터 스트리밍
def stream_article(topic: str) -> Iterable[PartialArticle]:
    return client.chat.completions.create_partial(
        model="gpt-4o",
        response_model=PartialArticle,
        stream=True,
        messages=[
            {"role": "user", "content": f"다음 주제에 대한 기사 개요를 작성: {topic}"}
        ]
    )

# 부분 결과가 도착하는 대로 소비
for partial in stream_article("2026년 재생 에너지 트렌드"):
    print(f"제목: {partial.title}")
    print(f"현재까지의 섹션 수: {len(partial.sections)}")
    print("---")

from pydantic import BaseModel, field_validator

class StrictDateRange(BaseModel):
    start_date: str = Field(description="YYYY-MM-DD 형식")
    end_date: str = Field(description="YYYY-MM-DD 형식, 시작일 이후여야 함")
    
    @field_validator('start_date', 'end_date')
    @classmethod
    def validate_date_format(cls, v):
        from datetime import datetime
        datetime.strptime(v, "%Y-%m-%d")
        return v
    
    @field_validator('end_date')
    @classmethod
    def validate_order(cls, end, info):
        start = info.data.get('start_date')
        if start and end <= start:
            raise ValueError("end_date는 start_date 이후여야 함")
        return end

# Instructor가 특정 검증 오류 피드백으로 재시도
def extract_date_range(text: str) -> StrictDateRange:
    return client.chat.completions.create(
        model="gpt-4o",
        response_model=StrictDateRange,
        max_retries=3,
        messages=[
            {"role": "user", "content": f"날짜 범위 추출: {text}"}
        ]
    )

# 모델이 처음에 날짜를 바꾸거나 잘못된 형식을 사용하더라도,
# Instructor가 구체적인 오류 메시지로 재질문
try:
    result = extract_date_range(
        "프로젝트는 2026년 3월 15일부터 2026년 1월 10일까지 진행되었습니다"
    )
    print(result)
except Exception as e:
    print(f"최대 재시도 횟수 후 실패: {e}")

from typing import Literal

class SupportTicket(BaseModel):
    customer_query: str
    category: Literal[
        "billing", 
        "technical_support", 
        "account_access",
        "feature_request",
        "refund",
        "general_inquiry"
    ]
    priority: Literal["low", "medium", "high", "urgent"]
    suggested_response: str

def classify_ticket(ticket_text: str) -> SupportTicket:
    return client.chat.completions.create(
        model="gpt-4o-mini",
        response_model=SupportTicket,
        messages=[
            {
                "role": "system",
                "content": "당신은 고객 지원 분류 전문가입니다."
            },
            {"role": "user", "content": f"이 티켓을 분류:\n\n{ticket_text}"}
        ]
    )

# 분류 결과는 항상 허용된 값 중 하나
ticket = classify_ticket(
    "이번 달 구독이 두 번 청구되었습니다. "
    "중복 청구를 즉시 환불해 주세요."
)
print(f"카테고리: {ticket.category}")  # 항상 "billing"
print(f"우선순위: {ticket.priority}")  # 항상 4개 값 중 하나

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import instructor
from openai import OpenAI

app = FastAPI(title="구조화된 LLM API")
client = instructor.from_openai(OpenAI())

# 요청 스키마
class ExtractionRequest(BaseModel):
    text: str
    extract_fields: list[str]

# 응답 스키마
class ExtractedData(BaseModel):
    entities: list[dict]
    relationships: list[dict]
    summary: str

@app.post("/extract", response_model=ExtractedData)
async def extract_entities(request: ExtractionRequest):
    """비구조화된 텍스트에서 구조화된 엔티티를 추출."""
    try:
        result = client.chat.completions.create(
            model="gpt-4o",
            response_model=ExtractedData,
            messages=[
                {
                    "role": "user",
                    "content": (
                        f"이 텍스트에서 엔티티를 추출. "
                        f"포커스: {', '.join(request.extract_fields)}\n\n"
                        f"텍스트: {request.text}"
                    )
                }
            ]
        )
        return result
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

# 실행: uvicorn main:app --reload

from typing import Type

class SearchQuery(BaseModel):
    """매개변수가 있는 생성된 검색 쿼리"""
    keywords: list[str]
    filters: dict[str, str]
    sort_by: Literal["relevance", "date", "price_asc", "price_desc"]
    
def generate_search(user_request: str) -> SearchQuery:
    return client.chat.completions.create(
        model="gpt-4o",
        response_model=SearchQuery,
        messages=[
            {
                "role": "user",
                "content": f"다음 요청에 대해 최적화된 검색 쿼리를 생성: {user_request}"
            }
        ]
    )

query = generate_search(
    "배터리 수명이 좋은 $100 이하 무선 이어폰을 찾아, 최신순"
)
print(query.keywords)  # ['wireless earbuds', 'bluetooth']
print(query.filters)   # {'max_price': '100'}
print(query.sort_by)   # 'date'

import logging
import instructor

# 상세 로깅 활성화
instructor.enable_logging()
logging.basicConfig(level=logging.DEBUG)

# 또는 특정 로거 구성
logger = logging.getLogger("instructor")
logger.setLevel(logging.INFO)

# 프로덕션용 파일 핸들러 추가
handler = logging.FileHandler("instructor.log")
handler.setFormatter(logging.Formatter(
    '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
))
logger.addHandler(handler)

# 모든 재시도 시도와 검증 오류가 기록됨
result = client.chat.completions.create(
    model="gpt-4o",
    response_model=UserProfile,
    max_retries=3,
    messages=[{"role": "user", "content": "추출: Jane, 25세, 예술 좋아함"}]
)