8장: 멀티모달 RAG 시스템 설계

# ColPali 모델 로드 (개념적 코드)
from colpali import ColPali
 
model = ColPali.from_pretrained("vidore/colpali-v1.3")
 
# 문서 페이지 인덱싱
page_embeddings = model.encode_documents([
    "page1.png", "page2.png", "page3.png"
])
 
# 쿼리 검색
query = "2023년 4분기 매출 차트"
query_embedding = model.encode_query(query)
 
# Late Interaction으로 유사도 계산
scores = model.score(query_embedding, page_embeddings)
# 가장 관련 높은 페이지 반환

from dataclasses import dataclass
from typing import Literal
 
@dataclass
class DocumentChunk:
    chunk_id: str
    page_number: int
    chunk_type: Literal["text", "table", "image", "chart"]
    content: str  # 텍스트 또는 캡션
    image_data: bytes | None  # 원본 이미지 (있는 경우)
    metadata: dict
 
def parse_multimodal_document(pdf_path: str, vlm_client) -> list[DocumentChunk]:
    """PDF를 멀티모달 청크로 파싱"""
    chunks = []
 
    for page_num, page_image in enumerate(render_pdf_pages(pdf_path)):
        # 1. 페이지 분석 — 영역 유형 식별
        regions = analyze_page_regions(vlm_client, page_image)
 
        for region in regions:
            if region.type == "text":
                chunks.append(DocumentChunk(
                    chunk_id=f"p{page_num}_text_{region.index}",
                    page_number=page_num,
                    chunk_type="text",
                    content=region.text,
                    image_data=None,
                    metadata={"position": region.bbox},
                ))
            elif region.type in ("table", "chart", "image"):
                # VLM으로 시각 영역 설명 생성
                caption = generate_region_caption(vlm_client, region.image)
                chunks.append(DocumentChunk(
                    chunk_id=f"p{page_num}_{region.type}_{region.index}",
                    page_number=page_num,
                    chunk_type=region.type,
                    content=caption,
                    image_data=region.image,
                    metadata={"position": region.bbox},
                ))
 
    return chunks

def process_table(table_image: bytes, vlm_client) -> dict:
    """표를 마크다운과 캡션으로 변환"""
    # VLM으로 표를 마크다운 변환
    markdown = vlm_client.analyze(table_image, prompt="""
이 표를 마크다운 형식으로 정확하게 변환해주세요.
모든 행과 열의 데이터를 포함하세요.""")
 
    # VLM으로 표 요약 생성
    summary = vlm_client.analyze(table_image, prompt="""
이 표의 내용을 2-3문장으로 요약해주세요.
핵심 수치와 트렌드를 포함하세요.""")
 
    return {
        "markdown": markdown,
        "summary": summary,
        "image": table_image,
    }

class MultimodalRAGRetriever:
    def __init__(self, text_store, image_store, page_store):
        self.text_store = text_store
        self.image_store = image_store
        self.page_store = page_store
 
    def retrieve(self, query: str, top_k: int = 5) -> list[DocumentChunk]:
        """멀티모달 검색"""
        # 텍스트 검색
        text_results = self.text_store.search(
            encode_text(query), top_k=top_k
        )
 
        # 이미지/차트 검색 (CLIP)
        image_results = self.image_store.search(
            encode_clip(query), top_k=top_k
        )
 
        # 페이지 검색 (ColPali)
        page_results = self.page_store.search(
            encode_colpali(query), top_k=top_k
        )
 
        # 결과 통합 및 중복 제거
        all_results = merge_and_deduplicate(
            text_results, image_results, page_results
        )
 
        # 리랭킹
        return rerank(query, all_results, top_k=top_k)

def generate_answer(
    query: str,
    retrieved_chunks: list[DocumentChunk],
    vlm_client,
) -> str:
    """검색된 멀티모달 청크로 답변 생성"""
    content = []
 
    for chunk in retrieved_chunks:
        if chunk.image_data:
            # 이미지가 있는 청크는 이미지와 캡션 모두 전달
            img_b64 = base64.standard_b64encode(chunk.image_data).decode()
            content.append({
                "type": "image",
                "source": {"type": "base64", "media_type": "image/png", "data": img_b64},
            })
            content.append({
                "type": "text",
                "text": f"[{chunk.chunk_type} — 페이지 {chunk.page_number}]\n{chunk.content}",
            })
        else:
            content.append({
                "type": "text",
                "text": f"[텍스트 — 페이지 {chunk.page_number}]\n{chunk.content}",
            })
 
    content.append({
        "type": "text",
        "text": f"""위 자료를 기반으로 다음 질문에 답변해주세요.
 
질문: {query}
 
답변 시 출처(페이지 번호)를 명시하고,
시각 자료(표, 차트)의 데이터를 활용하여 구체적으로 답변해주세요.""",
    })
 
    response = vlm_client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=4096,
        messages=[{"role": "user", "content": content}],
    )
    return response.content[0].text