Tworzenie aplikacji pytań i odpowiedzi z funkcją Multi-Modal RAG przy użyciu Gemini Pro

1. Wprowadzenie

Co to jest RAG

Retrieval Augmented Generation (RAG) to technika, która łączy możliwości dużych modeli językowych (LLM) ze zdolnością do pobierania odpowiednich informacji z zewnętrznych źródeł wiedzy. Oznacza to, że LLM nie polega tylko na wewnętrznych danych treningowych, ale może też uzyskiwać dostęp do aktualnych, szczegółowych informacji i wykorzystywać je podczas generowania odpowiedzi.

RAG zyskuje popularność z kilku powodów:

• Większa dokładność i trafność: RAG umożliwia dużym modelom językowym udzielanie dokładniejszych i trafniejszych odpowiedzi dzięki oparciu ich na informacjach pochodzących ze źródeł zewnętrznych. Jest to szczególnie przydatne w sytuacjach, w których aktualna wiedza jest kluczowa, np. podczas odpowiadania na pytania dotyczące bieżących wydarzeń lub dostarczania informacji na określone tematy.
• Mniej halucynacji: duże modele językowe mogą czasami generować odpowiedzi, które wydają się wiarygodne, ale w rzeczywistości są nieprawidłowe lub bezsensowne. RAG pomaga rozwiązać ten problem, weryfikując wygenerowane informacje na podstawie zewnętrznych źródeł.
• Większa adaptacyjność: RAG sprawia, że duże modele językowe lepiej dostosowują się do różnych dziedzin i zadań. Dzięki wykorzystaniu różnych źródeł wiedzy model LLM można łatwo dostosować do dostarczania informacji na wiele tematów.
• Lepsze wrażenia użytkowników: RAG może poprawić ogólne wrażenia użytkowników, dostarczając bardziej informatywnych, wiarygodnych i trafnych odpowiedzi.

Dlaczego warto korzystać z trybu multimodalnego

W dzisiejszym świecie, w którym mamy dostęp do ogromnych ilości danych, dokumenty często łączą tekst i obrazy, aby przekazywać informacje w sposób kompleksowy. Większość systemów generowania rozszerzonego przez wyszukiwanie w zapisanych informacjach (RAG) pomija jednak cenne informacje zawarte w obrazach. Wraz ze wzrostem popularności multimodalnych dużych modeli językowych (LLM) ważne jest, aby zbadać, w jaki sposób możemy wykorzystać treści wizualne wraz z tekstem w technologii RAG, co pozwoli nam lepiej zrozumieć krajobraz informacyjny.

2 opcje dla wielomodalnego RAG

• Multimodalne wektory dystrybucyjne – model multimodalnych wektorów dystrybucyjnych generuje 1408-wymiarowe wektory* na podstawie podanych danych wejściowych, które mogą obejmować kombinację danych obrazów, tekstu i wideo. Wektor dystrybucyjny obrazu i wektor dystrybucyjny tekstu znajdują się w tej samej przestrzeni semantycznej o tej samej liczbie wymiarów. Dzięki temu wektory te można stosować zamiennie w przypadkach użycia takich jak wyszukiwanie obrazu za pomocą tekstu lub wyszukiwanie filmu za pomocą obrazu. Obejrzyj to demo.

1. Używanie multimodalnych reprezentacji właściwościowych do umieszczania tekstu i obrazów
2. Pobieranie obu za pomocą wyszukiwania podobieństw
3. Przekazywanie pobranego obrazu w formie surowej i fragmentów tekstu do wielomodowego modelu LLM w celu syntezy odpowiedzi.

• Wektory dystrybucyjne tekstu –

1. Używanie multimodalnego modelu LLM do generowania podsumowań tekstowych obrazów
2. Umieszczanie i pobieranie tekstu
3. Przekazywanie fragmentów tekstu do modelu LLM w celu syntezy odpowiedzi

Czym jest Multi-Vector Retriever

Wyszukiwanie wielowektorowe wykorzystuje podsumowania sekcji dokumentu do pobierania oryginalnych treści na potrzeby syntezy odpowiedzi. Poprawia jakość RAG, zwłaszcza w przypadku zadań wymagających intensywnego korzystania z tabel, wykresów itp. Więcej informacji znajdziesz na blogu Langchain.

Co utworzysz

Przypadek użycia: tworzenie systemu odpowiadania na pytania za pomocą Gemini Pro

Wyobraź sobie, że masz dokumenty zawierające złożone wykresy lub diagramy pełne informacji. Chcesz wyodrębnić te dane, aby odpowiedzieć na pytania lub zapytania.

W tym ćwiczeniu wykonasz te czynności:

• Wczytywanie danych za pomocą LangChain document_loaders
• Generowanie podsumowań tekstów za pomocą modelu gemini-pro Google
• Generowanie podsumowań obrazów za pomocą modelu gemini-pro-vision od Google
• Utwórz wyszukiwanie wielowektorowe za pomocą modelu textembedding-gecko Google z bazą danych wektorów Chroma Db
• Tworzenie łańcucha multimodalnej techniki RAG do odpowiadania na pytania

2. Zanim zaczniesz

1. W konsoli Google Cloud na stronie wyboru projektu wybierz lub utwórz projekt Google Cloud.
2. Sprawdź, czy w projekcie Google Cloud włączone są płatności. Dowiedz się, jak sprawdzić, czy w projekcie są włączone płatności.
3. Włącz wszystkie zalecane interfejsy API w panelu Vertex AI.
4. Otwórz notatnik Colab i zaloguj się na to samo konto, które jest obecnie aktywne na Google Cloud.

3. Tworzenie multimodalnej techniki RAG

W tym ćwiczeniu używamy pakietu Vertex AI SDK for Python i Langchain, aby pokazać, jak wdrożyć w Google Cloud „Opcję 2” opisaną tutaj.

Pełny kod znajdziesz w pliku Multi-modal RAG with Google Cloud w repozytorium.

4. Krok 1. Zainstaluj i zaimportuj zależności

!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

Wpisz identyfikator projektu i dokończ uwierzytelnianie.

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

from google.colab import auth
auth.authenticate_user()

Inicjowanie platformy Vertex AI

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

5. Krok 2. Przygotuj i wczytaj dane

Używamy pliku ZIP z podzbiorem wyodrębnionych obrazów i pliku PDF z tego posta na blogu. Jeśli chcesz przejść przez cały proces, użyj oryginalnego przykładu.

Najpierw pobierz dane.

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()

Wczytaj treść tekstową z dokumentu

from langchain_community.document_loaders import PyPDFLoader

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

Sprawdzanie treści z pierwszej strony

texts[0]

Powinny pojawić się dane wyjściowe.

Łączna liczba stron w dokumencie

len(texts)

Oczekiwane dane wyjściowe to

6. Krok 3. Generowanie podsumowań tekstu

Najpierw zaimportuj wymagane biblioteki

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

Otrzymywanie podsumowań tekstu

# 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]

Oczekiwane dane wyjściowe to

7. Krok 4. Generowanie podsumowań obrazów

Najpierw zaimportuj wymagane biblioteki

import base64
import os

from langchain_core.messages import HumanMessage

Generowanie podsumowań obrazów

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]

Powinny się wyświetlić dane wyjściowe podobne do tych:

8. Krok 5. Tworzenie wyszukiwania wielowektorowego

Wygenerujmy podsumowania tekstu i obrazu i zapiszmy je w bazie wektorowej ChromaDB.

Importowanie wymaganych bibliotek

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

Tworzenie wyszukiwania wielowektorowego

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. Krok 6. Tworzenie multimodalnej techniki RAG

1. Definiowanie funkcji narzędziowych

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. Określanie promptu graficznego właściwego dla domeny

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. Definiowanie łańcucha multimodalnego 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. Krok 7. Przetestuj zapytania

1. Pobieranie odpowiednich dokumentów

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. Uruchomienie RAG na tym samym zapytaniu

result = chain_multimodal_rag.invoke(query)

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

Przykładowe dane wyjściowe (mogą się różnić w zależności od wykonania kodu)

11. Czyszczenie danych

Aby uniknąć obciążenia konta Google Cloud opłatami za zasoby użyte w tym laboratorium, wykonaj te czynności:

1. W konsoli Google Cloud otwórz stronę Zarządzanie zasobami.
2. Z listy projektów wybierz projekt do usunięcia, a potem kliknij Usuń.
3. W oknie wpisz identyfikator projektu i kliknij Wyłącz, aby usunąć projekt.

12. Gratulacje

Gratulacje! Udało Ci się opracować multimodalną technikę RAG przy użyciu Gemini.