Mit Gemini Pro eine Frage-und-Antwort-App mit multimodalem RAG erstellen

1. Einführung

Was ist RAG?

Retrieval Augmented Generation (RAG) ist eine Technik, die die Leistungsfähigkeit von Large Language Models (LLMs) mit der Möglichkeit kombiniert, relevante Informationen aus externen Wissensquellen abzurufen. Das bedeutet, dass ein LLM sich nicht nur auf seine internen Trainingsdaten verlässt, sondern beim Generieren von Antworten auch auf aktuelle, spezifische Informationen zugreifen und diese einbeziehen kann.

RAG wird aus mehreren Gründen immer beliebter:

• Höhere Genauigkeit und Relevanz:RAG ermöglicht es LLMs, genauere und relevantere Antworten zu liefern, indem sie auf Fakten basieren, die aus externen Quellen abgerufen werden. Das ist besonders nützlich in Szenarien, in denen aktuelles Wissen entscheidend ist, z. B. beim Beantworten von Fragen zu aktuellen Ereignissen oder beim Bereitstellen von Informationen zu bestimmten Themen.
• Weniger Halluzinationen:LLMs können manchmal Antworten generieren, die plausibel erscheinen, aber tatsächlich falsch oder unsinnig sind. RAG trägt dazu bei, dieses Problem zu mindern, indem die generierten Informationen anhand externer Quellen überprüft werden.
• Höhere Anpassungsfähigkeit: RAG macht LLMs anpassungsfähiger an verschiedene Bereiche und Aufgaben. Durch die Nutzung verschiedener Wissensquellen kann ein LLM problemlos angepasst werden, um Informationen zu einer Vielzahl von Themen bereitzustellen.
• Verbesserte Nutzerfreundlichkeit:RAG kann die Nutzerfreundlichkeit insgesamt verbessern, indem es informativere, zuverlässigere und relevantere Antworten liefert.

Warum multimodal?

In der heutigen datenreichen Welt werden in Dokumenten häufig Text und Bilder kombiniert, um Informationen umfassend zu vermitteln. Die meisten RAG-Systeme (Retrieval-Augmented Generation) übersehen jedoch die wertvollen Informationen, die in Bildern enthalten sind. Da multimodale Large Language Models (LLMs) immer wichtiger werden, ist es entscheidend, zu untersuchen, wie wir visuelle Inhalte neben Text in RAG nutzen können, um ein besseres Verständnis der Informationslandschaft zu ermöglichen.

Zwei Optionen für multimodale RAG

• Multimodale Einbettungen: Das Modell für multimodale Einbettungen generiert 1.408 dimensionale Vektoren basierend auf der von Ihnen bereitgestellten Eingabe, die eine Kombination aus Bild-, Text- und Videodaten enthalten kann. Der Bild- und der Texteinbettungsvektor befinden sich im selben semantischen Bereich mit derselben Dimensionalität. Daher können diese Vektoren austauschbar für Anwendungsfälle wie die Bildsuche nach Text oder die Videosuche nach Bild verwendet werden. Demo ansehen

1. Multimodale Einbettungen zum Einbetten von Text und Bildern verwenden
2. Beide mit der Ähnlichkeitssuche abrufen
3. Sowohl das abgerufene Rohbild als auch die Textblöcke an das multimodale LLM übergeben, um eine Antwort zu generieren

• Texteinbettungen:

1. Multimodales LLM zum Generieren von Textzusammenfassungen der Bilder verwenden
2. Text einbetten und abrufen
3. Textblöcke zur Antwortsynthese an LLM übergeben

Was ist Multi-Vector Retriever?

Beim Abrufen mit mehreren Vektoren werden Zusammenfassungen der Dokumentabschnitte verwendet, um Originalinhalte für die Antwortsynthese abzurufen. Dadurch wird die Qualität von RAG verbessert, insbesondere bei Aufgaben, die Tabellen, Grafiken und Diagramme umfassen. Weitere Informationen finden Sie im Blog von Langchain.

Umfang

Anwendungsfall: Question-Answering-System mit Gemini Pro entwickeln

Stellen Sie sich vor, Sie haben Dokumente mit komplexen Grafiken oder Diagrammen, die viele Informationen enthalten. Sie möchten diese Daten extrahieren, um Fragen oder Anfragen zu beantworten.

In diesem Codelab führen Sie folgende Aufgaben aus:

• Daten mit LangChain laden document_loaders
• Textzusammenfassungen mit dem gemini-pro-Modell von Google erstellen
• Bildzusammenfassungen mit dem gemini-pro-vision-Modell von Google erstellen
• Multi-Vektor-Abruf mit dem textembedding-gecko-Modell von Google und Chroma DB als Vektorspeicher erstellen
• Multimodale RAG-Kette für die Beantwortung von Fragen entwickeln

2. Hinweis

1. Wählen Sie in der Google Cloud Console auf der Seite zur Projektauswahl ein Google Cloud-Projekt aus oder erstellen Sie eines.
2. Die Abrechnung für das Google Cloud-Projekt muss aktiviert sein. So prüfen Sie, ob die Abrechnung für ein Projekt aktiviert ist.
3. Alle empfohlenen APIs im Vertex AI-Dashboard aktivieren
4. Öffnen Sie Colab-Notebook und melden Sie sich mit demselben Konto an wie bei Ihrem derzeit aktiven Google Cloud-Konto.

3. Multimodale RAG erstellen

In diesem Codelab wird gezeigt, wie Sie mit dem Vertex AI SDK für Python und Langchain die hier beschriebene Option 2 mit Google Cloud implementieren.

Den vollständigen Code finden Sie in der Datei Multi-modal RAG with Google Cloud im referenzierten Repository.

4. Schritt 1: Abhängigkeiten installieren und importieren

!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

Projekt-ID eingeben und Authentifizierung abschließen

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

from google.colab import auth
auth.authenticate_user()

Vertex AI Platform initialisieren

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

5. Schritt 2: Daten vorbereiten und laden

Wir verwenden eine ZIP-Datei mit einer Teilmenge der extrahierten Bilder und PDFs aus diesem Blogpost. Wenn Sie den vollständigen Ablauf durchlaufen möchten, verwenden Sie bitte das Originalbeispiel.

Daten zuerst herunterladen

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

Textinhalte aus dem Dokument laden

from langchain_community.document_loaders import PyPDFLoader

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

Inhalte auf der ersten Seite prüfen

texts[0]

Es sollte eine Ausgabe angezeigt werden.

Gesamtzahl der Seiten im Dokument

len(texts)

Die erwartete Ausgabe ist

6. Schritt 3: Textzusammenfassungen erstellen

Erforderliche Bibliotheken importieren

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

Textzusammenfassungen erhalten

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

Die erwartete Ausgabe ist

7. Schritt 4: Bildzusammenfassungen erstellen

Erforderliche Bibliotheken importieren

import base64
import os

from langchain_core.messages import HumanMessage

Bildzusammenfassungen erstellen

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]

Die Ausgabe sollte in etwa so aussehen:

8. Schritt 5: Multi-Vector Retrieval erstellen

Wir generieren Text- und Bildzusammenfassungen und speichern sie in einem ChromaDB-Vektorspeicher.

Erforderliche Bibliotheken importieren

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

Multi-Vector Retrieval erstellen

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. Schritt 6: Multimodales RAG erstellen

1. Hilfsfunktionen definieren

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. Domainspezifischen Bild-Prompt definieren

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. Multimodale RAG-Kette definieren

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. Schritt 7: Abfragen testen

1. Relevante Dokumente abrufen

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 mit derselben Anfrage ausführen

result = chain_multimodal_rag.invoke(query)

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

Beispielausgabe (kann bei der Ausführung des Codes variieren)

11. Bereinigen

So vermeiden Sie, dass Ihrem Google Cloud-Konto die in diesem Codelab verwendeten Ressourcen in Rechnung gestellt werden:

1. Wechseln Sie in der Google Cloud Console zur Seite Ressourcen verwalten.
2. Wählen Sie in der Projektliste das Projekt aus, das Sie löschen möchten, und klicken Sie auf Löschen.
3. Geben Sie im Dialogfeld die Projekt-ID ein und klicken Sie auf Beenden, um das Projekt zu löschen.

12. Glückwunsch

Glückwunsch! Sie haben erfolgreich eine multimodale RAG mit Gemini entwickelt.