הטמעת אירועים

‫IngestEvents API מפריד בין קליטת אירועים לבין יצירת זיכרון ב-Memory Bank. אתם יכולים להזרים תוכן ללא הפסקה אל Memory Bank, והיצירה תתבצע אוטומטית כשהקריטריונים להפעלה יתקיימו.

כדי לבצע את השלבים שמוצגים במדריך הזה, קודם צריך לבצע את השלבים במאמר הגדרה של Memory Bank.

בדוגמאות שבדף הזה נעשה שימוש ב-SDK של agentplatform, שמספק את אפשרויות ההגדרה העדכניות ביותר של IngestEvents. מאתחלים את הלקוח באמצעות agentplatform:

import agentplatform

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

הסבר על הטמעת נתונים של אירועים

עם GenerateMemories API, תהליך יצירת זיכרון מופעל באופן מיידי. השיטה הזו מאפשרת לכם שליטה ישירה בתזמון של יצירת הזיכרון. ‫IngestEvents מציעה דפוס סטרימינג שמפריד בין קליטת אירועים לבין יצירת זיכרון:

  • הזנת נתונים רציפה: אתם יכולים להזרים הקשר ל-Memory Bank באופן רציף בכל פעם שנתונים חדשים זמינים.
  • ביטול כפילויות: Memory Bank מזהה באופן אוטומטי היסטוריית שיחות חופפת ומתעלם ממנה על סמך מזהי אירועים משותפים.
  • טריגרים אוטומטיים: היצירה מופעלת באופן אוטומטי רק כשמתקיימים תנאים ספציפיים, כמו סף של ספירת אירועים או זמן השהיה.
  • יצירה אסינכרונית: האירועים מעובדים ברקע. ה-API מחזיר באופן מיידי פעולה ממושכת (LRO), שאפשר לדגום או לעקוב אחריה עד שתנאי ההפעלה מתקיימים והזיכרונות נוצרים.

סטרימינג ופעולות ממושכות (LRO)

הסטרימים פועלים כמאגרי נתונים זמניים מבודדים שמצטברים בהם אירועים מכמה בקשות. כל זרם מוגדר באופן ייחודי על ידי השילוב של scope ו-stream_id. אם לא תספקו מזהה של זרם, Memory Bank ישתמש ב-default כמזהה. אירועים מועברים למצב ביניים ומועברים יחד רק אם יש להם בדיוק את אותו היקף ואותו מזהה, כדי להבטיח בידוד מוחלט במהלך יצירת הזיכרון.

קריאות מרובות להעברה לאותו הזרם מחזירות את אותו LRO עד שהפעלת יצירת הזיכרון מסתיימת. אחרי השלמת הפעולה, ה-LRO‏ IngestEvents מכיל את ה-LRO במורד הזרם של קריאה ל-API‏ GenerateMemories. כל בקשה נוספת להוספת נתונים לזרם תחזיר 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 כדי לבטל כפילויות בנתונים באופן אוטומטי. כשמספקים מזהה ייחודי לכל אירוע, רק אירועים חדשים מתווספים למאגר.

העברת הקשר בין חלונות יצירה

כברירת מחדל, בכל חלון יצירה מתבצע עיבוד רק של האירועים שהצטברו מאז השטיפה הקודמת. כדי לשמור על עקביות הזיכרונות בחלונות יצירה עוקבים, אפשר להעביר את ההקשר מחלון אחד לחלון הבא על ידי הגדרת overlap_event_count בgeneration_rule. האירועים האלה כוללים מחדש את N האירועים האחרונים שכבר עברו עיבוד בתחילת החלון הבא.

אירועים שהועברו לא נספרים כחלק מהתנאי להפעלת הטריגר. הערך של overlap_event_count חייב להיות מספר שלם חיובי שקטן מ-event_count, וצריך לציין את event_count או כלל שמבוסס על זמן (idle_duration או fixed_interval).

מקור הנתונים

עם IngestEvents, אתם מספקים תוכן על ידי שליחת אירועים ישירות בבקשה של האפליקציה באמצעות direct_contents_source במטען ייעודי (payload) ישיר.

בדוגמאות הבאות מוצגים טריגרים שונים.

דוגמה: טריגר של ספירת אירועים

בדוגמה הזו נעשה שימוש בזרם מותאם אישית, והטריגר מופעל כשמצטברים בזרם 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"}
)

דוגמה: שילוב של טריגרים מבוססי-אירועים וטריגרים מבוססי-זמן

בדוגמה הזו מוגדרים שני תנאי טריגר באותה בקשה. היצירה מתבצעת כשמצטברים 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): צמדי מפתח/ערך שמוחלים על הגרסאות של הזיכרון שנוצר. לדוגמה, אפשר להוסיף תווית לגרסה עם מקור הנתונים שלה, ומאוחר יותר לסנן את הגרסאות לפי התווית הזו.
  • תפוגה של גרסאות (revision_expire_time או revision_ttl): קובעת מתי יפוגו הגרסאות של הזיכרונות שנוצרו. אם לא מגדירים תאריך תפוגה, Memory Bank משתמש בהגדרות המכונה או באורך החיים (TTL) שמוגדר כברירת מחדל של 365 ימים.
  • השבתת העריכות (disable_memory_revisions): אם True, Memory Bank לא יוצר עריכות לזיכרונות שנוצרו מהבקשה הזו.

בדוגמה הבאה, התוויות מציינות את מקור הנתונים של הגרסאות שנוצרו, והן נשמרות למשך 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"}
)

המאמרים הבאים