Utilizzare RAG Engine sulla piattaforma agentica Gemini Enterprise nell'API Gemini Live

La Retrieval-Augmented Generation (RAG) è una tecnica utilizzata per recuperare e fornire informazioni pertinenti agli LLM per generare risposte verificabili. Le informazioni possono includere informazioni aggiornate, un argomento e un contesto o il dato di fatto.

Questa pagina mostra come utilizzare RAG Engine con l' API Gemini Live, che consente di specificare e recuperare informazioni dal corpus RAG.

Prerequisiti

Prima di poter utilizzare RAG Engine con l'API Gemini Live, devi completare i seguenti prerequisiti:

  1. Abilita l'API RAG in Agent Platform.

  2. Crea l'esempio di corpus RAG.

  3. Per caricare i file nel corpus RAG, consulta l'esempio di API per l'importazione di file RAG.

Configura

Puoi utilizzare RAG Engine con l'API Live specificando RAG Engine come strumento. Il seguente esempio di codice mostra come specificare RAG Engine come strumento:

Sostituisci le seguenti variabili:

  • YOUR_PROJECT_ID: l'ID del tuo Google Cloud progetto.
  • YOUR_CORPUS_ID: l'ID del tuo corpus.
  • YOUR_LOCATION: la regione in cui elaborare la richiesta.
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}"
      }
    }
  }
}

Utilizza Websocket per la comunicazione in tempo reale

Per abilitare la comunicazione in tempo reale tra un client e un server, devi utilizzare un Websocket. Questi esempi di codice mostrano come utilizzare un Websocket utilizzando l'API Python e l'SDK Python.

API Python 

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)}"))

SDK Python

Per scoprire come installare l'SDK AI generativa, consulta Installare una libreria:

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

Utilizza RAG Engine come archivio di contesto

Puoi utilizzare RAG Engine come archivio di contesto per l'API Gemini Live per archiviare il contesto della sessione, formare e recuperare i contesti precedenti correlati alla conversazione e arricchire il contesto attuale per la generazione del modello. Puoi anche sfruttare questa funzionalità per condividere i contesti tra le diverse sessioni dell'API Live.

RAG Engine supporta l'archiviazione e l'indicizzazione delle seguenti forme di dati dai contesti delle sessioni:

  • Testo
  • Discorso audio

Crea un corpus di tipo MemoryCorpus

Per archiviare e indicizzare i testi delle conversazioni dal contesto della sessione, devi creare un corpus RAG di tipo MemoryCorpus. Devi anche specificare un parser LLM nella configurazione del corpus di memoria utilizzato per analizzare i contesti delle sessioni archiviati dall'API Live per creare la memoria per l'indicizzazione.

Questo esempio di codice mostra come creare un corpus. Tuttavia, prima sostituisci le variabili con i valori.

# 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
           )
       )
   ),
)

Specifica il corpus di memoria per archiviare i contesti

Quando utilizzi il corpus di memoria con l'API Live, devi specificare il corpus di memoria come strumento di recupero e poi impostare store_context su true per consentire all'API Live di archiviare i contesti delle sessioni.

Questo esempio di codice mostra come specificare il corpus di memoria per archiviare i contesti. Tuttavia, prima sostituisci le variabili con i valori.

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

Passaggi successivi