멀티모달 실전 프로젝트 — 이미지 접근성 도우미

# main.py
from fastapi import FastAPI, File, UploadFile, Form
from fastapi.responses import StreamingResponse, FileResponse
import base64
import io
from openai import OpenAI

app = FastAPI()
client = OpenAI()

# 세션별 대화 이력 (실제로는 Redis 사용)
sessions: dict[str, list] = {}

@app.post("/analyze-image")
async def analyze_image(
    file: UploadFile = File(...),
    session_id: str = Form(default="default"),
    detail_level: str = Form(default="standard"),  # standard/detailed/brief
):
    """이미지를 분석하고 텍스트 설명 + 음성 반환"""
    image_data = await file.read()
    encoded = base64.b64encode(image_data).decode("utf-8")
    mime = file.content_type or "image/jpeg"

    # 상세도에 따른 프롬프트
    prompts = {
        "brief": "이 이미지를 한 문장으로 설명해주세요.",
        "standard": """이 이미지를 시각 장애인이 이해할 수 있도록 설명해주세요.
다음을 포함하세요:
- 전체적인 장면
- 주요 오브젝트와 위치
- 색상과 분위기
- 눈에 띄는 세부사항""",
        "detailed": """이 이미지를 시각 장애인을 위해 매우 상세하게 설명해주세요.
배경, 주제, 세부 사항, 텍스트 내용(있다면), 감정적 톤까지 포함해주세요.""",
    }

    # 대화 이력 초기화/유지
    if session_id not in sessions:
        sessions[session_id] = []

    messages = [
        {"role": "system", "content": "당신은 시각 장애인을 위한 이미지 설명 전문가입니다. 명확하고 생생하게 이미지를 설명합니다."},
        *sessions[session_id],
        {
            "role": "user",
            "content": [
                {"type": "text", "text": prompts[detail_level]},
                {"type": "image_url", "image_url": {
                    "url": f"data:{mime};base64,{encoded}"
                }},
            ]
        }
    ]

    # Vision 분석
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        max_tokens=500,
    )
    description = response.choices[0].message.content

    # 대화 이력 업데이트
    sessions[session_id].append({
        "role": "user",
        "content": [{"type": "text", "text": prompts[detail_level]}]
    })
    sessions[session_id].append({
        "role": "assistant",
        "content": description
    })

    # TTS 변환
    tts_response = client.audio.speech.create(
        model="tts-1",
        voice="nova",
        input=description,
    )

    return {
        "description": description,
        "audio_base64": base64.b64encode(tts_response.read()).decode("utf-8"),
        "session_id": session_id,
    }

@app.post("/follow-up")
async def follow_up_question(
    audio: UploadFile = File(...),
    session_id: str = Form(...),
):
    """음성 후속 질문 처리"""
    audio_data = await audio.read()

    # Whisper 음성 인식
    transcript = client.audio.transcriptions.create(
        model="whisper-1",
        file=("audio.webm", io.BytesIO(audio_data), "audio/webm"),
        language="ko",
    )
    question = transcript.text

    # 이전 대화 맥락 포함 답변
    if session_id not in sessions:
        return {"error": "세션을 찾을 수 없습니다."}

    messages = [
        {"role": "system", "content": "당신은 이미지에 대한 질문에 답하는 접근성 도우미입니다."},
        *sessions[session_id],
        {"role": "user", "content": question},
    ]

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        max_tokens=300,
    )
    answer = response.choices[0].message.content

    sessions[session_id].append({"role": "user", "content": question})
    sessions[session_id].append({"role": "assistant", "content": answer})

    tts = client.audio.speech.create(model="tts-1", voice="nova", input=answer)

    return {
        "question": question,
        "answer": answer,
        "audio_base64": base64.b64encode(tts.read()).decode("utf-8"),
    }

import httpx
import base64
from pathlib import Path

BASE_URL = "http://localhost:8000"

async def analyze(image_path: str, session_id: str = "user1") -> dict:
    async with httpx.AsyncClient() as http:
        with open(image_path, "rb") as f:
            response = await http.post(
                f"{BASE_URL}/analyze-image",
                files={"file": (Path(image_path).name, f, "image/jpeg")},
                data={"session_id": session_id, "detail_level": "standard"},
            )
    result = response.json()

    # 음성 저장
    audio_bytes = base64.b64decode(result["audio_base64"])
    Path("description.mp3").write_bytes(audio_bytes)

    print(f"설명:\n{result['description']}")
    return result

# 사용
import asyncio
asyncio.run(analyze("photo.jpg"))