擷取事件

IngestEvents API 可將事件擷取作業與 Memory Bank 中的記憶生成作業分離。你可以持續將內容串流至 Memory Bank，系統會在符合觸發條件時自動生成。

如要完成本指南中示範的步驟，請先按照「設定 Memory Bank」中的步驟操作。

本頁面的範例使用 agentplatform SDK，其中提供最新的 IngestEvents 設定選項。使用 agentplatform 初始化用戶端：

import agentplatform

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

瞭解事件擷取

使用 GenerateMemories API 時，系統會立即觸發記憶體生成作業。這種方法可讓您直接控制記憶體生成時間。IngestEvents 提供串流模式，可將事件擷取作業與記憶體產生作業分離：

  • 持續擷取：只要有新資料，您就能持續將情境串流至 Memory Bank。
  • 重複資料刪除：Memory Bank 會根據共用的事件 ID 自動偵測並忽略重疊的對話記錄。
  • 自動觸發條件：只有在符合特定條件時，系統才會自動觸發生成作業，例如達到事件計數門檻或閒置時間。
  • 非同步生成：系統會在背景處理事件。API 會立即傳回長時間執行的作業 (LRO)，您可以輪詢或監控這項作業，直到符合觸發條件並生成回憶集錦為止。

串流和長時間執行的作業 (LRO)

串流會做為獨立緩衝區，累積多個要求中的事件。每個串流都是由 scope 和 stream_id 的組合來定義。如未提供串流 ID，Memory Bank 會使用 default 做為 ID。只有在事件具有完全相同的範圍和 ID 時，才會一起暫存和「清除」事件，確保在記憶體生成期間嚴格隔離。

對同一串流進行多次擷取呼叫時，系統會傳回相同的 LRO，直到觸發並完成記憶體生成作業為止。完成後，這個 IngestEvents LRO 會包含 GenerateMemories API 呼叫的下游 LRO。後續對串流提出的任何擷取要求，都會傳回新的 LRO。

觸發記憶生成

如果未符合任何觸發條件，系統會在擷取最後一個事件的 24 小時後，自動清除串流中的所有待處理事件。這項限制可確保 Memory Bank 處理所有事件，包括已放棄的事件。

generation_trigger_config 會決定何時排清收集到的事件，並處理這些事件以產生記憶生成。您可以設定下列觸發條件：

  • 事件計數 (event_count)：累積的不重複事件數達到上限時，系統就會觸發生成作業。
  • 閒置時間 (idle_duration，以秒為單位，精確度為分鐘，也就是說，您必須指定 60 的倍數)：如果串流在指定時間後未收到新事件 (例如 "300s")，系統就會觸發生成作業。
  • 固定間隔 (fixed_interval，以秒為單位，精確度為分鐘，也就是說，您必須指定 60 的倍數)：系統會以固定節奏執行產生作業，並在每個指定時間間隔後輪詢差異事件 (例如 "300s")。
  • 強制清除 (force_flush)：立即清除所有待處理的事件。

您可以在同一個 generation_rule 中，將事件計數規則與時間規則合併。舉例來說，您可以一併設定 event_count 和 idle_duration。如果設定多個條件，只要符合其中一個條件，系統就會立即觸發生成作業 (以先發生的情況為準)。這兩項時間條件互斥，單一規則只能指定 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.agent_engines.memories.ingest_events(
    name=memory_bank.api_resource.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.agent_engines.memories.ingest_events(
    name=memory_bank.api_resource.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.agent_engines.memories.ingest_events(
    name=memory_bank.api_resource.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.agent_engines.memories.ingest_events(
    name=memory_bank.api_resource.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"}
)

範例：結合以事件為準和以時間為準的觸發條件

這個範例會在同一要求中設定兩個觸發條件。當累積 10 個不重複事件，或串流閒置 5 分鐘 (300s) 時，系統就會執行產生作業，以先到者為準。

字典

client.agent_engines.memories.ingest_events(
    name=memory_bank.api_resource.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.agent_engines.memories.ingest_events(
    name=memory_bank.api_resource.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.agent_engines.memories.ingest_events(
    name=memory_bank.api_resource.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.agent_engines.memories.ingest_events(
    name=memory_bank.api_resource.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.agent_engines.memories.ingest_events(
    name=memory_bank.api_resource.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.agent_engines.memories.ingest_events(
    name=memory_bank.api_resource.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)：套用至生成記憶體修訂版本的鍵/值組合。舉例來說，您可以為修訂版本加上資料來源標籤，然後依據該標籤篩選修訂版本。
  • 修訂版本到期日 (revision_expire_time 或 revision_ttl)：控管生成的記憶體修訂版本到期時間。如未設定到期日，Memory Bank 會使用執行個體設定或預設存留時間 (TTL) 365 天。
  • 停用修訂版本 (disable_memory_revisions)：如果 True，Memory Bank 不會為這項要求產生的記憶內容建立修訂版本。

以下範例會為產生的修訂版本加上資料來源標籤，並保留 30 天：

字典

client.agent_engines.memories.ingest_events(
    name=memory_bank.api_resource.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.agent_engines.memories.ingest_events(
    name=memory_bank.api_resource.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.agent_engines.memories.ingest_events(
    name=memory_bank.api_resource.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.agent_engines.memories.ingest_events(
    name=memory_bank.api_resource.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"}
)

後續步驟