Importare eventi

L'API IngestEvents separa l'importazione degli eventi dal salvataggio di informazioni in Memory Bank. Puoi trasmettere in streaming continuo i contenuti a Memory Bank e la generazione verrà eseguita automaticamente quando i criteri di attivazione sono soddisfatti.

Per completare i passaggi illustrati in questa guida, devi prima seguire i passaggi descritti in Configurazione di Memory Bank.

Gli esempi in questa pagina utilizzano l'SDK agentplatform, che fornisce le opzioni di configurazione IngestEvents più recenti. Inizializza il client con agentplatform:

import agentplatform

client = agentplatform.Client(
    project="PROJECT_ID",
    location="LOCATION",
)

Informazioni sull'importazione di eventi

Con l'API GenerateMemories, il salvataggio di informazioni viene attivato immediatamente. Questo metodo ti consente di controllare direttamente quando avviene il salvataggio di informazioni. IngestEvents offre un pattern di streaming che disaccoppia l'inserimento degli eventi dal salvataggio di informazioni:

  • Importazione continua: puoi trasmettere in streaming continuo il contesto a Memory Bank ogni volta che sono disponibili nuovi dati.
  • Deduplicazione: Memory Bank rileva e ignora automaticamente la cronologia delle conversazioni sovrapposte in base agli ID evento condivisi.
  • Trigger automatici: la generazione viene attivata automaticamente solo quando vengono soddisfatte condizioni specifiche, come soglie di conteggio degli eventi o tempo di inattività.
  • Generazione asincrona: gli eventi vengono elaborati in background. L'API restituisce immediatamente un'operazione a lunga esecuzione (LRO), che puoi eseguire il polling o monitorare finché non viene soddisfatta la condizione di attivazione e vengono generate le memorie.

Stream e operazioni a lunga esecuzione (LRO)

Gli stream fungono da buffer isolati che accumulano eventi in più richieste. Ogni stream è definito in modo univoco dalla combinazione di scope e stream_id. Se non fornisci un ID stream, Memory Bank utilizza default come ID. Gli eventi vengono organizzati e "svuotati" insieme solo se condividono esattamente lo stesso ambito e ID, garantendo un isolamento rigoroso durante il salvataggio di informazioni.

Più chiamate di importazione allo stesso stream restituiscono lo stesso LRO finché il salvataggio di informazioni non viene attivato e completato. Al termine, questa LRO IngestEvents contiene la LRO downstream per la chiamata API GenerateMemories. Eventuali richieste di importazione successive al flusso restituiranno una nuova operazione di lunga durata.

Attivazione del salvataggio di informazioni

Se non vengono soddisfatte condizioni di attivazione, tutti gli eventi in attesa in uno stream vengono eliminati automaticamente 24 ore dopo l'inserimento dell'ultimo evento. Questo limite garantisce che Memory Bank elabori tutti gli eventi, inclusi quelli abbandonati.

Il generation_trigger_config determina quando gli eventi raccolti vengono svuotati ed elaborati per il salvataggio di informazioni. Puoi configurare le seguenti condizioni del trigger:

  • Conteggio eventi (event_count): la generazione viene attivata una volta raggiunto il limite del numero di eventi unici accumulati.
  • Tempo di inattività (idle_duration, in secondi con granularità di un minuto, il che significa che devi specificare un multiplo di 60): la generazione viene attivata quando lo stream non ha ricevuto nuovi eventi dopo la durata specificata (ad esempio, "300s").
  • Intervallo fisso (fixed_interval, in secondi con granularità di minuti, il che significa che devi specificare un multiplo di 60): la generazione viene eseguita a un ritmo fisso, eseguendo il polling degli eventi delta dopo ogni durata specificata (ad esempio, "300s").
  • Forza svuotamento (force_flush): svuota immediatamente tutti gli eventi in attesa.

Puoi combinare una regola basata sul conteggio degli eventi con una regola basata sul tempo nello stesso generation_rule. Ad esempio, puoi impostare event_count insieme a idle_duration. Quando vengono configurate più condizioni, la generazione viene attivata non appena una di queste viene soddisfatta, a seconda di quale si verifica per prima. Le due condizioni basate sul tempo si escludono a vicenda: una singola regola può specificare idle_duration o fixed_interval, ma non entrambi. Il campo force_flush è indipendente da queste regole e svuota sempre immediatamente gli eventi in attesa.

L'API utilizza event_id per deduplicare automaticamente i dati. Fornendo un ID univoco per ogni evento, al buffer vengono aggiunti solo i nuovi eventi.

Trasferire il contesto tra le finestre di generazione

Per impostazione predefinita, ogni finestra di generazione elabora solo gli eventi accumulati dall'ultimo svuotamento. Per mantenere la coerenza dei ricordi nelle finestre di generazione consecutive, puoi trasferire il contesto da una finestra all'altra impostando overlap_event_count in generation_rule. In questo modo, all'inizio della finestra successiva vengono inclusi nuovamente gli ultimi N eventi già elaborati.

Gli eventi riportati non vengono conteggiati ai fini della condizione di attivazione. Il valore overlap_event_count deve essere un numero intero positivo inferiore a event_count e richiede event_count o una regola basata sul tempo (idle_duration o fixed_interval).

Origine dati

Con IngestEvents, fornisci i contenuti inviando gli eventi direttamente all'interno della richiesta dell'applicazione utilizzando direct_contents_source in un payload diretto.

I seguenti esempi mostrano diversi trigger.

Esempio: trigger Conteggio eventi

Questo esempio utilizza un flusso personalizzato e attiva la generazione quando nel flusso sono stati accumulati 10 eventi unici.

Dizionario

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"}
)

Basato sul corso

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"}
)

Esempio: trigger di durata inattività basato sul tempo

Questo esempio attiva la generazione quando lo stream non riceve un nuovo evento per 5 minuti (300s). Imposta anche un stream_id personalizzato.

Dizionario

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"}
)

Basato sul corso

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"}
)

Esempio: combinare trigger basati su eventi e sul tempo

Questo esempio configura due condizioni di attivazione nella stessa richiesta. La generazione viene eseguita quando si accumulano 10 eventi unici o lo stream è inattivo per 5 minuti (300s), a seconda di quale condizione si verifica per prima.

Dizionario

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"}
)

Basato sul corso

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"}
)

Esempio: eventi di forzatura del flush

Questo esempio ignora in modo incondizionato qualsiasi condizione di attivazione e forza immediatamente lo svuotamento e l'elaborazione di tutti gli eventi in attesa nello stream.

Dizionario

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"}
)

Basato sul corso

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"}
)

Esempio: trasferire il contesto tra le finestre

Questo esempio attiva la generazione ogni 10 eventi unici e include nuovamente gli ultimi 2 eventi già elaborati nella finestra successiva, in modo che i ricordi rimangano coerenti tra le finestre.

Dizionario

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"}
)

Basato sul corso

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"}
)

Configura le revisioni della memoria

Come GenerateMemories, IngestEvents crea revisioni della memoria per i ricordi che genera. Puoi configurare le revisioni risultanti da una richiesta di importazione impostando i seguenti campi in config:

  • Etichette di revisione (revision_labels): coppie chiave-valore applicate alle revisioni della memoria generate. Ad esempio, puoi etichettare una revisione con la relativa origine dati e in seguito filtrarla in base a questa etichetta.
  • Scadenza revisione (revision_expire_time o revision_ttl): controlla la scadenza delle revisioni dei ricordi generate. Se non imposti una scadenza, Memory Bank utilizza la configurazione dell'istanza o la durata (TTL) predefinita di 365 giorni.
  • Disattiva revisioni (disable_memory_revisions): se True, Memory Bank non crea revisioni per i ricordi generati da questa richiesta.

Il seguente esempio etichetta le revisioni generate con un'origine dati e le conserva per 30 giorni:

Dizionario

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"}
)

Basato sul corso

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"}
)

Allegare metadati ai ricordi generati

Puoi allegare metadati ai ricordi generati dagli eventi importati. I metadati ti consentono di archiviare informazioni strutturate (stringhe, numeri doppi, valori booleani o timestamp) insieme a un fatto della memoria, il che è utile per filtrare o gestire i cicli di vita dei tuoi ricordi.

Utilizza metadata_merge_strategy per controllare il modo in cui i metadati interagiscono con i ricordi esistenti durante il consolidamento. Puoi impostarlo su OVERWRITE (valore predefinito se non impostato), MERGE o REQUIRE_EXACT_MATCH. Per saperne di più, consulta Utilizzo dei metadati durante il consolidamento.

Dizionario

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"}
)

Basato sul corso

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"}
)

Passaggi successivi