본 가이드는 클로바 스튜디오와 랭체인(Langchain)을 활용하여 Multimodal RAG(멀티모달 검색 증강 생성) 시스템을 구축하는 방법을 안내합니다.최근 비전 모델의 상용화가 가속화되면서 기업들은 내부의 다양한 이미지 기반 데이터를 효율적으로 검색하고 활용하려는 니즈가 증가하고 있습니다. 특히 기존 텍스트 중심 RAG 시스템을 이미지 데이터까지 포함하도록 확장하는 사례가 늘어나는 추세입니다.이 글에서는 PDF 형식의 데이터를 기반으로 질의응답 기능을 제공하는 Multimodal RAG 시스템을 랭체인을 통해 구현해보겠습니다. 구현하고자 하는 Multimodal RAG 시스템의 구조도는 아래와 같습니다.
멀티모달 임베딩 없이도 구현 가능한 Multimodal RAG 구조를 소개합니다. 이 방식은 비전 모델을 활용해 이미지를 텍스트로 변환한 후, 해당 텍스트를 임베딩하여 검색에 활용하는 접근법입니다. LangChain 프레임워크를 통해 CLOVA Studio의 모델과 Chroma, FAISS와 같은 외부 벡터 데이터베이스를 효과적으로 연동할 수 있습니다.
전체 과정은 셀 단위로 실습할 수 있도록 구성되어 있습니다(파일명: multimodal_RAG.ipynb). 이 가이드의 핵심은 이미지 기반 문서를 다루는 기업들이 쉽게 도입할 수 있는 범용적인 멀티모달 RAG 구현 방식을 소개하는 데 있습니다.
버전 정보 아래 예제 코드는 Python 3.12.2 환경에서 실행 검증을 완료했으며, 최소 Python 3.9 이상의 버전이 필요합니다. 다음 지침을 참고하여 필요한 모든 모듈을 설치해주세요. requirements.txt
{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "## 사전준비\n", "### 1. 랭체인 패키지 설치" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "%pip install -qU openai langchain langchain-naver " ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### 2. 공통 모듈 import" ] }, { "cell_type": "code", "execution_count": 2, "metadata": {}, "outputs": [], "source": [ "import os\n", "import getpass\n", "import uuid\n", "import re\n", "from urllib.parse import urlparse\n", "import http\n", "import json\n", "import time" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### 3. API 키 발급 받기" ] }, { "cell_type": "code", "execution_count": 3, "metadata": {}, "outputs": [], "source": [ "os.environ[\"CLOVASTUDIO_API_KEY\"] = getpass.getpass(\"CLOVA Studio API Key: \")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 문서 전처리하기" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### 1. PDF 문서에서 텍스트와 이미지 추출하기 (Load)" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "%pip install pymupdf" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "import fitz # PyMuPDF\n", "from langchain_core.documents import Document\n", "\n", "def extract_documents_from_pdf(pdf_path: str, output_dir: str = \"data/extracted_images_문서\"):\n", " os.makedirs(output_dir, exist_ok=True)\n", "\n", " merged_text_path = os.path.join(output_dir, \"merged_text.txt\")\n", " merged_text = \"\"\n", "\n", " doc = fitz.open(pdf_path)\n", " documents = []\n", "\n", " for i, page in enumerate(doc):\n", " page_number = i + 1\n", " page_text = page.get_text(\"text\").strip()\n", " images_info = []\n", "\n", " # 이미지 추출\n", " for img_index, img in enumerate(page.get_images(full=True)):\n", " xref = img[0]\n", " base_image = doc.extract_image(xref)\n", " image_bytes = base_image[\"image\"]\n", " image_ext = base_image[\"ext\"]\n", " image_filename = f\"page_{page_number}_img_{img_index+1}.{image_ext}\"\n", " image_path = os.path.join(output_dir, image_filename)\n", "\n", " with open(image_path, \"wb\") as img_file:\n", " img_file.write(image_bytes)\n", "\n", " images_info.append(image_path)\n", "\n", " # LangChain Document로 변환\n", " documents.append(Document(\n", " page_content=page_text,\n", " metadata={\n", " \"source\": os.path.basename(pdf_path),\n", " \"page\": page_number,\n", " \"images\": \", \".join(images_info)\n", " }\n", " ))\n", "\n", " # 병합 텍스트 저장용\n", " merged_text += f\"\\n\\n--- Page {page_number} ---\\n\\n{page_text}\"\n", "\n", " # 전체 텍스트 저장\n", " with open(merged_text_path, \"w\", encoding=\"utf-8\") as f:\n", " f.write(merged_text)\n", "\n", " return documents, merged_text_path\n", "\n", "pdf_path = \"data/모델튜닝.pdf\"\n", "docs, merged_path = extract_documents_from_pdf(pdf_path)\n", "\n", "print(f\"추출된 문서 페이지 수: {len(docs)}\")\n", "print(f\"병합된 텍스트 경로: {merged_path}\")\n", "print(docs[0]) # 하나 확인" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "%pip install -qU Pillow" ] }, { "cell_type": "code", "execution_count": 7, "metadata": {}, "outputs": [], "source": [ "from PIL import Image\n", "from pathlib import Path\n", "import shutil\n", "\n", "def check_and_resize_image_to_outdir(\n", " path: Path,\n", " outdir: Path,\n", " allowed_formats=(\"PNG\", \"JPEG\", \"WEBP\", \"BMP\"),\n", " max_bytes=20 * 1024 * 1024,\n", " max_length=2240,\n", " max_ratio=4.5,\n", " save_format=\"PNG\"\n", "):\n", " try:\n", " # 용량 초과 확인\n", " if path.stat().st_size > max_bytes:\n", " print(f\"[✘] 용량 초과: {path.name}\")\n", " return\n", "\n", " with Image.open(path) as image:\n", " format = image.format.upper()\n", " if format not in allowed_formats:\n", " print(f\"[✘] 포맷 불가: {path.name} ({format})\")\n", " return\n", "\n", " w, h = image.size\n", " ratio = max(w, h) / min(w, h)\n", " needs_resize = max(w, h) > max_length or ratio > max_ratio\n", "\n", " if not needs_resize:\n", " # 조건 만족 → 그대로 복사\n", " dest = outdir / path.name\n", " shutil.copy(path, dest)\n", " print(f\"[✓] 조건 만족 → 복사됨: {path.name}\")\n", " return\n", "\n", " # 리사이즈 크기 계산\n", " if ratio > max_ratio:\n", " if w > h:\n", " new_w = min(w, max_length)\n", " new_h = int(new_w / max_ratio)\n", " else:\n", " new_h = min(h, max_length)\n", " new_w = int(new_h / max_ratio)\n", " else:\n", " if w >= h:\n", " new_w = min(w, max_length)\n", " new_h = int(h * (new_w / w))\n", " else:\n", " new_h = min(h, max_length)\n", " new_w = int(w * (new_h / h))\n", "\n", " resized = image.resize((new_w, new_h), Image.LANCZOS).convert(\"RGB\")\n", " dest = outdir / path.name\n", " resized.save(dest, format=save_format, optimize=True)\n", " print(f\"[✔] 리사이즈됨 → 저장됨: {dest.name} ({new_w}x{new_h})\")\n", "\n", " except Exception as e:\n", " print(f\"[✘] 처리 실패: {path.name} → {e}\")\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "from pathlib import Path\n", "\n", "input_dir = Path(\"data/extracted_images_문서\")\n", "output_dir = Path(\"data/filtered_images\")\n", "output_dir.mkdir(parents=True, exist_ok=True)\n", "\n", "valid_exts = [\".png\", \".jpg\", \".jpeg\", \".webp\", \".bmp\"]\n", "image_files = [p for p in input_dir.glob(\"*\") if p.suffix.lower() in valid_exts]\n", "\n", "print(f\"총 {len(image_files)}개의 이미지 처리 시작\")\n", "\n", "for img_path in image_files:\n", " check_and_resize_image_to_outdir(img_path, outdir=output_dir)\n" ] }, { "cell_type": "code", "execution_count": 9, "metadata": {}, "outputs": [], "source": [ "# 네이버 클라우드에서 발급받은 키를 입력하세요\n", "os.environ[\"AWS_ACCESS_KEY_ID\"] = getpass.getpass(\"NCP Access Key: \")\n", "os.environ[\"AWS_SECRET_ACCESS_KEY\"] = getpass.getpass(\"NCP Secret Key: \")\n", "\n", "# 기본 리전 설정\n", "os.environ[\"AWS_DEFAULT_REGION\"] = \"kr\"" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "%pip install boto3" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "from glob import glob\n", "import boto3\n", "from botocore.client import Config\n", "from botocore.exceptions import ClientError\n", "import mimetypes\n", "\n", "# 설정\n", "BUCKET_NAME = \"multi-rag\"\n", "LOCAL_FOLDER = \"data/filtered_images\"\n", "ENDPOINT_URL = \"https://kr.ncloudstorage.com\"\n", "REGION = os.environ[\"AWS_DEFAULT_REGION\"]\n", "\n", "ACCESS_KEY = os.environ[\"AWS_ACCESS_KEY_ID\"]\n", "SECRET_KEY = os.environ[\"AWS_SECRET_ACCESS_KEY\"]\n", "\n", "# boto3 클라이언트 초기화\n", "s3 = boto3.client(\n", " \"s3\",\n", " aws_access_key_id=ACCESS_KEY,\n", " aws_secret_access_key=SECRET_KEY,\n", " endpoint_url=ENDPOINT_URL,\n", " region_name=REGION,\n", " config=Config(signature_version=\"s3v4\")\n", ")\n", "\n", "# 1. 버킷 생성\n", "try:\n", " s3.head_bucket(Bucket=BUCKET_NAME)\n", " print(f\"이미 존재하는 버킷입니다: {BUCKET_NAME}\")\n", "except ClientError as e:\n", " if e.response['Error']['Code'] == '404':\n", " print(f\"버킷이 존재하지 않아 생성합니다: {BUCKET_NAME}\")\n", " s3.create_bucket(Bucket=BUCKET_NAME)\n", " else:\n", " raise\n", "\n", "# 2. 이미지 수집\n", "IMAGE_EXTENSIONS = (\"*.jpeg\", \"*.jpg\", \"*.png\", \"*.bmp\", \"*.webp\")\n", "image_files = []\n", "\n", "for ext in IMAGE_EXTENSIONS:\n", " image_files.extend(glob(os.path.join(LOCAL_FOLDER, ext)))\n", "\n", "print(f\"총 {len(image_files)}개 이미지 파일을 찾았습니다.\")\n", "\n", "# 3. 이미지 업로드 및 URL 저장\n", "url_list = [] # 결과 저장할 리스트\n", "\n", "for file_path in image_files:\n", " file_name = os.path.basename(file_path)\n", "\n", " try:\n", " # 업로드\n", " s3.upload_file(file_path, BUCKET_NAME, file_name)\n", "\n", " # MIME 타입 추정\n", " mime_type, _ = mimetypes.guess_type(file_name)\n", " if not mime_type:\n", " mime_type = \"application/octet-stream\"\n", "\n", " # Signed URL 생성\n", " signed_url = s3.generate_presigned_url(\n", " \"get_object\",\n", " Params={\n", " \"Bucket\": BUCKET_NAME,\n", " \"Key\": file_name,\n", " \"ResponseContentDisposition\": \"inline\",\n", " \"ResponseContentType\": mime_type\n", " },\n", " ExpiresIn=3600\n", " )\n", "\n", " print(f\"URL: {signed_url}\")\n", " url_list.append(signed_url)\n", "\n", " except ClientError as e:\n", " print(f\"업로드 실패: {e}\")\n", "\n", "\n", "print(\"모든 이미지 업로드 및 링크 생성 완료!\")\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Convert" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "from langchain_core.messages import SystemMessage, HumanMessage\n", "from langchain_naver import ChatClovaX\n", "\n", "chat_llm = ChatClovaX(\n", " model=\"HCX-005\"\n", ")\n", "\n", "# 이미지 URL\n", "image_url = url_list[-1]\n", "\n", "# System, User prompt 구성\n", "system_message = SystemMessage(\n", " content=(\n", " \"당신은 문서 내 다양한 형태의 이미지를 분석하여, 검색 기반 질문응답 시스템(RAG)에 활용 가능한 텍스트 설명을 생성하는 AI입니다.\"\n", " \"이미지는 인포그래픽, 표, 그래프, 코드 캡처, 다이어그램, 화면 구성 등 다양한 유형일 수 있으며, 다음 기준에 따라 요약을 작성하세요.\"\n", " \"- 이미지의 주제와 목적을 명확하게 파악하고 자연어로 요약합니다.\"\n", " \"- 이미지가 전달하는 구조나 흐름이 있다면 순차적으로 설명합니다. (예: 단계, 관계, 비교 등)\"\n", " \"- 표, 그래프, 수치 정보는 전체 흐름과 특징적인 차이만 요약하고, 수치 나열은 피합니다.\"\n", " \"- 코드 캡처인 경우 기능과 역할 중심으로 요약하며, 함수/변수/모듈명 등 핵심 정보만 포함합니다.\"\n", " \"- 시각적 요소(색상, 도형, 배치 등)는 정보 전달에 필요할 경우에만 간단히 설명합니다.\"\n", " \"- OCR로 추출된 텍스트가 있다면 핵심 내용 위주로 정리하여 포함합니다.\"\n", " \"- 설명은 검색 가능한 핵심 키워드를 포함하고, 감상이나 해석 없이 사실 중심 문장으로 구성해야 합니다.\"\n", " \"- 최종 출력은 3~5문장 이내의 단일 문단으로 구성되며, RAG 시스템의 컨텍스트로 직접 활용 가능해야 합니다.\"\n", " )\n", ")\n", "human_message = HumanMessage(content=[\n", " {\"type\": \"text\", \"text\": \"이 이미지는 문서 내 시각 자료입니다. 핵심 정보를 요약해 주세요.\"},\n", " {\"type\": \"image_url\", \"image_url\": {\"url\": image_url}}\n", " ])\n", "\n", "# 메시지 구성\n", "messages = [\n", " system_message,\n", " human_message\n", " ]\n", "\n", "# 파라미터 설정\n", "config={\n", " \"generation_config\": {\n", " \"temperature\": 0.25,\n", " \"repetition_penalty\": 1.1\n", " }\n", " }\n", "\n", "# 모델 호출\n", "response = chat_llm.invoke(messages,config)\n", "print(\"[CLOVA 응답]\\n\", response.content)\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# 결과를 저장할 딕셔너리\n", "image_summary_results = []\n", "\n", "# URL 반복 → 프롬프트 생성 → 모델 호출 → 딕셔너리 저장\n", "for url in url_list:\n", " file_name = os.path.basename(url)\n", " clean_filename = file_name.split(\"?\")[0]\n", " try:\n", " # URL만 바꿔서 human_message 재생성\n", " human_message.content[1][\"image_url\"][\"url\"] = url\n", " messages = [system_message, human_message]\n", " response = chat_llm.invoke(messages,config)\n", "\n", " # 결과 딕셔너리에 저장\n", " image_summary_results.append({clean_filename: response.content})\n", " print(f\"[✔] 저장 완료: {url}\")\n", "\n", " except Exception as e:\n", " print(f\"[✘] 실패: {url} → {e}\")\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "이미지를 document 변환" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "image_docs = []\n", "for item in image_summary_results:\n", " # 각 딕셔너리에서 파일명과 요약 텍스트 추출\n", " file_name = list(item.keys())[0]\n", " summary = item[file_name]\n", "\n", " # 정규식으로 페이지 번호 추출\n", " match = re.search(r'page_(\\d+)_img_\\d+\\.\\w+', file_name)\n", " page_number = int(match.group(1)) if match else None\n", "\n", " # LangChain Document 생성\n", " image_docs.append(Document(\n", " page_content=summary,\n", " metadata={\n", " \"source\": \"모델튜닝.pdf\",\n", " \"page\": page_number,\n", " \"images\": file_name\n", " }\n", " ))\n", "\n", "print(f\"총 {len(image_docs)}개의 Document 생성 완료\")\n", "print(image_docs[0].page_content)\n", "print(image_docs[0].metadata) # 하나 확인" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Chunking" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# -*- coding: utf-8 -*-\n", "\n", "class CompletionExecutor:\n", " def __init__(self, host, api_key, request_id):\n", " self._host = host\n", " self._api_key = api_key\n", " self._request_id = request_id\n", "\n", " def _send_request(self, completion_request):\n", " headers = {\n", " 'Content-Type': 'application/json; charset=utf-8',\n", " 'Authorization': self._api_key,\n", " 'X-NCP-CLOVASTUDIO-REQUEST-ID': self._request_id\n", " }\n", "\n", " conn = http.client.HTTPSConnection(self._host)\n", " conn.request('POST', '/testapp/v1/api-tools/segmentation', json.dumps(completion_request), headers)\n", " response = conn.getresponse()\n", " result = json.loads(response.read().decode(encoding='utf-8'))\n", " conn.close()\n", " return result\n", "\n", " def execute(self, completion_request):\n", " res = self._send_request(completion_request)\n", " if res['status']['code'] == '20000':\n", " return res['result']['topicSeg']\n", " else:\n", " print(\"[CLOVA 응답 오류]\", res['status'])\n", " return 'Error'\n", " \n", "file_path = \"data/extracted_images_문서/merged_text.txt\"\n", "\n", "with open(file_path, \"r\", encoding=\"utf-8\") as f:\n", " text_content = f.read()\n", "\n", "if __name__ == '__main__':\n", " completion_executor = CompletionExecutor(\n", " host='clovastudio.streahttp://m.ntruss.com',\n", " api_key=\"Bearer \"+os.environ[\"CLOVASTUDIO_API_KEY\"], # 여기 키 형식이 Bearer이 붙네요 \n", " request_id=str(uuid.uuid4())\n", " )\n", "\n", " chunked_docs = []\n", "\n", " for doc in docs: # docs는 페이지별로 추출한 Document 리스트\n", " segments = completion_executor.execute(\n", " # 이전 블로그 참고해 파라미터 설정\n", " {\"postProcessMaxSize\": 100, # 후처리 시 하나의 문단이 가질 수 있는 최대 글자 수 (예: 1000자 이하로 잘라줌)\n", " \"alpha\": -100, # 문단 나누기 민감도 조절 파라미터 (기본: 0.0 / -100으로 두면 자동 조정) - 값이 클수록 더 잘게 나뉘고, 작을수록 덜 나뉨\n", " \"segCnt\": -1, # 원하는 문단 개수 설정 (-1이면 자동 분할, 1 이상의 정수 입력 시 해당 개수로 고정)\n", " \"postProcessMinSize\": -1, # 후처리 시 하나의 문단이 가져야 할 최소 글자 수 (예: 300자 이상 유지)\n", " \"text\": doc.page_content, # 실제 분할할 원본 텍스트\n", " \"postProcess\": True} # 후처리 여부 설정 (True: 문단 길이 균일화 / False: 모델 출력 그대로 사용)\n", " )\n", "\n", " for seg in segments:\n", " chunked_docs.append(Document(\n", " page_content=' '.join(seg),\n", " metadata=doc.metadata\n", " )) \n", "\n", " print(chunked_docs)\n", " print(\"chunk 개수 :\",len(chunked_docs))" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# image_docs를 chunked_docs에 추가 (원본은 그대로 유지)\n", "combined_docs = chunked_docs + image_docs\n", "\n", "print(f\"전체 chunk 개수: {len(combined_docs)}\")\n", "print(combined_docs)" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# 샘플 청크 출력\n", "print(\"\\n샘플 청크 (처음 3개):\")\n", "for i, chunk in enumerate(combined_docs[:3], 0):\n", " print(f\"\\n청크 {i+1}:\")\n", " print(f\"내용: {chunk.page_content}\")\n", " print(f\"metadata: {chunk.metadata}\")\n", " print(f\"길이: {len(chunk.page_content)}자\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Embedding" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "from langchain_naver import ClovaXEmbeddings\n", " \n", "clovax_embeddings = ClovaXEmbeddings(model='bge-m3') # 임베딩 모델을 설정\n", "\n", "text = \"임베딩 사용 예제입니다~\"\n", " \n", "clovax_embeddings.embed_query(text)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Vector Store" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "#chroma 다운받기\n", "%pip install -qU langchain-chroma" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "import chromadb\n", "from langchain_chroma import Chroma\n", "\n", "\n", "# 임베딩 모델 정의\n", "clovax_embeddings = ClovaXEmbeddings(model='bge-m3')\n", "\n", "# 로컬 클라이언트 생성\n", "client = chromadb.PersistentClient(path=\"./Chroma_langchain_db123\")\n", "\n", "# 컬렉션 준비 (이름 중복 주의!)\n", "collection_name = \"clovastudiodatas_docs\"\n", "client.get_or_create_collection(\n", " name=collection_name,\n", " metadata={\"hnsw:space\": \"cosine\"}\n", ")\n", "\n", "# 벡터스토어 객체 생성\n", "vectorstore_Chroma = Chroma(\n", " client=client,\n", " collection_name=collection_name,\n", " embedding_function=clovax_embeddings\n", ")\n", "\n", "# 문서 추가: 최신 방식은 vectorstore.add_documents 사용\n", "print(\"Adding documents to Chroma vectorstore...\")\n", "for doc in combined_docs:\n", " try:\n", " vectorstore_Chroma.add_documents([doc])\n", " time.sleep(0.5) \n", " except Exception as e:\n", " print(f\"[✘] 실패: {doc.metadata} → {e}\")\n", "\n", "print(\"All documents have been added to the vectorstore.\")\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "#FAISS 다운로드\n", "%pip install -qU langchain-community faiss-cpu" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "import faiss\n", "from langchain_community.vectorstores import FAISS\n", "from langchain_community.docstore.in_memory import InMemoryDocstore\n", "\n", "# 임베딩 모델 정의\n", "clovax_embeddings = ClovaXEmbeddings(model='bge-m3')\n", "\n", "# FAISS 인덱스 생성 (1024는 bge-m3 차원 수에 맞춰야 함)\n", "index = faiss.IndexFlatIP(1024) # 내적 기반 검색\n", "\n", "# FAISS 벡터스토어 생성\n", "vectorstore_FAISS = FAISS(\n", " embedding_function=clovax_embeddings,\n", " index=index,\n", " docstore=InMemoryDocstore(),\n", " index_to_docstore_id={}\n", ")\n", "\n", "# 문서 일괄 추가 (자동 임베딩 처리)\n", "print(\"Adding documents to FAISS vectorstore...\")\n", "for doc in combined_docs:\n", " try:\n", " vectorstore_FAISS.add_documents([doc])\n", " time.sleep(0.5) \n", " except Exception as e:\n", " print(f\"[✘] 실패: {doc.metadata} → {e}\")\n", "print(\"All documents have been added to FAISS vectorstore.\")\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 3. 질의하기" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### 질문하기" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "from langchain_core.prompts import ChatPromptTemplate, SystemMessagePromptTemplate, HumanMessagePromptTemplate\n", "from langchain.chains import RetrievalQA\n", "\n", "# System 및 User 메시지를 나눠 구성\n", "system_template = (\n", " \"당신은 질문-답변(Question-Answering)을 수행하는 친절한 AI 어시스턴트입니다. 당신의 임무는 원래 가지고있는 지식은 모두 배제하고, 주어진 문맥(context) 에서 주어진 질문(question) 에 답하는 것입니다.\"\n", " \"만약, 주어진 문맥(context) 에서 답을 찾을 수 없다면, 답을 모른다면 `주어진 정보에서 질문에 대한 정보를 찾을 수 없습니다` 라고 답하세요.\"\n", ")\n", "user_template = (\n", " \"다음은 검색된 문서 내용입니다:\\n\\n{context}\\n\\n\"\n", " \"위 정보를 바탕으로 다음 질문에 답해주세요:\\n{question}\"\n", ")\n", "\n", "prompt_template = ChatPromptTemplate.from_messages([\n", " SystemMessagePromptTemplate.from_template(system_template),\n", " HumanMessagePromptTemplate.from_template(user_template),\n", "])\n", "\n", "# 원하는 vectorstore 선택해서 사용\n", "retriever = vectorstore_Chroma.as_retriever(\n", " search_type=\"similarity_score_threshold\",\n", " search_kwargs={\"score_threshold\": 0.1, \"k\": 3}\n", " )\n", "# retriever = vectorstore_FAISS.as_retriever(\n", "# search_type=\"similarity_score_threshold\",\n", "# search_kwargs={\"score_threshold\": 0.1, \"k\": 3}\n", "# )\n", "\n", "# Retrieval QA 체인 구성\n", "qa_chain = RetrievalQA.from_chain_type(\n", " llm=chat_llm,\n", " chain_type=\"stuff\",\n", " retriever=retriever,\n", " chain_type_kwargs={\"prompt\": prompt_template},\n", " return_source_documents=True\n", ")\n", "\n", "# 실행\n", "question = \"데이터셋 규모가 커질수록 2대륙의 오류 발생 확률은 어떻게 돼?\"\n", "result = qa_chain.invoke({\"query\": question})\n", "\n", "print(\"질문:\", question)\n", "print(\"응답:\", result[\"result\"]) # 모델의 실제 응답\n", "for i, doc in enumerate(result[\"source_documents\"]): # 답변시 참고 한 문서\n", " print(f\"\\n[출처 문서 {i+1}]\\n내용: {doc.page_content}\\n메타데이터: {doc.metadata}\")" ] } ], "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.12.2" } }, "nbformat": 4, "nbformat_minor": 2 }
1. 사전준비
① Langchain 패키지 설치 멀티모달 RAG 시스템 구현을 위해서는 LangChain 프레임워크와 CLOVA Studio API 연동이 필요합니다. 최근 출시된 langchain-naver 패키지를 통해 CLOVA Studio의 최신 비전 모델 HCX-005를 LangChain과 원활하게 연동할 수 있습니다. 아래 명령어로 LangChain 관련 패키지를 설치하세요.
%pip install -qU openai langchain langchain-naver
② 코드 공통 모듈 imports 필요한 기본 모듈들을 미리 import합니다.
import os
import getpass
import uuid
import re
from urllib.parse import urlparse
import http
import json
import time
③ API 키 발급 받기 CLOVA Studio의 API 키 발급이 필요합니다."프로필 > API 키 > 테스트 > 테스트 앱 발급" 경로를 통해 키를 발급받을 수 있습니다. 발급된 키는 한 번만 표시되어 재확인이 불가능하므로, 반드시 복사하여 별도로 안전하게 보관해야 합니다
"익스플로러 > 문단나누기, 임베딩 > 테스트 앱 생성"으로 이동하여, 적절한 이름의 테스트 앱을 생성합니다. 이 앱은 추후 chunking과 embedding 과정에서 활용됩니다.
발급받은 API KEY는 환경 변수로 저장하여 관리합니다.
os.environ["CLOVASTUDIO_API_KEY"] = getpass.getpass("CLOVA Studio API Key: ")
멀티모달 RAG 시스템에서는 텍스트뿐만 아니라 이미지 등 다양한 형태의 데이터를 분할하고, 이를 벡터로 변환하여 검색에 활용해야 합니다. PDF 문서는 일반 텍스트 외에도 그래프, 테이블과 같은 시각적 요소가 포함되어 있어, 텍스트 기반 RAG보다 더 정교하고 복합적인 전처리 과정이 요구됩니다.
① PDF 문서에서 텍스트와 이미지 추출하기 (Load)
PDF 파일에서 텍스트와 이미지를 각각 추출하는 과정이 필요합니다.현재 멀티모달 문서의 정보 추출을 지원하는 대표적인 라이브러리로는PyPDF,PyMuPDF,LlamaParse,Unstructured.io,TorchMultimodal등이 있습니다. 각 라이브러리는 텍스트 추출 정확도, 이미지 처리 방식, 구조 보존 여부 등 구현 방식에서 차이를 보입니다. 따라서 특정 라이브러리가 절대적으로 우수하다고 보긴 어렵습니다.본 예제에서는 PyMuPDF를 활용하여 페이지별로 이미지를 저장하고 텍스트를 구조화하는 방식을 채택했습니다.
%pip install pymupdf
아래 코드는 PDF 파일에서 추출한 이미지와 텍스트를 상대 경로 기준의 지정된 폴더(output_dir)에 저장하도록 구성되어 있습니다. 이 과정을 통해 각 페이지에 포함된 이미지와 텍스트 추출 결과를 직접 확인할 수 있습니다. 실행 후 전체 디렉토리 구조는 다음과 같이 구성됩니다.
텍스트는 페이지 단위로 정리되며, 이후 LangChain의 Document 객체로 변환되어 임베딩 처리에 활용됩니다. 이미지는 "page_{page_number}img{img_index}.{image_ext}" 형식으로 저장됩니다. 이러한 명명 규칙은 이미지가 어느 페이지에서 추출되었는지 쉽게 추적할 수 있게 하며, 이후 메타데이터로 활용하기에도 매우 효과적입니다. 이렇게 구성된 문서는 향후 검색 기반 질의응답(RAG) 시스템의 컨텍스트로 활용됩니다.
import fitz # PyMuPDF
from langchain_core.documents import Document
def extract_documents_from_pdf(pdf_path: str, output_dir: str = "data/extracted_images_문서"):
os.makedirs(output_dir, exist_ok=True)
merged_text_path = os.path.join(output_dir, "merged_text.txt")
merged_text = ""
doc = fitz.open(pdf_path)
documents = []
for i, page in enumerate(doc):
page_number = i + 1
page_text = page.get_text("text").strip()
images_info = []
# 이미지 추출
for img_index, img in enumerate(page.get_images(full=True)):
xref = img[0]
base_image = doc.extract_image(xref)
image_bytes = base_image["image"]
image_ext = base_image["ext"]
image_filename = f"page_{page_number}_img_{img_index+1}.{image_ext}"
image_path = os.path.join(output_dir, image_filename)
with open(image_path, "wb") as img_file:
img_file.write(image_bytes)
images_info.append(image_path)
# LangChain Document로 변환
documents.append(Document(
page_content=page_text,
metadata={
"source": os.path.basename(pdf_path),
"page": page_number,
"images": ", ".join(images_info)
}
))
# 병합 텍스트 저장용
merged_text += f"\n\n--- Page {page_number} ---\n\n{page_text}"
# 전체 텍스트 저장
with open(merged_text_path, "w", encoding="utf-8") as f:
f.write(merged_text)
return documents, merged_text_path
pdf_path = "data/모델튜닝.pdf" # 다른 파일로 테스트할 경우 알맞은 경로 입력
docs, merged_path = extract_documents_from_pdf(pdf_path)
print(f"추출된 문서 페이지 수: {len(docs)}")
print(f"병합된 텍스트 경로: {merged_path}")
print(docs[0]) # 하나 확인
결과
② 이미지 → 텍스트 요약하기 (Convert)
PDF에서 추출한 이미지는 단순히 저장하는 것만으로는 검색이나 응답 생성에 즉시 활용하기 어렵습니다. 이미지 내용을 검색 가능한 정보로 변환하기 위해, CLOVA Studio의 비전 모델을 활용하여 시각 정보를 텍스트로 요약합니다. 이렇게 생성된 설명은 이후 RAG 시스템에서 문맥(Context)으로 활용될 수 있도록 구성됩니다.
2.1) HyperCLOVA X 비전 모델을 위한 이미지 전처리 CLOVA 비전 모델 사용 시 다음과 같은 이미지 제한 사항이 있습니다.
입력 조건 및 업로드 사양
지원 포맷: PNG, JPEG, WEBP, BMP
파일 용량 제한: 이미지당 최대 20 MB
최대 사이즈: 긴 변 기준 2240px 이하
가로:세로 비율 제한: 1:5또는 5:1이하
참고:2025년 4월 17일 기준, CLOVA Studio의HCX-005 모델은 사용자 한 턴에최대 1장의 이미지만 입력할 수 있지만,요청 한 번에 최대 5장의 이미지를 포함한 메시지 입력은 가능합니다.
PDF에서 추출한 이미지는 해상도가 매우 크거나 가로·세로 비율이 비정상적으로 긴 경우가 많습니다. 이러한 이미지를 CLOVA Studio에 그대로 입력하면 'Invalid image ratio' 에러가 발생할 수 있습니다. 이러한 오류를 사전에 방지하기 위해 이미지를 검사하고 리사이즈하는 전처리 과정을 수행하는 것이 필요합니다.
%pip install -qU Pillow
아래 함수는 하나의 로컬 이미지 경로를 입력받아 조건을 만족하면 output_dir에 그대로 복사하고, 조건에 맞지 않으면 리사이즈 후 output_dir에 저장하는 역할을 수행합니다.
from PIL import Image
from pathlib import Path
import shutil
def check_and_resize_image_to_outdir(
path: Path,
outdir: Path,
allowed_formats=("PNG", "JPEG", "WEBP", "BMP"),
max_bytes=20 * 1024 * 1024,
max_length=2240,
max_ratio=4.5,
save_format="PNG"
):
try:
# 용량 초과 확인
if path.stat().st_size > max_bytes:
print(f"[✘] 용량 초과: {path.name}")
return
with Image.open(path) as image:
format = image.format.upper()
if format not in allowed_formats:
print(f"[✘] 포맷 불가: {path.name} ({format})")
return
w, h = image.size
ratio = max(w, h) / min(w, h)
needs_resize = max(w, h) > max_length or ratio > max_ratio
if not needs_resize:
# 조건 만족 → 그대로 복사
dest = outdir / path.name
shutil.copy(path, dest)
print(f"[✓] 조건 만족 → 복사됨: {path.name}")
return
# 리사이즈 크기 계산
if ratio > max_ratio:
if w > h:
new_w = min(w, max_length)
new_h = int(new_w / max_ratio)
else:
new_h = min(h, max_length)
new_w = int(new_h / max_ratio)
else:
if w >= h:
new_w = min(w, max_length)
new_h = int(h * (new_w / w))
else:
new_h = min(h, max_length)
new_w = int(w * (new_h / h))
resized = image.resize((new_w, new_h), Image.LANCZOS).convert("RGB")
dest = outdir / path.name
resized.save(dest, format=save_format, optimize=True)
print(f"[✔] 리사이즈됨 → 저장됨: {dest.name} ({new_w}x{new_h})")
except Exception as e:
print(f"[✘] 처리 실패: {path.name} → {e}")
아래 메인 실행 코드를 통해 CLOVA Studio의 비전 모델 기준에 적합한 안전한 이미지 셋을 구성할 수 있습니다.
📁cookbook/ ├── multimodal_RAG.ipynb/ ├── data/ │ ├── 모델튜닝.pdf │ ├── 스킬.pdf │ ├── extracted_images_문서/ ← PDF에서 추출된 원본 이미지 │ └── filtered_images/ ← 조건에 맞는 이미지가 저장되는 곳 (output_dir)
from pathlib import Path
input_dir = Path("data/extracted_images_문서")
output_dir = Path("data/filtered_images")
output_dir.mkdir(parents=True, exist_ok=True)
valid_exts = [".png", ".jpg", ".jpeg", ".webp", ".bmp"]
image_files = [p for p in input_dir.glob("*") if p.suffix.lower() in valid_exts]
print(f"총 {len(image_files)}개의 이미지 처리 시작")
for img_path in image_files:
check_and_resize_image_to_outdir(img_path, outdir=output_dir)
결과
2.2) Ncloud Storage 사용해서 이미지 저장하기 CLOVA Studio의 비전 모델은 로컬 경로의 이미지 파일이 아닌, 웹 URL 형태의 이미지를 입력값으로 지원합니다. 따라서 모델을 활용하기 위해서는 먼저 추출한 이미지를 객체 스토리지(Ncloud Storage, S3, 구글 드라이브 등)에 업로드한 후, 해당 URL을 수집하여 정리하는 작업이 필요합니다. 이번 cookbook에서는 Ncloud Storage를 사용하겠습니다. (참고 :Ncloud Storage 가이드)
참고: 2025년 4월 17일 기준,Ncloud Storage 상품은 OBT(Open Beta)로 제공됩니다. Ncloud Storage는 Open Beta 기간동안 용량 제한 없이무료로 이용이 가능합니다. Open Beta 기간 종료 후에는 저장된 데이터의 저장 용량과 API 요청에 대해 과금으로 전환됩니다. Open Beta 기간에 서비스 제공에 대한 SLA는 보장되지 않습니다.
# 네이버 클라우드에서 발급받은 키를 입력하세요
os.environ["AWS_ACCESS_KEY_ID"] = getpass.getpass("NCP Access Key: ")
os.environ["AWS_SECRET_ACCESS_KEY"] = getpass.getpass("NCP Secret Key: ")
# 기본 리전 설정
os.environ["AWS_DEFAULT_REGION"] = "kr"
Ncloud Object Storage는 Amazon S3 API와 호환되며,Python에서는 이를 제어하기 위해 'boto3' 라이브러리를 사용합니다.
%pip install boto3
다음은 새로운 버킷을 생성하고 전처리한 이미지를 모두 업로드하는 코드입니다. 버킷 이름은 최소 3자에서 최대 63자로 구성할 수 있으며, 소문자, 숫자 및 하이픈(-)만 포함해야 합니다. 예시에서는 'multi-rag'를 버킷 이름으로 사용했습니다.
from glob import glob
import boto3
from botocore.client import Config
from botocore.exceptions import ClientError
import mimetypes
# 설정
BUCKET_NAME = "multi-rag"
LOCAL_FOLDER = "data/filtered_images"
ENDPOINT_URL = "https://kr.ncloudstorage.com"
REGION = os.environ["AWS_DEFAULT_REGION"]
ACCESS_KEY = os.environ["AWS_ACCESS_KEY_ID"]
SECRET_KEY = os.environ["AWS_SECRET_ACCESS_KEY"]
# boto3 클라이언트 초기화
s3 = boto3.client(
"s3",
aws_access_key_id=ACCESS_KEY,
aws_secret_access_key=SECRET_KEY,
endpoint_url=ENDPOINT_URL,
region_name=REGION,
config=Config(signature_version="s3v4")
)
# 1. 버킷 생성
try:
s3.head_bucket(Bucket=BUCKET_NAME)
print(f"이미 존재하는 버킷입니다: {BUCKET_NAME}")
except ClientError as e:
if e.response['Error']['Code'] == '404':
print(f"버킷이 존재하지 않아 생성합니다: {BUCKET_NAME}")
s3.create_bucket(Bucket=BUCKET_NAME)
else:
raise
# 2. 이미지 수집
IMAGE_EXTENSIONS = ("*.jpeg", "*.jpg", "*.png", "*.bmp", "*.webp")
image_files = []
for ext in IMAGE_EXTENSIONS:
image_files.extend(glob(os.path.join(LOCAL_FOLDER, ext)))
print(f"총 {len(image_files)}개 이미지 파일을 찾았습니다.")
# 3. 이미지 업로드 및 URL 저장
url_list = [] # 결과 저장할 리스트
for file_path in image_files:
file_name = os.path.basename(file_path)
try:
# 업로드
s3.upload_file(file_path, BUCKET_NAME, file_name)
# MIME 타입 추정
mime_type, _ = mimetypes.guess_type(file_name)
if not mime_type:
mime_type = "application/octet-stream"
# Signed URL 생성
signed_url = s3.generate_presigned_url(
"get_object",
Params={
"Bucket": BUCKET_NAME,
"Key": file_name,
"ResponseContentDisposition": "inline",
"ResponseContentType": mime_type
},
ExpiresIn=3600 #1시간만
)
print(f"URL: {signed_url}")
url_list.append({signed_url})
except ClientError as e:
print(f"업로드 실패: {e}")
print("모든 이미지 업로드 및 링크 생성 완료!")
결과
2.3) 비전 모델 사용해서 이미지 요약 하기 PDF 문서를 분석하다 보면 인포그래픽, 그래프, 테이블, 코드 캡처와 같은 다양한 형태의 이미지들이 등장합니다. 이미지가 담고 있는 정보의 형식과 내용이 이미지별로 상이하기 때문에, 요약 단계에서도 이미지 유형에 맞는 접근법이 필요합니다. 프롬프트뿐만 아니라 함께 사용하는 파라미터 설정(config) 또한 요약 품질에 큰 영향을 미칩니다. 실험을 통해 비교적 안정적이고 일관된 결과를 보여준 설정값을 함께 제시했지만, 이미지 특성이나 문서 도메인에 따라 적절한 조합은 달라질 수 있습니다. 따라서 사용하는 프롬프트와 이미지 유형에 맞게 config 값을 직접 조정해보는 것을 권장합니다.
아래는 문서 내 이미지를 다룰 때 일반적으로 활용되는 범용 프롬프트와 파라미터 값의 예시입니다.
from langchain_core.messages import SystemMessage, HumanMessage
from langchain_naver import ChatClovaX
chat_llm = ChatClovaX(
model="HCX-005"
)
# 이미지 URL
image_url = url_list[-1]
# System, User prompt 구성
system_message = SystemMessage(
content=(
"당신은 문서 내 다양한 형태의 이미지를 분석하여, 검색 기반 질문응답 시스템(RAG)에 활용 가능한 텍스트 설명을 생성하는 AI입니다."
"이미지는 인포그래픽, 표, 그래프, 코드 캡처, 다이어그램, 화면 구성 등 다양한 유형일 수 있으며, 다음 기준에 따라 요약을 작성하세요."
"- 이미지의 주제와 목적을 명확하게 파악하고 자연어로 요약합니다."
"- 이미지가 전달하는 구조나 흐름이 있다면 순차적으로 설명합니다. (예: 단계, 관계, 비교 등)"
"- 표, 그래프, 수치 정보는 전체 흐름과 특징적인 차이만 요약하고, 수치 나열은 피합니다."
"- 코드 캡처인 경우 기능과 역할 중심으로 요약하며, 함수/변수/모듈명 등 핵심 정보만 포함합니다."
"- 시각적 요소(색상, 도형, 배치 등)는 정보 전달에 필요할 경우에만 간단히 설명합니다."
"- OCR로 추출된 텍스트가 있다면 핵심 내용 위주로 정리하여 포함합니다."
"- 설명은 검색 가능한 핵심 키워드를 포함하고, 감상이나 해석 없이 사실 중심 문장으로 구성해야 합니다."
"- 최종 출력은 3~5문장 이내의 단일 문단으로 구성되며, RAG 시스템의 컨텍스트로 직접 활용 가능해야 합니다."
)
)
human_message = HumanMessage(content=[
{"type": "text", "text": "이 이미지는 문서 내 시각 자료입니다. 핵심 정보를 요약해 주세요."},
{"type": "image_url", "image_url": {"url": image_url}}
])
# 메시지 구성
messages = [
system_message,
human_message
]
# 파라미터 설정
config={
"generation_config": {
"temperature": 0.25,
"repetition_penalty": 1.1
}
}
# 모델 호출
response = chat_llm.invoke(messages,config)
print("[CLOVA 응답]\n", response.content)
결과
[CLOVA 응답]
이 이미지는 'Tuning'이라는 제목 아래 파란색 계열의 그래픽으로 구성된 시각 자료를 보여줍니다. 배경은 짙은 남색이며 상단에는 한글로 된 설명문이 있습니다. 이 설명문은 프롬프트만으로 성능 향상에 한계가 있어 자체 조달한 커스텀 데이터를 활용한 모델 튜닝(Tuning) 과정을 거쳐 성능을 더욱 향상시킬 수 있다는 내용을 담고 있습니다.
하단의 파동 모양 그래프는 시간에 따른 성능 향상을 나타내며, 각 지점마다 '1차', '2차', '3차'라는 텍스트가 표시되어 여러 번의 개선 단계를 의미합니다. 또한 하단에는 원형의 아이콘이 세 개 있으며 각각 '엔지니어링'이라는 단어와 화살표가 연결되어 있어, 이러한 엔지니어링 작업이 반복되면서 성능이 점차 향상됨을 시각적으로 표현했습니다. 오른쪽 끝부분에는 '튜닝'이라고 적힌 큰 원형 버튼이 강조되어 있으며 이는 최종적인 성능 최적화를 상징합니다. 전체적으로 이 이미지는 특정 과정에서의 지속적인 개선 및 최적화의 중요성을 나타내는 것으로 보입니다.
아래는 각 이미지 유형에 특화된 프롬프트를 적용해 생성한 예시 결과로, 이미지의 유형에 따라 요약 방식이 어떻게 달라지는지 비교해볼 수 있도록 구성했습니다.
아래 코드는 모든 이미지에 대해 요약을 생성하는 코드입니다.
# 결과를 저장할 딕셔너리
image_summary_results = []
# URL 반복 → 프롬프트 생성 → 모델 호출 → 딕셔너리 저장
for url in url_list:
file_name = os.path.basename(url)
clean_filename = file_name.split("?")[0]
try:
# URL만 바꿔서 human_message 재생성
human_message.content[1]["image_url"]["url"] = url
messages = [system_message, human_message]
response = chat_llm.invoke(messages,config)
# 결과 딕셔너리에 저장
image_summary_results.append({clean_filename: response.content})
print(f"[✔] 저장 완료: {url}")
except Exception as e:
print(f"[✘] 실패: {url} → {e}")
결과
2.4) 이미지 요약 텍스트를 Document 형식으로 변환하기 아래 코드는 이미지로부터 생성된 설명 텍스트(content)와 이미지의 위치 정보 등 메타데이터를 함께 담아 LangChain의 Document 형식으로 변환하는 과정입니다. 이전 이미지 추출 단계에서 파일명에 페이지 번호가 포함되도록 구성했습니다. 이를 활용해 이미지의 위치 정보를 메타데이터로 구성할 수 있습니다.
이렇게 변환된 Document는 텍스트 문단과 동일한 방식으로 벡터 임베딩이 가능하며, 이미지에서 추출된 정보 역시 텍스트 기반 질의처럼 검색되고 응답에 반영될 수 있습니다. 이는 멀티모달 RAG 구조의 핵심적인 전처리 단계입니다.
image_docs = []
for item in image_summary_results:
# 각 딕셔너리에서 파일명과 요약 텍스트 추출
file_name = list(item.keys())[0]
summary = item[file_name]
# 정규식으로 페이지 번호 추출
match = re.search(r'page_(\d+)_img_\d+\.\w+', file_name)
page_number = int(match.group(1)) if match else None
# LangChain Document 생성
image_docs.append(Document(
page_content=summary,
metadata={
"source": "모델튜닝.pdf",
"page": page_number,
"images": file_name
}
))
print(f"총 {len(image_docs)}개의 Document 생성 완료")
print(image_docs[0].page_content)
print(image_docs[0].metadata) # 하나 확인
결과
③ 문단 나누기 (Chunking)
텍스트와 이미지 각각의 정보를 검색에 적합한 단위로 분할하는 것이 매우 중요합니다. 너무 긴 텍스트는 검색 정확도를 저하시키고, 지나치게 잘게 쪼개면 문맥이 단절될 수 있어 적절한 분할 기준이 필요합니다. 이번에는 Clova Studio에서 제공하는 문단 나누기 API를 활용하여 자연스럽고 의미 단위로 구분된 문서 청크를 생성합니다.
# -*- coding: utf-8 -*-
class CompletionExecutor:
def __init__(self, host, api_key, request_id):
self._host = host
self._api_key = api_key
self._request_id = request_id
def _send_request(self, completion_request):
headers = {
'Content-Type': 'application/json; charset=utf-8',
'Authorization': self._api_key,
'X-NCP-CLOVASTUDIO-REQUEST-ID': self._request_id
}
conn = http.client.HTTPSConnection(self._host)
conn.request('POST', '/testapp/v1/api-tools/segmentation', json.dumps(completion_request), headers)
response = conn.getresponse()
result = json.loads(response.read().decode(encoding='utf-8'))
conn.close()
return result
def execute(self, completion_request):
res = self._send_request(completion_request)
if res['status']['code'] == '20000':
return res['result']['topicSeg']
else:
print("[CLOVA 응답 오류]", res['status'])
return 'Error'
file_path = "data/extracted_images_문서/merged_text.txt"
with open(file_path, "r", encoding="utf-8") as f:
text_content = f.read()
if __name__ == '__main__':
completion_executor = CompletionExecutor(
host='clovastudio.stream.ntruss.com',
api_key="Bearer "+os.environ["CLOVASTUDIO_API_KEY"],
request_id=str(uuid.uuid4())
)
chunked_docs = []
for doc in docs: # docs는 페이지별로 추출한 Document 리스트
segments = completion_executor.execute(
# 이전 블로그 참고해 파라미터 설정
{"postProcessMaxSize": 100, # 후처리 시 하나의 문단이 가질 수 있는 최대 글자 수 (예: 1000자 이하로 잘라줌)
"alpha": -100, # 문단 나누기 민감도 조절 파라미터 (기본: 0.0 / -100으로 두면 자동 조정) - 값이 클수록 더 잘게 나뉘고, 작을수록 덜 나뉨
"segCnt": -1, # 원하는 문단 개수 설정 (-1이면 자동 분할, 1 이상의 정수 입력 시 해당 개수로 고정)
"postProcessMinSize": -1, # 후처리 시 하나의 문단이 가져야 할 최소 글자 수 (예: 300자 이상 유지)
"text": doc.page_content, # 실제 분할할 원본 텍스트
"postProcess": True} # 후처리 여부 설정 (True: 문단 길이 균일화 / False: 모델 출력 그대로 사용)
)
for seg in segments:
chunked_docs.append(Document(
page_content=' '.join(seg),
metadata=doc.metadata
))
print(chunked_docs)
print("chunk 개수 :",len(chunked_docs))
결과
3.2) 이미지 chunking 이미지 설명에 대해서는 일반 텍스트와 달리 별도로 chunking을 하지 않고 한 덩어리로 그대로 유지하는 것이 효과적입니다.
그 이유는 간단합니다. 이미지 설명 텍스트는 일반적으로 길이가 짧고, 하나의 이미지가 하나의 의미 단위를 담고 있기 때문입니다. 이러한 내용을 잘라서 나누면 오히려 문맥이 단절되거나 의미가 모호해질 수 있습니다. 한 개의 독립적인 청크로 처리하는 것이 검색 정확도 측면에서도 더 안정적인 결과를 보여줍니다. 따라서 이미지 설명은 별도 분할 없이, 1 이미지 요약 = 1 Document 형태로 구성하여 기존 텍스트 청크들과 함께 병합하여 사용합니다.
# image_docs를 chunked_docs에 추가 (원본은 그대로 유지)
combined_docs = chunked_docs + image_docs
print(f"전체 chunk 개수: {len(combined_docs)}")
텍스트와 이미지에 대해 chunking 작업을 수행한 결과, 총 19개의 청크(텍스트 10개 + 이미지 9개)가 생성되었습니다. 이제 실제로 생성된 청크 중 일부를 출력하여 어떤 형태로 구성되어 있는지 확인해보겠습니다. 아래는 생성된 청크 중 처음 3개의 샘플입니다.
# 샘플 청크 출력
print("\n샘플 청크 (처음 3개):")
for i, chunk in enumerate(combined_docs[:3], 0):
print(f"\n청크 {i+1}:")
print(f"내용: {chunk.page_content}")
print(f"metadata: {chunk.metadata}")
print(f"길이: {len(chunk.page_content)}자")
결과
④ 벡터 데이터 변환 (Embedding)
이제 문단 단위로 잘게 나눠진 텍스트와 이미지 설명을 벡터로 변환할 차례입니다.
이번 예제에서는 CLOVA Studio의 텍스트 임베딩 모델을 활용해 벡터를 생성합니다. langchain-naver의 ClovaXEmbeddings를 통해 CLOVA Studio의 임베딩 및 임베딩 v2 API를 손쉽게 활용할 수 있습니다. 임베딩 V2는 bge-m3 모델을 사용하며, 이 모델은 임베딩 과정에서 유사도 판단을 위해 코사인 거리(Cosine)를 거리 단위로 사용합니다. 모델을 설정하지 않으면 clir-emb-dolphin 모델이 기본값으로 지정되므로, ClovaXEmbeddings의 모델을 bge-m3로 명시적으로 설정해주어야 합니다.
from langchain_naver import ClovaXEmbeddings
clovax_embeddings = ClovaXEmbeddings(model='bge-m3') # 임베딩 모델을 설정
text = "임베딩 사용 예제입니다~"
clovax_embeddings.embed_query(text)
결과
⑤ 벡터 데이터 저장 (Vector Store)
임베딩된 벡터 데이터를 저장하고 나중에 효율적으로 검색하기 위해서는 벡터 저장소가 필요합니다.
이 예제에서는 로컬 환경에서 보다 쉽게 활용할 수 있는 Chroma와 FAISS를 사용했습니다. Langchain의Vector DB 비교 문서를 참고하여 자신의 개발 환경에 적합한 솔루션을 선택하면 됩니다.
전체 문서를 add_documents()로 한 번에 추가하면 내부적으로 많은 개별 임베딩 API 요청이 병렬로 실행되어 에러가 발생할 수 있습니다. 이를 방지하기 위해 요청 사이에 time.sleep() 간격을 두어 처리 속도를 조절함으로써 에러 발생률을 낮췄습니다.
5.1) Chroma Chroma는 Python 기반의 오픈소스 벡터 데이터베이스로, 사용이 간편하고 빠른 프로토타이핑에 적합합니다. 특히 로컬 환경에서도 빠르게 실행할 수 있어 개발 초기 단계에서 많이 활용됩니다.
#chroma 다운받기
%pip install -qU langchain-chroma
import chromadb
from langchain_chroma import Chroma
# 임베딩 모델 정의
clovax_embeddings = ClovaXEmbeddings(model='bge-m3')
# 로컬 클라이언트 생성
client = chromadb.PersistentClient(path="./chroma_langchain_db")
# 컬렉션 준비 (이름 중복 주의!)
collection_name = "clovastudiodatas_docs"
client.get_or_create_collection(
name=collection_name,
metadata={"hnsw:space": "cosine"}
)
# 벡터스토어 객체 생성
vectorstore_Chroma = Chroma(
client=client,
collection_name=collection_name,
embedding_function=clovax_embeddings
)
# 문서 추가: 최신 방식은 vectorstore.add_documents 사용
print("Adding documents to Chroma vectorstore...")
for doc in combined_docs:
try:
vectorstore_Chroma.add_documents([doc])
time.sleep(0.5)
except Exception as e:
print(f"[✘] 실패: {doc.metadata} → {e}")
print("All documents have been added to the vectorstore.")
결과
5.2) FAISS FAISS는 대규모 벡터 검색을 위한 라이브러리로, 속도와 확장성 측면에서 매우 강력한 성능을 제공합니다. 특히 대용량 문서를 처리할 때 효율적이며, 검색 정확도도 뛰어난 편입니다.
#FAISS 다운로드
%pip install -qU langchain-community faiss-cpu
import faiss
from langchain_community.vectorstores import FAISS
from langchain_community.docstore.in_memory import InMemoryDocstore
# 임베딩 모델 정의
clovax_embeddings = ClovaXEmbeddings(model='bge-m3')
# FAISS 인덱스 생성 (1024는 bge-m3 차원 수에 맞춰야 함)
index = faiss.IndexFlatIP(1024) # 내적 기반 검색
# FAISS 벡터스토어 생성
vectorstore_FAISS = FAISS(
embedding_function=clovax_embeddings,
index=index,
docstore=InMemoryDocstore(),
index_to_docstore_id={}
)
# 문서 일괄 추가 (자동 임베딩 처리)
print("Adding documents to FAISS vectorstore...")
for doc in combined_docs:
try:
vectorstore_FAISS.add_documents([doc])
time.sleep(0.5)
except Exception as e:
print(f"[✘] 실패: {doc.metadata} → {e}")
print("All documents have been added to FAISS vectorstore.")
결과
3. 질의 응답해보기
① 질문하기
문서 임베딩과 벡터 저장소 구성이 완료되었다면, 이제 실제 질문을 입력하고 관련 내용을 찾아 답변을 생성하는 과정을 진행해보겠습니다. LangChain에서는 RetrievalQA 체인을 통해 이 흐름을 간단하게 구현할 수 있습니다.
1.1) Retriever 생성하기 먼저 사용자 질문에 따라 연관 문서를 검색하고, 해당 문서를 기반으로 답변을 생성하는 체인을 구성합니다.
아래 코드는 시스템 프롬프트와 사용자 프롬프트를 구분하여 설정합니다. 시스템 프롬프트에서는 LLM이 기존 지식을 사용하지 않고 검색된 문서(context)에 기반하여 답변하도록 안내합니다. 사용자 프롬프트에는 문서 내용과 질문이 함께 전달됩니다. 질문을 입력하면 관련 문서를 검색한 후 답변을 생성합니다. 결과에는 답변뿐만 아니라 어떤 문서가 참조되었는지도 확인할 수 있습니다.
from langchain_core.prompts import ChatPromptTemplate, SystemMessagePromptTemplate, HumanMessagePromptTemplate
from langchain.chains import RetrievalQA
# System 및 User 메시지를 나눠 구성
system_template = (
"당신은 질문-답변(Question-Answering)을 수행하는 친절한 AI 어시스턴트입니다. 당신의 임무는 원래 가지고있는 지식은 모두 배제하고, 주어진 문맥(context) 에서 주어진 질문(question) 에 답하는 것입니다."
"만약, 주어진 문맥(context) 에서 답을 찾을 수 없다면, 답을 모른다면 `주어진 정보에서 질문에 대한 정보를 찾을 수 없습니다` 라고 답하세요."
)
user_template = (
"다음은 검색된 문서 내용입니다:\n\n{context}\n\n"
"위 정보를 바탕으로 다음 질문에 답해주세요:\n{question}"
)
prompt_template = ChatPromptTemplate.from_messages([
SystemMessagePromptTemplate.from_template(system_template),
HumanMessagePromptTemplate.from_template(user_template),
])
# 원하는 vectorstore 선택해서 사용
retriever = vectorstore_Chroma.as_retriever(
search_type="similarity_score_threshold",
search_kwargs={"score_threshold": 0.1, "k": 3}
)
# retriever = vectorstore_FAISS.as_retriever(
# search_type="similarity_score_threshold",
# search_kwargs={"score_threshold": 0.1, "k": 3}
# )
# Retrieval QA 체인 구성
qa_chain = RetrievalQA.from_chain_type(
llm=chat_llm,
chain_type="stuff",
retriever=retriever,
chain_type_kwargs={"prompt": prompt_template},
return_source_documents=True
)
# 실행
question = "데이터셋 규모가 커질수록 2대륙의 오류 발생 확률은 어떻게 돼?"
result = qa_chain.invoke({"query": question})
print("질문:", question)
print("응답:", result["result"]) # 모델의 실제 응답
for i, doc in enumerate(result["source_documents"]): # 답변시 참고 한 문서
print(f"\n[출처 문서 {i+1}]\n내용: {doc.page_content}\n메타데이터: {doc.metadata}")
② 답변확인 텍스트만 사용했을 때는 모델이 관련 정보를 찾지 못해 제한적인 답변을 제공했지만, 이미지 요약 텍스트까지 함께 활용했을 때는 훨씬 구체적이고 관련성 높은 답변을 생성할 수 있었습니다. 멀티모달 RAG 시스템이 어떻게 검색 품질과 응답 정확도를 향상시키는지 확인하실 수 있습니다.
맺음말
이번 cookbook에서는 텍스트뿐만 아니라 이미지 기반 정보를 함께 활용해 검색 정확도를 높일 수 있는 Multimodal RAG 시스템을 구성해보았습니다. 단순히 문서를 임베딩하는 단계를 넘어, 이미지 속 시각 정보를 비전 모델을 통해 요약하고 벡터 DB에 저장하여 검색에 활용함으로써 더욱 풍부한 질의응답이 가능해졌습니다. 특히 텍스트만으로는 충분히 대응하기 어려웠던 질문에도 이미지 기반 문서를 통해 정확한 답변을 도출할 수 있었던 예시를 통해, 비정형 데이터를 효과적으로 구조화하고 활용하는 방법의 가능성을 확인할 수 있었습니다.
이 가이드를 통해 멀티모달 RAG 시스템을 쉽게 구축하고 비전 모델의 실제 활용 흐름을 이해할 수 있기를 기대합니다.
