Cette page explique comment créer un connecteur personnalisé.
Avant de commencer
Avant de commencer, assurez-vous de disposer des éléments suivants :
Vérifiez si la facturation est activée sur votre Google Cloud projet.
Installez et initialisez la Google Cloud CLI. Assurez-vous qu'elle est authentifiée pour votre projet.
Obtenez un accès administrateur Discovery Engine pour votre Google Cloud projet.
Obtenez des identifiants d'accès pour votre source de données tierce (tels que des clés API ou une authentification de base de données).
Créez un plan de mappage de données clair. Il doit inclure les champs à indexer et la manière de représenter le contrôle des accès, y compris les identités tierces.
Créer un connecteur de base
Cette section explique comment créer un connecteur personnalisé dans la langue de votre choix. Les principes et les modèles présentés ici s'appliquent à n'importe quel système externe. Il vous suffit d'adapter les appels d'API et les transformations de données pour votre source spécifique dans la langue de votre choix afin de créer un connecteur de base.
Extraction de données
Pour commencer, récupérez les données de votre source de données tierce. Dans cet exemple, nous allons vous montrer comment extraire des posts à l'aide de la pagination. Pour les environnements de production, nous vous recommandons d'utiliser une approche de streaming pour les ensembles de données volumineux. Cela évite les problèmes de mémoire qui peuvent survenir lors du chargement de toutes les données en même temps.
Récupération
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},
)
resp.raise_for_status()
batch = resp.json()
posts.extend(batch)
if len(batch) < per_page:
break
page += 1
return posts
Transformer les données
Pour convertir vos données sources au format de document Discovery Engine, structurez-les comme indiqué dans l'exemple de charge utile suivant. Vous pouvez inclure autant de paires clé/valeur que nécessaire. Par exemple, vous pouvez inclure le contenu complet pour une recherche exhaustive. Vous pouvez également inclure des champs structurés pour une recherche à facettes ou une combinaison des deux.
Récupération
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
Récupérer ou créer un magasin d'identités
Pour gérer les identités et les groupes d'utilisateurs pour le contrôle des accès, vous devez récupérer ou créer un magasin d'identités. Cette fonction obtient un magasin d'identités existant par son ID, son projet et son emplacement. Si le magasin d'identités n'existe pas, il en crée un et renvoie un nouveau magasin d'identités vide.
Récupération
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 fonction get_or_create_ims_data_store utilise les variables clés suivantes :
project_id: ID de votre Google Cloud projet.location: emplacement du magasin de mappage d'identité. Google Cloudidentity_mapping_store_id: identifiant unique du magasin d'identités.client_ims: instance de discoveryengine.IdentityMappingStoreServiceClient utilisée pour interagir avec l'API du magasin d'identités.parent_ims: nom de ressource de l'emplacement parent, construit à l'aide de client_ims.location_path.name: nom complet de la ressource du magasin de mappage d'identité, utilisé pour GetIdentityMappingStoreRequest.
Ingérer le mappage d'identité dans le magasin d'identités
Pour charger des entrées de mappage d'identité dans le magasin d'identités spécifié, utilisez cette fonction. Elle prend une liste d'entrées de mappage d'identité et lance une opération d'importation en ligne. Cela est essentiel pour établir les relations entre les utilisateurs, les groupes et les identités externes nécessaires au contrôle des accès et à la personnalisation.
Récupération
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}")
result = operation.result()
return result
La fonction load_ims_data utilise les variables clés suivantes :
ims_store: objet discoveryengine.DataStore représentant le magasin de mappage d'identité dans lequel les données seront chargées.id_mapping_data: liste d'objets discoveryengine.IdentityMappingEntry, chacun contenant une identité externe et l'ID d'utilisateur ou de groupe correspondant.result: valeur renvoyée de type discoveryengine.DataStore.
Créer data store
Pour utiliser un connecteur personnalisé, vous devez initialiser un data store pour votre contenu. Utilisez default_collection pour les connecteurs personnalisés. Le paramètre IndustryVertical personnalise le comportement du data store pour des cas d'utilisation spécifiques. GENERIC convient à la plupart des scénarios. Toutefois, vous pouvez choisir une autre valeur pour un secteur d'activité spécifique, tel que MEDIA ou HEALTHCARE_FHIR. Configurez le nom à afficher et les autres propriétés pour qu'ils correspondent aux conventions et exigences de nommage de votre projet.
Récupération
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 fonction get_or_create_data_store utilise les variables clés suivantes :
project_id: ID de votre Google Cloud projet.location: emplacement du data storedisplay_name: nom à afficher lisible par l'utilisateur pour le data store.data_store_id: identifiant unique du data store.identity_mapping_store: nom de ressource du magasin de mappage d'identité à lier.result: valeur renvoyée de type discoveryengine.DataStore.
Importer des documents en ligne
Pour envoyer directement des documents à Discovery Engine, utilisez l'importation en ligne. Cette méthode utilise le mode de rapprochement incrémentiel par défaut et n'est pas compatible avec le mode de rapprochement complet. En mode incrémentiel, les nouveaux documents sont ajoutés et les documents existants sont mis à jour, mais les documents qui ne sont plus dans la source ne sont pas supprimés. Le mode de rapprochement complet synchronise le data store avec vos données sources, y compris en supprimant les documents qui ne sont plus présents dans la source.
Le rapprochement incrémentiel est idéal pour les systèmes tels qu'un CRM qui gèrent des modifications fréquentes et mineures des données. Au lieu de synchroniser l'ensemble de la base de données, n'envoyez que des modifications spécifiques, ce qui rend le processus plus rapide et plus efficace.
Nous vous recommandons d'effectuer une synchronisation complète initiale, suivie de synchronisations incrémentielles plus fréquentes.
Récupération
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 fonction upload_documents_inline utilise les variables clés suivantes :
project_id: ID de votre Google Cloud projet.location: emplacement du data storedata_store_id: ID du data store.branch_id: ID de la branche dans le data store (généralement "0").documents: liste d'objets discoveryengine.Document à importer.result: valeur renvoyée de type discoveryengine.ImportDocumentsMetadata.
Le champ uri de l'objet discoveryengine.Document est utilisé pour pointer vers la source du contenu ingérée elle-même, soit sous forme d'octets bruts, soit sous forme d'URI dans Google Cloud Storage. Il est différent de l'URI du contenu source tiers. L'URI du contenu source tiers doit être défini comme un champ dans la charge utile json_data du document. Par exemple, dans la fonction convert_posts_to_documents, le champ url de la charge utile remplit cet objectif.
Valider votre connecteur
Pour vérifier que votre connecteur fonctionne comme prévu, effectuez un test pour vous assurer que le flux de données circule correctement de la source vers Discovery Engine.
Récupération
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}")
Validez que le code de votre connecteur utilise les variables clés suivantes :
SITE: URL de base de la source de données tierce.PROJECT_ID: ID de votre Google Cloud projet.LOCATION: emplacement des ressources. Google CloudIDENTITY_MAPPING_STORE_ID: ID unique de votre magasin de mappage d'identité.DATA_STORE_ID: ID unique de votre data store.BRANCH_ID: ID de la branche dans le data store.posts: stocke les posts extraits de la source tierce.docs: stocke les documents convertis au format discoveryengine.Document.ims_store: objet discoveryengine.DataStore récupéré ou créé pour le mappage d'identité.RAW_IDENTITY_MAPPING_DATA: liste d'objets discoveryengine.IdentityMappingEntry.
Résultat attendu :
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.
Vous pouvez également voir votre data store dans laconsole Google Google Cloud à ce stade :
Lorsqu'une recherche est effectuée, les champs de la charge utile sont utilisés pour la recherche, et le champ URI de la charge utile est utilisé pour les citations. Les trois champs de propriété clés identifiés par Gemini Enterprise sont title, description et uri. Les champs de document correspondants peuvent avoir des noms différents, et leur mappage peut être effectué à l'aide de l'option Schéma -> Modifier dans la Google Cloud console.
Créer un connecteur avec Google Cloud importationStorage
Bien que l'importation en ligne fonctionne bien pour le développement, les connecteurs de production doivent utiliser Google Cloud Storage pour une meilleure évolutivité et pour activer le mode de rapprochement complet. Cette approche gère efficacement les ensembles de données volumineux et permet la suppression automatique des documents qui ne sont plus présents dans la source de données tierce.
Convertir des documents au format JSONL
Pour préparer les documents à l'importation groupée dans Discovery Engine, convertissez-les au format JSON Lines.
Récupération
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 fonction convert_documents_to_jsonl utilise la variable suivante :
documents: liste d'objets discoveryengine.Document à convertir.
Importer dans Google Cloud Storage
Pour activer une importation groupée efficace, préparez vos données dans Google Cloud Storage.
Récupération
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 fonction upload_jsonl_to_gcs utilise les variables clés suivantes :
jsonl: contenu de la chaîne au format JSONL à importer.bucket_name: nom du Google Cloud bucketStorage.blob_name: nom de l'objet blob dans le bucket spécifié.
Importer depuis Google Cloud Storage avec rapprochement complet
Pour effectuer une synchronisation complète des données à l'aide du mode de rapprochement complet, utilisez cette méthode. Cela garantit que votre data store reflète exactement la source de données tierce, en supprimant automatiquement tous les documents qui n'existent plus.
Lorsque vous importez des données depuis Google Cloud Storage, tenez compte des limites suivantes :
- Une seule requête d'importation peut contenir au maximum 100 fichiers, ou 100 000 fichiers si le paramètre
dataSchemaest défini surcontent. Chaque fichier peut atteindre 2 Go, ou 100 Mo si le paramètre
dataSchemaest défini surcontent.
Récupération
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 fonction import_documents_from_gcs utilise les variables clés suivantes :
project_id: ID de votre Google Cloud projet.location: emplacement du data storedata_store_id: ID du data store.branch_id: ID de la branche dans le data store (généralement "0").gcs_uri: URIStorage pointant vers le fichier JSONL. Google Cloud
Tester l'importation Google Cloud Storage
Pour vérifier le Google Cloud workflow d'importation basé surStorage, exécutez les éléments suivants :
Récupération
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")
Les variables clés suivantes sont utilisées pour tester l'importation Google Cloud Storage :
BUCKET: nom du bucketStorage. Google CloudBLOB: chemin d'accès au blob dans le bucket.SITE: URL de base de la source de données tierce.PROJECT_ID: ID de votre Google Cloud projet.LOCATION: emplacement des ressources (par exemple, "global"). Google CloudIDENTITY_MAPPING_STORE_ID: ID unique de votre magasin de mappage d'identité.DATA_STORE_ID: ID unique de votre data store.BRANCH_ID: ID de la branche dans le data store (généralement "0").jsonl_payload: documents convertis au format JSONL.gcs_uri: URIStorage du fichier JSONL importé. Google Cloud
Résultat attendu :
Shell
Fetched 20 posts and converted to 20 documents.
Uploaded to: gs://alex-de-bucket/wp/posts.jsonl
Imported: 20 documents
Gérer les autorisations
Pour gérer l'accès au niveau des documents dans les environnements d'entreprise, Gemini Enterprise est compatible avec les listes de contrôle d'accès (LCA) et le mappage d'identité, ce qui permet de limiter le contenu que les utilisateurs peuvent voir.
Activer les LCA dans le data store
Pour activer les LCA lors de la création de votre data store, exécutez les éléments suivants :
Récupération
# get_or_create_data_store()
"data_store": discoveryengine.DataStore(
display_name=data_store_id,
industry_vertical=discoveryengine.IndustryVertical.GENERIC,
acl_enabled=True, # ADDED
)
Ajouter des LCA aux documents
Pour calculer et inclure AclInfo lors de la transformation des documents, exécutez les éléments suivants :
Récupération
# 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"}
]
}]
),
)
Rendre le contenu public
Pour rendre un document accessible au public, définissez le champ readers comme suit :
Récupération
readers=[{"idp_wide": True}]
Valider les LCA
Pour vérifier que vos configurations de LCA fonctionnent comme prévu, tenez compte des points suivants :
Effectuez une recherche en tant qu'utilisateur qui n'a pas accès au document.
Inspectez la structure du document importé dans Cloud Storage et comparez-la à une référence.
JSON
{
"id": "108",
"jsonData": "{...}",
"aclInfo": {
"readers": [
{
"principals": [
{ "userId": "baklavainthebalkans@gmail.com" },
{ "userId": "cloudysanfrancisco@gmail.com" }
],
"idpWide": false
}
]
}
}
Utiliser un mappage d'identité
Utilisez le mappage d'identité dans les scénarios suivants :
Votre source de données tierce utilise des identités non Google.
Vous souhaitez référencer des groupes personnalisés (par exemple, wp-admins) au lieu d'utilisateurs individuels.
Votre API ne renvoie que des noms de groupes.
Vous devez regrouper manuellement les utilisateurs pour des raisons d'évolutivité ou de cohérence.
Votre système tiers implémente des LCA à l'aide de groupes non basés sur un fournisseur d'identité, et vous souhaitez respecter ces LCA à l'aide du data store personnalisé dans Gemini Enterprise.
Pour mapper des identités, procédez comme suit :
- Créez et liez le data store d'identité.
Importez des identités externes (par exemple,
external_group:wp-admins). N'incluez pas leexternal_group: prefixlors de l'importation, par exemple :JSON
{ "externalIdentity": "wp-admins", "userId": "user@example.com" }Dans les informations de LCA de votre document, définissez l'ID d'entité externe dans le
principal identifier. Lorsque vous référencez des groupes personnalisés, utilisez le préfixeexternal_group:dans le champgroupId.Le préfixe
external_group:est obligatoire pour les ID de groupe dans les informations de LCA du document lors de l'importation, mais il n'est pas utilisé lors de l'importation d'identités dans le magasin de mappage. Exemple de document avec mappage d'identité :JSON
{ "id": "108", "aclInfo": { "readers": [ { "principals": [ { "userId": "cloudysanfrancisco@gmail.com" }, { "groupId": "external_group:wp-admins" } ] } ] }, "structData": { "id": 108, "date": "2025-04-24T18:16:04", ... } }
Étape suivante
- Pour fournir une interface utilisateur permettant d'interroger vos données, créez une application dans Gemini Enterprise et associez-la au datastore Connecteur personnalisé existant.