En esta página, se describe cómo crear un conector personalizado.
Antes de comenzar
Antes de comenzar, asegúrate de tener lo siguiente:
Verifica si la facturación está habilitada en tu Google Cloud proyecto.
Instala e inicializa gcloud CLI. y asegúrate de que esté autenticada para tu proyecto.
Instala las bibliotecas de Python necesarias:
pip install google-cloud-discoveryengine requestsObtén acceso de administrador de Discovery Engine para tu Google Cloud proyecto de.
Obtén credenciales de acceso para tu fuente de datos de terceros (como claves de API o autenticación de base de datos).
Crea un plan claro de asignación de datos. Este debe incluir qué campos indexar y cómo representar el control de acceso, incluidas las identidades de terceros.
Crea un conector básico
En esta sección, se muestra cómo crear un conector personalizado en el idioma que elijas. Los principios y patrones que se muestran aquí se aplican a cualquier sistema externo. Simplemente adapta las llamadas a la API y las transformaciones de datos para tu fuente específica en el idioma que elijas para crear un conector básico.
Cómo obtener datos
Para comenzar, recupera datos de tu fuente de datos de terceros. En este ejemplo, se muestra cómo obtener publicaciones con paginación. Para entornos de producción, recomendamos usar un enfoque de transmisión para conjuntos de datos grandes. Esto evita los problemas de memoria que pueden ocurrir cuando se cargan todos los datos a la vez.
Python
from typing import List
import requests
def fetch_posts(base_url: str, per_page: int = 15) -> List[dict]:
#Fetch all posts from the given site.#
url = base_url.rstrip("/") + "/wp-json/wp/v2/posts"
posts: List[dict] = []
page = 1
while True:
resp = requests.get(
url,
params={"page": page, "per_page": per_page},
# headers={"Authorization": "Bearer YOUR_API_KEY"}, # Add auth if needed
)
resp.raise_for_status()
batch = resp.json()
posts.extend(batch)
if len(batch) < per_page:
break
page += 1
return posts
Transforma los datos
Para convertir tus datos de origen al formato de documento de Discovery Engine, estructúralos como se muestra en la siguiente carga útil de ejemplo. Puedes incluir tantos pares clave-valor como necesites. Por ejemplo, puedes incluir el contenido completo para una búsqueda integral. Como alternativa, puedes incluir campos estructurados para una búsqueda por facetas o una combinación de ambos.
Python
from typing import List
import json
from google.cloud import discoveryengine_v1 as discoveryengine
def convert_posts_to_documents(posts: List[dict]) -> List[discoveryengine.Document]:
# Convert WP posts into Discovery Engine Document messages.
docs: List[discoveryengine.Document] = []
for post in posts:
payload = {
"title": post.get("title", {}).get("rendered"),
"body": post.get("content", {}).get("rendered"),
"url": post.get("link"),
"author": post.get("author"),
"categories": post.get("categories"),
"tags": post.get("tags"),
"date": post.get("date"),
}
doc = discoveryengine.Document(
id=str(post["id"]),
json_data=json.dumps(payload),
)
docs.append(doc)
return docs
Recupera o crea un almacén de identidades
Para administrar las identidades y los grupos de usuarios para el control de acceso, debes recuperar o crear un almacén de identidades. Esta función obtiene un almacén de identidades existente por su ID, proyecto y ubicación. Si el almacén de identidades no existe, crea y muestra uno nuevo y vacío.
Python
from google.cloud import discoveryengine_v1 as discoveryengine
def get_or_create_ims_data_store(
project_id: str,
location: str,
identity_mapping_store_id: str,
) -> discoveryengine.DataStore:
"""Get or create a DataStore."""
# Initialize the client
client_ims = discoveryengine.IdentityMappingStoreServiceClient()
# Construct the parent resource name
parent_ims = client_ims.location_path(project=project_id, location=location)
try:
# Create the request object
name = f"projects/{project_id}/locations/{location}/identityMappingStores/{identity_mapping_store_id}"
request = discoveryengine.GetIdentityMappingStoreRequest(
name=name,
)
return client_ims.get_identity_mapping_store(request=request)
except:
# Create the IdentityMappingStore object (it can be empty for basic creation)
identity_mapping_store = discoveryengine.IdentityMappingStore()
# Create the request object
request = discoveryengine.CreateIdentityMappingStoreRequest(
parent=parent_ims,
identity_mapping_store=identity_mapping_store,
identity_mapping_store_id=identity_mapping_store_id,
)
return client_ims.create_identity_mapping_store(request=request)
La función get_or_create_ims_data_store usa las siguientes variables clave:
project_id: El ID de tu Google Cloud proyecto.location: La Google Cloud ubicación del almacén de asignación de identidades.identity_mapping_store_id: Un identificador único para el almacén de identidades.client_ims: Una instancia de discoveryengine.IdentityMappingStoreServiceClient que se usa para interactuar con la API del almacén de identidades.parent_ims: El nombre del recurso de la ubicación superior, construido con client_ims.location_path.name: El nombre completo del recurso del almacén de asignación de identidades, que se usa para GetIdentityMappingStoreRequest.
Transfiere la asignación de identidades al almacén de identidades
Para cargar entradas de asignación de identidades en el almacén de identidades especificado, usa esta función. Toma una lista de entradas de asignación de identidades y comienza una operación de importación intercalada. Esto es fundamental para establecer las relaciones de usuario, grupo y de identidad externa necesarias para el control de acceso y la personalización.
Python
Python
from google.cloud import discoveryengine_v1 as discoveryengine
def load_ims_data(
ims_store: discoveryengine.DataStore,
id_mapping_data: list[discoveryengine.IdentityMappingEntry],
) -> discoveryengine.DataStore:
"""Get the IMS data store."""
# Initialize the client
client_ims = discoveryengine.IdentityMappingStoreServiceClient()
# Create the InlineSource object
inline_source = discoveryengine.ImportIdentityMappingsRequest.InlineSource(
identity_mapping_entries=id_mapping_data
)
# Create the main request object
request_ims = discoveryengine.ImportIdentityMappingsRequest(
identity_mapping_store=ims_store.name,
inline_source=inline_source,
)
try:
# Create the InlineSource object, which holds your list of entries
operation = client_ims.import_identity_mappings(
request=request_ims,
)
result = operation.result()
return result
except Exception as e:
print(f"IMS Load Error: {e}")
return None
La función load_ims_data usa las siguientes variables clave:
ims_store: El objeto discoveryengine.DataStore que representa el almacén de asignación de identidades en el que se cargarán los datos.id_mapping_data: Una lista de objetos discoveryengine.IdentityMappingEntry, cada uno de los cuales contiene una identidad externa y su ID de usuario o grupo correspondiente.result: Valor de retorno del tipo discoveryengine.DataStore.
Crea un almacén de datos
Para usar un conector personalizado, debes inicializar un almacén de datos para tu contenido. Usa default_collection para conectores personalizados. El parámetro IndustryVertical personaliza el comportamiento del almacén de datos para casos de uso específicos. GENERIC es adecuado para la mayoría de los casos. Sin embargo, puedes elegir un valor diferente para una industria en particular, como MEDIA o HEALTHCARE_FHIR. Configura el nombre visible y otras propiedades para que se alineen con las convenciones y los requisitos de nomenclatura de tu proyecto.
Python
from google.cloud import discoveryengine_v1 as discoveryengine
def get_or_create_data_store(
project_id: str,
location: str,
display_name: str,
data_store_id: str,
identity_mapping_store: str,
) -> discoveryengine.DataStore:
"""Get or create a DataStore."""
client = discoveryengine.DataStoreServiceClient()
ds_name = client.data_store_path(project_id, location, data_store_id)
try:
result = client.get_data_store(request={"name": ds_name})
return result
except:
parent = client.collection_path(project_id, location, "default_collection")
operation = client.create_data_store(
request={
"parent": parent,
"data_store": discoveryengine.DataStore(
display_name=display_name,
acl_enabled=True,
industry_vertical=discoveryengine.IndustryVertical.GENERIC,
identity_mapping_store=identity_mapping_store,
),
"data_store_id": data_store_id,
}
)
result = operation.result()
return result
La función get_or_create_data_store usa las siguientes variables clave:
project_id: El ID de tu Google Cloud proyecto.location: La Google Cloud ubicación del almacén de datos.display_name: El nombre visible legible por humanos para el almacén de datos.data_store_id: Un identificador único para el almacén de datos.identity_mapping_store: El nombre del recurso del almacén de asignación de identidades que se vinculará.result: Valor de retorno del tipo discoveryengine.DataStore.
Sube documentos intercalados
Para enviar documentos directamente a Discovery Engine, usa la carga intercalada. Este método usa el modo de conciliación incremental de forma predeterminada y no admite el modo de conciliación completa. En el modo incremental, se agregan documentos nuevos y se actualizan los existentes, pero no se borran los documentos que ya no están en la fuente. El modo de conciliación completa sincroniza el almacén de datos con tus datos de origen, lo que incluye borrar los documentos que ya no están presentes en la fuente.
La conciliación incremental es ideal para sistemas como un CRM que manejan cambios frecuentes y pequeños en los datos. En lugar de sincronizar toda la base de datos, solo envía cambios específicos, lo que hace que el proceso sea más rápido y eficiente.
Como práctica recomendada, realiza una sincronización completa inicial, seguida de sincronizaciones incrementales más frecuentes.
Python
from typing import List
from google.cloud import discoveryengine_v1 as discoveryengine
def upload_documents_inline(
project_id: str,
location: str,
data_store_id: str,
branch_id: str,
documents: List[discoveryengine.Document],
) -> discoveryengine.ImportDocumentsMetadata:
"""Inline import of Document messages."""
client = discoveryengine.DocumentServiceClient()
parent = client.branch_path(
project=project_id,
location=location,
data_store=data_store_id,
branch=branch_id,
)
request = discoveryengine.ImportDocumentsRequest(
parent=parent,
inline_source=discoveryengine.ImportDocumentsRequest.InlineSource(
documents=documents,
),
)
operation = client.import_documents(request=request)
operation.result()
result = operation.metadata
return result
La función upload_documents_inline usa las siguientes variables clave:
project_id: El ID de tu Google Cloud proyecto.location: La Google Cloud ubicación del almacén de datos.data_store_id: El ID del almacén de datos.branch_id: El ID de la rama dentro del almacén de datos (por lo general, "0").documents: Una lista de objetos discoveryengine.Document que se subirán.result: Valor de retorno del tipo discoveryengine.ImportDocumentsMetadata.
El campo uri dentro del objeto discoveryengine.Document se usa para apuntar a la fuente de contenido transferida, ya sea como bytes sin procesar o como un URI en Google Cloud Storage. Esto es diferente del URI del contenido de origen de terceros. El URI del contenido de origen de terceros debe definirse como un campo dentro de la carga útil json_data del documento. Por ejemplo, en la función convert_posts_to_documents, el campo url de la carga útil cumple este propósito.
Valida tu conector
Para validar que tu conector funcione como se espera, realiza una prueba para garantizar el flujo de datos adecuado desde la fuente a Discovery Engine.
Python
from google.cloud import discoveryengine_v1 as discoveryengine
from google.api_core import exceptions as gcp_exceptions
SITE = "https://altostrat.com"
PROJECT_ID = "ucs-3p-connectors-testing"
LOCATION = "global"
IDENTITY_MAPPING_STORE_ID = "your-unique-ims-id17" # A unique ID for your new store
DATA_STORE_ID = "my-acl-ds-id1"
BRANCH_ID = "0"
posts = fetch_posts(SITE)
docs = convert_posts_to_documents(posts)
print(f"Fetched {len(posts)} posts and converted to {len(docs)} documents.")
try:
# Step #1: Retrieve an existing identity mapping store or create a new identity mapping store
ims_store = get_or_create_ims_data_store(PROJECT_ID, LOCATION, IDENTITY_MAPPING_STORE_ID)
print(f"STEP #1: IMS Store Retrieval/Creation: {ims_store}")
RAW_IDENTITY_MAPPING_DATA = [
discoveryengine.IdentityMappingEntry(
external_identity="external_id_1",
user_id="testuser1@example.com",
),
discoveryengine.IdentityMappingEntry(
external_identity="external_id_2",
user_id="testuser2@example.com",
),
discoveryengine.IdentityMappingEntry(
external_identity="external_id_2",
group_id="testgroup1@example.com",
)
]
# Step #2: Load IMS Data
response = load_ims_data(ims_store, RAW_IDENTITY_MAPPING_DATA)
print(
"\nStep #2: Load Data in IMS Store successful.", response
)
# Step #3: Create Entity Data Store & Bind IMS Data Store
data_store = get_or_create_data_store(PROJECT_ID, LOCATION, "my-acl-datastore", DATA_STORE_ID, ims_store.name)
print("\nStep #3: Entity Data Store Create Result: ", data_store)
metadata = upload_documents_inline(
PROJECT_ID, LOCATION, DATA_STORE_ID, BRANCH_ID, docs
)
print(f"Uploaded {metadata.success_count} documents inline.")
except gcp_exceptions.GoogleAPICallError as e:
print(f"\n--- API Call Failed ---")
print(f"Server Error Message: {e.message}")
print(f"Status Code: {e.code}")
except Exception as e:
print(f"An error occurred: {e}")
Valida que el código de tu conector use las siguientes variables clave:
SITE: La URL base de la fuente de datos de terceros.PROJECT_ID: Es el ID del Google Cloud proyecto de.LOCATION: La Google Cloud ubicación de los recursos.IDENTITY_MAPPING_STORE_ID: Un ID único para tu almacén de asignación de identidades.DATA_STORE_ID: Un ID único para tu almacén de datos.BRANCH_ID: El ID de la rama dentro del almacén de datos.posts: Almacena las publicaciones obtenidas de la fuente de terceros.docs: Almacena los documentos convertidos en formato discoveryengine.Document.ims_store: El objeto discoveryengine.DataStore recuperado o creado para la asignación de identidades.RAW_IDENTITY_MAPPING_DATA: Una lista de objetos discoveryengine.IdentityMappingEntry.
Resultado esperado:
Shell
Fetched 20 posts and converted to 20 documents.
STEP #1: IMS Store Retrieval/Creation: "projects/ <Project Number>/locations/global/identityMappingStores/your-unique-ims-id17"
Step #2: Load Data in IMS Store successful.
Step #3: Entity Data Store Create Result: "projects/ <Project Number>/locations/global/collections/default_collection/dataStores/my-acl-ds-id1"
display_name: "my-acl-datastore"
industry_vertical: GENERIC
create_time {
seconds: 1760906997
nanos: 192641000
}
default_schema_id: "default_schema"
acl_enabled: true
identity_mapping_store: "projects/ <Project Number>/locations/global/identityMappingStores/your-unique-ims-id17".
Uploaded 20 documents inline.
También puedes ver tu almacén de datos en la Google Google Cloud consola en este punto:
Cuando se realiza una búsqueda, se usan los campos de la carga útil para la búsqueda y el campo URI de la carga útil para las citas. Los tres campos de propiedad clave que identifica Gemini Enterprise son title, description y uri. Los campos de documento correspondientes pueden tener nombres diferentes, y su asignación se puede realizar con la opción Esquema -> Editar en la Google Cloud consola.
Crea un conector con la carga de Google Cloud Storage
Si bien la importación intercalada funciona bien para el desarrollo, los conectores de producción deben usar Google Cloud Storage para mejorar la escalabilidad y habilitar el modo de conciliación completa. Este enfoque maneja conjuntos de datos grandes de manera eficiente y admite el borrado automático de documentos que ya no están presentes en la fuente de datos de terceros.
Convierte documentos a JSONL
Para preparar documentos para la importación masiva a Discovery Engine, conviértelos al formato de líneas JSON.
Python
from typing import List
from google.cloud import discoveryengine_v1 as discoveryengine
def convert_documents_to_jsonl(
documents: List[discoveryengine.Document],
) -> str:
"""Serialize Document messages to JSONL."""
return "\n".join(
discoveryengine.Document.to_json(doc, indent=None)
for doc in documents
) + "\n"
La función convert_documents_to_jsonl usa la siguiente variable:
documents: Una lista de objetos discoveryengine.Document que se convertirán.
Sube a Google Cloud Storage
Para habilitar la importación masiva eficiente, organiza tus datos en Google Cloud Storage.
Python
from google.cloud import storage
def upload_jsonl_to_gcs(jsonl: str, bucket_name: str, blob_name: str) -> str:
"""Upload JSONL content to Google Cloud Storage."""
client = storage.Client()
bucket = client.bucket(bucket_name)
blob = bucket.blob(blob_name)
blob.upload_from_string(jsonl, content_type="application/json")
return f"gs://{bucket_name}/{blob_name}"
La función upload_jsonl_to_gcs usa las siguientes variables clave:
jsonl: El contenido de la cadena con formato JSONL que se subirá.bucket_name: El nombre del bucket de Google Cloud Storage.blob_name: El nombre del blob (objeto) dentro del bucket especificado.
Importa desde Google Cloud Storage con conciliación completa
Para realizar una sincronización de datos completa con el modo de conciliación completa, usa este método. Esto garantiza que tu almacén de datos refleje exactamente la fuente de datos de terceros, y quita automáticamente los documentos que ya no existen.
Cuando importes desde Google Cloud Storage, ten en cuenta las siguientes limitaciones:
- Una sola solicitud de importación puede contener hasta 100 archivos o 100,000 archivos si el parámetro
dataSchemase establece encontent. Cada archivo puede tener hasta 2 GB o 100 MB si el parámetro
dataSchemase establece encontent.
Python
from google.cloud import discoveryengine_v1 as discoveryengine
def import_documents_from_gcs(
project_id: str,
location: str,
data_store_id: str,
branch_id: str,
gcs_uri: str,
) -> discoveryengine.ImportDocumentsMetadata:
"""Bulk-import documents from Google Cloud Storage with FULL reconciliation mode."""
client = discoveryengine.DocumentServiceClient()
parent = client.branch_path(
project=project_id,
location=location,
data_store=data_store_id,
branch=branch_id,
)
gcs_source = discoveryengine.GcsSource(input_uris=[gcs_uri])
request = discoveryengine.ImportDocumentsRequest(
parent=parent,
gcs_source=gcs_source,
reconciliation_mode=
discoveryengine.ImportDocumentsRequest
.ReconciliationMode.FULL,
)
operation = client.import_documents(request=request)
operation.result()
return operation.metadata
La función import_documents_from_gcs usa las siguientes variables clave:
project_id: El ID de tu Google Cloud proyecto.location: La Google Cloud ubicación del almacén de datos.data_store_id: El ID del almacén de datos.branch_id: El ID de la rama dentro del almacén de datos (por lo general, "0").gcs_uri: El Google Cloud URI deStorage que apunta al archivo JSONL.
Prueba la carga de Google Cloud Storage
Para verificar el flujo de trabajo de importación basado en Google Cloud Storage, ejecuta lo siguiente:
Python
BUCKET = "your-existing-bucket"
BLOB = "path-to-any-blob/wp/posts.jsonl"
SITE = "https://altostrat.com"
PROJECT_ID = "ucs-3p-connectors-testing"
LOCATION = "global"
IDENTITY_MAPPING_STORE_ID = "your-unique-ims-id17" # A unique ID for your new store
DATA_STORE_ID = "your-data-store-id"
BRANCH_ID = "0"
posts = fetch_posts(SITE)
docs = convert_posts_to_documents(posts)
print(f"Fetched {len(posts)} posts and converted to {len(docs)} documents.")
jsonl_payload = convert_documents_to_jsonl(docs)
gcs_uri = upload_jsonl_to_gcs(jsonl_payload, BUCKET, BLOB)
print("Uploaded to:", gcs_uri)
metadata = import_documents_from_gcs(
PROJECT_ID, LOCATION, DATA_STORE_ID, BRANCH_ID, gcs_uri
)
print(f"Imported: {metadata.success_count} documents")
Las siguientes variables clave se usan para probar la carga de Google Cloud Storage:
BUCKET: El nombre del Google Cloud bucket deStorage.BLOB: La ruta de acceso al blob dentro del bucket.SITE: La URL base de la fuente de datos de terceros.PROJECT_ID: Es el ID del Google Cloud proyecto de.LOCATION: La Google Cloud ubicación de los recursos (p.ej., "global").IDENTITY_MAPPING_STORE_ID: Un ID único para tu almacén de asignación de identidades.DATA_STORE_ID: Un ID único para tu almacén de datos.BRANCH_ID: El ID de la rama dentro del almacén de datos (por lo general, "0").jsonl_payload: Los documentos convertidos al formato JSONL.gcs_uri: El Google Cloud URI deStorage del archivo JSONL subido.
Resultado esperado:
Shell
Fetched 20 posts and converted to 20 documents.
Uploaded to: gs://alex-de-bucket/wp/posts.jsonl
Imported: 20 documents
Administra permisos
Para administrar el acceso a nivel de documento en entornos empresariales, Gemini Enterprise admite listas de control de acceso (ACL) y asignación de identidades, lo que ayuda a limitar el contenido que pueden ver los usuarios.
Habilita las ACL en el almacén de datos
Para habilitar las ACL cuando crees tu almacén de datos, ejecuta lo siguiente:
Python
# get_or_create_data_store()
"data_store": discoveryengine.DataStore(
display_name=data_store_id,
industry_vertical=discoveryengine.IndustryVertical.GENERIC,
acl_enabled=True, # ADDED
)
Agrega ACL a los documentos
Para calcular e incluir AclInfo cuando transformes los documentos, ejecuta lo siguiente:
Python
# convert_posts_to_documents()
doc = discoveryengine.Document(
id=str(post["id"]),
json_data=json.dumps(payload),
acl_info=discoveryengine.Document.AclInfo(
readers=[{
"principals": [
{"user_id": "baklavainthebalkans@gmail.com"},
{"user_id": "cloudysanfrancisco@gmail.com"}
]
}]
),
)
Haz público el contenido
Para que un documento sea de acceso público, configura el campo readers de la siguiente manera:
Python
readers=[{"idp_wide": True}]
Valida las ACL
Para validar que tus configuraciones de ACL funcionen como se espera, considera lo siguiente:
Realiza una búsqueda como un usuario que no tiene acceso al documento.
Inspecciona la estructura del documento subido en Cloud Storage y compárala con una referencia.
JSON
{
"id": "108",
"jsonData": "{...}",
"aclInfo": {
"readers": [
{
"principals": [
{ "userId": "baklavainthebalkans@gmail.com" },
{ "userId": "cloudysanfrancisco@gmail.com" }
],
"idpWide": false
}
]
}
}
Usa la asignación de identidades
Usa la asignación de identidades en los siguientes casos:
Tu fuente de datos de terceros usa identidades que no son de Google.
Quieres hacer referencia a grupos personalizados (p. ej., wp-admins) en lugar de usuarios individuales.
Tu API solo muestra nombres de grupos.
Necesitas agrupar usuarios de forma manual para lograr coherencia o escalabilidad.
Tu sistema de terceros implementa ACL con grupos que no se basan en IDP y quieres respetar esas ACL con el almacén de datos personalizado en Gemini Enterprise.
Para asignar identidades, sigue estos pasos:
- Crea y vincula el almacén de datos de identidad.
Importa identidades externas (por ejemplo,
external_group:wp-admins). No incluyas elexternal_group: prefixcuando importes, por ejemplo:JSON
{ "externalIdentity": "wp-admins", "userId": "user@example.com" }En la información de ACL de tu documento, define el ID de entidad externa en el
principal identifier. Cuando hagas referencia a grupos personalizados, usa el prefijoexternal_group:en el campogroupId.El prefijo
external_group:es obligatorio para los IDs de grupo dentro de la información de ACL del documento durante la importación, pero no se usa cuando se importan identidades al almacén de asignación. Ejemplo de documento con asignación de identidades:JSON
{ "id": "108", "aclInfo": { "readers": [ { "principals": [ { "userId": "cloudysanfrancisco@gmail.com" }, { "groupId": "external_group:wp-admins" } ] } ] }, "structData": { "id": 108, "date": "2025-04-24T18:16:04", ... } }
¿Qué sigue?
- Para proporcionar una interfaz de usuario para consultar tus datos, crea una app en Gemini Enterprise y conéctala al almacén de datos Conector personalizado existente.