Menyerap peristiwa

IngestEvents API memisahkan penyerapan peristiwa dari pembuatan memori di Memory Bank. Anda dapat melakukan streaming konten secara berkelanjutan ke Memory Bank, dan pembuatan akan otomatis berjalan saat kriteria pemicu Anda terpenuhi.

Untuk menyelesaikan langkah-langkah yang ditunjukkan dalam panduan ini, Anda harus terlebih dahulu mengikuti langkah-langkah di Menyiapkan Memory Bank.

Contoh di halaman ini menggunakan agentplatform SDK, yang menyediakan opsi konfigurasi IngestEvents terbaru. Lakukan inisialisasi klien dengan agentplatform:

import agentplatform

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

Memahami penyerapan peristiwa

Dengan GenerateMemories API, pembuatan memori akan segera dipicu. Metode ini memberi Anda kontrol langsung atas waktu terjadinya pembuatan memori. IngestEvents menawarkan pola streaming yang memisahkan penyerapan peristiwa dari pembuatan memori:

  • Penyerapan berkelanjutan: Anda dapat melakukan streaming konteks secara berkelanjutan ke Memory Bank kapan pun data baru tersedia.
  • Penghapusan duplikat: Memory Bank otomatis mendeteksi dan mengabaikan histori percakapan yang tumpang-tindih berdasarkan ID peristiwa bersama.
  • Pemicu otomatis: Pembuatan otomatis dipicu hanya jika kondisi tertentu terpenuhi, seperti batas jumlah peristiwa atau waktu tidak ada aktivitas.
  • Pembuatan asinkron: Peristiwa diproses di latar belakang. API akan segera menampilkan operasi yang berjalan lama (LRO), yang dapat Anda lakukan polling atau pantau hingga kondisi pemicu terpenuhi dan memori dibuat.

Streaming dan operasi yang berjalan lama (LRO)

Streaming bertindak sebagai buffer terisolasi yang mengakumulasi peristiwa di beberapa permintaan. Setiap streaming ditentukan secara unik oleh kombinasi scope dan stream_id. Jika Anda tidak memberikan ID streaming, Memory Bank akan menggunakan default sebagai ID. Peristiwa hanya dipentaskan dan "dikosongkan" bersama jika memiliki cakupan dan ID yang sama persis, sehingga memastikan isolasi yang ketat selama pembuatan memori.

Beberapa panggilan penyerapan ke streaming yang sama akan menampilkan LRO yang sama hingga pembuatan memori dipicu dan selesai. Setelah selesai, LRO IngestEvents ini berisi LRO hilir untuk panggilan GenerateMemories API. Setiap permintaan penyerapan berikutnya ke streaming akan menampilkan LRO baru.

Memicu pembuatan memori

Jika tidak ada kondisi pemicu yang terpenuhi, semua peristiwa yang tertunda dalam streaming akan otomatis dikosongkan 24 jam setelah peristiwa terakhir diserap. Batas ini memastikan Memory Bank memproses semua peristiwa, termasuk peristiwa yang telah ditinggalkan.

generation_trigger_config menentukan kapan peristiwa yang dikumpulkan dikosongkan dan diproses untuk pembuatan memori. Anda dapat mengonfigurasi kondisi pemicu berikut:

  • Jumlah peristiwa (event_count): Pembuatan dipicu setelah jumlah peristiwa unik yang terakumulasi mencapai batas.
  • Waktu tidak ada aktivitas (idle_duration, dalam detik dengan granularitas menit, yang berarti Anda harus menentukan kelipatan 60): Pembuatan dipicu saat streaming tidak menerima peristiwa baru setelah durasi yang ditentukan (misalnya, "300s").
  • Interval tetap (fixed_interval, dalam detik dengan granularitas menit, yang berarti Anda harus menentukan kelipatan 60): Pembuatan dijalankan dengan ritme tetap, melakukan polling untuk peristiwa delta setelah setiap durasi yang ditentukan (misalnya, "300s").
  • Paksa melakukan flush (force_flush): Segera melakukan flush semua peristiwa yang tertunda.

Anda dapat menggabungkan aturan jumlah peristiwa dengan aturan berbasis waktu dalam generation_rule yang sama. Misalnya, Anda dapat menetapkan event_count bersama dengan idle_duration. Jika lebih dari satu kondisi dikonfigurasi, pembuatan akan dipicu segera setelah salah satu kondisi terpenuhi, mana pun yang terjadi lebih dulu. Dua kondisi berbasis waktu saling eksklusif: satu aturan dapat menentukan idle_duration atau fixed_interval, tetapi tidak keduanya. Kolom force_flush tidak bergantung pada aturan ini dan selalu mengosongkan peristiwa yang tertunda.

API menggunakan event_id untuk menghapus duplikat data secara otomatis. Dengan memberikan ID unik untuk setiap peristiwa, hanya peristiwa baru yang ditambahkan ke buffer.

Meneruskan konteks antar-periode pembuatan

Secara default, setiap periode pembuatan hanya memproses peristiwa yang terakumulasi sejak pengosongan sebelumnya. Untuk menjaga memori tetap koheren di seluruh periode pembuatan berturut-turut, Anda dapat meneruskan konteks dari satu periode ke periode berikutnya dengan menetapkan overlap_event_count di generation_rule. Hal ini akan menyertakan kembali N peristiwa terakhir yang telah diproses di awal periode berikutnya.

Peristiwa yang diteruskan tidak dihitung dalam kondisi pemicu. Nilai overlap_event_count harus berupa bilangan bulat positif yang kurang dari event_count, dan memerlukan event_count atau aturan berbasis waktu (idle_duration atau fixed_interval).

Sumber data

Dengan IngestEvents, Anda menyediakan konten dengan mengirimkan peristiwa langsung dalam permintaan aplikasi menggunakan direct_contents_source dalam payload langsung.

Contoh berikut menunjukkan pemicu yang berbeda.

Contoh: Pemicu jumlah peristiwa

Contoh ini menggunakan streaming kustom dan memicu pembuatan saat 10 peristiwa unik telah terakumulasi dalam streaming.

Kamus

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

Berbasis class

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

Contoh: Pemicu durasi tidak ada aktivitas berbasis waktu

Contoh ini memicu pembuatan saat streaming tidak menerima peristiwa baru selama 5 menit (300s). Contoh ini juga menetapkan stream_id kustom.

Kamus

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

Berbasis class

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

Contoh: Menggabungkan pemicu berbasis peristiwa dan berbasis waktu

Contoh ini mengonfigurasi dua kondisi pemicu dalam permintaan yang sama. Pembuatan berjalan saat 10 peristiwa unik terakumulasi atau streaming tidak ada aktivitas selama 5 menit (300s), mana pun yang terjadi lebih dulu.

Kamus

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

Berbasis class

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

Contoh: Melakukan flush peristiwa

Contoh ini tanpa syarat melewati kondisi pemicu apa pun dan segera memaksa semua peristiwa yang tertunda dalam streaming untuk dikosongkan dan diproses.

Kamus

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

Berbasis class

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

Contoh: Meneruskan konteks antar-periode

Contoh ini memicu pembuatan setiap 10 peristiwa unik dan menyertakan kembali 2 peristiwa terakhir yang telah diproses di periode berikutnya sehingga memori tetap koheren di seluruh periode.

Kamus

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

Berbasis class

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

Mengonfigurasi revisi memori

Seperti GenerateMemories, IngestEvents membuat revisi memori untuk memori yang dibuatnya. Anda dapat mengonfigurasi revisi yang dihasilkan dari permintaan penyerapan dengan menetapkan kolom berikut di config:

  • Label revisi (revision_labels): Pasangan nilai kunci yang diterapkan ke revisi memori yang dibuat. Misalnya, Anda dapat memberi label revisi dengan sumber datanya dan kemudian memfilter revisi berdasarkan label tersebut.
  • Masa berlaku revisi (revision_expire_time atau revision_ttl): Mengontrol kapan masa berlaku revisi memori yang dibuat akan berakhir. Jika Anda tidak menetapkan masa berlaku, Memory Bank akan menggunakan konfigurasi instance atau waktu default untuk aktif (TTL) selama 365 hari.
  • Nonaktifkan revisi (disable_memory_revisions): Jika True, Memory Bank tidak akan membuat revisi untuk memori yang dibuat dari permintaan ini.

Contoh berikut memberi label revisi yang dibuat dengan sumber data dan menyimpannya selama 30 hari:

Kamus

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

Berbasis class

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

Melampirkan metadata ke memori yang dibuat

Anda dapat melampirkan metadata ke memori yang dibuat dari peristiwa yang diserap. Metadata memungkinkan Anda menyimpan informasi terstruktur (string, ganda, boolean, atau stempel waktu) bersama dengan fakta memori, yang berguna untuk memfilter atau mengelola siklus proses memori Anda.

Gunakan metadata_merge_strategy untuk mengontrol cara metadata berinteraksi dengan memori yang ada selama konsolidasi. Anda dapat menetapkannya ke OVERWRITE (default jika tidak ditetapkan), MERGE, atau REQUIRE_EXACT_MATCH. Untuk mengetahui informasi selengkapnya, lihat Menggunakan metadata selama konsolidasi.

Kamus

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

Berbasis class

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

Langkah berikutnya