איך משתמשים ב-RAG Engine ב-Gemini Enterprise Agent Platform ב-Gemini Live API

‫Retrieval-augmented generation ‏(RAG) היא שיטה שמשמשת לאחזור מידע רלוונטי ולספק אותו למודלים גדולים של שפה (LLM) כדי ליצור תשובות שניתן לאמת. המידע יכול לכלול מידע עדכני, נושא והקשר או אמת קרקע (ground truth).

בדף הזה מוסבר איך משתמשים ב-RAG Engine עם Gemini Live API, שמאפשר לציין ולאחזר מידע מקורפוס ה-RAG.

דרישות מוקדמות

כדי להשתמש במנוע RAG עם Gemini Live API, צריך לבצע את הפעולות הבאות:

  1. מפעילים את RAG API ב-Agent Platform.

  2. יצירת מאגר המידע של RAG.

  3. כדי להעלות קבצים למאגר המידע של RAG, אפשר לעיין בדוגמה לייבוא קבצים של RAG באמצעות API.

הגדרה

כדי להשתמש ב-RAG Engine עם Live API, צריך לציין את RAG Engine ככלי. דוגמת הקוד הבאה מראה איך לציין את RAG Engine ככלי:

מחליפים את המשתנים הבאים:

  • ‫YOUR_PROJECT_ID: מזהה הפרויקט ב- Google Cloud .
  • ‫YOUR_CORPUS_ID: המזהה של הקורפוס.
  • ‫YOUR_LOCATION: האזור שבו הבקשה תעובד.
PROJECT_ID = "YOUR_PROJECT_ID"
RAG_CORPUS_ID = "YOUR_CORPUS_ID"
LOCATION = "YOUR_LOCATION"

TOOLS = {
  "retrieval": {
    "vertex_rag_store": {
        "rag_resources": {
        "rag_corpus": "projects/${PROJECT_ID}/locations/${LOCATION}/ragCorpora/${RAG_CORPUS_ID}"
      }
    }
  }
}

שימוש ב-Websocket לתקשורת בזמן אמת

כדי לאפשר תקשורת בזמן אמת בין לקוח לשרת, צריך להשתמש ב-Websocket. בדוגמאות הקוד האלה נסביר איך להשתמש ב-Websocket באמצעות API בשפת Python ו-Python SDK.

Python API 

CONFIG = {"response_modalities": ["TEXT"], "speech_config": { "language_code": "en-US" }}
headers = {
  "Content-Type": "application/json",
  "Authorization": f"Bearer {bearer_token[0]}",
}
HOST= "${LOCATION}-aiplatform.googleapis.com"
SERVICE_URL = f"wss://{HOST}/ws/google.cloud.aiplatform.v1beta1.LlmBidiService/BidiGenerateContent"
MODEL = "YOUR_MODEL_NAME"

# Connect to the server
async with connect(SERVICE_URL, additional_headers=headers) as ws:
  # Setup the session
  await ws.send(
json.dumps(
          {
              "setup": {
                  "model": MODEL,
                  "generation_config": CONFIG,
                  # Setup RAG as a retrieval tool
                  "tools": TOOLS,
              }
          }
      )
  )

  # Receive setup response
  raw_response = await ws.recv(decode=False)
  setup_response = json.loads(raw_response.decode("ascii"))

  # Send text message
  text_input = "What are popular LLMs?"
  display(Markdown(f"**Input:** {text_input}"))

  msg = {
      "client_content": {
          "turns": [{"role": "user", "parts": [{"text": text_input}]}],
          "turn_complete": True,
      }
  }

  await ws.send(json.dumps(msg))

  responses = []

  # Receive chunks of server response
  async for raw_response in ws:
      response = json.loads(raw_response.decode())
      server_content = response.pop("serverContent", None)
      if server_content is None:
          break

      model_turn = server_content.pop("modelTurn", None)
      if model_turn is not None:
          parts = model_turn.pop("parts", None)
          if parts is not None:
              display(Markdown(f"**parts >** {parts}"))
              responses.append(parts[0]["text"])

      # End of turn
      turn_complete = server_content.pop("turnComplete", None)
      if turn_complete:
          grounding_metadata = server_content.pop("groundingMetadata", None)
          if grounding_metadata is not None:
            grounding_chunks = grounding_metadata.pop("groundingChunks", None)
            if grounding_chunks is not None:
              for chunk in grounding_chunks:
                display(Markdown(f"**grounding_chunk >** {chunk}"))
          break

  # Print the server response
  display(Markdown(f"**Response >** {''.join(responses)}"))

Python SDK

כדי ללמוד איך להתקין את ה-SDK של ה-AI הגנרטיבי, קראו את המאמר התקנת ספרייה:

from google import genai
from google.genai import types
from google.genai.types import (Content, LiveConnectConfig, HttpOptions, Modality, Part,)
from IPython import display

MODEL = "YOUR_MODEL_NAME"

client = genai.Client(
  vertexai=True,
  project=PROJECT_ID,
  location=LOCATION
)

async with client.aio.live.connect(
  model=MODEL,
  config=LiveConnectConfig(response_modalities=[Modality.TEXT],
                            tools=TOOLS),
) as session:
  text_input = "\'What are core LLM techniques?\'"
  print("> ", text_input, "\n")
  await session.send_client_content(
      turns=Content(role="user", parts=[Part(text=text_input)])
  )

  async for message in session.receive()
      if message.text:
          display.display(display.Markdown(message.text))
          continue

שימוש במנוע RAG בתור מאגר ההקשר

אתם יכולים להשתמש ב-RAG Engine כמאגר ההקשרים של Gemini Live API כדי לאחסן את ההקשר של הסשן, ליצור ולאחזר הקשרים קודמים שקשורים לשיחה שלכם ולהעשיר את ההקשר הנוכחי ליצירת המודל. אתם יכולים גם להשתמש בתכונה הזו כדי לשתף הקשרים בין סשנים שונים של Live API.

מנוע RAG תומך באחסון ובאינדקס של סוגי הנתונים הבאים מהקשרי סשנים:

  • טקסט
  • דיבור באודיו

יצירת מאגר מסוג MemoryCorpus

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

דוגמת הקוד הזו ממחישה איך ליצור קורפוס. אבל קודם צריך להחליף את המשתנים בערכים.

# Currently supports Google first-party embedding models
EMBEDDING_MODEL = YOUR_EMBEDDING_MODEL  # Such as "publishers/google/models/text-embedding-005"
MEMORY_CORPUS_DISPLAY_NAME = YOUR_MEMORY_CORPUS_DISPLAY_NAME
LLM_PARSER_MODEL_NAME = YOUR_LLM_PARSER_MODEL_NAME  # Such as "projects/{project_id}/locations/{location}/publishers/google/models/gemini-3.8-flash"

memory_corpus = rag.create_corpus(
   display_name=MEMORY_CORPUS_DISPLAY_NAME,
   corpus_type_config=rag.RagCorpusTypeConfig(
       corpus_type_config=rag.MemoryCorpus(
           llm_parser=rag.LlmParserConfig(
               model_name=LLM_PARSER_MODEL_NAME,
           )
       )
   ),
   backend_config=rag.RagVectorDbConfig(
       rag_embedding_model_config=rag.RagEmbeddingModelConfig(
           vertex_prediction_endpoint=rag.VertexPredictionEndpoint(
               publisher_model=EMBEDDING_MODEL
           )
       )
   ),
)

הגדרת מאגר הזיכרון לאחסון הקשרים

כשמשתמשים בקורפוס הזיכרון עם Live API, צריך לציין את קורפוס הזיכרון ככלי אחזור ואז להגדיר את store_context כ-true כדי לאפשר ל-Live API לאחסן את ההקשרים של הסשן.

דוגמת הקוד הזו ממחישה איך לציין את מאגר הזיכרון לאחסון הקשרים. עם זאת, קודם צריך להחליף את המשתנים בערכים.

from google import genai
from google.genai import types
from google.genai.types import (Content, LiveConnectConfig, HttpOptions, Modality, Part)
from IPython import display

PROJECT_ID=YOUR_PROJECT_ID
LOCATION=YOUR_LOCATION
TEXT_INPUT=YOUR_TEXT_INPUT
MODEL_NAME=YOUR_MODEL_NAME

client = genai.Client(
   vertexai=True,
   project=PROJECT_ID,
   location=LOCATION,
)

memory_store=types.VertexRagStore(
   rag_resources=[
       types.VertexRagStoreRagResource(
           rag_corpus=memory_corpus.name
       )
   ],
   store_context=True
)

async with client.aio.live.connect(
   model=MODEL_NAME,
   config=LiveConnectConfig(response_modalities=[Modality.TEXT],
                            tools=[types.Tool(
                                retrieval=types.Retrieval(
                                    vertex_rag_store=memory_store))]),
) as session:
   text_input=TEXT_INPUT
   await session.send_client_content(
       turns=Content(role="user", parts=[Part(text=text_input)])
   )

   async for message in session.receive():
       if message.text:
           display.display(display.Markdown(message.text))
           continue

המאמרים הבאים