איך מתחילים ומנהלים סשנים בשידור חי

‫Gemini Live API מאפשר אינטראקציות קוליות וטקסטואליות עם זמן טעינה נמוך. הוא מעבד זרמים רציפים של אודיו או טקסט שנקראים סשנים כדי לספק תשובות מיידיות בדיבור שנשמע אנושי. הניהול של מחזור החיים של הסשן, מהלחיצה הראשונית ועד לסיום תקין, נשלט על ידי המפתח.

בדף הזה מוסבר איך מתחילים סשן שיחה עם מודלים של Gemini באמצעות Gemini Live API. אפשר להתחיל סשן באמצעות Vertex AI Studio,‏ Google Gen AI SDK או WebSockets.

בדף הזה מוסבר גם איך:

• הארכת סשן מעבר למגבלת הזמן שמוגדרת כברירת מחדל
• המשך של סשן קודם
• עדכון הוראות המערכת במהלך סשן
• הגדרת חלון ההקשר של סשן
• הפעלת תמלול לסשן
• הצפנת נתוני סשן באמצעות מפתח הצפנה בניהול הלקוח (CMEK)

משך החיים של הסשן

בלי דחיסה של חלון ההקשר, משך השיחות עם אודיו בלבד מוגבל ל-15 דקות, ומשך השיחות עם אודיו ווידאו מוגבל ל-2 דקות בגלל מגבלות על טוקנים. חריגה מהמגבלות האלה תגרום לסיום הסשן, אבל אפשר להשתמש בדחיסת חלון ההקשר כדי להאריך את הסשנים לזמן בלתי מוגבל.

משך החיים של חיבור מוגבל לכ-10 דקות בגלל מגבלות החיבור של WebSocket. כשהחיבור מסתיים, גם הסשן מסתיים. במקרה כזה, אפשר להגדיר סשן יחיד שיישאר פעיל בכמה חיבורים באמצעות חידוש סשן. תקבלו גם GoAway הודעה לפני שהחיבור יסתיים, כדי שתוכלו לבצע פעולות נוספות.

מספר מקסימלי של סשנים בו-זמניים

בתוכנית בתשלום לפי שימוש (PayGo), אפשר להפעיל עד 1,000 סשנים בו-זמנית לכל פרויקט. המגבלה הזו לא חלה על לקוחות שמשתמשים בהקצאת משאבים לפי התפוקה שנקבעה.

התחלת סשן

בכרטיסיות הבאות מוסבר איך להתחיל סשן שיחה בזמן אמת באמצעות Vertex AI Studio,‏ Google Gen AI SDK או WebSockets:

gcloud auth application-default login
      

import asyncio
from google import genai

# Replace the PROJECT_ID and LOCATION with your Project ID and location.
client = genai.Client(vertexai=True, project="PROJECT_ID", location="LOCATION")

# Configuration
MODEL = "gemini-3.8-live"
config = {
   "response_modalities": ["audio"],
}

async def main():
   # Establish WebSocket session
   async with client.aio.live.connect(model=MODEL, config=config) as session:
       print("Session established. Sending audio...")

if __name__ == "__main__":
    asyncio.run(main())
      

import asyncio
import websockets

# Replace the PROJECT_ID and LOCATION with your Project ID and location.
PROJECT_ID = "PROJECT_ID"
LOCATION = "LOCATION"

# Authentication
token_list = !gcloud auth application-default print-access-token
ACCESS_TOKEN = token_list[0]

# Configuration
MODEL_ID = "gemini-3.8-live"
MODEL = f"projects/{PROJECT_ID}/locations/{LOCATION}/publishers/google/models/{MODEL_ID}"
config = {
   "response_modalities": ["audio"],
}

# Construct the WSS URL
HOST = f"{LOCATION}-aiplatform.googleapis.com"
URI = f"wss://{HOST}/ws/google.cloud.aiplatform.v1.LlmBidiService/BidiGenerateContent"

async def main():
   headers = {"Authorization": f"Bearer {ACCESS_TOKEN}"}

   async with websockets.connect(URI, additional_headers=headers) as ws:
       print("Session established.")

       # Send Setup (Handshake)
       await ws.send(json.dumps({
           "setup": {
               "model": MODEL,
               "generation_config": config
           }
       }))
    # Send audio/video ...

if __name__ == "__main__":
    asyncio.run(main())
      

הארכת סשן

אורך הסשן המקסימלי של שיחה הוא 10 דקות כברירת מחדל, ואפשר להאריך אותו באמצעות התראות על סיום השיחה. goAwayהתראה (BidiGenerateContentServerMessage.goAway) נשלחת ללקוח 60 שניות לפני סיום הסשן.

כדי להאריך סשן מעבר למגבלת החיבור של 10 דקות, צריך להתחבר מחדש באמצעות חידוש הסשן. כשמקבלים התראה על goAway או כשהחיבור מסתיים מסיבות אחרות, אפשר להתחיל חיבור חדש באמצעות טוקן סשן שהתקבל במהלך הסשן. הפעולה הזו תחדש את הסשן עם ההקשר שלו, ללא שינוי, בחיבור החדש. אין הגבלה על מספר הפעמים שאפשר לעשות את זה. דוגמה להמשך סשן מופיעה במאמר המשך סשן קודם.

בדוגמה הבאה אפשר לראות איך לזהות סיום קרוב של סשן על ידי האזנה להתראה goAway:

async for response in session.receive():
    if response.go_away is not None:
        # The connection will soon be terminated
        print(response.go_away.time_left)
      

המשך של סשן קודם

ממשק Gemini Live API תומך בהמשך שיחה כדי למנוע מהמשתמש לאבד את ההקשר של השיחה במהלך ניתוק קצר (לדוגמה, מעבר מ-Wi-Fi ל-5G). אפשר לחזור לסשן קודם תוך 24 שעות. המשכת הסשן מתאפשרת באמצעות אחסון נתונים במטמון, כולל הנחיות טקסט, וידאו ואודיו, והפלט מהמודל. הפרטיות ברמת הפרויקט נאכפת לגבי הנתונים האלה שנשמרו במטמון.

כברירת מחדל, חידוש הסשן מושבת. כדי להפעיל את האפשרות להמשך הסשן, צריך להגדיר את השדה sessionResumption של ההודעה BidiGenerateContentSetup. אם האפשרות הזו מופעלת, השרת שולח מעת לעת הודעות SessionResumptionUpdate שמכילות session_id ואסימון המשך. אם החיבור ל-WebSocket מתנתק, הלקוח יכול להתחבר מחדש ולכלול את פרטי הכניסה האלה בהודעת ההגדרה החדשה. השרת משחזר את ההקשר הקודם, כך שהשיחה יכולה להימשך בצורה חלקה.

חלון החידוש הוא מוגבל (בדרך כלל כ-10 דקות). אם הלקוח לא מתחבר מחדש בפרק הזמן הזה, מצב הסשן נמחק כדי לפנות משאבי שרת.

בדוגמה הבאה מתבצע חיבור לשירות, מתקבל handle של חידוש סשן, מתבצע ניתוק מדומה ואז מתבצע חיבור מחדש באמצעות ה-handle כדי לחדש את הסשן:

import asyncio
from google import genai
from google.genai import types
import websockets

# Replace the PROJECT_ID and LOCATION with your Project ID and location.
client = genai.Client(vertexai=True, project="PROJECT_ID", location="LOCATION")

# Configuration
MODEL = "gemini-3.8-live"

async def resumable_session_example():
    """Demonstrates session resumption by connecting, disconnecting, and reconnecting."""
    session_handle = None

    print("Starting a new session...")
    try:
        async with client.aio.live.connect(
            model=MODEL,
            config=types.LiveConnectConfig(
                response_modalities=["audio"],
                session_resumption=types.SessionResumptionConfig(handle=None),
            ),
        ) as session:
            await session.send_content(
                content=types.Content(role="user", parts=[types.Part(text="Hello!")])
            )
            async for message in session.receive():
                if message.session_resumption_update:
                    update = message.session_resumption_update
                    if update.resumable and update.new_handle:
                        session_handle = update.new_handle
                        print(f"Received session handle: {session_handle}")
                        # For demonstration, we break to simulate a disconnect
                        # after receiving a handle.
                        break
                if message.server_content and message.server_content.turn_complete:
                    break
    except websockets.exceptions.WebSocketException as e:
        print(f"Initial connection failed: {e}")
        return

    if not session_handle:
        print("Did not receive a session handle. Cannot demonstrate resumption.")
        return

    print(f"\nSimulating disconnect and reconnecting with handle {session_handle}...")

    try:
        async with client.aio.live.connect(
            model=MODEL,
            config=types.LiveConnectConfig(
                response_modalities=["audio"],
                session_resumption=types.SessionResumptionConfig(handle=session_handle),
            ),
        ) as session:
            print("Successfully resumed session.")
            await session.send_content(
                content=types.Content(role="user", parts=[types.Part(text="I am back!")])
            )
            async for message in session.receive():
                if message.session_resumption_update:
                    update = message.session_resumption_update
                    if update.resumable and update.new_handle:
                        session_handle = update.new_handle
                        print(f"Received updated session handle: {session_handle}")
                if message.server_content:
                    print(f"Received server content: {message.server_content}")
                    if message.server_content.turn_complete:
                        break
            print("Resumed session finished.")
    except websockets.exceptions.WebSocketException as e:
        print(f"Failed to resume session: {e}")

if __name__ == "__main__":
    asyncio.run(resumable_session_example())
      

הפעלה של חידוש סשן חלק עם מצב שקוף

כשמפעילים את האפשרות להמשך הסשן, אפשר גם להפעיל את המצב השקוף כדי שהמשתמש יוכל להמשיך את הסשן בצורה חלקה יותר. כשמפעילים את מצב השקיפות, מוחזר באופן מפורש האינדקס של הודעת הלקוח שתואם לתמונת המצב של ההקשר. כך תוכלו לזהות איזו הודעת לקוח צריך לשלוח שוב, כשממשיכים את הסשן מנקודת ההמשך.

כדי להפעיל את מצב השקיפות:

config = {
   "response_modalities": ["audio"],
   "session_resumption_config": {
    "transparent": True,
   }
}
      

הצפנת נתוני סשן באמצעות CMEK

בחידוש סשן, נתוני סשן במטמון נשמרים במצב לא פעיל למשך עד 24 שעות. כדי להגן על הנתונים האלה באמצעות מפתח משלכם במקום באמצעותGoogle Cloud-powered encryption key, אתם יכולים להשתמש בפרופיל להצגת מודעות. פרופיל למילוי בקשות מקשר מפתח Cloud KMS לפרויקט, למיקום ולטווח של API. בזמן שפרופיל ההצגה קיים, Agent Platform מצפין את הנתונים באופן אוטומטי – הגדרות הסשן ופורמט הבקשה לא משתנים.

יש תמיכה ב-CMEK של פרופיל ההגשה ל-Gemini Live API באזורים מרובים us ו-eu. אין תמיכה באזור הגלובלי.

כדי להגדיר CMEK עבור Gemini Live API:

1. מפעילים בפרויקט את Cloud KMS API ואת Agent Platform API. לפני שמתחילים
2. יוצרים מפתח Cloud KMS באזור מרובה us או eu. ראו יצירה של אוסף מפתחות ומפתח.
3. נותנים לסוכן השירות של Gemini Enterprise Agent Platform את התפקיד roles/cloudkms.cryptoKeyEncrypterDecrypter במפתח. איך מעניקים הרשאות ל-Agent Platform
4. יוצרים פרופיל להצגת מודעות שבו הערך של scope הוא GEMINI_LIVE והערך של cmekConfig.encryptionSpec.kmsKeyName הוא המפתח שלכם. איך יוצרים פרופיל להצגת מודעות

עדכון הוראות המערכת במהלך סשן

ממשק Gemini Live API מאפשר לעדכן את הוראות המערכת במהלך סשן פעיל. אפשר להשתמש בזה כדי לשנות את התשובות של המודל, למשל לשנות את שפת התשובה או את הטון.

כדי לעדכן את הוראות המערכת באמצע השיחה, אפשר לשלוח תוכן טקסט עם התפקיד system. ההוראה המעודכנת למערכת תישאר בתוקף למשך הסשן שנותר.

session.send_client_content(
      content=types.Content(
          role="system", parts=[types.Part(text="new system instruction")]
      ),
      turn_complete=False
  )
      

הגדרת חלון ההקשר של הסשן

חלון ההקשר של Gemini Live API משמש לאחסון נתונים שמוזרמים בזמן אמת (25 טוקנים לשנייה (TPS) לאודיו ו-258 TPS לווידאו) ותוכן אחר, כולל קלט טקסט והפלט מהמודל. לכל המודלים של Gemini Live API יש מגבלה של 128 אלף טוקנים בחלון ההקשר.

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

כשדחיסת חלון ההקשר מופעלת, היא משתמשת בחלון הזזה בצד השרת כדי לקטוע את התורות הכי ישנות. כשהאסימונים המצטברים חורגים מאורך מקסימלי מוגדר (שמוגדר באמצעות פס ההזזה גודל התוכן המקסימלי ב-Vertex AI Studio, או trigger_tokens ב-API), השרת מצמצם באופן אוטומטי את התפניות הכי ישנות או מסכם אותן כדי לשמור על ההקשר במסגרת המגבלה. ב-ContextWindowCompressionConfig, אפשר להגדיר מנגנון של חלון נע ואת מספר הטוקנים שמוגדרים בפרמטר target_tokens שגורם לדחיסה.

כך, מבחינת המשתמש, משך הסשן יכול להיות אינסופי כי הזיכרון מנוהל כל הזמן. בלי דחיסה, יכול להיות שהמגבלות יחולו על סשנים עם אודיו בלבד אחרי כ-15 דקות.

האורך המינימלי והמקסימלי של חלון ההקשר וגודל היעד הם:

כדי להגדיר את חלון ההקשר:

config = {
   "response_modalities": ["audio"],
   # Configures compression
   "context_window_compression" : {
    "trigger_tokens": 10000,
    "sliding_window": {"target_tokens" : 512}
   }
}
      

הפעלת תמלול אודיו בסשן

אתם יכולים להפעיל תמלול גם לאודיו של הקלט וגם לאודיו של הפלט.

כדי לקבל תמלילים, צריך לעדכן את הגדרות הסשן. צריך להוסיף את האובייקטים input_audio_transcription ו-output_audio_transcription ולוודא ש-text כלול ב-response_modalities.

כדי לשפר את איכות התמלול של זיהוי דיבור (ASR) אוטומטי רב-לשוני, אפשר לספק רמזים לגבי השפה באמצעות השדה language_codes בתוך input_audio_transcription או output_audio_transcription. מומלץ לספק רמזים כדי לשפר את איכות התמלול, כי זה מפחית את הסיכון לזיהוי שפה שגוי, במיוחד בהנחיות קצרות. בשדה language_codes אפשר לציין רשימה של קודי שפה בתקן BCP-47 (לדוגמה, "en-US",‏ "es-US").

config = {
    "response_modalities": ["audio", "text"],
    "input_audio_transcription": {
        "language_codes": ["en-US"]
    },
    "output_audio_transcription": {},
}

מעבד את התשובה

בדוגמת הקוד הבאה אפשר לראות איך להתחבר באמצעות הסשן שהוגדר ולחלץ את חלקי הטקסט (תמלילים) לצד נתוני האודיו.

# Receive Output Loop
async for message in session.receive():
    server_content = message.server_content
    if server_content:
        # Handle Model Turns (Audio + Text)
        model_turn = server_content.model_turn
        if model_turn and model_turn.parts:
            for part in model_turn.parts:
                # Handle Text (Transcriptions)
                if part.text:
                    print(f"Transcription: {part.text}")
                # Handle Audio
                if part.inline_data:
                    audio_data = part.inline_data.data
                    # Process audio bytes...
                    pass

        # Check for turn completion
        if server_content.turn_complete:
            print("Turn complete.")

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

• שליחת סטרימינג של אודיו ווידאו
• שיטות מומלצות לשימוש ב-Gemini Live API
• עיצוב פרומפטים מולטי-מודאליים
• מבוא לקריאה להפעלת פונקציות