11장: 실전 프로젝트 — 멀티모달 AI 애플리케이션 구축

import fitz  # PyMuPDF
from dataclasses import dataclass
from pathlib import Path
 
@dataclass
class ParsedPage:
    page_number: int
    text: str
    images: list[bytes]
    page_image: bytes  # 전체 페이지 렌더링
 
class PDFParser:
    def __init__(self, dpi: int = 150):
        self.dpi = dpi
 
    def parse(self, pdf_path: str | Path) -> list[ParsedPage]:
        """PDF를 페이지별로 파싱"""
        doc = fitz.open(str(pdf_path))
        pages = []
 
        for page_num in range(len(doc)):
            page = doc[page_num]
 
            # 텍스트 추출
            text = page.get_text("text")
 
            # 이미지 추출
            images = []
            for img_info in page.get_images(full=True):
                xref = img_info[0]
                img_data = doc.extract_image(xref)
                if img_data:
                    images.append(img_data["image"])
 
            # 전체 페이지 이미지 렌더링
            pix = page.get_pixmap(dpi=self.dpi)
            page_image = pix.tobytes("png")
 
            pages.append(ParsedPage(
                page_number=page_num + 1,
                text=text,
                images=images,
                page_image=page_image,
            ))
 
        doc.close()
        return pages

import anthropic
import base64
from dataclasses import dataclass
from typing import Literal
 
@dataclass
class MultimodalChunk:
    chunk_id: str
    page_number: int
    chunk_type: Literal["text", "table", "chart", "image", "diagram"]
    text_content: str
    image_data: bytes | None
    caption: str | None
 
class MultimodalChunker:
    def __init__(self, vlm_client: anthropic.Anthropic):
        self.vlm = vlm_client
 
    def chunk_page(self, page: "ParsedPage") -> list[MultimodalChunk]:
        """페이지를 멀티모달 청크로 분할"""
        chunks = []
 
        # 텍스트 청크 (단락 단위)
        paragraphs = self._split_paragraphs(page.text)
        for i, para in enumerate(paragraphs):
            if len(para.strip()) > 50:  # 짧은 텍스트 제외
                chunks.append(MultimodalChunk(
                    chunk_id=f"p{page.page_number}_text_{i}",
                    page_number=page.page_number,
                    chunk_type="text",
                    text_content=para.strip(),
                    image_data=None,
                    caption=None,
                ))
 
        # 이미지 청크 — VLM으로 분석하여 캡션 생성
        for i, img_bytes in enumerate(page.images):
            caption = self._generate_caption(img_bytes)
            img_type = self._classify_image(img_bytes)
 
            chunks.append(MultimodalChunk(
                chunk_id=f"p{page.page_number}_{img_type}_{i}",
                page_number=page.page_number,
                chunk_type=img_type,
                text_content=caption,
                image_data=img_bytes,
                caption=caption,
            ))
 
        return chunks
 
    def _generate_caption(self, image_data: bytes) -> str:
        """VLM으로 이미지 캡션 생성"""
        img_b64 = base64.standard_b64encode(image_data).decode()
        response = self.vlm.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=512,
            messages=[{
                "role": "user",
                "content": [
                    {"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": img_b64}},
                    {"type": "text", "text": "이 이미지의 내용을 3-5문장으로 상세히 설명해주세요. 표가 있다면 주요 데이터를, 차트가 있다면 트렌드를 포함해주세요."},
                ],
            }],
        )
        return response.content[0].text
 
    def _classify_image(self, image_data: bytes) -> str:
        """이미지 유형 분류"""
        img_b64 = base64.standard_b64encode(image_data).decode()
        response = self.vlm.messages.create(
            model="claude-haiku-4-5-20251001",
            max_tokens=50,
            messages=[{
                "role": "user",
                "content": [
                    {"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": img_b64}},
                    {"type": "text", "text": "이 이미지의 유형을 하나만 답해주세요: table, chart, diagram, photo, screenshot"},
                ],
            }],
        )
        result = response.content[0].text.strip().lower()
        valid_types = {"table", "chart", "diagram", "photo", "screenshot"}
        return result if result in valid_types else "image"
 
    def _split_paragraphs(self, text: str) -> list[str]:
        """텍스트를 단락으로 분할"""
        paragraphs = text.split("\n\n")
        return [p for p in paragraphs if p.strip()]

import numpy as np
 
class HybridRetriever:
    def __init__(self, text_store, image_store, text_weight=0.6, image_weight=0.4):
        self.text_store = text_store
        self.image_store = image_store
        self.text_weight = text_weight
        self.image_weight = image_weight
 
    def search(
        self,
        query: str,
        text_embedding: np.ndarray,
        clip_embedding: np.ndarray,
        top_k: int = 5,
    ) -> list[dict]:
        """텍스트 + 이미지 하이브리드 검색"""
        # 텍스트 검색
        text_results = self.text_store.query(
            query_embeddings=[text_embedding.tolist()],
            n_results=top_k * 2,
        )
 
        # 이미지 검색 (CLIP)
        image_results = self.image_store.query(
            query_embeddings=[clip_embedding.tolist()],
            n_results=top_k * 2,
        )
 
        # 점수 통합
        candidates = {}
 
        for i, doc_id in enumerate(text_results["ids"][0]):
            distance = text_results["distances"][0][i]
            score = 1 / (1 + distance)  # 거리 → 유사도
            candidates[doc_id] = {
                "text_score": score,
                "image_score": 0,
                "metadata": text_results["metadatas"][0][i],
            }
 
        for i, doc_id in enumerate(image_results["ids"][0]):
            distance = image_results["distances"][0][i]
            score = 1 / (1 + distance)
            if doc_id in candidates:
                candidates[doc_id]["image_score"] = score
            else:
                candidates[doc_id] = {
                    "text_score": 0,
                    "image_score": score,
                    "metadata": image_results["metadatas"][0][i],
                }
 
        # 가중 결합 점수
        for doc_id, data in candidates.items():
            data["combined_score"] = (
                self.text_weight * data["text_score"] +
                self.image_weight * data["image_score"]
            )
 
        # 정렬 및 상위 K개 반환
        sorted_results = sorted(
            candidates.items(),
            key=lambda x: x[1]["combined_score"],
            reverse=True,
        )
 
        return [
            {"chunk_id": doc_id, **data}
            for doc_id, data in sorted_results[:top_k]
        ]

import anthropic
import base64
 
class MultimodalAnswerGenerator:
    def __init__(self, client: anthropic.Anthropic):
        self.client = client
 
    def generate(
        self,
        query: str,
        retrieved_chunks: list["MultimodalChunk"],
        max_images: int = 3,
    ) -> str:
        """검색된 청크로 멀티모달 답변 생성"""
        content = []
        image_count = 0
 
        for chunk in retrieved_chunks:
            # 이미지가 있는 청크
            if chunk.image_data and image_count < max_images:
                img_b64 = base64.standard_b64encode(chunk.image_data).decode()
                content.append({
                    "type": "text",
                    "text": f"--- 참고 자료 (페이지 {chunk.page_number}, {chunk.chunk_type}) ---",
                })
                content.append({
                    "type": "image",
                    "source": {"type": "base64", "media_type": "image/png", "data": img_b64},
                })
                if chunk.caption:
                    content.append({"type": "text", "text": f"설명: {chunk.caption}"})
                image_count += 1
            else:
                # 텍스트 청크 또는 이미지 예산 초과
                content.append({
                    "type": "text",
                    "text": f"--- 참고 자료 (페이지 {chunk.page_number}) ---\n{chunk.text_content}",
                })
 
        # 질문 및 지시사항
        content.append({
            "type": "text",
            "text": f"""---
 
위 참고 자료를 기반으로 다음 질문에 답변해주세요.
 
질문: {query}
 
답변 지침:
- 참고 자료에 근거하여 답변하세요
- 표나 차트의 구체적 수치를 인용해주세요
- 출처(페이지 번호)를 명시해주세요
- 참고 자료에서 찾을 수 없는 내용은 "문서에서 확인할 수 없습니다"라고 답하세요""",
        })
 
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=4096,
            messages=[{"role": "user", "content": content}],
        )
        return response.content[0].text

from fastapi import FastAPI, UploadFile, File
from pydantic import BaseModel
 
app = FastAPI(title="멀티모달 문서 Q&A")
 
class QuestionRequest(BaseModel):
    document_id: str
    question: str
 
class AnswerResponse(BaseModel):
    answer: str
    sources: list[dict]
    cost_estimate_usd: float
 
@app.post("/documents/upload")
async def upload_document(file: UploadFile = File(...)):
    """PDF 문서 업로드 및 인덱싱"""
    # 1. PDF 파싱
    pages = parser.parse(file.file)
 
    # 2. 멀티모달 청킹
    all_chunks = []
    for page in pages:
        chunks = chunker.chunk_page(page)
        all_chunks.extend(chunks)
 
    # 3. 임베딩 생성 및 벡터 DB 저장
    for chunk in all_chunks:
        text_emb = text_embedder.embed(chunk.text_content)
        text_store.add(chunk.chunk_id, text_emb, chunk.text_content, chunk.metadata)
 
        if chunk.image_data:
            clip_emb = image_embedder.embed(chunk.image_data)
            image_store.add(chunk.chunk_id, clip_emb, chunk.metadata)
 
    return {"document_id": doc_id, "chunks": len(all_chunks)}
 
@app.post("/documents/ask", response_model=AnswerResponse)
async def ask_question(request: QuestionRequest):
    """문서에 대한 질문 답변"""
    # 1. 쿼리 임베딩
    text_emb = text_embedder.embed(request.question)
    clip_emb = image_embedder.embed_text(request.question)
 
    # 2. 하이브리드 검색
    results = retriever.search(request.question, text_emb, clip_emb, top_k=5)
 
    # 3. 청크 로드
    chunks = load_chunks(results)
 
    # 4. 답변 생성
    answer = generator.generate(request.question, chunks)
 
    return AnswerResponse(
        answer=answer,
        sources=[{"page": c.page_number, "type": c.chunk_type} for c in chunks],
        cost_estimate_usd=estimate_cost(chunks),
    )