在 Gemini Live API 中使用 Vertex AI RAG 引擎

檢索增強生成 (RAG) 技術可檢索並提供相關資訊給 LLM，生成可驗證的回覆。這些資訊可能包括最新資訊、主題和背景資訊，或真值。

本頁面說明如何搭配使用 Vertex AI RAG Engine 和 Gemini Live API，指定及擷取 RAG 語料庫中的資訊。

必要條件

如要搭配多模態 Live API 使用 Vertex AI RAG 引擎，請先完成下列必要條件：

  1. 在 Vertex AI 中啟用 RAG API。

  2. 建立 RAG 語料庫 範例。

  3. 如要將檔案上傳至 RAG 語料庫，請參閱「匯入 RAG 檔案範例 API」。

設定

如要搭配 Live API 使用 Vertex AI RAG 引擎，請將 Vertex AI RAG 引擎指定為工具。下列程式碼範例說明如何將 Vertex AI RAG Engine 指定為工具：

請替換下列變數：

  • YOUR_PROJECT_ID： Google Cloud 專案的 ID。
  • YOUR_CORPUS_ID：語料庫的 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。這些程式碼範例示範如何使用 Python API 和 Python SDK，透過 Websocket 執行作業。

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="gemini-2.0-flash-exp"

# 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

如要瞭解如何安裝生成式 AI SDK，請參閱「安裝程式庫」：

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

MODEL="gemini-2.0-flash-exp"

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

將 Vertex AI RAG 引擎做為脈絡資訊儲存庫

您可以將 Vertex AI RAG Engine 做為 Gemini Live API 的脈絡儲存空間，儲存工作階段脈絡，以便形成及擷取與對話相關的過往脈絡，並豐富模型生成作業的當前脈絡。您也可以利用這項功能，在不同的 Live API 工作階段之間共用內容。

Vertex AI RAG Engine 支援儲存及索引來自工作階段脈絡的下列形式資料：

  • 文字
  • 語音

建立 MemoryCorpus 類型的語料庫

如要儲存及索引工作階段情境中的對話文字，您必須建立 MemoryCorpus 類型的 RAG 語料庫。您也必須在記憶體語料庫設定中指定 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-2.5-pro-preview-05-06"

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  # Such as "gemini-2.0-flash-exp"

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

後續步驟