LLM Structured Output 실전 가이드 — JSON Mode, Tool Use, Pydantic 스키마 검증

from openai import OpenAI
from pydantic import BaseModel
from typing import List, Optional

client = OpenAI()

# 방법 1: JSON Mode (기본)
response = client.chat.completions.create(
    model="gpt-4o-2024-08-06",
    messages=[
        {"role": "system", "content": "응답을 JSON으로 반환하세요."},
        {"role": "user", "content": "서울의 유명 맛집 3곳 추천해줘"}
    ],
    response_format={"type": "json_object"}
)
# JSON은 보장되지만, 스키마는 보장되지 않음

# 방법 2: Structured Outputs (스키마 보장)
class Restaurant(BaseModel):
    name: str
    cuisine: str
    price_range: str
    rating: float
    address: str

class RestaurantList(BaseModel):
    restaurants: List[Restaurant]
    total_count: int

response = client.beta.chat.completions.parse(
    model="gpt-4o-2024-08-06",
    messages=[
        {"role": "system", "content": "서울의 유명 맛집을 추천해주세요."},
        {"role": "user", "content": "한식 맛집 3곳"}
    ],
    response_format=RestaurantList
)

result = response.choices[0].message.parsed
print(result.restaurants[0].name)  # 타입 안전!

import anthropic
from typing import List

client = anthropic.Anthropic()

# Anthropic은 Tool Use를 활용한 Structured Output
response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    tools=[
        {
            "name": "extract_restaurants",
            "description": "맛집 정보를 구조화된 형식으로 추출",
            "input_schema": {
                "type": "object",
                "properties": {
                    "restaurants": {
                        "type": "array",
                        "items": {
                            "type": "object",
                            "properties": {
                                "name": {"type": "string"},
                                "cuisine": {"type": "string"},
                                "price_range": {
                                    "type": "string",
                                    "enum": ["$", "$$", "$$$", "$$$$"]
                                },
                                "rating": {"type": "number"},
                                "address": {"type": "string"}
                            },
                            "required": ["name", "cuisine", "price_range"]
                        }
                    },
                    "total_count": {"type": "integer"}
                },
                "required": ["restaurants", "total_count"]
            }
        }
    ],
    tool_choice={"type": "tool", "name": "extract_restaurants"},
    messages=[
        {"role": "user", "content": "서울 한식 맛집 3곳 추천해줘"}
    ]
)

# Tool Use 결과에서 구조화된 데이터 추출
tool_use = next(
    block for block in response.content
    if block.type == "tool_use"
)
restaurants = tool_use.input["restaurants"]

import google.generativeai as genai

genai.configure(api_key="YOUR_API_KEY")

model = genai.GenerativeModel(
    "gemini-2.0-flash",
    generation_config=genai.GenerationConfig(
        response_mime_type="application/json",
        response_schema={
            "type": "object",
            "properties": {
                "restaurants": {
                    "type": "array",
                    "items": {
                        "type": "object",
                        "properties": {
                            "name": {"type": "string"},
                            "cuisine": {"type": "string"},
                            "rating": {"type": "number"}
                        }
                    }
                }
            }
        }
    )
)

response = model.generate_content("서울 한식 맛집 3곳")
import json
data = json.loads(response.text)

from anthropic import Anthropic

client = Anthropic()

response = client.messages.create(
    model="claude-opus-5",
    max_tokens=16000,
    messages=[{"role": "user", "content": "..."}],
    output_config={
        "format": {
            "type": "json_schema",
            "schema": {
                "type": "object",
                "properties": {
                    "name": {"type": "string"},
                    "plan": {"type": "string"},
                },
                "required": ["name", "plan"],
                "additionalProperties": False,
            },
        }
    },
)

from pydantic import BaseModel

class Account(BaseModel):
    name: str
    plan: str

response = client.messages.parse(
    model="claude-opus-5",
    max_tokens=16000,
    messages=[{"role": "user", "content": "..."}],
    output_format=Account,
)

account = response.parsed_output  # 검증을 통과한 Account 인스턴스

tools = [
    {
        "name": "extract_restaurants",
        "description": "맛집 정보를 구조화된 형식으로 추출",
        "strict": True,  # input_schema 안이 아니라 여기
        "input_schema": {
            "type": "object",
            "properties": {
                "restaurants": {"type": "array", "items": {"type": "object"}},
                "total_count": {"type": "integer"},
            },
            "required": ["restaurants", "total_count"],
            "additionalProperties": False,
        },
    }
]

# Chat Completions — json_schema가 한 겹 더 중첩되고, 이름이 안쪽에 있다
"response_format": {
    "type": "json_schema",
    "json_schema": {
        "name": "person",
        "strict": True,
        "schema": {"type": "object", "properties": {}},
    },
}

# Responses API — text.format 아래에 평평하게 놓이고, 이름이 type과 같은 레벨이다
"text": {
    "format": {
        "type": "json_schema",
        "name": "person",
        "strict": True,
        "schema": {"type": "object", "properties": {}},
    }
}

from openai import OpenAI

client = OpenAI()

# Chat Completions — beta 경로가 아니라 여기
completion = client.chat.completions.parse(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "한식 맛집 3곳"}],
    response_format=RestaurantList,
)
result = completion.choices[0].message.parsed

# Responses API — 새 프로젝트 권장 경로
response = client.responses.parse(
    model="gpt-4o-mini",
    input=[{"role": "user", "content": "한식 맛집 3곳"}],
    text_format=RestaurantList,
)
# 파싱 결과를 꺼내는 속성 이름은 SDK 버전에 따라 다릅니다.
# 정확한 API는 사용 중인 버전의 문서에서 확인하세요.

from google import genai

client = genai.Client()

interaction = client.interactions.create(
    model="gemini-3.6-flash",
    input=prompt,
    response_format={
        "type": "text",
        "mime_type": "application/json",
        "schema": Recipe.model_json_schema(),  # Pydantic 모델을 그대로 넘길 수 없다
    },
)

recipe = Recipe.model_validate_json(interaction.output_text)

from google import genai
from google.genai import types

client = genai.Client(api_key="...")

# 레거시 경로 — 계속 지원되며, 여기서는 Pydantic 모델을 그대로 넘길 수 있다
response = client.models.generate_content(
    model="gemini-3.6-flash",
    contents=prompt,
    config=types.GenerateContentConfig(
        response_mime_type="application/json",
        response_schema=CountryInfo,
    ),
)

from pydantic import BaseModel, Field, validator
from typing import List, Optional, Literal
from enum import Enum
import json

class PriceRange(str, Enum):
    CHEAP = "$"
    MODERATE = "$$"
    EXPENSIVE = "$$$"
    VERY_EXPENSIVE = "$$$$"

class Restaurant(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    cuisine: str = Field(..., description="음식 종류")
    price_range: PriceRange
    rating: float = Field(..., ge=0.0, le=5.0)
    address: Optional[str] = None
    tags: List[str] = Field(default_factory=list, max_length=10)

    @validator('rating')
    def round_rating(cls, v):
        return round(v, 1)

class RestaurantResponse(BaseModel):
    restaurants: List[Restaurant] = Field(..., min_length=1, max_length=20)
    query: str
    total_count: int

# LLM 응답 파싱 + 검증
def parse_llm_response(raw_json: str) -> RestaurantResponse:
    """LLM 응답을 파싱하고 Pydantic으로 검증"""
    try:
        data = json.loads(raw_json)
        return RestaurantResponse(**data)
    except json.JSONDecodeError as e:
        raise ValueError(f"Invalid JSON: {e}")
    except Exception as e:
        raise ValueError(f"Schema validation failed: {e}")

from tenacity import retry, stop_after_attempt, retry_if_exception_type

class StructuredOutputParser:
    def __init__(self, client, model: str, schema: type[BaseModel]):
        self.client = client
        self.model = model
        self.schema = schema

    @retry(
        stop=stop_after_attempt(3),
        retry=retry_if_exception_type(ValueError)
    )
    def parse(self, prompt: str) -> BaseModel:
        """스키마 검증 실패 시 에러 메시지를 포함하여 재시도"""
        schema_json = self.schema.model_json_schema()

        messages = [
            {
                "role": "system",
                "content": f"다음 JSON 스키마에 맞게 응답하세요:\n{json.dumps(schema_json, indent=2)}"
            },
            {"role": "user", "content": prompt}
        ]

        # 이전 시도의 에러가 있으면 포함
        if hasattr(self, '_last_error'):
            messages.append({
                "role": "user",
                "content": f"이전 응답에서 에러가 발생했습니다: {self._last_error}\n올바른 JSON으로 다시 응답해주세요."
            })

        response = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            response_format={"type": "json_object"}
        )

        raw = response.choices[0].message.content
        try:
            data = json.loads(raw)
            result = self.schema(**data)
            if hasattr(self, '_last_error'):
                del self._last_error
            return result
        except Exception as e:
            self._last_error = str(e)
            raise ValueError(str(e))

# 사용
parser = StructuredOutputParser(client, "gpt-4o", RestaurantResponse)
result = parser.parse("서울 한식 맛집 3곳 추천")

import litellm
from pydantic import BaseModel

class ExtractedInfo(BaseModel):
    summary: str
    key_points: list[str]
    sentiment: str
    confidence: float

# OpenAI
response = litellm.completion(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Kubernetes 1.35 릴리스 요약해줘"}],
    response_format=ExtractedInfo
)

# Anthropic (자동으로 Tool Use 변환)
response = litellm.completion(
    model="claude-sonnet-4-20250514",
    messages=[{"role": "user", "content": "Kubernetes 1.35 릴리스 요약해줘"}],
    response_format=ExtractedInfo
)

# Gemini
response = litellm.completion(
    model="gemini/gemini-2.0-flash",
    messages=[{"role": "user", "content": "Kubernetes 1.35 릴리스 요약해줘"}],
    response_format=ExtractedInfo
)

# 동일한 코드로 3개 프로바이더 사용 가능!

# pip install instructor
import instructor
from openai import OpenAI
from pydantic import BaseModel
from typing import List

client = instructor.from_openai(OpenAI())

class Step(BaseModel):
    explanation: str
    output: str

class MathSolution(BaseModel):
    steps: List[Step]
    final_answer: str
    confidence: float

# Pydantic 모델을 직접 response_model로 사용
solution = client.chat.completions.create(
    model="gpt-4o",
    response_model=MathSolution,
    messages=[
        {"role": "user", "content": "2x + 5 = 15를 풀어줘"}
    ],
    max_retries=3  # 자동 재시도
)

print(solution.steps[0].explanation)
print(f"답: {solution.final_answer}")

# Anthropic도 동일하게 지원
import anthropic

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

solution = anthropic_client.messages.create(
    model="claude-sonnet-4-20250514",
    response_model=MathSolution,
    max_tokens=1024,
    messages=[
        {"role": "user", "content": "3x - 7 = 20을 풀어줘"}
    ]
)

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

app = FastAPI()
client = instructor.from_openai(OpenAI())

class ProductReview(BaseModel):
    sentiment: str  # positive, negative, neutral
    score: float
    key_phrases: List[str]
    summary: str
    language: str

class ReviewRequest(BaseModel):
    text: str
    model: str = "gpt-4o-mini"

@app.post("/analyze", response_model=ProductReview)
async def analyze_review(request: ReviewRequest):
    try:
        result = client.chat.completions.create(
            model=request.model,
            response_model=ProductReview,
            messages=[
                {
                    "role": "system",
                    "content": "제품 리뷰를 분석하세요."
                },
                {"role": "user", "content": request.text}
            ],
            max_retries=2
        )
        return result
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

import asyncio
from typing import List
from openai import AsyncOpenAI
import instructor

async_client = instructor.from_openai(AsyncOpenAI())

class ExtractedEntity(BaseModel):
    name: str
    entity_type: str
    confidence: float

class EntityExtractionResult(BaseModel):
    entities: List[ExtractedEntity]
    text_length: int

async def extract_entities(text: str) -> EntityExtractionResult:
    return await async_client.chat.completions.create(
        model="gpt-4o-mini",
        response_model=EntityExtractionResult,
        messages=[
            {"role": "system", "content": "텍스트에서 엔티티를 추출하세요."},
            {"role": "user", "content": text}
        ]
    )

async def batch_extract(texts: List[str], concurrency: int = 5):
    """동시성 제한으로 배치 처리"""
    semaphore = asyncio.Semaphore(concurrency)

    async def limited_extract(text):
        async with semaphore:
            return await extract_entities(text)

    tasks = [limited_extract(text) for text in texts]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    successes = [r for r in results if not isinstance(r, Exception)]
    failures = [r for r in results if isinstance(r, Exception)]

    print(f"성공: {len(successes)}, 실패: {len(failures)}")
    return successes

# 실행
texts = ["서울에 새로운 AI 스타트업이...", "삼성전자가 반도체...", ...]
results = asyncio.run(batch_extract(texts))

# 같은 사고를 세 경로에서 각각 잡는 법

# Chat Completions
if completion.choices[0].finish_reason == "length":
    raise RuntimeError("출력이 잘렸다 — max_tokens를 올려야 한다")

refusal = completion.choices[0].message.refusal
if refusal:
    raise RuntimeError(f"모델이 거부했다: {refusal}")

# Responses API — finish_reason이 아예 없다
if response.status == "incomplete":
    if response.incomplete_details.reason == "max_output_tokens":
        raise RuntimeError("출력이 잘렸다 — max_output_tokens를 올려야 한다")

# Anthropic
if message.stop_reason == "max_tokens":
    raise RuntimeError("JSON이 잘렸다 — max_tokens를 올려야 한다")
if message.stop_reason == "refusal":
    raise RuntimeError("거부 응답 — 출력이 스키마와 맞지 않을 수 있다")

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

class TicketTriage(BaseModel):
    category: Literal["billing", "bug", "feature_request", "account", "other"]
    severity: Literal["p0", "p1", "p2", "p3"]
    summary: str = Field(description="한 문장 요약, 60자 이내")
    needs_human: bool
    suggested_owner: Optional[str] = None

from anthropic import Anthropic

client = Anthropic()

TICKET = """
결제가 두 번 청구됐습니다. 카드 명세서에 8월 3일자로 같은 금액이
두 건 찍혀 있어요. 주문번호는 A-91823입니다. 환불 부탁드립니다.
"""

response = client.messages.parse(
    model="claude-sonnet-5",
    max_tokens=2000,
    messages=[
        {"role": "user", "content": f"다음 지원 티켓을 분류하세요.\n\n{TICKET}"}
    ],
    output_format=TicketTriage,
)

if response.stop_reason == "max_tokens":
    raise RuntimeError("잘림 — max_tokens를 올려라")

triage = response.parsed_output
print(triage.category, triage.severity, triage.needs_human)

{
  "category": "billing",
  "severity": "p1",
  "summary": "8월 3일 동일 금액 이중 청구, 주문번호 A-91823, 환불 요청",
  "needs_human": true,
  "suggested_owner": "billing-ops"
}

from openai import OpenAI

client = OpenAI()

response = client.responses.parse(
    model="gpt-4o-mini",
    input=[{"role": "user", "content": f"다음 지원 티켓을 분류하세요.\n\n{TICKET}"}],
    text_format=TicketTriage,
)

if response.status == "incomplete":
    if response.incomplete_details.reason == "max_output_tokens":
        raise RuntimeError("잘림 — max_output_tokens를 올려라")

from litellm import completion, get_supported_openai_params, supports_response_schema

resp = completion(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "..."}],
    response_format=EventsList,
)

# 여기까지는 문자열이다. 검증은 내 몫.
events = EventsList.model_validate_json(resp.choices[0].message.content)