使用 Private Service Connect 设置 Vector Search

def vector_search_create_index_endpoint_private_service_connect(
    project: str, location: str, display_name: str, project_allowlist: list[str]
) -> aiplatform.MatchingEngineIndexEndpoint:
    """Create a vector search index endpoint with Private Service Connect enabled.

    Args:
        project (str): Required. Project ID
        location (str): Required. The region name
        display_name (str): Required. The index endpoint display name
        project_allowlist (list[str]): Required. A list of projects from which
            the forwarding rule will be able to target the service attachment.

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

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

    return index_endpoint

def vector_search_deploy_psc_automation_index(
    project: str,
    location: str,
    index_name: str,
    index_endpoint_name: str,
    deployed_index_id: str,
    psc_automation_configs: Sequence[Tuple[str, str]],
) -> None:
    """Deploy a vector search index to an index endpoint using PSC automation.

    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.
        psc_automation_config (Sequence[Tuple[str, str]]): Required. A list of
          (project_id, network) pairs where PSC endpoints will be setup for the
          deployed index. Example:
          [("123", "{projects/123/global/networks/my-network1"),
          ("123", "{projects/123/global/networks/my-network2")]
    """
    # 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 with PSC automation enabled.
    index_endpoint.deploy_index(
        index=index,
        deployed_index_id=deployed_index_id,
        psc_automation_configs=psc_automation_configs,
    )

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)