L'API IngestEvents dissocie l'ingestion d'événements de la génération de la mémoire dans Memory Bank. Vous pouvez diffuser du contenu en continu vers Memory Bank. La génération s'exécute automatiquement lorsque vos critères de déclenchement sont remplis.
Pour suivre les étapes décrites dans ce guide, vous devez d'abord suivre les étapes de la section Configurer Memory Bank.
Les exemples de cette page utilisent le SDK agentplatform, qui fournit les dernières options de configuration IngestEvents. Initialisez le client avec agentplatform :
import agentplatform
client = agentplatform.Client(
project="PROJECT_ID",
location="LOCATION",
)
Comprendre l'ingestion d'événements
Avec l'API GenerateMemories, la génération de mémoire est déclenchée immédiatement. Cette méthode vous permet de contrôler directement le moment où la génération de mémoire a lieu. IngestEvents offre un modèle de streaming qui dissocie l'ingestion d'événements de la génération de la mémoire :
- Ingestion continue : vous pouvez diffuser du contexte en continu vers Memory Bank chaque fois que de nouvelles données sont disponibles.
- Déduplication : Memory Bank détecte et ignore automatiquement l'historique des conversations qui se chevauchent en fonction des ID d'événement partagés.
- Déclencheurs automatisés : la génération n'est déclenchée automatiquement que lorsque des conditions spécifiques sont remplies, comme des seuils de nombre d'événements ou un temps d'inactivité.
- Génération asynchrone : les événements sont traités en arrière-plan. L'API renvoie immédiatement une opération de longue durée (LRO), que vous pouvez interroger ou surveiller jusqu'à ce que la condition de déclenchement soit remplie et que les mémoires soient générées.
Flux et opérations de longue durée (LRO)
Les flux agissent comme des tampons isolés qui accumulent des événements sur plusieurs requêtes. Chaque flux est défini de manière unique par la combinaison de son scope et de son stream_id. Si vous ne fournissez pas d'ID de flux, Memory Bank utilise default comme ID. Les événements ne sont mis en scène et "vidés" ensemble que s'ils partagent exactement le même champ d'application et le même ID, ce qui garantit une isolation stricte lors de la génération de la mémoire.
Plusieurs appels d'ingestion vers le même flux renvoient la même LRO jusqu'à ce que la génération de mémoire soit déclenchée et terminée. Une fois l'opération terminée, cette LRO IngestEvents contient la LRO en aval pour l'appel d'API GenerateMemories. Toute requête d'ingestion ultérieure vers le flux renverra une nouvelle LRO.
Déclencher la génération de la mémoire
Si aucune condition de déclenchement n'est remplie, tous les événements en attente dans un flux sont automatiquement vidés 24 heures après l'ingestion du dernier événement. Cette limite garantit que Memory Bank traite tous les événements, y compris ceux qui ont été abandonnés.
Le generation_trigger_config détermine le moment où les événements collectés sont vidés et traités pour la génération de la mémoire. Vous pouvez configurer les conditions de déclenchement suivantes :
- Nombre d'événements (
event_count) : la génération est déclenchée une fois que le nombre d'événements uniques accumulés atteint la limite. - Temps d'inactivité (
idle_duration, en secondes avec une granularité à la minute, ce qui signifie que vous devez spécifier un multiple de 60) : la génération est déclenchée lorsque le flux n'a pas reçu de nouveaux événements après la durée spécifiée (par exemple,"300s"). - Intervalle fixe (
fixed_interval, en secondes avec une granularité à la minute, ce qui signifie que vous devez spécifier un multiple de 60) : la génération s'exécute à un rythme fixe, en interrogeant les événements delta après chaque durée spécifiée (par exemple,"300s"). - Forcer la vidange (
force_flush) : videz immédiatement tous les événements en attente.
Vous pouvez combiner une règle de nombre d'événements avec une règle basée sur le temps dans le même generation_rule. Par exemple, vous pouvez définir event_count avec idle_duration. Lorsque plusieurs conditions sont configurées, la génération est déclenchée dès que l'une d'elles est remplie, quelle que soit la première. Les deux conditions basées sur le temps s'excluent mutuellement : une seule règle peut spécifier idle_duration ou fixed_interval, mais pas les deux. Le champ force_flush est indépendant de ces règles et vide toujours immédiatement les événements en attente.
L'API utilise event_id pour dédupliquer automatiquement les données. En fournissant un ID unique pour chaque événement, seuls les nouveaux événements sont ajoutés au tampon.
Transférer le contexte entre les fenêtres de génération
Par défaut, chaque fenêtre de génération ne traite que les événements accumulés depuis la vidange précédente. Pour que les mémoires restent cohérentes entre les fenêtres de génération consécutives, vous pouvez transférer le contexte d'une fenêtre à l'autre en définissant overlap_event_count dans le generation_rule. Cela inclut à nouveau les N derniers événements déjà traités au début de la fenêtre suivante.
Les événements transférés ne sont pas pris en compte dans la condition de déclenchement. La valeur overlap_event_count doit être un entier positif inférieur à event_count, et elle nécessite event_count ou une règle basée sur le temps (idle_duration ou fixed_interval).
Source de données
Avec IngestEvents, vous fournissez du contenu en envoyant des événements directement dans votre requête d'application à l'aide de direct_contents_source dans une charge utile directe.
Les exemples suivants présentent différents déclencheurs.
Exemple : Déclencheur de nombre d'événements
Cet exemple utilise un flux personnalisé et déclenche la génération lorsque 10 événements uniques ont été accumulés dans le flux.
Dictionnaire
client.memory_banks.ingest_events(
name=memory_bank.name,
stream_id="my-custom-stream",
direct_contents_source={
"events": [
{
"content": {
"role": "user",
"parts": [{"text": "My favorite color is blue."}]
},
"event_id": "event-1" # Optional, used for deduplication.
}
]
},
generation_trigger_config={
"generation_rule": {
"event_count": 10
}
},
scope={"user_id": "123"}
)
Basé sur une classe
from google import genai
from agentplatform import types
client.memory_banks.ingest_events(
name=memory_bank.name,
stream_id="my-custom-stream",
direct_contents_source=types.IngestionDirectContentsSource(
events=[
types.IngestionDirectContentsSourceEvent(
content=genai.types.Content(
role="user",
parts=[genai.types.Part.from_text(text="My favorite color is blue.")]
),
event_id="event-1" # Optional, used for deduplication.
)
]
),
generation_trigger_config=types.MemoryGenerationTriggerConfig(
generation_rule=types.MemoryGenerationTriggerConfigGenerationTriggerRule(
event_count=10
)
),
scope={"user_id": "123"}
)
Exemple : Déclencheur de durée d'inactivité basé sur le temps
Cet exemple déclenche la génération lorsque le flux n'a pas reçu de nouvel événement pendant cinq minutes (300s). Il définit également un stream_id personnalisé.
Dictionnaire
client.memory_banks.ingest_events(
name=memory_bank.name,
stream_id="my-custom-stream",
direct_contents_source={
"events": [
{
"content": {
"role": "user",
"parts": [{"text": "I will be back in a bit."}]
},
"event_id": "event-2" # Optional, used for deduplication.
}
]
},
generation_trigger_config={
"generation_rule": {
"idle_duration": "300s"
}
},
scope={"user_id": "123"}
)
Basé sur une classe
from google import genai
from agentplatform import types
client.memory_banks.ingest_events(
name=memory_bank.name,
stream_id="my-custom-stream",
direct_contents_source=types.IngestionDirectContentsSource(
events=[
types.IngestionDirectContentsSourceEvent(
content=genai.types.Content(
role="user",
parts=[genai.types.Part.from_text(text="I will be back in a bit.")]
),
event_id="event-2" # Optional, used for deduplication.
)
]
),
generation_trigger_config=types.MemoryGenerationTriggerConfig(
generation_rule=types.MemoryGenerationTriggerConfigGenerationTriggerRule(
idle_duration="300s"
)
),
scope={"user_id": "123"}
)
Exemple : Combiner des déclencheurs basés sur des événements et sur le temps
Cet exemple configure deux conditions de déclenchement dans la même requête. La génération s'exécute lorsque 10 événements uniques sont accumulés ou que le flux est inactif pendant cinq minutes (300s), selon la première occurrence.
Dictionnaire
client.memory_banks.ingest_events(
name=memory_bank.name,
stream_id="my-custom-stream",
direct_contents_source={
"events": [
{
"content": {
"role": "user",
"parts": [{"text": "Let's keep this conversation going."}]
},
"event_id": "event-3" # Optional, used for deduplication.
}
]
},
generation_trigger_config={
"generation_rule": {
"event_count": 10,
"idle_duration": "300s"
}
},
scope={"user_id": "123"}
)
Basé sur une classe
from google import genai
from agentplatform import types
client.memory_banks.ingest_events(
name=memory_bank.name,
stream_id="my-custom-stream",
direct_contents_source=types.IngestionDirectContentsSource(
events=[
types.IngestionDirectContentsSourceEvent(
content=genai.types.Content(
role="user",
parts=[genai.types.Part.from_text(text="Let's keep this conversation going.")]
),
event_id="event-3" # Optional, used for deduplication.
)
]
),
generation_trigger_config=types.MemoryGenerationTriggerConfig(
generation_rule=types.MemoryGenerationTriggerConfigGenerationTriggerRule(
event_count=10,
idle_duration="300s"
)
),
scope={"user_id": "123"}
)
Exemple : Forcer la vidange des événements
Cet exemple contourne inconditionnellement toutes les conditions de déclenchement et force immédiatement la vidange et le traitement de tous les événements en attente dans le flux.
Dictionnaire
client.memory_banks.ingest_events(
name=memory_bank.name,
stream_id="my-custom-stream",
direct_contents_source={
"events": [
{
"content": {
"role": "user",
"parts": [{"text": "That's all for today!"}]
},
}
]
},
config={
"force_flush": True
},
scope={"user_id": "123"}
)
Basé sur une classe
from google import genai
from agentplatform import types
client.memory_banks.ingest_events(
name=memory_bank.name,
stream_id="my-custom-stream",
direct_contents_source=types.IngestionDirectContentsSource(
events=[
types.IngestionDirectContentsSourceEvent(
content=genai.types.Content(
role="user",
parts=[genai.types.Part.from_text(text="That's all for today!")]
),
)
]
),
config=types.IngestEventsConfig(
force_flush=True
),
scope={"user_id": "123"}
)
Exemple : Transférer le contexte entre les fenêtres
Cet exemple déclenche la génération tous les 10 événements uniques et inclut à nouveau les deux derniers événements déjà traités dans la fenêtre suivante afin que les mémoires restent cohérentes entre les fenêtres.
Dictionnaire
client.memory_banks.ingest_events(
name=memory_bank.name,
stream_id="my-custom-stream",
direct_contents_source={
"events": [
{
"content": {
"role": "user",
"parts": [{"text": "Let's continue where we left off."}]
},
}
]
},
generation_trigger_config={
"generation_rule": {
"event_count": 10,
"overlap_event_count": 2
}
},
scope={"user_id": "123"}
)
Basé sur une classe
from google import genai
from agentplatform import types
client.memory_banks.ingest_events(
name=memory_bank.name,
stream_id="my-custom-stream",
direct_contents_source=types.IngestionDirectContentsSource(
events=[
types.IngestionDirectContentsSourceEvent(
content=genai.types.Content(
role="user",
parts=[genai.types.Part.from_text(text="Let's continue where we left off.")]
),
)
]
),
generation_trigger_config=types.MemoryGenerationTriggerConfig(
generation_rule=types.MemoryGenerationTriggerConfigGenerationTriggerRule(
event_count=10,
overlap_event_count=2
)
),
scope={"user_id": "123"}
)
Configurer les révisions de mémoire
Comme GenerateMemories, IngestEvents crée des révisions de mémoire pour les mémoires qu'il génère. Vous pouvez configurer les révisions résultant d'une requête d'ingestion en définissant les champs suivants dans le config :
- Libellés de révision (
revision_labels) : paires clé-valeur appliquées aux révisions de mémoire générées. Par exemple, vous pouvez libeller une révision avec sa source de données, puis filtrer les révisions par ce libellé. - Expiration de la révision (
revision_expire_timeourevision_ttl) : contrôle le moment où les révisions de mémoire générées expirent. Si vous ne définissez pas d'expiration, Memory Bank utilise la configuration de l'instance ou la valeur TTL (Time To Live) par défaut de 365 jours. - Désactiver les révisions (
disable_memory_revisions) : si la valeur estTrue, Memory Bank ne crée pas de révisions pour les mémoires générées à partir de cette requête.
L'exemple suivant libelle les révisions générées avec une source de données et les conserve pendant 30 jours :
Dictionnaire
client.memory_banks.ingest_events(
name=memory_bank.name,
stream_id="my-custom-stream",
direct_contents_source={
"events": [
{
"content": {
"role": "user",
"parts": [{"text": "My favorite color is blue."}]
},
}
]
},
generation_trigger_config={
"generation_rule": {
"event_count": 10
}
},
config={
"revision_labels": {"data_source": "321"},
# Persist memory revisions for 30 days after they're created.
"revision_ttl": f"{30 * 60 * 60 * 24}s"
},
scope={"user_id": "123"}
)
Basé sur une classe
from google import genai
from agentplatform import types
client.memory_banks.ingest_events(
name=memory_bank.name,
stream_id="my-custom-stream",
direct_contents_source=types.IngestionDirectContentsSource(
events=[
types.IngestionDirectContentsSourceEvent(
content=genai.types.Content(
role="user",
parts=[genai.types.Part.from_text(text="My favorite color is blue.")]
),
)
]
),
generation_trigger_config=types.MemoryGenerationTriggerConfig(
generation_rule=types.MemoryGenerationTriggerConfigGenerationTriggerRule(
event_count=10
)
),
config=types.IngestEventsConfig(
revision_labels={"data_source": "321"},
# Persist memory revisions for 30 days after they're created.
revision_ttl=f"{30 * 60 * 60 * 24}s"
),
scope={"user_id": "123"}
)
Associer des métadonnées aux mémoires générées
Vous pouvez associer des métadonnées aux mémoires générées à partir d'événements ingérés. Les métadonnées vous permettent de stocker des informations structurées (chaînes, doubles, booléens ou horodatages) avec un fait de mémoire, ce qui est utile pour filtrer ou gérer les cycles de vie de vos mémoires.
Utilisez metadata_merge_strategy pour contrôler l'interaction des métadonnées avec les mémoires existantes lors de la consolidation. Vous pouvez définir la valeur sur OVERWRITE (valeur par défaut si elle n'est pas définie), MERGE ou REQUIRE_EXACT_MATCH. Pour en savoir plus, consultez la section Utiliser des métadonnées lors de la consolidation.
Dictionnaire
client.memory_banks.ingest_events(
name=memory_bank.name,
stream_id="my-custom-stream",
direct_contents_source={
"events": [
{
"content": {
"role": "user",
"parts": [{"text": "My favorite color is blue."}]
},
}
]
},
generation_trigger_config={
"generation_rule": {
"event_count": 10
}
},
config={
"metadata": {
"my_string_key": {"string_value": "my_string_value"},
"my_boolean_key": {"bool_value": True}
},
"metadata_merge_strategy": "MERGE"
},
scope={"user_id": "123"}
)
Basé sur une classe
from google import genai
from agentplatform import types
client.memory_banks.ingest_events(
name=memory_bank.name,
stream_id="my-custom-stream",
direct_contents_source=types.IngestionDirectContentsSource(
events=[
types.IngestionDirectContentsSourceEvent(
content=genai.types.Content(
role="user",
parts=[genai.types.Part.from_text(text="My favorite color is blue.")]
),
)
]
),
generation_trigger_config=types.MemoryGenerationTriggerConfig(
generation_rule=types.MemoryGenerationTriggerConfigGenerationTriggerRule(
event_count=10
)
),
config=types.IngestEventsConfig(
metadata={
"my_string_key": types.MemoryMetadataValue(string_value="my_string_value"),
"my_boolean_key": types.MemoryMetadataValue(bool_value=True)
},
metadata_merge_strategy=types.MemoryMetadataMergeStrategy.MERGE
),
scope={"user_id": "123"}
)