A API IngestEvents separa a transferência de eventos da geração de dados de memória no Memory Bank. Você pode transmitir conteúdo continuamente para o Memory Bank, e a geração será executada automaticamente quando os critérios de acionamento forem atendidos.
Para concluir as etapas demonstradas neste guia, primeiro siga as instruções em Configurar o Memory Bank.
Os exemplos nesta página usam o SDK agentplatform, que oferece as opções de configuração mais recentes do IngestEvents. Inicialize o cliente com agentplatform:
import agentplatform
client = agentplatform.Client(
project="PROJECT_ID",
location="LOCATION",
)
Entender a ingestão de eventos
Com a API GenerateMemories, a geração de dados de memória é acionada imediatamente. Esse método oferece controle direto sobre quando a geração de dados de memória ocorre. O IngestEvents oferece um padrão de streaming que separa a ingestão de eventos da geração de dados de memória:
- Ingestão contínua: você pode transmitir contexto continuamente para o Memory Bank sempre que novos dados estiverem disponíveis.
- Deduplicação: o Memory Bank detecta e ignora automaticamente o histórico de conversas sobrepostas com base em IDs de eventos compartilhados.
- Acionadores automáticos: a geração é acionada automaticamente apenas quando condições específicas são atendidas, como limiares de contagem de eventos ou tempo ocioso.
- Geração assíncrona: os eventos são processados em segundo plano. A API retorna imediatamente uma operação de longa duração (LRO, na sigla em inglês), que você pode pesquisar ou monitorar até que a condição de acionamento seja atendida e as recordações sejam geradas.
Streams e operações de longa duração (LROs)
Os fluxos atuam como buffers isolados que acumulam eventos em várias solicitações. Cada stream é definida de forma exclusiva pela combinação de scope e stream_id. Se você não fornecer um ID de stream, o Memory Bank usará default como ID. Os eventos só são organizados e "transferidos" juntos se compartilharem exatamente o mesmo escopo e ID, garantindo um isolamento estrito durante a geração de dados de memória.
Várias chamadas de ingestão para o mesmo fluxo retornam a mesma LRO até que a geração de dados de memória seja acionada e concluída. Após a conclusão, essa LRO IngestEvents contém a LRO downstream para a chamada de API GenerateMemories. Qualquer solicitação de ingestão subsequente para o fluxo vai retornar um novo LRO.
Acionamento da geração de dados de memória
Se nenhuma condição de acionamento for atendida, todos os eventos pendentes em um stream serão liberados automaticamente 24 horas após a ingestão do último evento. Esse limite garante que o Memory Bank processe todos os eventos, incluindo os abandonados.
O generation_trigger_config determina quando os eventos coletados são liberados e processados para geração de dados de memória. É possível configurar as seguintes condições de acionamento:
- Contagem de eventos (
event_count): a geração é acionada quando o número de eventos únicos acumulados atinge o limite. - Tempo de inatividade (
idle_duration, em segundos com granularidade de minutos, ou seja, é necessário especificar um múltiplo de 60): a geração é acionada quando o fluxo não recebe novos eventos após a duração especificada (por exemplo,"300s"). - Intervalo fixo (
fixed_interval, em segundos com granularidade de minuto, ou seja, você precisa especificar um múltiplo de 60): a geração é executada em um ritmo fixo, pesquisando eventos delta após cada duração especificada (por exemplo,"300s"). - Forçar transferência (
force_flush): transfere todos os eventos pendentes imediatamente.
É possível combinar uma regra de contagem de eventos com uma regra baseada em tempo no mesmo generation_rule. Por exemplo, você pode definir event_count com idle_duration. Quando mais de uma condição é configurada, a geração é acionada assim que uma delas é atendida, o que ocorrer primeiro. As duas condições baseadas em tempo são mutuamente exclusivas: uma única regra pode especificar idle_duration ou fixed_interval, mas não ambos. O campo force_flush é independente dessas regras e sempre libera eventos pendentes imediatamente.
A API usa o event_id para eliminar a duplicação de dados automaticamente. Ao fornecer um ID exclusivo para cada evento, apenas os novos são adicionados ao buffer.
Transferir contexto entre janelas de geração
Por padrão, cada janela de geração processa apenas os eventos acumulados desde a limpeza anterior. Para manter as memórias coerentes em janelas de geração consecutivas, você pode transferir o contexto de uma janela para a próxima definindo overlap_event_count no generation_rule. Isso inclui novamente os últimos N eventos já processados no início da próxima janela.
Os eventos transferidos não contam para a condição de acionamento. O valor de overlap_event_count precisa ser um número inteiro positivo menor que event_count e exige event_count ou uma regra baseada em tempo (idle_duration ou fixed_interval).
Fonte de dados
Com o IngestEvents, você fornece conteúdo enviando eventos diretamente na solicitação do aplicativo usando direct_contents_source em uma carga direta.
Os exemplos a seguir mostram diferentes acionadores.
Exemplo: acionador de contagem de eventos
Este exemplo usa um fluxo personalizado e aciona a geração quando 10 eventos únicos são acumulados nele.
Dicionário
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"}
)
Com base em classes
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"}
)
Exemplo: acionador de duração de inatividade com base no tempo
Este exemplo aciona a geração quando o stream não recebe um novo evento há 5 minutos (300s). Ele também define um stream_id personalizado.
Dicionário
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"}
)
Com base em classes
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"}
)
Exemplo: combinar gatilhos baseados em eventos e em tempo
Este exemplo configura duas condições de acionamento na mesma solicitação. A geração é executada quando 10 eventos únicos são acumulados ou quando o stream fica inativo por 5 minutos (300s), o que ocorrer primeiro.
Dicionário
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"}
)
Com base em classes
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"}
)
Exemplo: forçar o flush de eventos
Este exemplo ignora incondicionalmente todas as condições de acionamento e força imediatamente o descarte e o processamento de todos os eventos pendentes no fluxo.
Dicionário
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"}
)
Com base em classes
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"}
)
Exemplo: manter o contexto entre janelas
Este exemplo aciona a geração a cada 10 eventos únicos e inclui novamente os dois últimos eventos já processados na próxima janela para que as recordações permaneçam coerentes entre as janelas.
Dicionário
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"}
)
Com base em classes
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"}
)
Configurar revisões de memória
Assim como GenerateMemories, IngestEvents cria revisões de memória para as recordações geradas. É possível configurar as revisões resultantes de uma solicitação de ingestão definindo os seguintes campos no config:
- Rótulos de revisão (
revision_labels): pares de chave-valor aplicados às revisões de memória geradas. Por exemplo, é possível rotular uma revisão com a fonte de dados dela e depois filtrar revisões por esse rótulo. - Validade da revisão (
revision_expire_timeourevision_ttl): controla quando as revisões de memória geradas expiram. Se você não definir uma validade, o Memory Bank usará a configuração da instância ou o time to live (TTL) padrão de 365 dias. - Desativar revisões (
disable_memory_revisions): seTrue, o Memory Bank não vai criar revisões para as recordações geradas com base nessa solicitação.
O exemplo a seguir rotula as revisões geradas com uma fonte de dados e as mantém por 30 dias:
Dicionário
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"}
)
Com base em classes
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"}
)
Anexar metadados às recordações geradas
É possível anexar metadados às recordações geradas com base em eventos ingeridos. Os metadados permitem armazenar informações estruturadas (strings, doubles, booleanos ou carimbos de data/hora) junto com um fato da memória, o que é útil para filtrar ou gerenciar os ciclos de vida das suas recordações.
Use metadata_merge_strategy para controlar como os metadados interagem com as memórias atuais durante a consolidação. Você pode definir como OVERWRITE (padrão quando não definido), MERGE ou REQUIRE_EXACT_MATCH. Para mais informações, consulte Como usar metadados durante a consolidação.
Dicionário
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"}
)
Com base em classes
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"}
)