LLM 구조화된 출력과 Constrained Decoding 실전 가이드: JSON Schema부터 프로덕션 적용까지

import torch
import numpy as np

def apply_grammar_mask(logits: torch.Tensor, allowed_token_ids: set, vocab_size: int) -> torch.Tensor:
    """
    Constrained Decoding의 핵심: 문법적으로 허용된 토큰만 남기고
    나머지 토큰의 logit을 -inf로 마스킹한다.
    """
    mask = torch.full((vocab_size,), float('-inf'), dtype=logits.dtype, device=logits.device)
    allowed_ids = torch.tensor(list(allowed_token_ids), dtype=torch.long, device=logits.device)
    mask[allowed_ids] = 0.0
    masked_logits = logits + mask
    return masked_logits

# 예시: JSON 객체 시작 직후, 키 문자열만 허용
vocab_size = 32000
logits = torch.randn(vocab_size)  # 모델이 출력한 raw logits

# 현재 FSM 상태에서 허용되는 토큰: 쌍따옴표(")로 시작하는 키
allowed_tokens = {345, 678, 1234}  # tokenizer에서 '"name', '"age' 등에 해당하는 ID
masked = apply_grammar_mask(logits, allowed_tokens, vocab_size)

# softmax 적용 후 허용된 토큰만 양의 확률을 가짐
probs = torch.softmax(masked, dim=-1)
assert probs[0].item() == 0.0  # 허용되지 않은 토큰의 확률은 0

from openai import OpenAI
from pydantic import BaseModel, Field
from typing import List, Optional
from enum import Enum

client = OpenAI()

# Pydantic 모델로 출력 스키마 정의
class Severity(str, Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

class SecurityFinding(BaseModel):
    vulnerability_type: str = Field(description="취약점 유형 (예: SQL Injection, XSS)")
    severity: Severity = Field(description="심각도")
    affected_component: str = Field(description="영향받는 컴포넌트")
    description: str = Field(description="취약점 상세 설명")
    remediation: str = Field(description="권장 조치")
    cvss_score: Optional[float] = Field(default=None, ge=0.0, le=10.0, description="CVSS 점수")

class SecurityReport(BaseModel):
    findings: List[SecurityFinding] = Field(description="발견된 취약점 목록")
    overall_risk: Severity = Field(description="전체 위험 수준")
    summary: str = Field(description="보안 분석 요약")
    scan_timestamp: str = Field(description="스캔 시각 (ISO 8601)")

# Structured Outputs로 호출 - 스키마 100% 보장
response = client.beta.chat.completions.parse(
    model="gpt-4o-2024-08-06",
    messages=[
        {
            "role": "system",
            "content": "당신은 보안 분석 전문가입니다. 주어진 코드를 분석하여 보안 취약점을 보고하세요."
        },
        {
            "role": "user",
            "content": """다음 Flask 코드의 보안 취약점을 분석해주세요:

@app.route('/search')
def search():
    query = request.args.get('q')
    result = db.execute(f"SELECT * FROM users WHERE name = '{query}'")
    return render_template_string(f"<h1>Results for {query}</h1>")
"""
        }
    ],
    response_format=SecurityReport
)

report = response.choices[0].message.parsed
print(f"전체 위험 수준: {report.overall_risk.value}")
for finding in report.findings:
    print(f"  [{finding.severity.value}] {finding.vulnerability_type}: {finding.description}")

import anthropic
import json

client = anthropic.Anthropic()

# Tool Use를 활용한 구조화된 출력
response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=2048,
    tools=[
        {
            "name": "analyze_code_security",
            "description": "코드의 보안 취약점을 구조화된 형식으로 분석합니다",
            "input_schema": {
                "type": "object",
                "properties": {
                    "findings": {
                        "type": "array",
                        "items": {
                            "type": "object",
                            "properties": {
                                "vulnerability_type": {"type": "string"},
                                "severity": {
                                    "type": "string",
                                    "enum": ["low", "medium", "high", "critical"]
                                },
                                "affected_component": {"type": "string"},
                                "description": {"type": "string"},
                                "remediation": {"type": "string"}
                            },
                            "required": ["vulnerability_type", "severity", "description", "remediation"]
                        }
                    },
                    "overall_risk": {
                        "type": "string",
                        "enum": ["low", "medium", "high", "critical"]
                    },
                    "summary": {"type": "string"}
                },
                "required": ["findings", "overall_risk", "summary"]
            }
        }
    ],
    tool_choice={"type": "tool", "name": "analyze_code_security"},
    messages=[
        {
            "role": "user",
            "content": "다음 코드의 보안 취약점을 분석해주세요:\n\n@app.route('/login', methods=['POST'])\ndef login():\n    username = request.form['username']\n    password = request.form['password']\n    query = f\"SELECT * FROM users WHERE username='{username}' AND password='{password}'\"\n    user = db.execute(query).fetchone()\n    if user:\n        session['user'] = username\n        return redirect('/dashboard')"
        }
    ]
)

# Tool Use 결과에서 구조화된 데이터 추출
tool_use_block = next(block for block in response.content if block.type == "tool_use")
report = tool_use_block.input
print(f"전체 위험: {report['overall_risk']}")
print(f"발견 항목: {len(report['findings'])}건")

from vllm import LLM, SamplingParams
from pydantic import BaseModel, Field
from typing import List, Optional
import json

# Pydantic 모델로 출력 스키마 정의
class ExtractedEntity(BaseModel):
    name: str = Field(description="엔티티 이름")
    entity_type: str = Field(description="엔티티 유형: PERSON, ORG, LOCATION, DATE, PRODUCT")
    confidence: float = Field(ge=0.0, le=1.0, description="신뢰도")
    context: Optional[str] = Field(default=None, description="원문에서의 관련 문맥")

class EntityExtractionResult(BaseModel):
    entities: List[ExtractedEntity] = Field(description="추출된 엔티티 목록")
    source_language: str = Field(description="원문 언어")

# vLLM 엔진 초기화
llm = LLM(
    model="meta-llama/Llama-3.1-8B-Instruct",
    max_model_len=4096,
    gpu_memory_utilization=0.85,
    guided_decoding_backend="xgrammar",  # 또는 "outlines"
)

# JSON Schema를 가이드로 사용하는 SamplingParams
sampling_params = SamplingParams(
    temperature=0.1,
    top_p=0.95,
    max_tokens=1024,
    guided_decoding={
        "json_object": EntityExtractionResult.model_json_schema()
    }
)

# 배치 추론
texts = [
    "삼성전자 이재용 회장이 2026년 3월 서울에서 AI 반도체 전략을 발표했다.",
    "Apple CEO Tim Cook announced the new M5 chip at WWDC 2026 in Cupertino.",
    "소프트뱅크 손정의 회장이 도쿄에서 ARM 기반 AI 인프라 투자를 발표했다.",
]

prompts = [
    f"다음 텍스트에서 엔티티(사람, 조직, 장소, 날짜, 제품)를 추출하세요:\n\n{text}"
    for text in texts
]

outputs = llm.generate(prompts, sampling_params)

for output in outputs:
    result = json.loads(output.outputs[0].text)
    parsed = EntityExtractionResult(**result)
    print(f"언어: {parsed.source_language}")
    for entity in parsed.entities:
        print(f"  [{entity.entity_type}] {entity.name} (신뢰도: {entity.confidence})")
    print("---")

# vLLM 서버 실행 (guided decoding 활성화)
vllm serve meta-llama/Llama-3.1-8B-Instruct \
    --guided-decoding-backend xgrammar \
    --max-model-len 4096 \
    --port 8000

from openai import OpenAI

# vLLM 서버에 연결
client = OpenAI(base_url="http://localhost:8000/v1", api_key="dummy")

# extra_body로 guided decoding 전달
response = client.chat.completions.create(
    model="meta-llama/Llama-3.1-8B-Instruct",
    messages=[
        {"role": "system", "content": "텍스트에서 엔티티를 추출하세요."},
        {"role": "user", "content": "Google의 Sundar Pichai CEO가 Mountain View에서 Gemini 3.0을 공개했다."}
    ],
    extra_body={
        "guided_json": {
            "type": "object",
            "properties": {
                "entities": {
                    "type": "array",
                    "items": {
                        "type": "object",
                        "properties": {
                            "name": {"type": "string"},
                            "type": {"type": "string", "enum": ["PERSON", "ORG", "LOCATION", "PRODUCT"]},
                            "confidence": {"type": "number"}
                        },
                        "required": ["name", "type", "confidence"]
                    }
                }
            },
            "required": ["entities"]
        }
    }
)

import json
result = json.loads(response.choices[0].message.content)
print(json.dumps(result, indent=2, ensure_ascii=False))

# TGI 서버 실행
docker run --gpus all -p 8080:80 \
    ghcr.io/huggingface/text-generation-inference:latest \
    --model-id meta-llama/Llama-3.1-8B-Instruct \
    --max-input-tokens 2048 \
    --max-total-tokens 4096

import requests
import json

# TGI의 grammar 파라미터로 JSON Schema 전달
response = requests.post(
    "http://localhost:8080/generate",
    json={
        "inputs": "서울의 유명 관광지 3곳을 추천해주세요.",
        "parameters": {
            "max_new_tokens": 512,
            "temperature": 0.7,
            "grammar": {
                "type": "json",
                "value": {
                    "type": "object",
                    "properties": {
                        "places": {
                            "type": "array",
                            "items": {
                                "type": "object",
                                "properties": {
                                    "name": {"type": "string"},
                                    "category": {"type": "string"},
                                    "description": {"type": "string"}
                                },
                                "required": ["name", "category", "description"]
                            },
                            "minItems": 3,
                            "maxItems": 3
                        }
                    },
                    "required": ["places"]
                }
            }
        }
    }
)

result = json.loads(response.json()["generated_text"])
for place in result["places"]:
    print(f"{place['name']} ({place['category']}): {place['description']}")

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from pydantic import BaseModel, Field
from typing import List, Literal

# 출력 스키마 정의
class DatabaseQuery(BaseModel):
    """자연어를 SQL 쿼리로 변환한 결과"""
    sql: str = Field(description="생성된 SQL 쿼리")
    tables_used: List[str] = Field(description="사용된 테이블 목록")
    query_type: Literal["SELECT", "INSERT", "UPDATE", "DELETE"] = Field(description="쿼리 타입")
    explanation: str = Field(description="쿼리 설명")
    estimated_complexity: Literal["low", "medium", "high"] = Field(description="쿼리 복잡도")

# OpenAI 모델에 구조화된 출력 적용
openai_llm = ChatOpenAI(model="gpt-4o", temperature=0)
openai_structured = openai_llm.with_structured_output(DatabaseQuery)

# Anthropic 모델에도 동일하게 적용
anthropic_llm = ChatAnthropic(model="claude-sonnet-4-20250514", temperature=0)
anthropic_structured = anthropic_llm.with_structured_output(DatabaseQuery)

# 동일한 인터페이스로 호출
question = "지난 30일 동안 가장 매출이 높은 상위 10개 제품과 해당 카테고리를 보여줘"

result_openai = openai_structured.invoke(question)
result_anthropic = anthropic_structured.invoke(question)

print(f"[OpenAI] SQL: {result_openai.sql}")
print(f"[OpenAI] 테이블: {result_openai.tables_used}")
print(f"[OpenAI] 복잡도: {result_openai.estimated_complexity}")
print()
print(f"[Anthropic] SQL: {result_anthropic.sql}")
print(f"[Anthropic] 테이블: {result_anthropic.tables_used}")
print(f"[Anthropic] 복잡도: {result_anthropic.estimated_complexity}")

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

client = OpenAI()

# 1단계: Function Calling으로 정보 수집
tools = [
    {
        "type": "function",
        "function": {
            "name": "search_database",
            "description": "데이터베이스에서 제품 정보를 검색합니다",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "검색 쿼리"},
                    "category": {"type": "string", "enum": ["electronics", "clothing", "food"]},
                    "limit": {"type": "integer", "default": 10}
                },
                "required": ["query"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_price_history",
            "description": "제품의 가격 이력을 조회합니다",
            "parameters": {
                "type": "object",
                "properties": {
                    "product_id": {"type": "string"},
                    "days": {"type": "integer", "default": 30}
                },
                "required": ["product_id"]
            }
        }
    }
]

# 2단계: 최종 응답은 Structured Output으로 반환
class ProductAnalysis(BaseModel):
    product_name: str
    current_price: float
    price_trend: str  # "rising", "falling", "stable"
    recommendation: str
    confidence: float = Field(ge=0.0, le=1.0)

class AnalysisReport(BaseModel):
    analyses: List[ProductAnalysis]
    market_summary: str
    generated_at: str

# 멀티턴 대화로 Function Calling 후 Structured Output
messages = [
    {"role": "system", "content": "제품 시장 분석 전문가입니다. 도구를 사용하여 정보를 수집하고 분석 보고서를 작성하세요."},
    {"role": "user", "content": "최신 노트북 시장 동향을 분석해주세요."}
]

# Function Calling 단계
response = client.chat.completions.create(
    model="gpt-4o",
    messages=messages,
    tools=tools,
    tool_choice="auto"
)

# 도구 호출 결과를 처리한 후 최종 Structured Output 요청
# (도구 실행 및 결과 피드백 로직 생략)

# 최종 분석 보고서를 Structured Output으로 생성
final_response = client.beta.chat.completions.parse(
    model="gpt-4o-2024-08-06",
    messages=messages,  # 전체 대화 이력 포함
    response_format=AnalysisReport
)

report = final_response.choices[0].message.parsed
print(f"시장 요약: {report.market_summary}")
for analysis in report.analyses:
    print(f"  {analysis.product_name}: {analysis.price_trend} (신뢰도: {analysis.confidence})")

1차: Constrained Decoding (100% 스키마 보장)
  ↓ 실패 시
2차: JSON Mode + Pydantic 검증
  ↓ 실패 시
3차: 자유 텍스트 + LLM 기반 재파싱
  ↓ 실패 시
4차: 기본값 반환 + 알림

response = client.beta.chat.completions.parse(
    model="gpt-4o-2024-08-06",
    messages=messages,
    response_format=MySchema
)

if response.choices[0].message.refusal:
    print(f"응답 거부: {response.choices[0].message.refusal}")
elif response.choices[0].message.parsed:
    result = response.choices[0].message.parsed
    # 정상 처리