d01 - 03. Structed Output - 04. streamnig structured output v2

04. 스트리밍 Structured Outputs v2

이 노트북은 구조화 출력과 스트리밍을 함께 사용합니다. 이벤트를 실시간으로 받으면서도 마지막에는 Pydantic 스키마에 맞는 결과를 확인할 수 있습니다.

# 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단계. 출력 스키마와 입력 준비

스트리밍 결과도 최종적으로는 같은 스키마를 따라야 합니다. 따라서 체크리스트 항목과 전체 체크리스트 구조를 먼저 정의합니다.

# 출시 점검 항목 하나를 표현하는 스키마입니다.
class ChecklistItem(BaseModel):
    # 우선순위를 P1, P2, P3 같은 문자열로 저장합니다.
    priority: str = Field(description="우선순위")
    # 담당자 또는 담당 팀을 저장합니다.
    owner: str = Field(description="담당자")
    # 해야 할 작업을 저장합니다.
    task: str = Field(description="점검 항목")
    # 완료 기준을 저장합니다.
    done_criteria: str = Field(description="완료 기준")

# 출시 체크리스트 전체를 표현하는 스키마입니다.
class StreamingReleaseChecklist(BaseModel):
    # 출시 대상을 저장합니다.
    release_name: str = Field(description="출시 대상")
    # 전체 요약을 저장합니다.
    summary: str = Field(description="전체 요약")
    # 점검 항목 목록을 저장합니다.
    items: list[ChecklistItem] = Field(description="점검 항목 목록")
    # 잠재 장애 요소 목록을 저장합니다.
    blockers: list[str] = Field(description="잠재 장애 요소")
    # 즉시 실행할 후속 조치 목록을 저장합니다.
    next_actions: list[str] = Field(description="즉시 실행할 후속 조치")

# 모델의 역할 지시문입니다.
INSTRUCTIONS = "당신은 출시 준비 점검 도우미입니다. 사용자 요청을 구조화된 체크리스트로 정리하세요."
# 구조화할 사용자 요청입니다.
PROMPT = "이번 주 금요일에 신규 결제 기능을 출시합니다. 사전 점검 체크리스트를 우선순위, 담당자, 완료 기준까지 포함해 정리해 주세요."

 

 

 

3단계. Responses API 직접 호출

client.responses.stream(...)을 직접 호출하고 이벤트를 순회합니다. 스트림 종료 후 get_final_response()에서 구조화 결과를 확인합니다.

# 스트리밍 시작 시간을 기록합니다.
started = time.perf_counter()
# 이벤트 종류별 개수를 저장할 dict입니다.
event_counts: dict[str, int] = {}
# 텍스트 델타 조각을 누적할 리스트입니다.
text_chunks: list[str] = []
# 첫 텍스트 델타가 도착한 시간을 저장합니다.
first_text_delta_seconds = None
# 최종 응답 객체를 저장할 변수입니다.
final_response = None
# responses.stream을 직접 호출합니다.
stream = client.responses.stream(
    # 스트리밍 구조화 출력을 만들 모델입니다.
    model=STRUCTURED_MODEL,
    # 모델의 역할과 출력 기준입니다.
    instructions=INSTRUCTIONS,
    # 출시 점검 요청입니다.
    input=[{"role": "user", "content": PROMPT}],
    # 스트리밍 완료 후 이 Pydantic 스키마로 파싱됩니다.
    text_format=StreamingReleaseChecklist,
)
# 스트림은 context manager로 열어야 최종 응답을 안전하게 받을 수 있습니다.
with stream as session:
    # 스트리밍 이벤트를 하나씩 순회합니다.
    for event in session:
        # 이벤트 타입 이름을 읽습니다.
        event_type = str(getattr(event, "type", "unknown"))
        # 이벤트 타입별 개수를 누적합니다.
        event_counts[event_type] = event_counts.get(event_type, 0) + 1
        # 텍스트 델타 이벤트이면 조각을 누적합니다.
        if event_type == "response.output_text.delta":
            # 이번 이벤트의 텍스트 조각입니다.
            delta = str(getattr(event, "delta", ""))
            # 첫 델타 도착 시간을 한 번만 기록합니다.
            if delta and first_text_delta_seconds is None:
                first_text_delta_seconds = time.perf_counter() - started
            # 텍스트 조각을 누적합니다.
            text_chunks.append(delta)
    # 스트림 종료 후 최종 response 객체를 가져옵니다.
    final_response = session.get_final_response()
# 전체 스트리밍 시간을 계산합니다.
stream_total_seconds = time.perf_counter() - started
# 최종 응답의 output_parsed를 dict로 정규화합니다.
parsed = normalize_data(extract_parsed_response(final_response))
# dict 형태인지 확인합니다.
if not isinstance(parsed, dict):
    raise RuntimeError("스트리밍 구조화 결과 파싱에 실패했습니다.")
# 점검 항목 수를 계산합니다.
item_count = len(parsed.get("items", []))
# 이벤트와 결과를 출력합니다.
print(f"이벤트 종류 수: {len(event_counts)}")
print(f"첫 텍스트 델타 도착: {first_text_delta_seconds}")
print(f"점검 항목 수: {item_count}")
print(json.dumps(parsed, ensure_ascii=False, indent=2))

이벤트 종류 수: 9
첫 텍스트 델타 도착: 17.15637899999274
점검 항목 수: 20
{
  "release_name": "신규 결제 기능 — 이번 주 금요일 출시",
  "summary": "이번 주 금요일 프로덕션 배포 예정인 신규 결제 기능의 사전 점검 체크리스트입니다. 우선순위(critical/high/medium/low), 담당자(역할), 완료 기준을 포함합니다.",
  "items": [
    {
      "priority": "Critical",
      "owner": "엔지니어링 리드",
      "task": "코드 최종 검토 및 머지(코드 프리즈)",
      "done_criteria": "모든 관련 PR이 리뷰 완료 및 머지되어 코드 프리즈 상태. CI 빌드 성공, 주요 리뷰 코멘트 처리."
    },
    {
      "priority": "Critical",
      "owner": "DevOps / SRE",
      "task": "CI/CD 파이프라인 및 배포 스크립트 검증",
      "done_criteria": "배포 스테이지(스테이징/프로덕션)에서 파이프라인 실행 성공, 배포 스크립트 롤백 동작 검증."
    },
    {
      "priority": "Critical",
      "owner": "백엔드/DB 담당",
      "task": "데이터베이스 마이그레이션 준비 및 롤백 검증",
      "done_criteria": "마이그레이션 스크립트 리뷰 완료, 스테이징에서 적용 및 롤백 테스트 성공, 다운타임/잠재 영향 문서화."
    },
    {
      "priority": "Critical",
      "owner": "결제 연동 엔지니어",
      "task": "결제 게이트웨이(샌드박스/프로덕션) 키 및 시크릿 확인",
      "done_criteria": "프로덕션용 API 키/시크릿이 시크릿 매니저에 등록되어 있고 권한 검증 완료, 키 유효성 확인(샘플 호출 성공)."
    },
    {
      "priority": "Critical",
      "owner": "QA(결제 담당)",
      "task": "결제 통합 테스트(결제/취소/환불/중복/에러 시나리오)",
      "done_criteria": "샌드박스에서 모든 시나리오 통과, 프로덕션 샌드위치(소액) 결제 테스트 수행 및 결과 기록."
    },
    {
      "priority": "High",
      "owner": "QA",
      "task": "기능 테스트 및 회귀 테스트",
      "done_criteria": "핵심 사용자 흐름(결제 생성→완료, 실패 처리 등) 통과, 회귀 테스트 실패 없음 또는 승인된 이슈만 존재."
    },
    {
      "priority": "High",
      "owner": "SRE / 퍼포먼스 엔지니어",
      "task": "성능/부하 테스트(피크 트래픽 시나리오)",
      "done_criteria": "목표 TPS/응답시간/오류율 기준 충족, 병목 발생 시 완화 방안 문서화."
    },
    {
      "priority": "Critical",
      "owner": "보안팀",
      "task": "보안 리뷰 및 취약점 스캔(SAST/DAST/종합 스캔)",
      "done_criteria": "중/치명적(High/Critical) 취약점이 해결되거나 공식적인 예외 승인 문서 존재."
    },
    {
      "priority": "Critical",
      "owner": "컴플라이언스/법무",
      "task": "PCI 및 관련 규정 준수 확인",
      "done_criteria": "필요한 PCI 요건 충족 여부 검토 완료, 미결사항이 있을 경우 배포 전 해결 또는 예외 승인 확보."
    },
    {
      "priority": "High",
      "owner": "SRE",
      "task": "모니터링, 대시보드 및 알림 구성",
      "done_criteria": "결제 성공률, 오류율, 지연(latency), 큐 길이 등 대시보드 구성, 임계치 기반 알림(Pager/Slack) 설정 완료."
    },
    {
      "priority": "High",
      "owner": "백엔드",
      "task": "로그 및 추적(트랜잭션 ID 포함) 설정 확인",
      "done_criteria": "모든 결제 트랜잭션에 고유 ID가 부여되고 로그/트레이스가 수집되어 샘플 조회로 확인됨."
    },
    {
      "priority": "Critical",
      "owner": "엔지니어링 리드 / SRE",
      "task": "롤백 및 비상복구(타임라인/책임자) 계획 수립",
      "done_criteria": "명확한 롤백 절차 문서화(조건, 명령, 담당자), 롤백 시나리오 드라이런 완료."
    },
    {
      "priority": "High",
      "owner": "프로덕트 매니저 / 엔지니어",
      "task": "피처 플래그 및 점진적 배포(카나리) 계획",
      "done_criteria": "피처 플래그 준비 및 초기 퍼센트 롤아웃 계획 수립, 모니터링 기준과 자동 중단 조건 정의."
    },
    {
      "priority": "Medium",
      "owner": "고객지원팀",
      "task": "고객지원(FAQ/스크립트) 및 내부 교육 준비",
      "done_criteria": "CS용 Q&A, 에러 케이스별 대응 스크립트 작성 및 담당자 교육 완료."
    },
    {
      "priority": "High",
      "owner": "재무(파이낸스)",
      "task": "회계/정산 흐름 및 환불 정책 확인",
      "done_criteria": "정산 주기, 수수료 처리, 환불 플로우가 문서화되어 있고 테스트 케이스로 검증됨."
    },
    {
      "priority": "Medium",
      "owner": "법무",
      "task": "서비스 약관/결제 관련 고지 사항 업데이트",
      "done_criteria": "변경사항이 반영된 약관 및 고지가 준비되어 배포 계획에 포함됨."
    },
    {
      "priority": "Medium",
      "owner": "엔지니어링(결제 담당)",
      "task": "제3자(PSP) 연락처 및 SLA/에스컬레이션 루트 확인",
      "done_criteria": "핫라인/에스컬레이션 연락처 확인 완료, 장애 시 연락 프로세스 문서화."
    },
    {
      "priority": "High",
      "owner": "SRE / 엔지니어링",
      "task": "운영 드라이런(릴리스 전 시뮬레이션, 롤백 포함)",
      "done_criteria": "릴리스 드라이런을 통해 주요 단계(배포→검증→롤백) 실행, 이슈 없거나 해결책 문서화."
    },
    {
      "priority": "Medium",
      "owner": "프로덕트 매니저 / 커뮤니케이션",
      "task": "릴리스 커뮤니케이션(내부/외부) 준비",
      "done_criteria": "릴리스 노트, 고객 공지 텍스트, 내부 알림(채널/시간) 준비 및 승인 완료."
    },
    {
      "priority": "High",
      "owner": "온콜 및 담당자 명시(팀별)",
      "task": "포스트 릴리스 검증 체크(배포 후 0h/1h/6h/24h 점검 리스트)",
      "done_criteria": "각 시점별 검증 항목(결제 성공률, 오류 증가 여부 등)과 담당자 할당, 결과 리포트 양식 준비."
    }
  ],
  "blockers": [
    "치명적 또는 고심각도(High/Critical) 버그 미해결",
    "프로덕션 결제 키/시크릿 미등록 또는 유효성 실패",
    "결제 게이트웨이 인증/인티그레이션 검증 실패",
    "PCI/규정 준수 관련 미해결 사항",
    "마이그레이션(또는 DB 변경)에 대한 롤백 미검증",
    "모니터링·알림이 미설정 또는 대시보드 미구성"
  ],
  "next_actions": [
    "코드 프리즈(머지 마감) 선언 및 관련 PR 최종 머지",
    "풀 테스트(회귀 + 결제 통합) 실행 및 결과 공유",
    "프로덕션 API 키/시크릿 최종 확인 및 접근 권한 검증",
    "프로덕션 백업(데이터/설정) 수행",
    "릴리스 윈도우 및 온콜 담당자(연락처 포함) 최종 확정",
    "릴리스 드라이런(배포→검증→롤백) 수행",
    "릴리스 노트/고객 공지 최종 검토 및 배포 스케줄링",
    "배포 시작 1시간 전 스탠드업(관련 담당자 참석)으로 상태 점검"
  ]
}

 

 

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

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

# 사람이 읽기 쉬운 체크리스트 텍스트를 구성합니다.
human_lines = [f"출시 대상: {parsed.get('release_name', '')}", f"요약: {parsed.get('summary', '')}", "", "[점검 항목]"]
# 점검 항목을 한 줄씩 추가합니다.
for item in parsed.get("items", []):
    human_lines.append(f"- {item.get('priority', '')} | {item.get('owner', '')} | {item.get('task', '')} (완료 기준: {item.get('done_criteria', '')})")
# 잠재 장애 요소 제목을 추가합니다.
human_lines.append("\n[잠재 장애 요소]")
# 잠재 장애 요소를 추가합니다.
human_lines.extend([f"- {item}" for item in parsed.get("blockers", [])])
# 후속 조치 제목을 추가합니다.
human_lines.append("\n[즉시 실행 조치]")
# 후속 조치를 추가합니다.
human_lines.extend([f"- {item}" for item in parsed.get("next_actions", [])])
# 최종 텍스트를 만듭니다.
human_text = "\n".join(human_lines) + "\n"
# 텍스트 파일 경로입니다.
output_text_path = OUTPUT_DIR / "04_streaming_v2_final_output.txt"
# 구조화 JSON 파일 경로입니다.
structured_path = OUTPUT_DIR / "04_streaming_v2_structured.json"
# 이벤트 JSON 파일 경로입니다.
events_path = OUTPUT_DIR / "04_streaming_v2_events.json"
# 메타데이터 파일 경로입니다.
metadata_path = OUTPUT_DIR / "04_streaming_v2_meta.json"
# 텍스트 결과를 저장합니다.
output_text_path.write_text(human_text, encoding="utf-8")
# 구조화 결과를 저장합니다.
structured_path.write_text(json.dumps(parsed, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
# 이벤트 개수를 저장합니다.
events_path.write_text(json.dumps(event_counts, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
# usage를 dict로 변환합니다.
usage = usage_to_dict(getattr(final_response, "usage", None))
# 메타데이터를 구성합니다.
metadata = {"example": "04.streaming_structured_output_v2", "requested_model": STRUCTURED_MODEL, "actual_model": final_response.model, "event_type_count": len(event_counts), "item_count": item_count, "first_text_delta_seconds": None if first_text_delta_seconds is None else round(first_text_delta_seconds, 4), "stream_total_seconds": round(stream_total_seconds, 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), "events_path": workspace_path(events_path)}
# 메타데이터를 저장합니다.
metadata_path.write_text(json.dumps(metadata, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
# 저장 위치를 출력합니다.
print(f"텍스트 저장: {workspace_path(output_text_path)}")
print(f"구조화 JSON 저장: {workspace_path(structured_path)}")
print(f"이벤트 저장: {workspace_path(events_path)}")

텍스트 저장: 03.structured_output/output/04_streaming_v2_final_output.txt
구조화 JSON 저장: 03.structured_output/output/04_streaming_v2_structured.json
이벤트 저장: 03.structured_output/output/04_streaming_v2_events.json