La API de IngestEvents desacopla la ingesta de eventos de la generación de memoria en Memory Bank. Puedes transmitir contenido de forma continua a Memory Bank, y la generación se ejecutará automáticamente cuando se cumplan tus criterios de activación.
Para completar los pasos que se muestran en esta guía, primero debes seguir los pasos que se indican en Configura Memory Bank.
En los ejemplos de esta página, se usa el SDK de agentplatform, que proporciona las opciones de configuración más recientes de IngestEvents. Inicializa el cliente con agentplatform:
import agentplatform
client = agentplatform.Client(
project="PROJECT_ID",
location="LOCATION",
)
Información sobre la ingesta de eventos
Con la API de GenerateMemories, la generación de memoria se activa de inmediato. Este método te brinda control directo sobre cuándo se produce la generación de memoria. IngestEvents ofrece un patrón de transmisión que desacopla la ingesta de eventos de la generación de memoria:
- Ingesta continua: Puedes transmitir contexto de forma continua a Memory Bank cada vez que haya datos nuevos disponibles.
- Anulación de duplicados: Memory Bank detecta y omite automáticamente el historial de conversaciones superpuestas en función de los IDs de eventos compartidos.
- Activadores automatizados: La generación se activa automáticamente solo cuando se cumplen condiciones específicas, como los umbrales de recuento de eventos o el tiempo de inactividad.
- Generación asíncrona: Los eventos se procesan en segundo plano. La API muestra de inmediato una operación de larga duración (LRO), que puedes sondear o supervisar hasta que se cumpla la condición de activación y se generen los recuerdos.
Transmisiones y operaciones de larga duración (LRO)
Las transmisiones actúan como búferes aislados que acumulan eventos en varias solicitudes. Cada transmisión se define de forma única por la combinación de su scope y stream_id. Si no proporcionas un ID de transmisión, Memory Bank usa default como ID. Los eventos solo se almacenan y se "vacían" juntos si comparten exactamente el mismo alcance y el mismo ID, lo que garantiza un aislamiento estricto durante la generación de memoria.
Varias llamadas de ingesta a la misma transmisión muestran la misma LRO hasta que se activa y finaliza la generación de memoria. Una vez completada, esta LRO de IngestEvents contiene la LRO descendente para la llamada a la API de GenerateMemories. Cualquier solicitud de ingesta posterior a la transmisión mostrará una LRO nueva.
Cómo activar la generación de memoria
Si no se cumplen las condiciones de activación, todos los eventos pendientes de una transmisión se vacían automáticamente 24 horas después de que se ingiere el último evento. Este límite garantiza que Memory Bank procese todos los eventos, incluidos los que se abandonaron.
generation_trigger_config determina cuándo se vacían y procesan los eventos recopilados para la generación de memoria. Puedes configurar las siguientes condiciones de activación:
- Recuento de eventos (
event_count): La generación se activa una vez que la cantidad de eventos únicos acumulados alcanza el límite. - Tiempo de inactividad (
idle_duration, en segundos con granularidad de minutos, lo que significa que debes especificar un múltiplo de 60): La generación se activa cuando la transmisión no recibe eventos nuevos después de la duración especificada (por ejemplo,"300s"). - Intervalo fijo (
fixed_interval, en segundos con granularidad de minutos, lo que significa que debes especificar un múltiplo de 60): La generación se ejecuta con un ritmo fijo y sondea los eventos delta después de cada duración especificada (por ejemplo,"300s"). - Forzar vaciado (
force_flush): Vacía todos los eventos pendientes de inmediato.
Puedes combinar una regla de recuento de eventos con una regla basada en el tiempo en la misma generation_rule. Por ejemplo, puedes configurar event_count junto con idle_duration. Cuando se configura más de una condición, la generación se activa en cuanto se cumple cualquiera de ellas, lo que ocurra primero. Las dos condiciones basadas en el tiempo son mutuamente excluyentes: una sola regla puede especificar idle_duration o fixed_interval, pero no ambas. El campo force_flush es independiente de estas reglas y siempre vacía los eventos pendientes de inmediato.
La API usa el event_id para anular automáticamente la duplicación de datos. Si proporcionas un ID único para cada evento, solo se agregarán eventos nuevos al búfer.
Transfiere el contexto entre las ventanas de generación
De forma predeterminada, cada ventana de generación solo procesa los eventos que se acumularon desde el vaciado anterior. Para mantener la coherencia de los recuerdos en las ventanas de generación consecutivas, puedes transferir el contexto de una ventana a la siguiente configurando overlap_event_count en generation_rule. Esto vuelve a incluir los últimos N eventos ya procesados al comienzo de la siguiente ventana.
Los eventos transferidos no se toman en cuenta para la condición de activación. El valor overlap_event_count debe ser un número entero positivo menor que event_count, y requiere event_count o una regla basada en el tiempo (idle_duration o fixed_interval).
Fuente de datos
Con IngestEvents, proporcionas contenido enviando eventos directamente dentro de la solicitud de tu aplicación con direct_contents_source en una carga útil directa.
En los siguientes ejemplos, se muestran diferentes activadores.
Ejemplo: Activador de recuento de eventos
En este ejemplo, se usa una transmisión personalizada y se activa la generación cuando se acumulan 10 eventos únicos en la transmisión.
Diccionario
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"}
)
Basado en clases
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"}
)
Ejemplo: Activador de duración de inactividad basado en el tiempo
En este ejemplo, se activa la generación cuando la transmisión no recibe un evento nuevo durante 5 minutos (300s). También establece un stream_id personalizado.
Diccionario
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"}
)
Basado en clases
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"}
)
Ejemplo: Combina activadores basados en eventos y en el tiempo
En este ejemplo, se configuran dos condiciones de activación en la misma solicitud. La generación se ejecuta cuando se acumulan 10 eventos únicos o cuando la transmisión está inactiva durante 5 minutos (300s), lo que ocurra primero.
Diccionario
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"}
)
Basado en clases
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"}
)
Ejemplo: Forzar el vaciado de eventos
En este ejemplo, se omite de forma incondicional cualquier condición de activación y se obliga de inmediato a vaciar y procesar todos los eventos pendientes de la transmisión.
Diccionario
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"}
)
Basado en clases
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"}
)
Ejemplo: Transfiere el contexto entre las ventanas
En este ejemplo, se activa la generación cada 10 eventos únicos y se vuelven a incluir los últimos 2 eventos ya procesados en la siguiente ventana para que los recuerdos sigan siendo coherentes en todas las ventanas.
Diccionario
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"}
)
Basado en clases
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 revisiones de memoria
Al igual que GenerateMemories, IngestEvents crea revisiones de memoria para los recuerdos que genera. Puedes configurar las revisiones que resultan de una solicitud de ingesta configurando los siguientes campos en config:
- Etiquetas de revisión (
revision_labels): Pares clave-valor que se aplican a las revisiones de memoria generadas. Por ejemplo, puedes etiquetar una revisión con su fuente de datos y, luego, filtrar las revisiones por esa etiqueta. - Vencimiento de la revisión (
revision_expire_timeorevision_ttl): Controla cuándo vencen las revisiones de memoria generadas. Si no estableces un vencimiento, Memory Bank usa la configuración de la instancia o el tiempo de actividad (TTL) predeterminado de 365 días. - Inhabilitar revisiones (
disable_memory_revisions): Si esTrue, Memory Bank no crea revisiones para los recuerdos que se generan a partir de esta solicitud.
En el siguiente ejemplo, se etiquetan las revisiones generadas con una fuente de datos y se conservan durante 30 días:
Diccionario
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"}
)
Basado en clases
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"}
)
Adjunta metadatos a los recuerdos generados
Puedes adjuntar metadatos a los recuerdos que se generan a partir de eventos ingeridos. Los metadatos te permiten almacenar información estructurada (cadenas, dobles, booleanos o marcas de tiempo) junto con un hecho de memoria, lo que es útil para filtrar o administrar los ciclos de vida de tus recuerdos.
Usa metadata_merge_strategy para controlar cómo interactúan los metadatos con los recuerdos existentes durante la consolidación. Puedes configurarlo como OVERWRITE (predeterminado cuando no se configura), MERGE o REQUIRE_EXACT_MATCH. Para obtener más información, consulta Cómo usar metadatos durante la consolidación.
Diccionario
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"}
)
Basado en clases
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"}
)