イベントを取り込む

IngestEvents API は、メモリバンクでのイベントの取り込みとメモリーの生成を分離します。コンテンツをメモリバンクに継続的にストリーミングできます。トリガー条件が満たされると、生成が自動的に実行されます。

このガイドで説明する手順を完了するには、まず メモリバンクの設定の手順に沿って操作する必要があります。

このページの例では、最新の IngestEvents 構成オプションを提供する agentplatform SDK を使用します。agentplatform でクライアントを初期化します。

import agentplatform

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

イベントの取り込みについて

GenerateMemories API を使用すると、メモリーの生成がすぐにトリガーされます。このメソッドを使用すると、メモリーの生成のタイミングを直接制御できます。IngestEvents は、イベントの取り込みとメモリ生成を分離するストリーミング パターンを提供します。

  • 継続的な取り込み: 新しいデータが利用可能になったら、いつでもコンテキストをメモリバンクに継続的にストリーミングできます。
  • 重複除去: メモリバンクは、共有イベント ID に基づいて、重複する会話履歴を自動的に検出して無視します。
  • 自動トリガー: 生成は、イベント数のしきい値やアイドル時間など、特定の条件が満たされた場合にのみ自動的にトリガーされます。
  • 非同期生成: イベントはバックグラウンドで処理されます。API は長時間実行オペレーション（LRO）をすぐに返します。トリガー条件が満たされ、メモリが生成されるまで、この LRO をポーリングまたはモニタリングできます。

ストリームと長時間実行オペレーション（LRO）

ストリームは、複数のリクエストにわたってイベントを蓄積する分離されたバッファとして機能します。各ストリームは、scope と stream_id の組み合わせによって一意に定義されます。ストリーム ID を指定しない場合、メモリバンク は ID として default を使用します。イベントは、まったく同じスコープと ID を共有している場合にのみステージングされ、まとめて「フラッシュ」されます。これにより、メモリ生成中に厳密な分離が保証されます。

同じストリームへの複数の取り込み呼び出しは、メモリーの生成がトリガーされて完了するまで同じ LRO を返します。完了すると、この IngestEvents LRO には、GenerateMemories API 呼び出しの下流 LRO が含まれます。ストリームへの後続の取り込みリクエストは、新しい LRO を返します。

メモリーの生成をトリガーする

トリガー条件が満たされない場合、ストリーム内の保留中のイベントはすべて、最後のイベントが取り込まれてから 24 時間後に自動的にフラッシュされます。この上限により、メモリバンク は破棄されたイベントを含むすべてのイベントを処理します。

generation_trigger_config は、収集されたイベントがフラッシュされ、メモリーの生成のために処理されるタイミングを決定します。次のトリガー条件を構成できます。

  • イベント数 （event_count）: 累積された一意のイベント数が上限に達すると、生成がトリガーされます。
  • 非アクティブ時間（idle_duration、分単位の粒度で秒単位。つまり、60 の倍数を指定する必要があります）: 指定した期間（"300s" など）が経過してもストリームが新しいイベントを受信しなかった場合に、生成がトリガーされます。
  • 固定間隔（fixed_interval、分単位の粒度で秒単位。つまり、60 の倍数を指定する必要があります）: 指定した期間（"300s" など）ごとに差分イベントをポーリングし、固定のリズムで生成を実行します。
  • 強制フラッシュ （force_flush）: 保留中のイベントをすべてすぐにフラッシュします。

イベント数ルールと時間ベースのルールは、同じ generation_rule で組み合わせることができます。たとえば、event_count を idle_duration と組み合わせて設定できます。複数の条件が構成されている場合、いずれかの条件が満たされるとすぐに生成がトリガーされます。2 つの時間ベースの条件は相互に排他的です。1 つのルールで指定できるのは idle_duration または fixed_interval のいずれか一方です。両方を指定することはできません。force_flush フィールドはこれらのルールとは独立しており、保留中のイベントを常にすぐにフラッシュします。

API は event_id を使用してデータを自動的に重複除去します。各イベントに一意の ID を指定することで、新しいイベントのみがバッファに追加されます。

生成ウィンドウ間でコンテキストを引き継ぐ

デフォルトでは、各生成ウィンドウは、前回のフラッシュ以降に蓄積されたイベントのみを処理します。連続する生成ウィンドウ間でメモリの一貫性を維持するには、generation_rule で overlap_event_count を設定して、あるウィンドウから次のウィンドウにコンテキストを引き継ぎます。これにより、次のウィンドウの開始時に、すでに処理された最後の N 個のイベントが再度含まれます。

引き継がれたイベントは、トリガー条件の対象にはなりません。overlap_event_count の値は、event_count より小さい正の整数にする必要があります。また、event_count または時間ベースのルール（idle_duration または fixed_interval）が必要です。

データソース

IngestEvents を使用すると、直接ペイロードの direct_contents_source を使用して、アプリケーション リクエスト内でイベントを直接送信することでコンテンツを提供できます。

次の例は、さまざまなトリガーを示しています。

例: イベント数トリガー

この例では、カスタム ストリームを使用し、ストリームに 10 個の一意のイベントが蓄積されると生成をトリガーします。

辞書

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

クラスベース

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

例: 時間ベースのアイドル時間トリガー

この例では、ストリームが 5 分（300s）間新しいイベントを受信しなかった場合に生成をトリガーします。また、カスタム stream_id も設定します。

辞書

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

クラスベース

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

例: イベントベースのトリガーと時間ベースのトリガーを組み合わせる

この例では、同じリクエストで 2 つのトリガー条件を構成します。10 個の一意のイベントが蓄積されるか、ストリームが 5 分（300s）間アイドル状態になると、生成が実行されます。

辞書

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

クラスベース

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

例: イベントを強制的にフラッシュする

この例では、トリガー条件を無条件でバイパスし、ストリーム内の保留中のイベントをすべてすぐにフラッシュして処理します。

辞書

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

クラスベース

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

例: ウィンドウ間でコンテキストを引き継ぐ

この例では、10 個の一意のイベントごとに生成をトリガーし、すでに処理された最後の 2 つのイベントを次のウィンドウに再度含めることで、ウィンドウ間でメモリの一貫性を維持します。

辞書

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

クラスベース

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

メモリ リビジョンを構成する

GenerateMemories と同様に、IngestEvents は生成されたメモリのメモリ リビジョンを作成します。取り込みリクエストの結果として生成されるリビジョンを構成するには、config で次のフィールドを設定します。

  • リビジョン ラベル （revision_labels）: 生成されたメモリ リビジョンに適用される Key-Value ペア。たとえば、リビジョンにデータソースのラベルを付け、後でそのラベルでリビジョンをフィルタできます。
  • リビジョンの有効期限（revision_expire_time または revision_ttl）: 生成されたメモリ リビジョンの有効期限を制御します。有効期限を設定しない場合、メモリバンクはインスタンス構成またはデフォルトの有効期間（TTL）である 365 日を使用します。
  • リビジョンを無効にする（disable_memory_revisions）：Trueの場合、メモリバンクはこのリクエストから生成されたメモリのリビジョンを作成しません。

次の例では、生成されたリビジョンにデータソースのラベルを付け、30 日間保持します。

辞書

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

クラスベース

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

生成されたメモリにメタデータを添付する

取り込まれたイベントから生成されたメモリにメタデータを添付できます。メタデータを使用すると、メモリファクトとともに構造化情報（文字列、倍精度浮動小数点数、ブール値、タイムスタンプ）を保存できます。これは、メモリのフィルタリングやライフサイクルの管理に役立ちます。

metadata_merge_strategy を使用して、統合中にメタデータが既存のメモリとどのように連携するかを制御します。OVERWRITE（設定されていない場合のデフォルト）、MERGE、REQUIRE_EXACT_MATCH のいずれかに設定できます。詳細については、統合中にメタデータを使用するをご覧ください。

辞書

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

クラスベース

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

次のステップ