פיתוח אפליקציה של שאלות ותשובות עם RAG רב-שלבי בעזרת Gemini Pro

1. מבוא

מהו RAG

יצירת טקסט באמצעות אחזור משופר (RAG) היא טכניקה שמשלבת את העוצמה של מודלים גדולים של שפה (LLM) עם היכולת לאחזר מידע רלוונטי ממקורות ידע חיצוניים. המשמעות היא ש-LLM לא מסתמך רק על נתוני האימון הפנימיים שלו, אלא יכול גם לגשת למידע ספציפי עדכני ולשלב אותו בתשובות שהוא יוצר.

RAG צוברת פופולריות מכמה סיבות:

• דיוק ורלוונטיות משופרים: RAG מאפשר למודלים גדולים של שפה (LLMs) לספק תשובות מדויקות ורלוונטיות יותר על ידי התבססות על מידע עובדתי שאוחזר ממקורות חיצוניים. האפשרות הזו שימושית במיוחד בתרחישים שבהם ידע עדכני חיוני, למשל כדי לענות על שאלות לגבי אירועים אקטואליים או לספק מידע בנושאים ספציפיים.
• הפחתת הזיות: לפעמים מודלים גדולים של שפה יכולים ליצור תשובות שנראות הגיוניות אבל בפועל הן שגויות או לא הגיוניות. כדי לצמצם את הבעיה הזו, נעזרים ב-RAG לאימות המידע שנוצר מול מקורות חיצוניים.
• התאמה טובה יותר: RAG מאפשר למודלים גדולים של שפה (LLMs) להתאים את עצמם לדומיינים ולמשימות שונים. בעזרת מקורות ידע שונים, אפשר להתאים אישית בקלות את ה-LLM כדי לספק מידע על מגוון רחב של נושאים.
• חוויית משתמש משופרת: RAG יכול לשפר את חוויית המשתמש הכוללת על ידי מתן תשובות מפורטות, מהימנות ורלוונטיות יותר.

למה כדאי להשתמש במודל רב-מודלי

בעולם של ימינו, שמלא בנתונים, מסמכים משלבים לעיתים קרובות טקסט ותמונות כדי להעביר מידע באופן מקיף. עם זאת, רוב המערכות ליצירת תמונות עם שיפור אחזור (RAG) מתעלמות מהתובנות החשובות שמוטמעות בתמונות. מודל שפה גדול (LLM) רב-מודלי צובר תאוצה, ולכן חשוב לבדוק איך אפשר להשתמש בתוכן חזותי לצד טקסט ב-RAG, כדי לקבל הבנה עמוקה יותר של תמונת המידע.

שתי אפשרויות ל-RAG רב-מודלי

• מיקומים מוטמעים (embeddings) מרובי-מודולים – מודל המיקומים המוטמעים (embeddings) מרובי-המודולים יוצר וקטורים* ב-1, 408 מימדים על סמך הקלט שאתם מספקים, שיכול לכלול שילוב של נתוני תמונות, טקסט וסרטונים. וקטור הטמעת התמונה ווקטור הטמעת הטקסט נמצאים באותו מרחב סמנטי עם אותה מימדות. לכן, אפשר להשתמש בווקטורים האלה באופן חלופי בתרחישי שימוש כמו חיפוש תמונה לפי טקסט או חיפוש סרטון לפי תמונה. כדאי לעיין בהדגמה הזו.

1. שימוש בהטמעה רב-מודאלית להטמעת טקסט ותמונות
2. אחזור של שניהם באמצעות חיפוש לפי דמיון
3. מעבירים את התמונה הגולמית ואת קטעי הטקסט שאוחזרו לLLM רב-מודלי לצורך סינתזה של תשובה

• הטמעת טקסט –

1. שימוש ב-LLM רב-מודלי כדי ליצור סיכומי טקסט של התמונות
2. הטמעת טקסט ואחזור שלו
3. העברת קטעי טקסט ל-LLM לצורך סינתזה של תשובה

מהו Multi-Vector Retriever

באחזור בכמה וקטורים נעשה שימוש בסיכומים של קטעי המסמך כדי לאחזר תוכן מקורי לצורך סינתזה של תשובות. הוא משפר את איכות ה-RAG, במיוחד במשימות שמתבצעות בהן פעולות רבות כמו טבלאות, גרפים, תרשימים וכו'. פרטים נוספים זמינים בבלוג של Langchain.

מה תפַתחו

תרחיש לדוגמה: פיתוח מערכת לתשובות לשאלות באמצעות Gemini Pro

נניח שיש לכם מסמכים שמכילים תרשימים או דיאגרמות מורכבים עם מידע רב. אתם רוצים לחלץ את הנתונים האלה כדי לענות על שאלות או שאילתות.

ב-codelab הזה תלמדו:

• טעינה של נתונים באמצעות LangChain document_loaders
• יצירת סיכומי טקסט באמצעות מודל gemini-pro של Google
• יצירת סיכומי תמונות באמצעות המודל gemini-pro-vision של Google
• יצירת אחזור של וקטוריים מרובים באמצעות מודל textembedding-gecko של Google עם Croma Db בתור מאגר וקטוריים
• פיתוח שרשרת RAG רב-מודאלית למתן תשובות לשאלות

2. לפני שמתחילים

1. בדף לבחירת הפרויקט במסוף Google Cloud, בוחרים או יוצרים פרויקט ב-Google Cloud.
2. מוודאים שהחיוב מופעל בפרויקט שלכם ב-Google Cloud. כך בודקים אם החיוב מופעל בפרויקט
3. הפעלה של כל ממשקי ה-API המומלצים ממסך הבקרה של Vertex AI
4. פותחים את Colab Notebook ונכנסים לאותו חשבון שבו אתם משתמשים ב-Google Cloud.

3. בניית RAG מרובה מצבים

בשיעור ה-Codelab הזה נעשה שימוש ב-Vertex AI SDK for Python וב-Langchain כדי להדגים איך להטמיע את 'אפשרות 2' שמתוארת כאן באמצעות Google Cloud.

אפשר לעיין בקוד המלא בקובץ Multi-modal RAG with Google Cloud במאגר שצוין.

4. שלב 1: התקנה וייבוא של יחסי התלות

!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

מזינים את מזהה הפרויקט ומסיימים את האימות

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

from google.colab import auth
auth.authenticate_user()

איך מפעילים את פלטפורמת Vertex AI

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

5. שלב 2: הכנה וטעינה של נתונים

אנחנו משתמשים בקובץ zip עם קבוצת משנה של התמונות והקובץ ה-PDF שחולצו מהפוסט הזה בבלוג. אם רוצים לפעול לפי התהליך המלא, צריך להשתמש בדוגמה המקורית.

קודם מורידים את הנתונים

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

טעינת תוכן הטקסט מהמסמך

from langchain_community.document_loaders import PyPDFLoader

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

בודקים את התוכן מהדף הראשון

texts[0]

הפלט אמור להיראות כך

סה"כ דפים במסמך

len(texts)

הפלט הצפוי הוא

6. שלב 3: יצירת סיכומי טקסט

קודם מייבאים את הספריות הנדרשות

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

קבלת סיכומי טקסט

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

הפלט הצפוי הוא

7. שלב 4: יצירת סיכומי תמונות

קודם מייבאים את הספריות הנדרשות

import base64
import os

from langchain_core.messages import HumanMessage

יצירת סיכומים של תמונות

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]

הפלט אמור להיראות כך

8. שלב 5: פיתוח אחזור באמצעות רכיבים מרובים

נלמד ליצור סיכומי טקסט ותמונות ולשמור אותם במאגר וקטורים של ChromaDB.

ספריות נדרשות לייבוא

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

יצירת אחזור בכמה ווקטורים

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. שלב 6: יצירת תרשים RAG רב-מודלי

1. הגדרה של פונקציות שימושיות

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. הגדרת הנחיה ספציפית לדומיין לבחירת תמונה

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. הגדרה של שרשרת 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. שלב 7: בודקים את השאילתות

1. אחזור מסמכים רלוונטיים

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 שלנו על אותה שאילתה

result = chain_multimodal_rag.invoke(query)

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

פלט לדוגמה (עשוי להשתנות כשמריצים את הקוד)

11. הסרת המשאבים

כדי להימנע מחיובים בחשבון Google Cloud על המשאבים שבהם השתמשתם בקודלאב הזה, עליכם לפעול לפי השלבים הבאים:

1. נכנסים לדף Manage resources במסוף Google Cloud.
2. ברשימת הפרויקטים, בוחרים את הפרויקט שרוצים למחוק ולוחצים על Delete.
3. כדי למחוק את הפרויקט, כותבים את מזהה הפרויקט בתיבת הדו-שיח ולוחצים על Shut down.

12. מזל טוב

מעולה! פיתחתם בהצלחה ניתוח RAG רב-מודלי באמצעות Gemini.