将 Agent Platform Vector Search 与 RAG 引擎搭配使用

本页面介绍了如何将 RAG 引擎连接到 Gemini Enterprise Agent Platform Vector Search。

您也可以使用 RAG Engine 搭配 Agent Platform Vector Search 笔记本进行操作。

Agent Platform Vector Search 设置

Agent Platform Vector Search 基于 Google 研究开发的 Vector Search 技术。借助 Vector Search，您可以利用为 Google 搜索、YouTube 和 Google Play 等 Google 产品奠定基础的基础设施。

如需与 RAG 引擎集成，您需要一个空的 Vector Search 搜索索引。

设置 Vertex AI SDK

如需设置 Vertex AI SDK，请参阅设置。

创建 Vector Search 索引

如需创建与 RAG 语料库兼容的 Vector Search 索引，该索引必须满足以下条件：

  1. IndexUpdateMethod 必须为 STREAM_UPDATE，请参阅创建流索引。

  2. 距离衡量类型必须明确设置为以下其中一种类型：

    • DOT_PRODUCT_DISTANCE
    • COSINE_DISTANCE

  3. 向量的维度必须与您计划在 RAG 语料库中使用的嵌入模型一致。其他参数可以根据您的选择进行调整，这些选择决定了是否可以调整其他参数。

Python

如需了解如何安装或更新 Vertex AI SDK for Python，请参阅安装 Vertex AI SDK for Python。 如需了解详情，请参阅 Python API 参考文档。 

def vector_search_create_streaming_index(
    project: str, location: str, display_name: str, gcs_uri: Optional[str] = None
) -> aiplatform.MatchingEngineIndex:
    """Create a vector search index.

    Args:
        project (str): Required. Project ID
        location (str): Required. The region name
        display_name (str): Required. The index display name
        gcs_uri (str): Optional. The Google Cloud Storage uri for index content

    Returns:
        The created MatchingEngineIndex.
    """
    # Initialize the Vertex AI client
    aiplatform.init(project=project, location=location)

    # Create Index
    index = aiplatform.MatchingEngineIndex.create_tree_ah_index(
        display_name=display_name,
        contents_delta_uri=gcs_uri,
        description="Matching Engine Index",
        dimensions=100,
        approximate_neighbors_count=150,
        leaf_node_embedding_count=500,
        leaf_nodes_to_search_percent=7,
        index_update_method="STREAM_UPDATE",  # Options: STREAM_UPDATE, BATCH_UPDATE
        distance_measure_type=aiplatform.matching_engine.matching_engine_index_config.DistanceMeasureType.DOT_PRODUCT_DISTANCE,
    )

    return index

创建 Vector Search 索引端点

RAG 引擎支持公共端点。

Python

如需了解如何安装或更新 Vertex AI SDK for Python，请参阅安装 Vertex AI SDK for Python。 如需了解详情，请参阅 Python API 参考文档。 

def vector_search_create_index_endpoint(
    project: str, location: str, display_name: str
) -> None:
    """Create a vector search index endpoint.

    Args:
        project (str): Required. Project ID
        location (str): Required. The region name
        display_name (str): Required. The index endpoint display name
    """
    # Initialize the Vertex AI client
    aiplatform.init(project=project, location=location)

    # Create Index Endpoint
    index_endpoint = aiplatform.MatchingEngineIndexEndpoint.create(
        display_name=display_name,
        public_endpoint_enabled=True,
        description="Matching Engine Index Endpoint",
    )

    print(index_endpoint.name)

将索引部署到索引端点

在执行最近邻搜索之前，必须将索引部署到索引端点。

Python

如需了解如何安装或更新 Vertex AI SDK for Python，请参阅安装 Vertex AI SDK for Python。 如需了解详情，请参阅 Python API 参考文档。 

def vector_search_deploy_index(
    project: str,
    location: str,
    index_name: str,
    index_endpoint_name: str,
    deployed_index_id: str,
) -> None:
    """Deploy a vector search index to a vector search index endpoint.

    Args:
        project (str): Required. Project ID
        location (str): Required. The region name
        index_name (str): Required. The index to update. A fully-qualified index
          resource name or a index ID.  Example:
          "projects/123/locations/us-central1/indexes/my_index_id" or
          "my_index_id".
        index_endpoint_name (str): Required. Index endpoint to deploy the index
          to.
        deployed_index_id (str): Required. The user specified ID of the
          DeployedIndex.
    """
    # Initialize the Vertex AI client
    aiplatform.init(project=project, location=location)

    # Create the index instance from an existing index
    index = aiplatform.MatchingEngineIndex(index_name=index_name)

    # Create the index endpoint instance from an existing endpoint.
    index_endpoint = aiplatform.MatchingEngineIndexEndpoint(
        index_endpoint_name=index_endpoint_name
    )

    # Deploy Index to Endpoint
    index_endpoint = index_endpoint.deploy_index(
        index=index, deployed_index_id=deployed_index_id
    )

    print(index_endpoint.deployed_indexes)

如果您是首次将索引部署到索引端点，则需要大约 30 分钟才能自动构建并启动后端，然后才能存储索引。首次部署后，索引会在几秒钟内准备就绪。如需查看索引部署的状态，请打开 Vector Search 控制台，选择索引端点标签页，然后选择您的索引端点。

确定索引和索引端点的资源名称，其格式如下：

  • projects/${PROJECT_NUMBER}/locations/${LOCATION_ID}/indexes/${INDEX_ID}
  • projects/${PROJECT_NUMBER}/locations/${LOCATION_ID}/indexEndpoints/${INDEX_ENDPOINT_ID}。

在 RAG 引擎中使用 Agent Platform Vector Search

设置 Vector Search 实例后，按照本部分中的步骤将 Vector Search 实例设置为 RAG 应用的向量数据库。

设置向量数据库以创建 RAG 语料库

创建 RAG 语料库时，仅指定完整的 INDEX_ENDPOINT_NAME 和 INDEX_NAME。确保索引和索引端点资源名称都使用数字 ID。RAG 语料库会进行创建，并自动与 Vector Search 索引关联。系统会根据条件执行验证。如果不满足任何一项要求，则请求会被拒绝。

Python

试用此示例之前，请按照《Agent Platform 快速入门：使用客户端库》中的 Python 设置说明进行操作。

如需向代理平台进行身份验证，请设置应用默认凭据。 如需了解详情，请参阅为本地开发环境设置身份验证。 


import agentplatform
from agentplatform import types

# TODO(developer): Update and un-comment below lines
# PROJECT_ID = "your-project-id"
# vector_search_index_name = "projects/{PROJECT_ID}/locations/{LOCATION}/indexes/{INDEX_ID}"
# vector_search_index_endpoint_name = "projects/{PROJECT_ID}/locations/{LOCATION}/indexEndpoints/{INDEX_ENDPOINT_ID}"
# display_name = "test_corpus"
# description = "Corpus Description"

# Initialize Agent Platform client once per session
client = agentplatform.Client(project=PROJECT_ID, location="us-central1")

# Configure embedding model (Optional)
embedding_model_config = types.RagEmbeddingModelConfig(
    vertex_prediction_endpoint=types.RagEmbeddingModelConfigVertexPredictionEndpoint(
        endpoint="publishers/google/models/text-embedding-005"
    )
)

# Configure Vector DB
vector_db = types.RagVectorDbConfigVertexVectorSearch(
    index=vector_search_index_name,
    index_endpoint=vector_search_index_endpoint_name
)

corpus = client.rag.create_corpus(
    rag_corpus=types.RagCorpus(
        display_name=display_name,
        description=description,
        rag_vector_db_config=types.RagVectorDbConfig(
            rag_embedding_model_config=embedding_model_config,
            vertex_vector_search=vector_db,
        ),
    )
)
print(corpus)
# Example response:
# RagCorpus(name='projects/1234567890/locations/us-central1/ragCorpora/1234567890',
# display_name='test_corpus', description='Corpus Description', embedding_model_config=...
# ...

REST

  # TODO(developer): Update and un-comment the following lines:
  # CORPUS_DISPLAY_NAME = "YOUR_CORPUS_DISPLAY_NAME"
  # Full index/indexEndpoint resource name
  # Index: projects/${PROJECT_NUMBER}/locations/${LOCATION_ID}/indexes/${INDEX_ID}
  # IndexEndpoint: projects/${PROJECT_NUMBER}/locations/${LOCATION_ID}/indexEndpoints/${INDEX_ENDPOINT_ID}
  # INDEX_RESOURCE_NAME = "YOUR_INDEX_ENDPOINT_RESOURCE_NAME"
  # INDEX_NAME = "YOUR_INDEX_RESOURCE_NAME"
  # Call CreateRagCorpus API to create a new RagCorpus
  curl -X POST -H "Authorization: Bearer $(gcloud auth print-access-token)" -H "Content-Type: application/json" https://${LOCATION_ID}-aiplatform.googleapis.com/v1beta1/projects/${PROJECT_NUMBER}/locations/${LOCATION_ID}/ragCorpora -d '{
        "display_name" : '\""${CORPUS_DISPLAY_NAME}"\"',
        "rag_vector_db_config" : {
                "vertex_vector_search": {
                  "index":'\""${INDEX_NAME}"\"'
              "index_endpoint":'\""${INDEX_ENDPOINT_NAME}"\"'
                }
          }
    }'

  # Call ListRagCorpora API to verify the RagCorpus is created successfully
  curl -sS -X GET \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $(gcloud auth print-access-token)" \
  "https://${LOCATION_ID}-aiplatform.googleapis.com/v1beta1/projects/${PROJECT_NUMBER}/locations/${LOCATION_ID}/ragCorpora"

使用 RAG API 导入文件

使用 ragFiles.import API 方法将文件从 Cloud Storage 或 Google 云端硬盘导入到 Vector Search 索引。这些文件会嵌入到 Vector Search 索引并存储在其中。

REST

# TODO(developer): Update and uncomment the following lines:
# RAG_CORPUS_ID = "your-rag-corpus-id"
#
# Google Cloud Storage bucket/file location.
# For example, "gs://rag-fos-test/"
# GCS_URIS= "your-gcs-uris"

# Call ImportRagFiles API to embed files and store in the BigQuery table
curl -X POST \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://us-central1-aiplatform.googleapis.com/v1beta1/projects/${PROJECT_NUMBER}/locations/us-central1/ragCorpora/${RAG_CORPUS_ID}/ragFiles:import \
-d '{
  "import_rag_files_config": {
    "gcs_source": {
      "uris": '\""${GCS_URIS}"\"'
    },
    "rag_file_chunking_config": {
      "chunk_size": 512
    }
  }
}'

# Call ListRagFiles API to verify the files are imported successfully
curl -X GET \
-H "Content-Type: application/json" \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
https://us-central1-aiplatform.googleapis.com/v1beta1/projects/${PROJECT_NUMBER}/locations/us-central1/ragCorpora/${RAG_CORPUS_ID}/ragFiles

Python

如需了解如何安装或更新 Vertex AI SDK for Python，请参阅安装 Vertex AI SDK for Python。 如需了解详情，请参阅 Python API 参考文档。 


import agentplatform
from agentplatform import types

from google.genai import types as genai_types

# TODO(developer): Update and un-comment below lines
# PROJECT_ID = "your-project-id"
# corpus_name = "projects/{PROJECT_ID}/locations/us-central1/ragCorpora/{rag_corpus_id}"

# Supports Google Cloud Storage and Google Drive Links
# paths = ["https://drive.google.com/file/d/123", "gs://my_bucket/my_files_dir/*"]

# Initialize Agent Platform client once per session
client = agentplatform.Client(project=PROJECT_ID, location="us-central1")

response = client.rag.import_files(
    name=corpus_name,
    import_config=types.ImportRagFilesConfig(
        gcs_source=genai_types.GcsSource(uris=[paths[1]]),
        google_drive_source=types.GoogleDriveSource(
            resource_ids=[
                types.GoogleDriveSourceResourceId(
                    resource_id=paths[0],
                    resource_type=types.ResourceType.RESOURCE_TYPE_FILE
                )
            ]
        ), # optional
        rag_file_transformation_config=types.RagFileTransformationConfig(
            rag_file_chunking_config=types.RagFileChunkingConfig(
                chunk_size=512,
                chunk_overlap=100,
            )
        ), # optional
        max_embedding_requests_per_min=900, # optional
    )
)

print(f"Imported {response.imported_rag_files_count} files.")
# Example response:
# Imported 2 files.

使用 RAG API 检索相关上下文

文件导入完成后，您可以使用 RetrieveContexts API 从 Vector Search 索引检索相关上下文。

REST

# TODO(developer): Update and uncomment the following lines:
# RETRIEVAL_QUERY="your-retrieval-query"
#
# Full RAG corpus resource name
# Format:
# "projects/${PROJECT_NUMBER}/locations/us-central1/ragCorpora/${RAG_CORPUS_ID}"
# RAG_CORPUS_RESOURCE="your-rag-corpus-resource"

# Call RetrieveContexts API to retrieve relevant contexts
curl -X POST \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $(gcloud auth print-access-token)" \
https://us-central1-aiplatform.googleapis.com/v1beta1/projects/${PROJECT_NUMBER}/locations/us-central1:retrieveContexts \
  -d '{
    "vertex_rag_store": {
      "rag_resources": {
          "rag_corpus": '\""${RAG_CORPUS_RESOURCE}"\"',
        },
    },
    "query": {
      "text": '\""${RETRIEVAL_QUERY}"\"',
      "similarity_top_k": 10
    }
  }'

Python

如需了解如何安装或更新 Vertex AI SDK for Python，请参阅安装 Vertex AI SDK for Python。 如需了解详情，请参阅 Python API 参考文档。 


import agentplatform

from agentplatform import types
from google.genai import types as genai_types

# TODO(developer): Update and un-comment below lines
# PROJECT_ID = "your-project-id"
# corpus_name = "projects/[PROJECT_ID]/locations/us-central1/ragCorpora/[rag_corpus_id]"

# Initialize Agent Platform client once per session
client = agentplatform.Client(project=PROJECT_ID, location="us-east4")

response = client.rag.retrieve_contexts(
    vertex_rag_store=genai_types.VertexRagStore(
        rag_resources=[
            genai_types.VertexRagStoreRagResource(
                rag_corpus=corpus_name,
                # Optional: supply IDs from `rag.list_files()`.
                # rag_file_ids=["rag-file-1", "rag-file-2", ...],
            )
        ],
    ),
    query=types.RagQuery(
        text="Hello World!",
        rag_retrieval_config=genai_types.RagRetrievalConfig(
            top_k=10,
            filter=genai_types.RagRetrievalConfigFilter(
                vector_distance_threshold=0.5
            ),
        ),
    )
)
print(response)
# Example response:
# contexts {
#   contexts {
#     source_uri: "gs://your-bucket-name/file.txt"
#     text: "....
#   ....

使用 Agent Platform Gemini API 生成内容

如需使用 Gemini 模型生成内容，请调用 Agent Platform GenerateContent API。通过在请求中指定 RAG_CORPUS_RESOURCE，API 会自动从 Vector Search 索引中检索数据。

REST

# TODO(developer): Update and uncomment the following lines:
# MODEL_ID=gemini-2.5-flash
# GENERATE_CONTENT_PROMPT="your-generate-content-prompt"

# GenerateContent with contexts retrieved from the FeatureStoreOnline index
curl -X POST \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json"  https://us-central1-aiplatform.googleapis.com/v1beta1/projects/${PROJECT_NUMBER}/locations/us-central1/publishers/google/models/${MODEL_ID}:generateContent \
-d '{
  "contents": {
    "role": "user",
    "parts": {
      "text": '\""${GENERATE_CONTENT_PROMPT}"\"'
    }
  },
  "tools": {
    "retrieval": {
      "vertex_rag_store": {
        "rag_resources": {
            "rag_corpus": '\""${RAG_CORPUS_RESOURCE}"\"',
          },
        "similarity_top_k": 8,
        "vector_distance_threshold": 0.32
      }
    }
  }
}'

Python

如需了解如何安装或更新 Vertex AI SDK for Python，请参阅安装 Vertex AI SDK for Python。 如需了解详情，请参阅 Python API 参考文档。 


from google import genai
from google.genai import types as genai_types

# TODO(developer): Update and un-comment below lines
# PROJECT_ID = "your-project-id"
# corpus_name = "projects/{PROJECT_ID}/locations/us-central1/ragCorpora/{rag_corpus_id}"

rag_retrieval_tool = genai_types.Tool(
    retrieval=genai_types.Retrieval(
        vertex_rag_store=genai_types.VertexRagStore(
            rag_resources=[
                genai_types.VertexRagStoreRagResource(
                    rag_corpus=corpus_name
                )
            ],
            rag_retrieval_config=genai_types.RagRetrievalConfig(
                top_k=10,
                filter=genai_types.RagRetrievalConfigFilter(
                    vector_distance_threshold=0.5
                ),
            ),
        ),
    )
)

# Create a GenAI SDK client to make a generate_content request
genai_client = genai.Client(enterprise=True, project=PROJECT_ID, location="us-central1")

response = genai_client.models.generate_content(
    model="gemini-2.5-pro",
    contents="Why is the sky blue?",
    config=genai_types.GenerateContentConfig(
        tools=[rag_retrieval_tool]
    )
)
print(response.text)
# Example response:
#   The sky appears blue due to a phenomenon called Rayleigh scattering.
#   Sunlight, which contains all colors of the rainbow, is scattered
#   by the tiny particles in the Earth's atmosphere....
#   ...

后续步骤