Compila una app de preguntas y respuestas con RAG multimodal con Gemini Pro

1. Introducción

¿Qué es la RAG?

La generación de aumento de recuperación (RAG) es una técnica que combina el poder de los modelos de lenguaje grandes (LLM) con la capacidad de recuperar información relevante de fuentes de conocimiento externas. Esto significa que un LLM no solo depende de sus datos de entrenamiento internos, sino que también puede acceder e incorporar información actualizada y específica cuando genera respuestas.

La RAG está ganando popularidad por varias razones:

• Mayor precisión y relevancia: La RAG permite que los LLM proporcionen respuestas más precisas y relevantes basándolas en información fáctica recuperada de fuentes externas. Esto es particularmente útil en situaciones en las que es fundamental tener conocimiento actualizado, como responder preguntas sobre eventos actuales o proporcionar información sobre temas específicos.
• Reducción de alucinaciones: En ocasiones, los LLM pueden generar respuestas que parecen creíbles, pero que en realidad son incorrectas o sin sentido. RAG ayuda a mitigar este problema mediante la verificación de la información generada con respecto a fuentes externas.
• Mayor adaptabilidad: La RAG hace que los LLM se adapten más a diferentes dominios y tareas. Cuando aprovechas diferentes fuentes de conocimiento, un LLM se puede personalizar fácilmente para proporcionar información sobre una amplia variedad de temas.
• Experiencia del usuario mejorada: El formato RAV puede mejorar la experiencia general del usuario, ya que proporciona respuestas más informativas, confiables y relevantes.

Por qué los modelos multimodales

En el mundo actual lleno de datos, los documentos suelen combinar texto e imágenes para transmitir información de manera integral. Sin embargo, la mayoría de los sistemas de generación aumentada de recuperación (RAG) pasan por alto las estadísticas valiosas bloqueadas en las imágenes. A medida que los modelos grandes de lenguaje multimodales (LLM) adquieren protagonismo, es crucial explorar cómo podemos aprovechar el contenido visual junto con el texto en la RAG, lo que desbloquea una comprensión más profunda del panorama de la información.

Dos opciones para el RAG multimodal

• Incorporaciones multimodales: El modelo de incorporaciones multimodales genera vectores de 1, 408 dimensiones* basados en la entrada que proporciones, que pueden incluir una combinación de datos de imagen, texto y video. El vector de incorporación de imagen y el vector de incorporación de texto están en el mismo espacio semántico con la misma dimensionalidad. En consecuencia, estos vectores se pueden usar indistintamente para casos de uso como la búsqueda de imagen por texto o de video por imagen. Observa esta demostración.

1. Usa la incorporación multimodal para incorporar imágenes y texto
2. Recupera ambos con la búsqueda de similitudes
3. Pasa la imagen sin procesar y los fragmentos de texto recuperados al LLM multimodal para la síntesis de respuestas.

• Incorporaciones de texto:

1. Usa un LLM multimodal para generar resúmenes textuales de las imágenes
2. Texto incorporado y recuperación
3. Pasa los fragmentos de texto al LLM para la síntesis de respuestas.

¿Qué es el recuperador de varios vectores?

La recuperación multivectorial emplea resúmenes de las secciones del documento a fin de recuperar el contenido original para la síntesis de respuestas. Mejora la calidad del RAG en especial para las tareas intensivas con tablas, gráficos, cuadros, etcétera. Obtén más detalles en el blog de Langchain.

Qué compilarás

Caso de uso: Desarrollo de un sistema de respuesta a preguntas con Gemini Pro

Imagina que tienes documentos que contienen gráficos o diagramas complejos repletos de información. Quieres extraer estos datos para responder preguntas o consultas.

En este codelab, realizarás lo siguiente:

• Cargando datos con LangChain document_loaders
• Genera resúmenes de texto con el modelo gemini-pro de Google
• Genera resúmenes de imágenes con el modelo gemini-pro-vision de Google
• Crea la recuperación multivectorial con el modelo textembedding-gecko de Google con Croma Db como almacén de vectores
• Desarrollar una cadena RAG multimodal para responder preguntas

2. Antes de comenzar

1. En la página del selector de proyectos de la consola de Google Cloud, selecciona o crea un proyecto de Google Cloud.
2. Asegúrate de que la facturación esté habilitada para tu proyecto de Google Cloud. Obtén información sobre cómo verificar si la facturación está habilitada en un proyecto.
3. Habilita todas las APIs recomendadas desde el panel de Vertex AI
4. Abre el Notebook de Colab y accede a la misma cuenta que tienes en tu cuenta de Google Cloud activa actual.

3. Cómo compilar la RAG multimodal

En este codelab, se usa el SDK de Vertex AI para Python y Langchain para demostrar cómo implementar la “Opción 2” que se describe aquí con Google Cloud.

Puedes consultar el código completo en el archivo RAG multimodal con Google Cloud en el repositorio al que se hace referencia.

4. Paso 1: Instala y, luego, importa las dependencias

!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

Ingresa el ID del proyecto y completa la autenticación

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

from google.colab import auth
auth.authenticate_user()

Inicializa Vertex AI Platform

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

5. Paso 2: Prepara y carga los datos

Usamos un archivo ZIP con un subconjunto de las imágenes extraídas y un PDF de esta entrada de blog. Si quieres seguir el flujo completo, usa el ejemplo original.

Primero, descarga los datos

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

Carga el contenido de texto del 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]

Revisa el contenido de la primera página

texts[0]

Deberías ver el resultado

Total de páginas del documento

len(texts)

El resultado esperado es

6. Paso 3: Genera resúmenes de texto

Primero importa las bibliotecas requeridas

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

Obtener resúmenes de texto

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

El resultado esperado es

7. Paso 4: Genera resúmenes de imágenes

Primero importa las bibliotecas requeridas

import base64
import os

from langchain_core.messages import HumanMessage

Generar resúmenes de imágenes

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]

Deberías ver un resultado como este

8. Paso 5: Compila la recuperación de varios vectores

Generemos resúmenes de imágenes y texto, y los guardemos en un almacén de vectores de ChromaDB.

Importar bibliotecas

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

Crear recuperación multivectorial

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. Paso 6: Cómo compilar la RAG multimodal

1. Cómo definir funciones de utilidad

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. Define el mensaje de imagen específico 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. Definición de la cadena RAG multimodal

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. Paso 7: Prueba tus consultas

1. Recupera documentos relevantes

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. Ejecuta nuestro RAG en la misma consulta

result = chain_multimodal_rag.invoke(query)

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

Resultado de muestra (puede variar cuando ejecutas el código)

11. Limpia

Sigue estos pasos para evitar que se apliquen cargos a tu cuenta de Google Cloud por los recursos que usaste en este codelab:

1. En la consola de Google Cloud, ve a la página Administrar recursos.
2. En la lista de proyectos, elige el proyecto que deseas borrar y haz clic en Borrar.
3. En el diálogo, escribe el ID del proyecto y, luego, haz clic en Cerrar para borrarlo.

12. Felicitaciones

¡Felicitaciones! Desarrollaste con éxito un RAG multimodal con Gemini.