d01 - 03. Structed Output - 03. nested array extraction v2

03. 중첩 객체와 배열 추출 v2

이 노트북은 긴 교육 계획 메모에서 팀, 세션, 성공 지표처럼 계층이 있는 데이터를 추출합니다. 중첩 구조를 다룰 수 있어야 실제 업무 문서를 자동화에 연결할 수 있습니다.

# 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 키, 모델명, 저장 폴더를 먼저 준비합니다. 구조화 출력은 결과 저장과 검증까지 이어지므로 경로와 환경을 초기에 확정하는 것이 중요합니다.

# 환경 변수와 모델명을 읽기 위해 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 클래스로 나누면 복잡한 문서도 안정적인 구조로 받을 수 있습니다.

# 하나의 교육 세션을 표현하는 하위 스키마입니다.
class SessionInfo(BaseModel):
    # 세션 주제를 저장합니다.
    topic: str = Field(description="세션 주제")
    # 세션 시간을 시간 단위 숫자로 저장합니다.
    duration_hours: float = Field(description="세션 시간")
    # 담당 강사 이름을 저장합니다.
    instructor: str = Field(description="담당 강사")
    # 필수 참석 여부를 저장합니다.
    mandatory: bool = Field(description="필수 여부")

# 한 팀의 교육 계획을 표현하는 중간 스키마입니다.
class TeamPlan(BaseModel):
    # 팀 이름을 저장합니다.
    team_name: str = Field(description="팀 이름")
    # 분기 교육 목표를 저장합니다.
    quarterly_goal: str = Field(description="분기 목표")
    # 팀의 세션 목록을 저장합니다.
    sessions: list[SessionInfo] = Field(description="세션 목록")
    # 교육 성공 지표 목록을 저장합니다.
    success_metrics: list[str] = Field(description="성공 지표")

# 전체 교육 로드맵을 표현하는 최상위 스키마입니다.
class TrainingRoadmap(BaseModel):
    # 대상 분기를 저장합니다.
    quarter: str = Field(description="대상 분기")
    # 팀별 계획 목록을 저장합니다.
    teams: list[TeamPlan] = Field(description="팀별 계획")
    # 공통 준비사항 목록을 저장합니다.
    shared_requirements: list[str] = Field(description="공통 준비사항")
    # 리스크와 주의사항 목록을 저장합니다.
    risk_notes: list[str] = Field(description="리스크 및 주의사항")

# 모델의 역할 지시문입니다.
INSTRUCTIONS = "당신은 교육 운영 PM입니다. 사용자 문장에서 팀별 교육 계획을 추출하고, 반드시 지정된 구조를 준수하세요."
# 중첩 구조로 추출할 원문입니다.
PROMPT = "2026년 2분기 교육 계획 메모입니다. 백엔드팀은 장애 대응 훈련(3시간, 김선우, 필수)과 성능 튜닝 워크숍(2시간, 박하늘, 선택)을 진행합니다. 성공 지표는 평균 복구 시간 20% 단축, 장애 재발률 15% 감소입니다. 프론트엔드팀은 접근성 개선 세션(2시간, 이도윤, 필수)과 디자인 시스템 실습(2시간, 정시온, 필수)을 진행합니다. 성공 지표는 접근성 점수 90점 이상, 컴포넌트 재사용률 30% 증가입니다. 공통 준비사항은 사전 과제 제출, 실습 계정 준비, 교육 후 회고 작성입니다. 주의사항은 일정 변경 가능성과 강사 일정 충돌입니다."

 

 

3단계. Responses API 직접 호출

text_format=TrainingRoadmap을 전달해 최상위 스키마뿐 아니라 내부 리스트와 하위 객체까지 함께 파싱합니다.

# API 호출 시작 시간을 기록합니다.
started = time.perf_counter()
# Pydantic 중첩 스키마를 text_format으로 전달해 parse를 호출합니다.
response = client.responses.parse(
    # 구조화 추출에 사용할 모델입니다.
    model=STRUCTURED_MODEL,
    # 모델의 역할과 추출 규칙입니다.
    instructions=INSTRUCTIONS,
    # 중첩 구조로 추출할 원문입니다.
    input=[{"role": "user", "content": PROMPT}],
    # 최상위 Pydantic 스키마입니다.
    text_format=TrainingRoadmap,
    # 복잡한 구조가 잘리지 않도록 토큰 한도를 넉넉히 둡니다.
    max_output_tokens=1000,
    reasoning={"effort": "low"} if STRUCTURED_MODEL.lower().startswith("gpt-5") else None,
)
# 호출 소요 시간을 계산합니다.
elapsed = time.perf_counter() - started
# 파싱된 Pydantic 객체를 JSON 저장 가능한 dict로 바꿉니다.
parsed = normalize_data(extract_parsed_response(response))
# dict 형태인지 확인합니다.
if not isinstance(parsed, dict):
    raise RuntimeError("중첩 구조 파싱에 실패했습니다.")
# 상위 필수 필드를 검사합니다.
missing_top_level = missing_fields(parsed, ["quarter", "teams", "shared_requirements", "risk_notes"])
# 팀 수를 계산합니다.
team_count = len(parsed.get("teams", []))
# 전체 세션 수를 계산합니다.
session_count = sum(len(team.get("sessions", [])) for team in parsed.get("teams", []))
# 구조화 결과를 출력합니다.
print(json.dumps(parsed, ensure_ascii=False, indent=2))
# 팀 수와 세션 수를 출력합니다.
print(f"팀 수: {team_count}, 세션 수: {session_count}")

{
  "quarter": "2026년 2분기",
  "teams": [
    {
      "team_name": "백엔드팀",
      "quarterly_goal": "장애 대응 역량 강화 및 시스템 성능 향상",
      "sessions": [
        {
          "topic": "장애 대응 훈련",
          "duration_hours": 3.0,
          "instructor": "김선우",
          "mandatory": true
        },
        {
          "topic": "성능 튜닝 워크숍",
          "duration_hours": 2.0,
          "instructor": "박하늘",
          "mandatory": false
        }
      ],
      "success_metrics": [
        "평균 복구 시간 20% 단축",
        "장애 재발률 15% 감소"
      ]
    },
    {
      "team_name": "프론트엔드팀",
      "quarterly_goal": "접근성 개선 및 디자인 시스템 도입 촉진",
      "sessions": [
        {
          "topic": "접근성 개선 세션",
          "duration_hours": 2.0,
          "instructor": "이도윤",
          "mandatory": true
        },
        {
          "topic": "디자인 시스템 실습",
          "duration_hours": 2.0,
          "instructor": "정시온",
          "mandatory": true
        }
      ],
      "success_metrics": [
        "접근성 점수 90점 이상",
        "컴포넌트 재사용률 30% 증가"
      ]
    }
  ],
  "shared_requirements": [
    "사전 과제 제출",
    "실습 계정 준비",
    "교육 후 회고 작성"
  ],
  "risk_notes": [
    "일정 변경 가능성",
    "강사 일정 충돌"
  ]
}
팀 수: 2, 세션 수: 4

 

 

 

4단계. 결과 저장과 검증 요약

구조화 출력은 이후 코드가 바로 사용할 데이터가 됩니다. 텍스트, JSON, 메타데이터를 함께 저장하면 사람이 읽는 결과와 프로그램이 처리하는 결과를 모두 확인할 수 있습니다.

# 팀별 계획을 사람이 읽기 쉬운 보고서로 바꿉니다.
report_lines = ["# 03 예제 실행 보고서", "", f"- 분기: {parsed.get('quarter', '')}", f"- 팀 수: {team_count}", f"- 전체 세션 수: {session_count}", "", "## 팀별 계획"]
# 각 팀을 순회하며 보고서 행을 추가합니다.
for team in parsed.get("teams", []):
    report_lines.append(f"- 팀: {team.get('team_name', '')}")
    report_lines.append(f"  목표: {team.get('quarterly_goal', '')}")
    for session in team.get("sessions", []):
        report_lines.append(f"  세션: {session.get('topic', '')} / {session.get('duration_hours', '')}시간 / 강사 {session.get('instructor', '')} / 필수 {session.get('mandatory', False)}")
    report_lines.append(f"  성공 지표: {', '.join(team.get('success_metrics', []))}")
# 최종 보고서 텍스트를 만듭니다.
report_text = "\n".join(report_lines) + "\n"
# 텍스트 파일 경로입니다.
output_text_path = OUTPUT_DIR / "03_nested_v2_final_output.txt"
# 구조화 JSON 파일 경로입니다.
structured_path = OUTPUT_DIR / "03_nested_v2_structured.json"
# 보고서 파일 경로입니다.
report_path = OUTPUT_DIR / "03_nested_v2_report.md"
# 메타데이터 파일 경로입니다.
metadata_path = OUTPUT_DIR / "03_nested_v2_meta.json"
# 텍스트 결과를 저장합니다.
output_text_path.write_text(response.output_text or "구조화 응답 생성 완료\n", encoding="utf-8")
# 구조화 결과를 저장합니다.
structured_path.write_text(json.dumps(parsed, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
# 보고서를 저장합니다.
report_path.write_text(report_text, encoding="utf-8")
# usage를 dict로 변환합니다.
usage = usage_to_dict(response.usage)
# 메타데이터를 구성합니다.
metadata = {"example": "03.nested_array_extraction_v2", "requested_model": STRUCTURED_MODEL, "actual_model": response.model, "team_count": team_count, "session_count": session_count, "missing_top_level_count": len(missing_top_level), "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), "report_path": workspace_path(report_path)}
# 메타데이터를 저장합니다.
metadata_path.write_text(json.dumps(metadata, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
# 저장 위치를 출력합니다.
print(f"구조화 JSON 저장: {workspace_path(structured_path)}")
print(f"보고서 저장: {workspace_path(report_path)}")
print(f"메타데이터 저장: {workspace_path(metadata_path)}")

구조화 JSON 저장: 03.structured_output/output/03_nested_v2_structured.json
보고서 저장: 03.structured_output/output/03_nested_v2_report.md
메타데이터 저장: 03.structured_output/output/03_nested_v2_meta.json