Crea un'app di domande e risposte con una RAG multimodale utilizzando Gemini Pro

1. Introduzione

Che cos'è la RAG

La Retrieval Augmented Generation (RAG) è una tecnica che combina la potenza dei modelli linguistici di grandi dimensioni (LLM) con la capacità di recuperare informazioni pertinenti da fonti di conoscenza esterne. Ciò significa che un LLM non si basa solo sui dati di addestramento interni, ma può anche accedere e incorporare informazioni specifiche e aggiornate quando genera risposte.

La RAG sta guadagnando popolarità per diversi motivi:

• Maggiore accuratezza e pertinenza: la RAG consente agli LLM di fornire risposte più accurate e pertinenti basandole su informazioni oggettive recuperate da fonti esterne. Ciò è particolarmente utile negli scenari in cui la conoscenza aggiornata è fondamentale, ad esempio per rispondere a domande su eventi attuali o fornire informazioni su argomenti specifici.
• Allucinazioni ridotte: a volte gli LLM possono generare risposte che sembrano plausibili ma che in realtà sono errate o prive di senso. La RAG contribuisce a mitigare questo problema verificando le informazioni generate rispetto a fonti esterne.
• Maggiore adattabilità: la RAG rende gli LLM più adattabili a diversi domini e attività. Sfruttando diverse fonti di conoscenza, un LLM può essere facilmente personalizzato per fornire informazioni su un'ampia gamma di argomenti.
• Esperienza utente migliorata: la RAG può migliorare l'esperienza utente complessiva fornendo risposte più informative, affidabili e pertinenti.

Perché multimodale

Nel mondo odierno, ricco di dati, i documenti spesso combinano testo e immagini per trasmettere informazioni in modo completo. Tuttavia, la maggior parte dei sistemi di Retrieval Augmented Generation (RAG) trascura le preziose informazioni contenute nelle immagini. Man mano che i modelli linguistici di grandi dimensioni (LLM) multimodali acquistano importanza, è fondamentale esplorare come possiamo sfruttare i contenuti visivi insieme al testo nella RAG, sbloccando una comprensione più approfondita del panorama informativo.

Due opzioni per la RAG multimodale

• Incorporamenti multimodali: il modello di incorporamenti multimodali genera vettori a 1408 dimensioni* in base all'input fornito, che può includere una combinazione di dati di immagini, testo e video. Il vettore di embedding delle immagini e il vettore di embedding del testo si trovano nello stesso spazio semantico con la stessa dimensionalità. Di conseguenza, questi vettori possono essere utilizzati in modo intercambiabile per casi d'uso come la ricerca di immagini per testo o la ricerca di video per immagini. Dai un'occhiata a questa demo.

1. Utilizza l'incorporamento multimodale per incorporare testo e immagini
2. Recuperali entrambi utilizzando la ricerca di similarità
3. Trasmetti sia l'immagine grezza recuperata sia i blocchi di testo all'LLM multimodale per la sintesi delle risposte

• Incorporamenti di testo -

1. Utilizza l'LLM multimodale per generare riepiloghi di testo delle immagini
2. Incorpora e recupera il testo
3. Trasmetti i blocchi di testo all'LLM per la sintesi delle risposte

Che cos'è il recupero multi-vettore

Il recupero multi-vettore utilizza i riepiloghi delle sezioni del documento per recuperare i contenuti originali per la sintesi delle risposte. Migliora la qualità della RAG, in particolare per le attività che richiedono un uso intensivo di tabelle, grafici, diagrammi e così via. Scopri maggiori dettagli nel blog di Langchain.

Che cosa creerai

Caso d'uso: sviluppo di un sistema di domande e risposte utilizzando Gemini Pro

Immagina di avere documenti contenenti grafici o diagrammi complessi ricchi di informazioni. Vuoi estrarre questi dati per rispondere a domande o query.

In questo codelab eseguirai le seguenti operazioni:

• Caricamento dei dati utilizzando LangChain document_loaders
• Generazione di riepiloghi di testo utilizzando il modello gemini-pro di Google
• Generazione di riepiloghi di immagini utilizzando il modello gemini-pro-vision di Google
• Creazione di un recupero multi-vettore utilizzando il modello textembedding-gecko di Google con Croma Db come archivio vettoriale
• Sviluppo di una catena RAG multimodale per il question answering

2. Prima di iniziare

1. Nella console Google Cloud, nella pagina di selezione del progetto, seleziona o crea un progetto Google Cloud.
2. Verifica che la fatturazione sia attivata per il tuo progetto Google Cloud. Scopri come verificare se la fatturazione è attivata per un progetto.
3. Abilita tutte le API consigliate dalla dashboard di Vertex AI
4. Apri notebook Colab e accedi con lo stesso account del tuo account Google Cloud attivo corrente.

3. Creazione di una RAG multimodale

Questo codelab utilizza SDK Vertex AI Python e Langchain per mostrare come implementare l'opzione 2 descritta qui con Google Cloud.

Puoi fare riferimento al codice completo nel file Multi-modal RAG with Google Cloud dal repository di riferimento.

4. Passaggio 1: installa e importa le dipendenze

!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

Inserisci l'ID progetto e completa l'autenticazione

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

from google.colab import auth
auth.authenticate_user()

Inizializza la piattaforma Vertex AI

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

5. Passaggio 2: prepara e carica i dati

Utilizziamo un file ZIP con un sottoinsieme delle immagini e dei PDF estratti da questo post del blog. Se vuoi seguire il flusso completo, utilizza l'esempio originale example.

Scarica prima i dati

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

Carica i contenuti di testo dal documento

from langchain_community.document_loaders import PyPDFLoader

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

Controlla i contenuti della prima pagina

texts[0]

Dovresti vedere l'output

Totale pagine nel documento

len(texts)

L'output previsto è

6. Passaggio 3: genera riepiloghi di testo

Importa prima le librerie richieste

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

Ottieni riepiloghi di testo

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

L'output previsto è

7. Passaggio 4: genera riepiloghi di immagini

Importa prima le librerie richieste

import base64
import os

from langchain_core.messages import HumanMessage

Genera riepiloghi di immagini

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]

Dovresti vedere un output simile a questo

8. Passaggio 5: crea un recupero multi-vettore

Generiamo riepiloghi di testo e immagini e salviamoli in un archivio vettoriale ChromaDB.

Importa le librerie richieste

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

Crea un recupero multi-vettore

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. Passaggio 6: crea una RAG multimodale

1. Definisci le funzioni di utilità

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. Definisci un prompt di immagine specifico del dominio

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. Definisci la catena RAG multimodale

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. Passaggio 7: prova le query

1. Recupera i documenti pertinenti

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. Esegui la nostra RAG sulla stessa query

result = chain_multimodal_rag.invoke(query)

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

Output di esempio (può variare quando esegui il codice)

11. Libera spazio

Per evitare che al tuo account Google Cloud vengano addebitati costi relativi alle risorse utilizzate in questo codelab, segui questi passaggi:

1. Nella console Google Cloud, vai alla pagina Gestisci risorse.
2. Nell'elenco dei progetti, seleziona il progetto che vuoi eliminare, quindi fai clic su Elimina.
3. Nella finestra di dialogo, digita l'ID progetto, quindi fai clic su Chiudi per eliminare il progetto.

12. Complimenti

Complimenti! Hai sviluppato con successo una RAG multimodale utilizzando Gemini.