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

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

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

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

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

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

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

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

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

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

התחלת סשן

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

המסוף

  1. פותחים את Vertex AI Studio > Stream realtime.
  2. כדי להתחיל את השיחה, לוחצים על התחלת סשן.

כדי לסיים את הסשן, לוחצים על סיום הסשן.

Python

לפני שמתחילים, צריך לבצע אימות ב-Gemini Enterprise Agent Platform באמצעות מפתח API או פרטי כניסה שמוגדרים כברירת מחדל באפליקציה (ADC): 

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-live-2.5-flash-native-audio"
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())

Python

כשמשתמשים ב-WebSockets, החיבור נוצר באמצעות לחיצת יד רגילה של WebSocket. נקודת הקצה היא אזורית ומשתמשת באסימוני bearer מסוג OAuth 2.0 לצורך אימות. בתרחיש הזה, אסימון האימות מועבר בדרך כלל בכותרות של WebSocket (למשל Authorization: Bearer [TOKEN]). 

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-live-2.5-flash-native-audio"
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:

Python

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 כדי לחדש את הסשן:

Python

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-live-2.5-flash-native-audio"

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())

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

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

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

Python

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

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

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

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

Python

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 דקות לפני שתגיעו למגבלות הקשיחות.

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

הגדרה (דגל API) ערך מינימלי ערך מקסימלי
אורך חלון ההקשר המקסימלי (trigger_tokens) 5,000 ‫128,000
גודל ההקשר של היעד (target_tokens) 0 ‫128,000

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

המסוף

  1. פותחים את Vertex AI Studio > Stream realtime.
  2. לוחצים כדי לפתוח את התפריט מתקדם.
  3. בקטע Session Context, משתמשים בפס ההזזה Max context size כדי להגדיר את גודל ההקשר לערך בין 5,000 ל-128,000.
  4. (אופציונלי) באותו קטע, משתמשים בפס ההזזה גודל משטח המגע כדי להגדיר את גודל משטח המגע לערך בין 0 ל-128,000.

Python

מגדירים את השדות context_window_compression.trigger_tokens ו-context_window_compression.sliding_window.target_tokens בהודעת ההגדרה: 

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

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