Gemini Pro를 사용하여 멀티모달 RAG로 Q&A 앱 빌드하기

1. 소개

RAG란 무엇인가요?

검색 증강 생성 (RAG)은 대규모 언어 모델 (LLM)의 성능과 외부 지식 소스에서 관련 정보를 검색하는 기능을 결합한 기법입니다. 즉, LLM은 내부 학습 데이터에만 의존할 뿐만 아니라 응답을 생성할 때 특정 최신 정보에 액세스하고 통합할 수도 있습니다.

RAG가 인기를 얻고 있는 이유에는 여러 가지가 있습니다.

• 정확성 및 관련성 향상: RAG를 사용하면 LLM이 외부 소스에서 가져온 사실에 기반한 정보를 기반으로 하여 더 정확하고 관련성 높은 응답을 제공할 수 있습니다. 이는 시사에 대한 질문에 답하거나 특정 주제에 대한 정보를 제공하는 등 최신 지식이 중요한 시나리오에서 특히 유용합니다.
• 할루시네이션 감소: LLM은 타당해 보이지만 실제로는 부정확하거나 말이 되지 않는 대답을 생성할 수 있습니다. RAG는 외부 소스를 기준으로 생성된 정보를 확인하여 이 문제를 완화하는 데 도움이 됩니다.
• 적응성 향상: RAG를 사용하면 LLM이 다양한 분야와 작업에 맞게 적응할 수 있습니다. 다양한 지식 소스를 활용하면 LLM을 쉽게 맞춤설정하여 광범위한 주제에 관한 정보를 제공할 수 있습니다.
• 사용자 환경 향상: RAG는 더 유익하고 안정적이며 관련성이 높은 응답을 제공하여 전반적인 사용자 경험을 개선할 수 있습니다.

멀티 모달을 사용해야 하는 이유

데이터가 풍부한 오늘날의 환경에서는 문서를 포괄적으로 전달하기 위해 텍스트와 이미지를 결합하는 경우가 많습니다. 그러나 대부분의 검색 증강 생성 (RAG) 시스템은 이미지 안에 갇혀 있는 중요한 정보를 간과합니다. 멀티모달 대규모 언어 모델 (LLM)이 주목을 받으면서 RAG의 텍스트와 함께 시각적 콘텐츠를 활용하여 정보 환경을 더 깊이 이해할 수 있는 방법을 탐색하는 것이 중요합니다.

멀티모달 RAG를 위한 두 가지 옵션

• 멀티모달 임베딩 - 멀티모달 임베딩 모델은 제공된 입력을 기반으로 1,408차원 벡터* 를 생성하며, 여기에는 이미지, 텍스트, 동영상 데이터의 조합이 포함될 수 있습니다. 이미지 임베딩 벡터와 텍스트 임베딩 벡터가 동일한 차원을 가진 동일한 시맨틱 공간에 있습니다. 따라서 이러한 벡터는 텍스트로 이미지를 검색하거나 이미지로 동영상을 검색하는 등의 사용 사례에 서로 바꿔서 사용할 수 있습니다. 이 데모를 살펴보세요.

1. 멀티모달 임베딩을 사용하여 텍스트 및 이미지 삽입
2. 유사성 검색을 사용하여 둘 다 검색
3. 검색된 원시 이미지와 텍스트 청크를 모두 멀티모달 LLM에 전달하여 답변 합성을 수행합니다.

• 텍스트 임베딩 -

1. 멀티모달 LLM을 사용하여 이미지의 텍스트 요약 생성
2. 텍스트 삽입 및 검색
3. 답변 합성을 위해 텍스트 척을 LLM에 전달

멀티벡터 리트리버란 무엇인가요?

다중 경로 검색은 문서 섹션의 요약을 사용하여 답변 합성을 위해 원본 콘텐츠를 검색합니다. 특히 표, 그래프, 차트 등 집약적인 작업에서 RAG의 품질이 향상됩니다. 자세한 내용은 Langchain 블로그.

빌드할 항목

사용 사례: Gemini Pro를 사용하여 질의 응답 시스템 개발

정보로 가득한 복잡한 그래프나 다이어그램이 포함된 문서가 있다고 상상해 보세요. 질문이나 쿼리에 답하기 위해 이 데이터를 추출하고자 합니다.

이 Codelab에서는 다음 작업을 수행합니다.

• LangChain document_loaders을 사용하여 데이터 로드
• Google의 gemini-pro 모델을 사용하여 텍스트 요약 생성
• Google의 gemini-pro-vision 모델을 사용하여 이미지 요약 생성
• Croma 데이터베이스를 벡터 저장소로 사용하는 Google의 textembedding-gecko 모델을 사용하여 다중 벡터 검색 만들기
• 질의 응답용 멀티모달 RAG 체인 개발

2. 시작하기 전에

1. Google Cloud 콘솔의 프로젝트 선택기 페이지에서 Google Cloud 프로젝트를 선택하거나 만듭니다.
2. Google Cloud 프로젝트에 결제가 사용 설정되어 있는지 확인합니다. 프로젝트에 결제가 사용 설정되어 있는지 확인하는 방법을 알아보세요.
3. Vertex AI 대시보드에서 모든 권장 API 사용 설정
4. Colab 노트북을 열고 현재 활성 상태인 Google Cloud 계정과 동일한 계정으로 로그인합니다.

3. 멀티 모달 RAG 빌드

이 Codelab에서는 Python용 Vertex AI SDK와 Langchain을 사용하여 Google Cloud에서 여기에 설명된 '옵션 2'를 구현하는 방법을 보여줍니다.

참조된 저장소에서 Google Cloud를 사용한 멀티모달 RAG 파일에서 전체 코드를 참고할 수 있습니다.

4. 1단계: 종속 항목 설치 및 가져오기

!pip install -U --quiet langchain langchain_community chromadb  langchain-google-vertexai
!pip install --quiet "unstructured[all-docs]" pypdf pillow pydantic lxml pillow matplotlib chromadb tiktoken

프로젝트 ID를 입력하고 인증을 완료하세요.

#TODO : ENter project and location
PROJECT_ID = ""
REGION = "us-central1"

from google.colab import auth
auth.authenticate_user()

Vertex AI Platform 초기화

import vertexai
vertexai.init(project = PROJECT_ID , location = REGION)

5. 2단계: 데이터 준비 및 로드하기

이 블로그 게시물에서 추출한 이미지 및 PDF의 하위 집합이 포함된 ZIP 파일을 사용합니다. 전체 흐름을 따르려면 원래 예시를 사용하세요.

먼저 데이터 다운로드

import logging
import zipfile
import requests

logging.basicConfig(level=logging.INFO)

data_url = "https://storage.googleapis.com/benchmarks-artifacts/langchain-docs-benchmarking/cj.zip"
result = requests.get(data_url)
filename = "cj.zip"
with open(filename, "wb") as file:
   file.write(result.content)

with zipfile.ZipFile(filename, "r") as zip_ref:
   zip_ref.extractall()

문서에서 텍스트 콘텐츠 로드

from langchain_community.document_loaders import PyPDFLoader

loader = PyPDFLoader("./cj/cj.pdf")
docs = loader.load()
tables = []
texts = [d.page_content for d in docs]

첫 페이지의 콘텐츠 확인하기

texts[0]

출력은

문서의 총 페이지 수

len(texts)

예상 출력은 다음과 같습니다.

6. 3단계: 텍스트 요약 생성

필수 라이브러리 먼저 가져오기

from langchain_google_vertexai import VertexAI , ChatVertexAI , VertexAIEmbeddings
from langchain.prompts import PromptTemplate
from langchain_core.messages import AIMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableLambda

텍스트 요약 가져오기

# Generate summaries of text elements
def generate_text_summaries(texts, tables, summarize_texts=False):
   """
   Summarize text elements
   texts: List of str
   tables: List of str
   summarize_texts: Bool to summarize texts
   """

   # Prompt
   prompt_text = """You are an assistant tasked with summarizing tables and text for retrieval. \
   These summaries will be embedded and used to retrieve the raw text or table elements. \
   Give a concise summary of the table or text that is well optimized for retrieval. Table or text: {element} """
   prompt = PromptTemplate.from_template(prompt_text)
   empty_response = RunnableLambda(
       lambda x: AIMessage(content="Error processing document")
   )
   # Text summary chain
   model = VertexAI(
       temperature=0, model_name="gemini-pro", max_output_tokens=1024
   ).with_fallbacks([empty_response])
   summarize_chain = {"element": lambda x: x} | prompt | model | StrOutputParser()

   # Initialize empty summaries
   text_summaries = []
   table_summaries = []

   # Apply to text if texts are provided and summarization is requested
   if texts and summarize_texts:
       text_summaries = summarize_chain.batch(texts, {"max_concurrency": 1})
   elif texts:
       text_summaries = texts

   # Apply to tables if tables are provided
   if tables:
       table_summaries = summarize_chain.batch(tables, {"max_concurrency": 1})

   return text_summaries, table_summaries


# Get text summaries
text_summaries, table_summaries = generate_text_summaries(
   texts, tables, summarize_texts=True
)

text_summaries[0]

예상 출력은 다음과 같습니다.

7. 4단계: 이미지 요약 생성

필수 라이브러리 먼저 가져오기

import base64
import os

from langchain_core.messages import HumanMessage

이미지 요약 생성

def encode_image(image_path):
   """Getting the base64 string"""
   with open(image_path, "rb") as image_file:
       return base64.b64encode(image_file.read()).decode("utf-8")


def image_summarize(img_base64, prompt):
   """Make image summary"""
   model = ChatVertexAI(model_name="gemini-pro-vision", max_output_tokens=1024)

   msg = model(
       [
           HumanMessage(
               content=[
                   {"type": "text", "text": prompt},
                   {
                       "type": "image_url",
                       "image_url": {"url": f"data:image/jpeg;base64,{img_base64}"},
                   },
               ]
           )
       ]
   )
   return msg.content


def generate_img_summaries(path):
   """
   Generate summaries and base64 encoded strings for images
   path: Path to list of .jpg files extracted by Unstructured
   """

   # Store base64 encoded images
   img_base64_list = []

   # Store image summaries
   image_summaries = []

   # Prompt
   prompt = """You are an assistant tasked with summarizing images for retrieval. \
   These summaries will be embedded and used to retrieve the raw image. \
   Give a concise summary of the image that is well optimized for retrieval."""

   # Apply to images
   for img_file in sorted(os.listdir(path)):
       if img_file.endswith(".jpg"):
           img_path = os.path.join(path, img_file)
           base64_image = encode_image(img_path)
           img_base64_list.append(base64_image)
           image_summaries.append(image_summarize(base64_image, prompt))

   return img_base64_list, image_summaries


# Image summaries
img_base64_list, image_summaries = generate_img_summaries("./cj")

len(img_base64_list)

len(image_summaries)

image_summaries[0]

다음과 같은 출력이 표시됩니다.

8. 5단계: 다중 벡터 검색 빌드

텍스트 및 이미지 요약을 생성하여 ChromaDB 벡터 저장소에 저장해 보겠습니다.

가져오기에는 라이브러리가 필요합니다.

import uuid
from langchain.retrievers.multi_vector import MultiVectorRetriever
from langchain.storage import InMemoryStore
from langchain_community.vectorstores import Chroma
from langchain_core.documents import Document

다중 벡터 검색 만들기

def create_multi_vector_retriever(
   vectorstore, text_summaries, texts, table_summaries, tables, image_summaries, images
):
   """
   Create retriever that indexes summaries, but returns raw images or texts
   """

   # Initialize the storage layer
   store = InMemoryStore()
   id_key = "doc_id"

   # Create the multi-vector retriever
   retriever = MultiVectorRetriever(
       vectorstore=vectorstore,
       docstore=store,
       id_key=id_key,
   )

   # Helper function to add documents to the vectorstore and docstore
   def add_documents(retriever, doc_summaries, doc_contents):
       doc_ids = [str(uuid.uuid4()) for _ in doc_contents]
       summary_docs = [
           Document(page_content=s, metadata={id_key: doc_ids[i]})
           for i, s in enumerate(doc_summaries)
       ]
       retriever.vectorstore.add_documents(summary_docs)
       retriever.docstore.mset(list(zip(doc_ids, doc_contents)))

   # Add texts, tables, and images
   # Check that text_summaries is not empty before adding
   if text_summaries:
       add_documents(retriever, text_summaries, texts)
   # Check that table_summaries is not empty before adding
   if table_summaries:
       add_documents(retriever, table_summaries, tables)
   # Check that image_summaries is not empty before adding
   if image_summaries:
       add_documents(retriever, image_summaries, images)

   return retriever


# The vectorstore to use to index the summaries
vectorstore = Chroma(
   collection_name="mm_rag_cj_blog",
   embedding_function=VertexAIEmbeddings(model_name="textembedding-gecko@latest"),
)

# Create retriever
retriever_multi_vector_img = create_multi_vector_retriever(
   vectorstore,
   text_summaries,
   texts,
   table_summaries,
   tables,
   image_summaries,
   img_base64_list,
)
 

9. 6단계: 멀티 모달 RAG 빌드

1. 유틸리티 함수 정의

import io
import re

from IPython.display import HTML, display
from langchain_core.runnables import RunnableLambda, RunnablePassthrough
from PIL import Image


def plt_img_base64(img_base64):
   """Disply base64 encoded string as image"""
   # Create an HTML img tag with the base64 string as the source
   image_html = f'<img src="data:image/jpeg;base64,{img_base64}" />'
   # Display the image by rendering the HTML
   display(HTML(image_html))


def looks_like_base64(sb):
   """Check if the string looks like base64"""
   return re.match("^[A-Za-z0-9+/]+[=]{0,2}$", sb) is not None


def is_image_data(b64data):
   """
   Check if the base64 data is an image by looking at the start of the data
   """
   image_signatures = {
       b"\xFF\xD8\xFF": "jpg",
       b"\x89\x50\x4E\x47\x0D\x0A\x1A\x0A": "png",
       b"\x47\x49\x46\x38": "gif",
       b"\x52\x49\x46\x46": "webp",
   }
   try:
       header = base64.b64decode(b64data)[:8]  # Decode and get the first 8 bytes
       for sig, format in image_signatures.items():
           if header.startswith(sig):
               return True
       return False
   except Exception:
       return False


def resize_base64_image(base64_string, size=(128, 128)):
   """
   Resize an image encoded as a Base64 string
   """
   # Decode the Base64 string
   img_data = base64.b64decode(base64_string)
   img = Image.open(io.BytesIO(img_data))

   # Resize the image
   resized_img = img.resize(size, Image.LANCZOS)

   # Save the resized image to a bytes buffer
   buffered = io.BytesIO()
   resized_img.save(buffered, format=img.format)

   # Encode the resized image to Base64
   return base64.b64encode(buffered.getvalue()).decode("utf-8")


def split_image_text_types(docs):
   """
   Split base64-encoded images and texts
   """
   b64_images = []
   texts = []
   for doc in docs:
       # Check if the document is of type Document and extract page_content if so
       if isinstance(doc, Document):
           doc = doc.page_content
       if looks_like_base64(doc) and is_image_data(doc):
           doc = resize_base64_image(doc, size=(1300, 600))
           b64_images.append(doc)
       else:
           texts.append(doc)
   if len(b64_images) > 0:
       return {"images": b64_images[:1], "texts": []}
   return {"images": b64_images, "texts": texts}

2. 도메인별 이미지 프롬프트 정의

def img_prompt_func(data_dict):
   """
   Join the context into a single string
   """
   formatted_texts = "\n".join(data_dict["context"]["texts"])
   messages = []

   # Adding the text for analysis
   text_message = {
       "type": "text",
       "text": (
           "You are financial analyst tasking with providing investment advice.\n"
           "You will be given a mixed of text, tables, and image(s) usually of charts or graphs.\n"
           "Use this information to provide investment advice related to the user question. \n"
           f"User-provided question: {data_dict['question']}\n\n"
           "Text and / or tables:\n"
           f"{formatted_texts}"
       ),
   }
   messages.append(text_message)
   # Adding image(s) to the messages if present
   if data_dict["context"]["images"]:
       for image in data_dict["context"]["images"]:
           image_message = {
               "type": "image_url",
               "image_url": {"url": f"data:image/jpeg;base64,{image}"},
           }
           messages.append(image_message)
   return [HumanMessage(content=messages)]

3. 멀티 모달 RAG 체인 정의

def multi_modal_rag_chain(retriever):
   """
   Multi-modal RAG chain
   """

   # Multi-modal LLM
   model = ChatVertexAI(
       temperature=0, model_name="gemini-pro-vision", max_output_tokens=1024
   )

   # RAG pipeline
   chain = (
       {
           "context": retriever | RunnableLambda(split_image_text_types),
           "question": RunnablePassthrough(),
       }
       | RunnableLambda(img_prompt_func)
       | model
       | StrOutputParser()
   )

   return chain


# Create RAG chain
chain_multimodal_rag = multi_modal_rag_chain(retriever_multi_vector_img)

10. 7단계: 쿼리 테스트

1. 관련 문서 가져오기

query = "What are the EV / NTM and NTM rev growth for MongoDB, Cloudflare, and Datadog?"
docs = retriever_multi_vector_img.get_relevant_documents(query, limit=1)

# We get relevant docs
len(docs)

docs

         You may get similar output 

plt_img_base64(docs[3])

2. 동일한 쿼리로 RAG 실행

result = chain_multimodal_rag.invoke(query)

from IPython.display import Markdown as md
md(result)

샘플 출력 (코드를 실행하는 경우 달라질 수 있음)

11. 삭제

이 Codelab에서 사용한 리소스 비용이 Google Cloud 계정에 청구되지 않도록 하려면 다음 단계를 따르세요.

1. Google Cloud 콘솔에서 리소스 관리 페이지로 이동합니다.
2. 프로젝트 목록에서 삭제할 프로젝트를 선택하고 삭제를 클릭합니다.
3. 대화상자에서 프로젝트 ID를 입력하고 종료를 클릭하여 프로젝트를 삭제합니다.

12. 축하합니다

수고하셨습니다 Gemini를 사용하여 멀티모달 RAG를 성공적으로 개발했습니다.