MultiModal RAG With GPT-4 Vision and LangChain 정리

https://youtu.be/6D9mpFCPeI8?si=P45ND9OjfPKsdaUq

 

 

---

 

 

LLM의 기능을 강화시키는 RAG는 Something to Vector 동작이 근간을 이루고 있다.

• 텍스트의 경우는 EmbeddingModel 써서 간단히 벡터로 변환할 수 있음.
• 그러나 PDF의 경우... 고려할 게 많다.
  • Text, Table, Images...
  • 등장 순서나 구성방식도 정보를 포함하고 있다.

 

텍스트는 ChatModel을 활용하고, 이미지는 GPT-4 Vision 모델을 활용하면, pdf에 있는 데이터를 벡터화할 수 있다

• pdf의 text, table, image 내용을 Summarize
• Raw Document도 DocumentStore에 저장하고 값을 받아온다
  • 영상에서는 제작자가 '아직 image + text를 OpenAI에 single request로 처리할 수 있는 방법을 찾지 못했다'고 안내함.
• Similarity Search 수행해서 얻어낸 값을 OpenAI api로 전달.

github repository: https://github.com/Coding-Crashkurse/Multimodal-RAG-With-OpenAI/tree/main

Data Loading

!pip install langchain unstructured[all-docs] pydantic lxml openai chromadb tiktoken
# unstructured: to extract all the relevant docs from pdf
# tiktoken: for creating our tokens.

from typing import Any
import os
from unstructured.partition.pdf import partition_pdf
import pytesseract
import os

pytesseract.pytesseract.tesseract_cmd = r'C:\Program Files\Tesseract-OCR\tesseract.exe'

input_path = os.getcwd()
output_path = os.path.join(os.getcwd(), "output")

# Get elements
raw_pdf_elements = partition_pdf(
    filename=os.path.join(input_path, "test.pdf"),
    extract_images_in_pdf=True,
    infer_table_structure=True,
    chunking_strategy="by_title",
    max_characters=4000,
    new_after_n_chars=3800,
    combine_text_under_n_chars=2000,
    image_output_dir_path=output_path,
)

unstructured 패키지는 이미지 인식을 위해 tesseract를 사용한다.

• pytessaract binary를 다운받아서, cmd 경로에 추가해준다.

unstructured 패키지의 partition_pdf 메소드를 사용해서 raw data를 가져온다.

• chunking strategy로 'by title' 설정.
• 추출한 이미지를 별도의 directory에 저장하는 image_output_dir_path 옵션이 있다.

 

object 확인해보면, unstructured.documents 객체가 리스트 형태로 들어가 있고, table 같은 object가 별도로 만들어진 것도 확인할 수 있다.

---

이제, 각 element별로 분리하고, 필요한 작업을 수행한다

• 이미지: base64 인코딩

import base64

text_elements = []
table_elements = []
image_elements = []

# Function to encode images
def encode_image(image_path):
    with open(image_path, "rb") as image_file:
        return base64.b64encode(image_file.read()).decode('utf-8')

for element in raw_pdf_elements:
    if 'CompositeElement' in str(type(element)):
        text_elements.append(element)
    elif 'Table' in str(type(element)):
        table_elements.append(element)

table_elements = [i.text for i in table_elements]
text_elements = [i.text for i in text_elements]

# Tables
print(len(table_elements))

# Text
print(len(text_elements))

# images
for image_file in os.listdir(output_path):
    if image_file.endswith(('.png', '.jpg', '.jpeg')):
        image_path = os.path.join(output_path, image_file)
        encoded_image = encode_image(image_path)
        image_elements.append(encoded_image)
print(len(image_elements))

 

• 이미지는 output 디렉토리에 저장되어 있고
• table은 3개, text는 24개로 리스트에 저장된 걸 확인할 수 있다.
• 파일로 저장된 이미지는 읽어서 base64로 변환한 뒤 리스트에 저장한다.

이제, 각각의 element를 summarize한다.

from langchain.chat_models import ChatOpenAI
from langchain.schema.messages import HumanMessage, AIMessage

## 영상 녹화 시점에서는 gpt3와 gpt4-vision를 다르게 정의하고, payload도 분리함. 같은 방식으로는 동작하지 않기 때문.
chain_gpt_35 = ChatOpenAI(model="gpt-3.5-turbo", max_tokens=1024)
chain_gpt_4_vision = ChatOpenAI(model="gpt-4-vision-preview", max_tokens=1024)

# Function for text summaries
def summarize_text(text_element):
    prompt = f"Summarize the following text:\n\n{text_element}\n\nSummary:"
    response = chain_gpt_35.invoke([HumanMessage(content=prompt)])
    return response.content

# Function for table summaries
def summarize_table(table_element):
    prompt = f"Summarize the following table:\n\n{table_element}\n\nSummary:"
    response = chain_gpt_35.invoke([HumanMessage(content=prompt)])
    return response.content

# Function for image summaries
def summarize_image(encoded_image):
    prompt = [
        AIMessage(content="You are a bot that is good at analyzing images."),
        HumanMessage(content=[
            {"type": "text", "text": "Describe the contents of this image."},
            {
                "type": "image_url",
                "image_url": {
                    "url": f"data:image/jpeg;base64,{encoded_image}"
                },
            },
        ])
    ]
    response = chain_gpt_4_vision.invoke(prompt)
    return response.content


# 각각의 element마다 summary 실행한다
# Processing table elements with feedback and sleep
# 강의 예시에서 굳이 element를 0:2로 제한한 이유: api 호출비용이 비싸서
table_summaries = []
for i, te in enumerate(table_elements[0:2]):
    summary = summarize_table(te)
    table_summaries.append(summary)
    print(f"{i + 1}th element of tables processed.")# Processing text elements with feedback and sleep

text_summaries = []
for i, te in enumerate(text_elements[0:2]):
    summary = summarize_text(te)
    text_summaries.append(summary)
    print(f"{i + 1}th element of texts processed.")

# Processing image elements with feedback and sleep
image_summaries = []
for i, ie in enumerate(image_elements[0:2]):
    summary = summarize_image(ie)
    image_summaries.append(summary)
    print(f"{i + 1}th element of images processed.")

Multi-vector retriever

reference: https://python.langchain.com/v0.1/docs/modules/data_connection/retrievers/multi_vector/#summary

• 강의 예시로는 inMemoryDB인 ChromaDB 사용.

import uuid

from langchain.embeddings import OpenAIEmbeddings
from langchain.retrievers.multi_vector import MultiVectorRetriever
from langchain.schema.document import Document
from langchain.storage import InMemoryStore
from langchain.vectorstores import Chroma

# Initialize the vector store and storage layer
vectorstore = Chroma(collection_name="summaries", embedding_function=OpenAIEmbeddings())
store = InMemoryStore()
id_key = "doc_id"

# Initialize the retriever
retriever = MultiVectorRetriever(vectorstore=vectorstore, docstore=store, id_key=id_key)

# Function to add documents to the retriever
def add_documents_to_retriever(summaries, original_contents):
    ## document별 uuid를 생성해서, id로 명시
    doc_ids = [str(uuid.uuid4()) for _ in summaries]

    # langchain에서 사용하는 Document 컴포넌트로 변경해준다.
    summary_docs = [
        Document(page_content=s, metadata={id_key: doc_ids[i]})
        for i, s in enumerate(summaries)
    ]
    # vector store에 저장한다
    retriever.vectorstore.add_documents(summary_docs)
    ## raw data도 저장한다. vector store와 doc store 간 Link는 아까 만든 uuid.
    retriever.docstore.mset(list(zip(doc_ids, original_contents)))

# Add text summaries
add_documents_to_retriever(text_summaries, text_elements)

# Add table summaries
add_documents_to_retriever(table_summaries, table_elements)

# Add image summaries
# image와 text를 한번에 전달하는 방법이 없던 시절이라서 이렇게 했음.
add_documents_to_retriever(image_summaries, image_summaries) # hopefully real images soon

Table Retrieval

# We can retrieve this table
retriever.get_relevant_documents(
    "What do you see on the images in the database?"
)

 

세 개의 docs를 토대로, 가장 유사한 vector store 정보를 찾아서 응답해준다.

• 단, image 정보를 그대로 주는 게 아니라, llm이 이미지를 요약한 텍스트를 리턴해준다는 단점이 있음.

from langchain.schema.runnable import RunnablePassthrough
from langchain.prompts import ChatPromptTemplate
from langchain.schema.output_parser import StrOutputParser

template = """Answer the question based only on the following context, which can include text, images and tables:
{context}
Question: {question}
"""
prompt = ChatPromptTemplate.from_template(template)

model = ChatOpenAI(temperature=0, model="gpt-3.5-turbo")

chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | model
    | StrOutputParser()
)



chain.invoke(
     "What do you see on the images in the database?"
)