d01 - 03. Structed Output - 01. responese structed basic v2

01. Responses API Structured Outputs 기본 v2

이 노트북은 responses.parse와 Pydantic 스키마를 사용해 모델 응답을 바로 Python 객체로 받는 방법을 다룹니다. 자연어 문장을 사람이 다시 파싱하지 않아도 되므로 저장, 검증, 후속 자동화가 쉬워집니다.

# Colab 또는 로컬 노트북 실행 환경을 구분하기 위해 sys를 가져옵니다.
import sys
# 패키지 설치 명령을 현재 Python 커널에서 실행하기 위해 subprocess를 가져옵니다.
import subprocess
# 패키지 설치 여부를 확인하기 위해 importlib.util을 가져옵니다.
import importlib.util
# 환경 변수에서 API 키를 읽고 설정하기 위해 os를 가져옵니다.
import os
# API 키를 화면에 노출하지 않고 입력받기 위해 getpass를 가져옵니다.
import getpass
# .env 파일 위치를 다루기 위해 pathlib의 Path를 가져옵니다.
from pathlib import Path

# google.colab 모듈이 있으면 현재 런타임이 Google Colab이라고 판단합니다.
IN_COLAB = "google.colab" in sys.modules
# 노트북에서 사용하는 import 이름과 pip 패키지 이름을 짝지어 둡니다.
REQUIRED_PACKAGES = {
    "openai": "openai>=2.26.0",
    "dotenv": "python-dotenv>=1.2.2",
    "pydantic": "pydantic>=2.11.0",
    "PIL": "pillow>=12.1.1",
    "requests": "requests>=2.32.5",
    "numpy": "numpy>=2.3.3",
    "websockets": "websockets>=15.0.1",
    "websocket": "websocket-client>=1.8.0",
    "nest_asyncio": "nest-asyncio>=1.6.0",
}

# 현재 커널에서 import할 수 없는 패키지만 설치하는 함수입니다.
def ensure_package(import_name: str, package_name: str) -> None:
    # 이미 import 가능한 패키지는 설치를 건너뜁니다.
    if importlib.util.find_spec(import_name) is not None:
        # 설치가 필요 없음을 호출자에게 조용히 알리고 돌아갑니다.
        return
    # Colab에서는 현재 노트북 커널의 Python에 패키지를 설치해야 합니다.
    subprocess.check_call([sys.executable, "-m", "pip", "install", "-q", package_name])

# Colab 기본 런타임에는 일부 OpenAI 실습 패키지가 없을 수 있으므로 먼저 준비합니다.
if IN_COLAB:
    # 실습 전체에서 필요한 패키지를 하나씩 확인하고 부족한 것만 설치합니다.
    for import_name, package_name in REQUIRED_PACKAGES.items():
        # 누락된 패키지를 현재 Colab 런타임에 설치합니다.
        ensure_package(import_name, package_name)

# 패키지 설치 이후 .env 로딩 기능을 사용할 수 있도록 load_dotenv를 가져옵니다.
from dotenv import load_dotenv

# 현재 작업 폴더의 .env 파일이 있으면 로컬 실행처럼 환경 변수로 올립니다.
load_dotenv(Path.cwd() / ".env")
# OPENAI_API_KEY가 이미 있으면 그대로 사용하고, 없으면 Colab Secrets 또는 입력으로 보완합니다.
if not os.getenv("OPENAI_API_KEY"):
    # Colab Secrets에서 OPENAI_API_KEY를 읽어 볼 변수를 준비합니다.
    secret_key = None
    # Colab이 아닌 로컬 환경에서는 google.colab import가 실패할 수 있으므로 예외를 허용합니다.
    try:
        # Colab Secrets의 userdata API를 가져옵니다.
        from google.colab import userdata
        # Secrets에 저장된 OPENAI_API_KEY 값을 읽습니다.
        secret_key = userdata.get("OPENAI_API_KEY")
    except Exception:
        # Colab Secrets를 사용할 수 없으면 이후 수동 입력 단계로 넘어갑니다.
        secret_key = None
    # Secrets에서 키를 찾았다면 현재 런타임 환경 변수로 설정합니다.
    if secret_key:
        # OpenAI SDK가 자동으로 읽을 수 있도록 표준 환경 변수 이름에 저장합니다.
        os.environ["OPENAI_API_KEY"] = secret_key
    else:
        # 키가 없으면 노트북 실행자가 직접 입력하도록 요청합니다.
        entered_key = getpass.getpass("OPENAI_API_KEY를 입력하세요: ").strip()
        # 빈 문자열이 아닌 값을 입력한 경우에만 환경 변수로 등록합니다.
        if entered_key:
            # OpenAI SDK가 자동으로 읽을 수 있도록 표준 환경 변수 이름에 저장합니다.
            os.environ["OPENAI_API_KEY"] = entered_key

# API 키가 끝까지 없으면 다음 OpenAI API 호출이 실패하므로 명확한 오류를 냅니다.
if not os.getenv("OPENAI_API_KEY"):
    # Colab에서는 Secrets 또는 입력, 로컬에서는 .env 또는 환경 변수를 설정해야 함을 알려 줍니다.
    raise RuntimeError("OPENAI_API_KEY가 없습니다. Colab Secrets, 수동 입력, 또는 .env 파일로 설정해 주세요.")

# 이후 셀에서 Colab 여부를 참고할 수 있도록 간단히 출력합니다.
print(f"개발환경: {'Colab' if IN_COLAB else '로컬'}")
# API 키를 직접 출력하지 않고 준비 완료 여부만 알려 줍니다.
print("OPENAI_API_KEY 준비 완료")

 

1단계. 실행 환경 준비

구조화 출력은 API 응답뿐 아니라 Pydantic 스키마, JSON 저장, 필드 검증이 함께 필요합니다. 먼저 실행 환경을 준비해 이후 단계가 같은 경로와 같은 모델 설정을 사용하게 합니다.

# 환경 변수와 모델명을 읽기 위해 os를 가져옵니다.
import os
# 구조화 결과와 메타데이터를 JSON으로 저장하기 위해 json을 가져옵니다.
import json
# API 호출 시간을 측정하기 위해 time을 가져옵니다.
import time
# 파일 경로를 안전하게 다루기 위해 Path를 가져옵니다.
from pathlib import Path
# 일부 함수와 응답 객체의 느슨한 타입을 표현하기 위해 Any를 가져옵니다.
from typing import Any, Literal
# .env 파일에서 OPENAI_API_KEY를 로드하기 위해 load_dotenv를 가져옵니다.
from dotenv import load_dotenv
# OpenAI API를 호출하는 공식 SDK 클라이언트입니다.
from openai import OpenAI
# Structured Outputs에서 Python 클래스로 스키마를 정의하기 위해 BaseModel과 Field를 가져옵니다.
from pydantic import BaseModel, Field

# 현재 실행 위치를 확인합니다.
CURRENT_DIR = Path.cwd()
# 루트에서 실행 중이면 03.structured_output 폴더를 선택합니다.
if (CURRENT_DIR / "03.structured_output").exists():
    CURRICULUM_ROOT = CURRENT_DIR / "03.structured_output"
# 실습 폴더 안에서 실행 중이면 현재 위치를 사용합니다.
elif CURRENT_DIR.name == "03.structured_output":
    CURRICULUM_ROOT = CURRENT_DIR
# 그 밖의 경우 상대 경로를 해석합니다.
else:
    CURRICULUM_ROOT = Path("03.structured_output").resolve()
# 워크스페이스 루트는 실습 폴더의 상위 폴더입니다.
WORKSPACE_ROOT = CURRICULUM_ROOT.parent
# 결과 저장 폴더 경로입니다.
OUTPUT_DIR = CURRICULUM_ROOT / "output"
# 결과 저장 폴더가 없으면 생성합니다.
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
# .env 파일에서 환경 변수를 읽습니다.
load_dotenv(WORKSPACE_ROOT / ".env")
# API 키가 없으면 명확한 메시지로 중단합니다.
if not os.getenv("OPENAI_API_KEY"):
    raise RuntimeError("OPENAI_API_KEY가 없습니다. 워크스페이스 루트의 .env 파일을 확인해 주세요.")
# OpenAI 클라이언트를 생성합니다.
client = OpenAI()
# Structured Outputs 실습에 사용할 모델입니다.
STRUCTURED_MODEL = os.getenv("OPENAI_STRUCTURED_MODEL", "gpt-5-mini")
# 필요할 때 사용할 보조 모델명입니다.
FALLBACK_MODEL = os.getenv("OPENAI_STRUCTURED_FALLBACK_MODEL", "gpt-4.1-mini")

# 워크스페이스 기준 상대 경로를 반환합니다.
def workspace_path(path: Path) -> str:
    # Windows에서도 읽기 쉬운 / 구분자 문자열로 변환합니다.
    return path.resolve().relative_to(WORKSPACE_ROOT.resolve()).as_posix()

# Pydantic 객체, dict, list를 JSON 저장 가능한 형태로 바꿉니다.
def normalize_data(value: Any) -> Any:
    # Pydantic 모델은 model_dump()로 dict가 됩니다.
    if hasattr(value, "model_dump"):
        return value.model_dump()
    # dict는 값들을 재귀적으로 변환합니다.
    if isinstance(value, dict):
        return {key: normalize_data(item) for key, item in value.items()}
    # list는 각 원소를 재귀적으로 변환합니다.
    if isinstance(value, list):
        return [normalize_data(item) for item in value]
    # 그 밖의 값은 그대로 반환합니다.
    return value

# usage 객체를 dict로 변환합니다.
def usage_to_dict(usage: Any) -> dict[str, Any]:
    # usage가 없으면 빈 dict를 반환합니다.
    if usage is None:
        return {}
    # Pydantic 객체면 model_dump()로 변환합니다.
    if hasattr(usage, "model_dump"):
        return usage.model_dump()
    # 이미 dict면 그대로 반환합니다.
    if isinstance(usage, dict):
        return usage
    # 예상 밖 형태는 빈 dict로 처리합니다.
    return {}

# 긴 텍스트를 한 줄 미리보기로 줄입니다.
def preview_text(value: str, limit: int = 220) -> str:
    # 공백과 줄바꿈을 정리합니다.
    compact = " ".join(str(value).split())
    # 길이가 짧으면 그대로 반환합니다.
    if len(compact) <= limit:
        return compact
    # 길면 말줄임표와 함께 앞부분만 반환합니다.
    return compact[: limit - 3] + "..."

# JSON 텍스트를 안전하게 파싱합니다.
def parse_json_text(raw_text: str) -> tuple[Any, str | None]:
    # 문자열 앞뒤 공백을 제거합니다.
    text = str(raw_text or "").strip()
    # 비어 있으면 오류 메시지를 반환합니다.
    if not text:
        return None, "빈 문자열입니다."
    # Markdown 코드블록으로 감싸진 JSON도 처리합니다.
    if text.startswith("```"):
        lines = text.splitlines()
        text = "\n".join(lines[1:-1]).strip() if len(lines) >= 3 else text
    # json.loads로 파싱을 시도합니다.
    try:
        return json.loads(text), None
    # 실패하면 예외를 밖으로 던지지 않고 메시지로 반환합니다.
    except Exception as exc:
        return None, str(exc)

# Responses parse 응답에서 구조화 파싱 결과를 찾습니다.
def extract_parsed_response(response: Any) -> Any:
    # SDK 버전에 따라 최상위 output_parsed에 파싱 결과가 들어올 수 있습니다.
    direct = getattr(response, "output_parsed", None)
    # 최상위 값이 있으면 그대로 반환합니다.
    if direct is not None:
        return direct
    # 최상위에 없으면 output 배열의 content block을 확인합니다.
    for item in getattr(response, "output", []) or []:
        # content 속성을 안전하게 읽습니다.
        content = getattr(item, "content", None)
        # content가 list가 아니면 다음 항목으로 넘어갑니다.
        if not isinstance(content, list):
            continue
        # content block을 하나씩 확인합니다.
        for block in content:
            # block 안의 parsed 속성을 읽습니다.
            parsed = getattr(block, "parsed", None)
            # parsed 값이 있으면 반환합니다.
            if parsed is not None:
                return parsed
    # 어느 위치에서도 찾지 못하면 None을 반환합니다.
    return None

# 필수 필드 누락 여부를 확인합니다.
def missing_fields(payload: dict[str, Any], required: list[str]) -> list[str]:
    # 누락된 필드명을 담을 리스트입니다.
    missing: list[str] = []
    # 필수 필드를 하나씩 검사합니다.
    for key in required:
        # 키가 없으면 누락으로 기록합니다.
        if key not in payload:
            missing.append(key)
            continue
        # 값이 None이면 누락으로 봅니다.
        if payload[key] is None:
            missing.append(key)
        # 문자열이 공백뿐이면 누락으로 봅니다.
        elif isinstance(payload[key], str) and not payload[key].strip():
            missing.append(key)
        # 리스트나 dict가 비어 있으면 누락으로 봅니다.
        elif isinstance(payload[key], (list, dict)) and not payload[key]:
            missing.append(key)
    # 누락된 필드명 목록을 반환합니다.
    return missing

# 실행 위치와 모델 정보를 출력합니다.
print(f"실습 폴더: {CURRICULUM_ROOT}")
# 저장 폴더를 출력합니다.
print(f"결과 저장 폴더: {OUTPUT_DIR}")
# 요청 모델을 출력합니다.
print(f"요청 모델: {STRUCTURED_MODEL}")

실습 폴더: c:\Users\ai4nu\2026-ai-tech\AS.1.1_0601\03.structured_output
결과 저장 폴더: c:\Users\ai4nu\2026-ai-tech\AS.1.1_0601\03.structured_output\output
요청 모델: gpt-5-mini

 

2단계. Pydantic 스키마 정의

모델에게 원하는 응답 형태를 말로만 설명하면 필드명이 흔들릴 수 있습니다. Pydantic 클래스를 스키마로 전달하면 어떤 필드가 필요한지 코드 수준에서 명확해집니다.

# 교육 공지 구조를 표현하는 Pydantic 모델입니다.
class TrainingNotice(BaseModel):
    # 공지 제목을 저장하는 문자열 필드입니다.
    title: str = Field(description="공지 제목")
    # 교육 대상자를 저장하는 문자열 필드입니다.
    target_audience: str = Field(description="공지 대상")
    # 교육 일정을 저장하는 문자열 필드입니다.
    schedule: str = Field(description="교육 일정")
    # 준비물을 여러 개 담는 문자열 리스트 필드입니다.
    required_items: list[str] = Field(description="준비물 목록")
    # 문의처를 저장하는 문자열 필드입니다.
    contact: str = Field(description="문의처")
    # 전체 공지를 요약하는 문자열 필드입니다.
    summary: str = Field(description="공지 요약")

# 구조화할 작업의 모델 역할 지시문입니다.
INSTRUCTIONS = "당신은 사내 교육 공지 작성 도우미입니다. 사용자 요청을 반드시 구조화된 JSON으로 정리하고, 모호한 표현은 안전한 기본값으로 보완하세요."
# 사용자가 모델에게 요청하는 실제 입력입니다.
PROMPT = "신입 개발자 온보딩 교육 안내를 작성해 주세요. 교육 제목, 대상, 일정, 준비물, 문의처, 요약을 구조화해서 알려주세요."

 

3단계. client.responses.parse(...) 직접 호출

이 단계가 핵심입니다. text_format=TrainingNotice를 전달하면 SDK가 모델 응답을 스키마에 맞춰 파싱하고, 결과를 response.output_parsed에 담아 줍니다.

# API 호출 시작 시간을 기록합니다.
started = time.perf_counter()
# Responses API의 parse 메서드를 직접 호출합니다.
response = client.responses.parse(
    # 구조화 응답을 생성할 모델입니다.
    model=STRUCTURED_MODEL,
    # 모델의 역할과 출력 규칙을 지정합니다.
    instructions=INSTRUCTIONS,
    # 사용자 요청을 메시지 배열 형태로 전달합니다.
    input=[{"role": "user", "content": PROMPT}],
    # Pydantic 모델을 구조화 출력 형식으로 전달합니다.
    text_format=TrainingNotice,
    # 출력이 잘리지 않도록 최대 출력 토큰을 지정합니다.
    max_output_tokens=900,
    reasoning={"effort": "low"} if STRUCTURED_MODEL.lower().startswith("gpt-5") else None,
)
# API 호출 소요 시간을 계산합니다.
elapsed = time.perf_counter() - started
# output_parsed는 TrainingNotice 객체로 파싱된 결과입니다.
parsed_notice = extract_parsed_response(response)
# JSON 저장을 위해 Pydantic 객체를 dict로 바꿉니다.
parsed = normalize_data(parsed_notice)
# 파싱 결과가 dict인지 확인합니다.
if not isinstance(parsed, dict):
    raise RuntimeError("구조화 응답 파싱에 실패했습니다.")
# 파싱된 구조를 출력합니다.
print(json.dumps(parsed, ensure_ascii=False, indent=2))
# 실제 응답 모델을 출력합니다.
print(f"실제 모델: {response.model}")
# 호출 시간을 출력합니다.
print(f"실행 시간: {elapsed:.4f}초")

{
  "title": "신입 개발자 온보딩 교육",
  "target_audience": "신입 개발자 (입사 3개월 이내, 개발 직군)",
  "schedule": "2026-06-10 09:30–17:30 (1일) — 장소: 본사 교육실(회의실 A) 및 온라인 병행(Zoom). 필요 시 추가 기술 세션은 추후 일정 조정",
  "required_items": [
    "개인 노트북(회사 지급 노트북이 없는 경우)",
    "사원증 또는 신분증",
    "회사 이메일 계정 및 GitHub 계정(사전 생성 권장)",
    "충전기 및 필기도구",
    "사전 배포된 사전 학습 자료(이메일로 송부 예정)"
  ],
  "contact": "인사팀 담당자: 김민수 / 이메일: hr@example.com / 내선: 1234 / 개발 온보딩 담당: 박지은(Dev Lead) / 이메일: dev-onboard@example.com",
  "summary": "회사 소개 및 조직 문화, 개발 프로세스(Git 워크플로우, 코드리뷰), 개발 환경 설정(로컬 환경, CI/CD 접근), 주요 코드베이스 및 아키텍처 개요, 보안·컴플라이언스 기본 교육, 실습 과제(간단한 기능 구현 및 PR 제출), 멘토 매칭 및 Q&A 시간. 교육 후 체크리스트 기반 온보딩 진행과 1:1 멘토링이 제공됩니다."
}
실제 모델: gpt-5-mini-2025-08-07
실행 시간: 10.6269초

 

4단계. 사람이 읽는 텍스트와 JSON 저장

구조화 결과는 프로그램이 쓰기 좋은 JSON으로 저장하고, 교육생이 빠르게 읽을 수 있는 텍스트도 함께 만듭니다. 두 형태를 비교하면 Structured Outputs의 장점이 더 잘 보입니다.

# 사람이 읽기 쉬운 텍스트를 줄 단위로 구성합니다.
human_text = "\n".join([f"제목: {parsed.get('title', '')}", f"대상: {parsed.get('target_audience', '')}", f"일정: {parsed.get('schedule', '')}", f"준비물: {', '.join(parsed.get('required_items', []))}", f"문의처: {parsed.get('contact', '')}", f"요약: {parsed.get('summary', '')}"]) + "\n"
# 텍스트 결과 파일 경로입니다.
output_text_path = OUTPUT_DIR / "01_basic_v2_final_output.txt"
# 구조화 JSON 파일 경로입니다.
structured_path = OUTPUT_DIR / "01_basic_v2_structured.json"
# 메타데이터 파일 경로입니다.
metadata_path = OUTPUT_DIR / "01_basic_v2_meta.json"
# 텍스트 결과를 저장합니다.
output_text_path.write_text(human_text, encoding="utf-8")
# 구조화 결과를 JSON으로 저장합니다.
structured_path.write_text(json.dumps(parsed, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
# usage 객체를 dict로 변환합니다.
usage = usage_to_dict(response.usage)
# 실행 메타데이터를 만듭니다.
metadata = {"example": "01.responses_structured_basic_v2", "requested_model": STRUCTURED_MODEL, "actual_model": response.model, "schema_name": "TrainingNotice", "parse_success": True, "elapsed_seconds": round(elapsed, 4), "input_tokens": usage.get("input_tokens"), "output_tokens": usage.get("output_tokens"), "total_tokens": usage.get("total_tokens"), "output_path": workspace_path(output_text_path), "structured_path": workspace_path(structured_path)}
# 메타데이터를 저장합니다.
metadata_path.write_text(json.dumps(metadata, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
# 저장 위치를 출력합니다.
print(f"텍스트 저장: {workspace_path(output_text_path)}")
# 구조화 JSON 저장 위치를 출력합니다.
print(f"구조화 JSON 저장: {workspace_path(structured_path)}")
# 메타데이터 저장 위치를 출력합니다.
print(f"메타데이터 저장: {workspace_path(metadata_path)}")

텍스트 저장: 03.structured_output/output/01_basic_v2_final_output.txt
구조화 JSON 저장: 03.structured_output/output/01_basic_v2_structured.json
메타데이터 저장: 03.structured_output/output/01_basic_v2_meta.json