Ereignisse aufnehmen

Mit der IngestEvents API wird die Aufnahme von Ereignissen vom Merken von Informationen in der Memory Bank entkoppelt. Sie können Inhalte kontinuierlich an Memory Bank streamen. Die Generierung erfolgt automatisch, wenn Ihre Auslösekriterien erfüllt sind.

Wenn Sie dieser Anleitung folgen möchten, sollten Sie zuerst die Schritte unter Für Memory Bank einrichten ausführen.

In den Beispielen auf dieser Seite wird das agentplatform SDK verwendet, das die neuesten IngestEvents-Konfigurationsoptionen bietet. Initialisieren Sie den Client mit agentplatform:

import agentplatform

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

Ereignisaufnahme

Mit der GenerateMemories API wird das Merken von Informationen sofort ausgelöst. Mit dieser Methode haben Sie die direkte Kontrolle darüber, wann Erinnerungen generiert werden. IngestEvents bietet ein Streamingmuster, das die Ereignisaufnahme vom Merken von Informationen entkoppelt:

  • Kontinuierliche Aufnahme: Sie können Kontext kontinuierlich in Memory Bank streamen, sobald neue Daten verfügbar sind.
  • Deduplizierung: Memory Bank erkennt und ignoriert automatisch sich überschneidende Unterhaltungsverläufe basierend auf gemeinsamen Ereignis-IDs.
  • Automatisierte Auslöser: Die Generierung wird nur automatisch ausgelöst, wenn bestimmte Bedingungen erfüllt sind, z. B. Grenzwerte für die Ereignisanzahl oder Leerlaufzeiten.
  • Asynchrone Generierung: Ereignisse werden im Hintergrund verarbeitet. Die API gibt sofort einen Vorgang mit langer Ausführungszeit (long-running operation, LRO) zurück, den Sie abfragen oder überwachen können, bis die Triggerbedingung erfüllt ist und Erinnerungen generiert werden.

Streams und Vorgänge mit langer Ausführungszeit (Long-Running Operations, LROs)

Streams fungieren als isolierte Puffer, in denen Ereignisse aus mehreren Anfragen gesammelt werden. Jeder Stream wird eindeutig durch die Kombination aus scope und stream_id definiert. Wenn Sie keine Stream-ID angeben, verwendet Memory Bank default als ID. Ereignisse werden nur dann gemeinsam bereitgestellt und „geleert“, wenn sie denselben Bereich und dieselbe ID haben. So wird eine strikte Isolation während des Merkens von Informationen gewährleistet.

Mehrere Ingestion-Aufrufe für denselben Stream geben denselben LRO zurück, bis das Merken von Informationen ausgelöst und abgeschlossen ist. Nach Abschluss enthält dieser IngestEvents-LRO den Downstream-LRO für den GenerateMemories-API-Aufruf. Für alle nachfolgenden Aufnahmeanfragen für den Stream wird ein neuer LRO zurückgegeben.

Merken von Informationen auslösen

Wenn keine Triggerbedingungen erfüllt sind, werden alle ausstehenden Ereignisse in einem Stream 24 Stunden nach der Erfassung des letzten Ereignisses automatisch geleert. Dieses Limit sorgt dafür, dass Memory Bank alle Ereignisse verarbeitet, auch solche, die abgebrochen wurden.

Der generation_trigger_config bestimmt, wann die erfassten Ereignisse geleert und für das Merken von Informationen verarbeitet werden. Sie können die folgenden Triggerbedingungen konfigurieren:

  • Ereignisanzahl (event_count): Die Generierung wird ausgelöst, sobald die Anzahl der eindeutigen Ereignisse das Limit erreicht.
  • Inaktive Zeit (idle_duration, in Sekunden mit einer Granularität von einer Minute, d. h., Sie müssen ein Vielfaches von 60 angeben): Die Generierung wird ausgelöst, wenn im Stream nach der angegebenen Dauer keine neuen Ereignisse empfangen wurden (z. B. "300s").
  • Festes Intervall (fixed_interval in Sekunden mit Minutengranularität, d. h. Sie müssen ein Vielfaches von 60 angeben): Die Generierung erfolgt in einem festen Rhythmus. Nach jeder angegebenen Dauer (z. B. "300s") werden Delta-Ereignisse abgefragt.
  • Sofortiges Leeren erzwingen (force_flush): Alle ausstehenden Ereignisse werden sofort geleert.

Sie können eine Regel für die Anzahl der Ereignisse mit einer zeitbasierten Regel im selben generation_rule kombinieren. Sie können beispielsweise event_count zusammen mit idle_duration festlegen. Wenn mehrere Bedingungen konfiguriert sind, wird die Generierung ausgelöst, sobald eine der Bedingungen erfüllt ist, je nachdem, welche Bedingung zuerst erfüllt wird. Die beiden zeitbasierten Bedingungen schließen sich gegenseitig aus: In einer einzelnen Regel kann entweder idle_duration oder fixed_interval angegeben werden, aber nicht beides. Das Feld force_flush ist unabhängig von diesen Regeln und leert ausstehende Ereignisse immer sofort.

Die API verwendet die event_id, um Daten automatisch zu deduplizieren. Wenn Sie für jedes Ereignis eine eindeutige ID angeben, werden dem Puffer nur neue Ereignisse hinzugefügt.

Kontext zwischen Generierungsfenstern übertragen

Standardmäßig werden in jedem Generierungszeitraum nur die Ereignisse verarbeitet, die seit dem letzten Flush angefallen sind. Damit Erinnerungen über aufeinanderfolgende Generierungsfenster hinweg konsistent bleiben, können Sie Kontext von einem Fenster in das nächste übertragen, indem Sie overlap_event_count in generation_rule festlegen. Dadurch werden die letzten N bereits verarbeiteten Ereignisse am Anfang des nächsten Fensters wieder einbezogen.

Übertragene Ereignisse werden nicht auf die Triggerbedingung angerechnet. Der Wert overlap_event_count muss eine positive Ganzzahl sein, die kleiner als event_count ist. Außerdem ist entweder event_count oder eine zeitbasierte Regel (idle_duration oder fixed_interval) erforderlich.

Datenquelle

Mit IngestEvents stellen Sie Inhalte bereit, indem Sie Ereignisse direkt in Ihrem Anwendungsantrag über direct_contents_source in einer direkten Nutzlast senden.

Die folgenden Beispiele zeigen verschiedene Trigger.

Beispiel: Trigger für Ereignisanzahl

In diesem Beispiel wird ein benutzerdefinierter Stream verwendet und die Generierung wird ausgelöst, wenn 10 eindeutige Ereignisse im Stream zusammengefasst wurden.

Wörterbuch

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

Klassenbasiert

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

Beispiel: Zeitbasierter Trigger für die Leerlaufdauer

In diesem Beispiel wird die Generierung ausgelöst, wenn seit 5 Minuten (300s) kein neues Ereignis für den Stream empfangen wurde. Außerdem wird ein benutzerdefinierter stream_id festgelegt.

Wörterbuch

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

Klassenbasiert

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

Beispiel: Ereignisbasierte und zeitbasierte Trigger kombinieren

In diesem Beispiel werden zwei Triggerbedingungen in derselben Anfrage konfiguriert. Die Generierung wird ausgeführt, wenn entweder 10 eindeutige Ereignisse zusammenkommen oder der Stream 5 Minuten lang inaktiv ist (300s), je nachdem, was zuerst eintritt.

Wörterbuch

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

Klassenbasiert

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

Beispiel: Ereignisse erzwingen

In diesem Beispiel werden alle Triggerbedingungen bedingungslos umgangen und alle ausstehenden Ereignisse im Stream werden sofort geleert und verarbeitet.

Wörterbuch

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

Klassenbasiert

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

Beispiel: Kontext zwischen Fenstern übertragen

In diesem Beispiel wird die Generierung alle 10 eindeutigen Ereignisse ausgelöst. Die letzten 2 bereits verarbeiteten Ereignisse werden in das nächste Fenster aufgenommen, damit die Erinnerungen über die Fenster hinweg zusammenhängend bleiben.

Wörterbuch

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

Klassenbasiert

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

Arbeitsspeicherrevisionen konfigurieren

Wie bei GenerateMemories werden bei IngestEvents Erinnerungsüberarbeitungen für die generierten Erinnerungen erstellt. Sie können die Überarbeitungen, die sich aus einer Aufnahmeanfrage ergeben, konfigurieren, indem Sie die folgenden Felder in der config festlegen:

  • Überarbeitungslabels (revision_labels): Schlüssel/Wert-Paare, die auf die generierten Speicherüberarbeitungen angewendet werden. Sie können beispielsweise eine Überarbeitung mit ihrer Datenquelle kennzeichnen und später Überarbeitungen nach diesem Label filtern.
  • Ablauf von Überarbeitungen (revision_expire_time oder revision_ttl): Steuert, wann die generierten Überarbeitungen von Erinnerungen ablaufen. Wenn Sie keine Ablaufzeit festlegen, verwendet Memory Bank die Instanzkonfiguration oder die Standard-TTL (Time-to-Live) von 365 Tagen.
  • Überarbeitungen deaktivieren (disable_memory_revisions): Wenn True, Memory Bank erstellt keine Überarbeitungen für die Erinnerungen, die aus dieser Anfrage generiert werden.

Im folgenden Beispiel werden die generierten Überarbeitungen mit einer Datenquelle gekennzeichnet und 30 Tage lang aufbewahrt:

Wörterbuch

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

Klassenbasiert

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

Metadaten an generierte Erinnerungen anhängen

Sie können den Erinnerungen, die aus aufgenommenen Ereignissen generiert werden, Metadaten anhängen. Mit Metadaten können Sie strukturierte Informationen (Strings, Doubles, boolesche Werte oder Zeitstempel) zusammen mit einem Memory-Fakt speichern. Das ist nützlich, um Ihre Erinnerungen zu filtern oder ihren Lebenszyklus zu verwalten.

Mit metadata_merge_strategy können Sie steuern, wie die Metadaten während der Konsolidierung mit vorhandenen Erinnerungen interagieren. Sie können den Wert auf OVERWRITE (Standardeinstellung, wenn nicht festgelegt), MERGE oder REQUIRE_EXACT_MATCH festlegen. Weitere Informationen finden Sie unter Metadaten bei der Konsolidierung verwenden.

Wörterbuch

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

Klassenbasiert

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

Nächste Schritte