5장: AI 서비스 API 설계 패턴

from fastapi import FastAPI, BackgroundTasks
from enum import Enum
import uuid
 
 
class BatchStatus(str, Enum):
    pending = "pending"
    running = "running"
    completed = "completed"
    failed = "failed"
    cancelled = "cancelled"
 
 
class BatchRequest(BaseModel):
    model: str
    inputs: list[CompletionInput]
    metadata: dict[str, str] | None = None
 
 
class BatchResponse(BaseModel):
    id: str
    status: BatchStatus
    total: int
    completed: int
    failed: int
    results: list[CompletionResult] | None = None
    created_at: str
    completed_at: str | None = None
    # 폴링 간격 힌트
    retry_after: int | None = None
 
 
@app.post("/api/v1/batches", status_code=202)
async def create_batch(
    request: BatchRequest,
    background_tasks: BackgroundTasks,
) -> BatchResponse:
    batch_id = f"batch_{uuid.uuid4().hex[:12]}"
    
    # 배치 작업을 백그라운드에서 처리
    background_tasks.add_task(
        process_batch, batch_id, request
    )
    
    return BatchResponse(
        id=batch_id,
        status=BatchStatus.pending,
        total=len(request.inputs),
        completed=0,
        failed=0,
        created_at=datetime.now().isoformat(),
        retry_after=5,  # 5초 후 다시 확인 권장
    )
 
 
@app.get("/api/v1/batches/{batch_id}")
async def get_batch(batch_id: str) -> BatchResponse:
    batch = await batch_store.get(batch_id)
    if not batch:
        raise HTTPException(status_code=404, detail="배치를 찾을 수 없습니다")
    
    response = BatchResponse(**batch)
    
    # 상태에 따른 폴링 간격 힌트
    if batch["status"] == "running":
        response.retry_after = 5
    
    return response

class BatchRequestWithWebhook(BaseModel):
    model: str
    inputs: list[CompletionInput]
    webhook_url: str  # 완료 시 결과를 보낼 URL
    webhook_secret: str | None = None  # HMAC 서명용
 
 
@app.post("/api/v1/batches", status_code=202)
async def create_batch_with_webhook(
    request: BatchRequestWithWebhook,
    background_tasks: BackgroundTasks,
) -> BatchResponse:
    batch_id = f"batch_{uuid.uuid4().hex[:12]}"
    
    background_tasks.add_task(
        process_and_notify, batch_id, request
    )
    
    return BatchResponse(
        id=batch_id,
        status=BatchStatus.pending,
        total=len(request.inputs),
        completed=0,
        failed=0,
        created_at=datetime.now().isoformat(),
    )
 
 
async def process_and_notify(
    batch_id: str,
    request: BatchRequestWithWebhook,
):
    result = await process_batch(batch_id, request)
    
    # 웹훅으로 결과 전달
    payload = result.model_dump_json()
    
    headers = {"Content-Type": "application/json"}
    if request.webhook_secret:
        signature = hmac.new(
            request.webhook_secret.encode(),
            payload.encode(),
            hashlib.sha256,
        ).hexdigest()
        headers["X-Webhook-Signature"] = f"sha256={signature}"
    
    async with httpx.AsyncClient() as client:
        await client.post(
            request.webhook_url,
            content=payload,
            headers=headers,
            timeout=30.0,
        )

from sse_starlette.sse import EventSourceResponse
 
 
@app.get("/api/v1/batches/{batch_id}/stream")
async def stream_batch_status(batch_id: str):
    async def event_generator():
        while True:
            batch = await batch_store.get(batch_id)
            
            yield {
                "event": "status",
                "data": json.dumps({
                    "status": batch["status"],
                    "completed": batch["completed"],
                    "total": batch["total"],
                }),
            }
            
            if batch["status"] in ("completed", "failed", "cancelled"):
                yield {
                    "event": "done",
                    "data": json.dumps(batch),
                }
                break
            
            await asyncio.sleep(2)
    
    return EventSourceResponse(event_generator())

class TokenUsage(BaseModel):
    prompt_tokens: int
    completion_tokens: int
    total_tokens: int
    
    # 세부 분류 (선택적)
    prompt_tokens_details: PromptTokensDetail | None = None
    completion_tokens_details: CompletionTokensDetail | None = None
 
 
class PromptTokensDetail(BaseModel):
    cached_tokens: int = 0    # 캐시된 프롬프트 토큰
    text_tokens: int = 0      # 텍스트 토큰
    image_tokens: int = 0     # 이미지 토큰
    audio_tokens: int = 0     # 오디오 토큰
 
 
class CompletionTokensDetail(BaseModel):
    text_tokens: int = 0
    reasoning_tokens: int = 0  # 추론 토큰 (o1 계열)
 
 
class CompletionResponse(BaseModel):
    id: str
    model: str
    choices: list[Choice]
    usage: TokenUsage
    
    # 비용 정보 (선택적이지만 권장)
    cost: CostInfo | None = None
 
 
class CostInfo(BaseModel):
    prompt_cost: float
    completion_cost: float
    total_cost: float
    currency: str = "USD"

{
  "model": "claude-4-vision",
  "messages": [
    {
      "role": "user",
      "content": [
        {
          "type": "text",
          "text": "이 아키텍처 다이어그램을 분석해주세요"
        },
        {
          "type": "image",
          "source": {
            "type": "base64",
            "media_type": "image/png",
            "data": "iVBORw0KGgoAAAANSUhEUg..."
          }
        }
      ]
    }
  ]
}

{
  "model": "claude-4-vision",
  "messages": [
    {
      "role": "user",
      "content": [
        {
          "type": "text",
          "text": "이 PDF 문서를 요약해주세요"
        },
        {
          "type": "document",
          "source": {
            "type": "url",
            "url": "https://files.example.com/uploads/abc123.pdf"
          }
        }
      ]
    }
  ]
}

from fastapi import UploadFile, File
 
 
@app.post("/api/v1/files")
async def upload_file(
    file: UploadFile = File(...),
    purpose: str = "assistants",
) -> FileResponse:
    # 파일 크기 검증
    if file.size > 100 * 1024 * 1024:  # 100MB 제한
        raise APIError(
            status_code=413,
            error_type="file_too_large",
            message="파일 크기는 100MB를 초과할 수 없습니다",
        )
    
    # 허용된 MIME 타입 검증
    allowed_types = {
        "image/png", "image/jpeg", "image/webp", "image/gif",
        "application/pdf",
        "audio/mp3", "audio/wav", "audio/ogg",
        "text/plain", "text/csv",
    }
    if file.content_type not in allowed_types:
        raise APIError(
            status_code=415,
            error_type="unsupported_media_type",
            message=f"지원하지 않는 파일 형식입니다: {file.content_type}",
        )
    
    # 파일 저장 및 메타데이터 반환
    file_id = await file_store.save(file, purpose)
    
    return FileResponse(
        id=file_id,
        filename=file.filename,
        size=file.size,
        content_type=file.content_type,
        purpose=purpose,
        created_at=datetime.now().isoformat(),
    )

{
  "model": "claude-4",
  "messages": [
    {
      "role": "user",
      "content": "서울의 현재 날씨를 알려주세요"
    }
  ],
  "tools": [
    {
      "type": "function",
      "function": {
        "name": "get_weather",
        "description": "지정된 도시의 현재 날씨를 조회합니다",
        "parameters": {
          "type": "object",
          "properties": {
            "city": {
              "type": "string",
              "description": "도시 이름"
            },
            "unit": {
              "type": "string",
              "enum": ["celsius", "fahrenheit"],
              "description": "온도 단위"
            }
          },
          "required": ["city"]
        }
      }
    }
  ]
}

# 1단계: 초기 요청 → 도구 호출 응답
initial_response = {
    "id": "resp_abc123",
    "model": "claude-4",
    "choices": [{
        "index": 0,
        "message": {
            "role": "assistant",
            "content": None,
            "tool_calls": [{
                "id": "call_xyz789",
                "type": "function",
                "function": {
                    "name": "get_weather",
                    "arguments": '{"city": "서울", "unit": "celsius"}'
                }
            }]
        },
        "finish_reason": "tool_calls"
    }],
    "usage": {"prompt_tokens": 120, "completion_tokens": 25, "total_tokens": 145}
}
 
# 2단계: 도구 결과를 포함한 후속 요청
followup_request = {
    "model": "claude-4",
    "messages": [
        {"role": "user", "content": "서울의 현재 날씨를 알려주세요"},
        {
            "role": "assistant",
            "content": None,
            "tool_calls": [{
                "id": "call_xyz789",
                "type": "function",
                "function": {
                    "name": "get_weather",
                    "arguments": '{"city": "서울", "unit": "celsius"}'
                }
            }]
        },
        {
            "role": "tool",
            "tool_call_id": "call_xyz789",
            "content": '{"temperature": 18, "condition": "맑음", "humidity": 45}'
        }
    ],
    "tools": [...]  # 동일한 도구 정의
}
 
# 3단계: 최종 응답
final_response = {
    "id": "resp_def456",
    "model": "claude-4",
    "choices": [{
        "index": 0,
        "message": {
            "role": "assistant",
            "content": "서울의 현재 날씨는 맑음이며, 기온은 18도, 습도는 45%입니다."
        },
        "finish_reason": "stop"
    }],
    "usage": {"prompt_tokens": 180, "completion_tokens": 35, "total_tokens": 215}
}

class BatchInput(BaseModel):
    custom_id: str  # 클라이언트 측 식별자
    method: str     # "POST"
    url: str        # "/v1/chat/completions"
    body: dict      # 요청 본문
 
 
class BatchOutput(BaseModel):
    custom_id: str
    status_code: int
    body: dict
    error: dict | None = None
 
 
@app.post("/api/v1/batches")
async def create_batch(
    inputs: list[BatchInput],
    completion_window: str = "24h",  # 완료 보장 시간
    metadata: dict[str, str] | None = None,
) -> BatchResponse:
    """
    배치 API는 개별 요청 대비 50% 할인된 가격으로 제공됩니다.
    최대 50,000건의 요청을 하나의 배치로 처리할 수 있습니다.
    """
    batch_id = f"batch_{uuid.uuid4().hex[:12]}"
    
    await batch_processor.enqueue(
        batch_id=batch_id,
        inputs=inputs,
        completion_window=completion_window,
        metadata=metadata,
    )
    
    return BatchResponse(
        id=batch_id,
        status="pending",
        total=len(inputs),
        completed=0,
        failed=0,
    )

{
  "model": "claude-4",
  "messages": [
    {
      "role": "user",
      "content": "다음 기술 블로그 포스트의 메타데이터를 추출하세요: ..."
    }
  ],
  "response_format": {
    "type": "json_schema",
    "json_schema": {
      "name": "blog_metadata",
      "strict": true,
      "schema": {
        "type": "object",
        "properties": {
          "title": { "type": "string" },
          "summary": { "type": "string", "maxLength": 200 },
          "tags": {
            "type": "array",
            "items": { "type": "string" },
            "maxItems": 5
          },
          "difficulty": {
            "type": "string",
            "enum": ["beginner", "intermediate", "advanced"]
          },
          "estimated_reading_minutes": { "type": "integer", "minimum": 1 }
        },
        "required": ["title", "summary", "tags", "difficulty"],
        "additionalProperties": false
      }
    }
  }
}

from pydantic import BaseModel, Field
from typing import Literal
 
 
class BlogMetadata(BaseModel):
    title: str
    summary: str = Field(max_length=200)
    tags: list[str] = Field(max_length=5)
    difficulty: Literal["beginner", "intermediate", "advanced"]
    estimated_reading_minutes: int = Field(ge=1)
 
 
# 응답을 Pydantic 모델로 검증
response = await ai_client.complete(request)
content = response.choices[0].message.content
metadata = BlogMetadata.model_validate_json(content)

ERROR_TAXONOMY = {
    # 클라이언트 오류 (4xx)
    "invalid_request": (400, "요청 형식이 올바르지 않습니다"),
    "authentication_error": (401, "인증에 실패했습니다"),
    "permission_denied": (403, "이 리소스에 접근할 권한이 없습니다"),
    "not_found": (404, "요청한 리소스를 찾을 수 없습니다"),
    "content_policy_violation": (400, "콘텐츠 정책을 위반하는 요청입니다"),
    "context_length_exceeded": (400, "모델의 컨텍스트 길이를 초과했습니다"),
    "invalid_tool_call": (400, "도구 호출 형식이 올바르지 않습니다"),
    
    # 레이트 리밋 (429)
    "rate_limit_exceeded": (429, "요청 한도를 초과했습니다"),
    "token_limit_exceeded": (429, "토큰 한도를 초과했습니다"),
    "budget_exceeded": (429, "월간 예산을 초과했습니다"),
    
    # 서버 오류 (5xx)
    "model_error": (500, "모델 추론 중 오류가 발생했습니다"),
    "model_overloaded": (503, "모델이 과부하 상태입니다"),
    "timeout": (504, "추론 시간이 초과되었습니다"),
}