‫Google משתמשת בטכנולוגיית AI כדי לתרגם תוכן לשפה המועדפת עליך. בתרגומים כאלו עשויות להיות שגיאות.

סטרימינג דו-כיווני

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

סקירה כללית

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

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

Google Gen AI SDK תומך בפריסה של סוכן מרוחק עם שיטות שאילתות דו-כיווניות. כדי לפרוס נציג עם יכולת דו-כיוונית, מגדירים את מצב שרת הנציג EXPERIMENTAL כשמשתמשים ב-SDK או כששולחים קריאה ל-Agent Platform API.

פיתוח סוכן

כשמפתחים נציג, צריך לפעול לפי השלבים הבאים כדי להטמיע סטרימינג דו-כיווני:

הגדרה של שיטת שאילתה דו-כיוונית להעברה בסטרימינג
בדיקת שיטת השאילתה של סטרימינג דו-כיווני
רישום שיטות בהתאמה אישית (אופציונלי)

הגדרה של שיטת שאילתה דו-כיוונית להעברה בסטרימינג

כדי שהנציג יוכל לתמוך ב-bidi, צריך להגדיר שיטת bidi_stream_query שמקבלת באופן אסינכרוני בקשות סטרימינג כקלט ומוציאה תגובות סטרימינג. לדוגמה, התבנית הבאה מרחיבה את התבנית הבסיסית כדי להעביר בסטרימינג בקשות ותשובות, ואפשר לפרוס אותה ב-Gemini Enterprise Agent Platform:

import asyncio
from typing import Any, AsyncIterable

class BidiStreamingAgent(StreamingAgent):

    async def bidi_stream_query(
        self,
        request_queue: asyncio.Queue[Any]
    ) -> AsyncIterable[Any]:
        from langchain.load.dump import dumpd

        while True:
            request = await request_queue.get()
            # This is just an illustration, you're free to use any termination mechanism.
            if request == "END":
                break
            for chunk in self.graph.stream(request):
                yield dumpd(chunk)

agent = BidiStreamingAgent(
    model=model,                # Required.
    tools=[get_exchange_rate],  # Optional.
    project="PROJECT_ID",
    location="LOCATION",
)
agent.set_up()

כשמשתמשים ב-API לסטרימינג דו-כיווני, חשוב לזכור את הדברים הבאים:

‫asyncio.Queue: אתם יכולים להוסיף כל סוג נתונים לתור הבקשות הזה כדי להמתין לשליחה ל-API של המודל.
זמן קצוב מקסימלי לתפוגה: הזמן הקצוב המקסימלי לתפוגה של שאילתת סטרימינג דו-כיוונית הוא 10 דקות. אם לנציג שלכם נדרש זמן עיבוד ארוך יותר, כדאי לפצל את המשימה לחלקים קטנים יותר ולהשתמש בסשן או בזיכרון כדי לשמור על המצב.
ויסות צריכת התוכן: כשצורכים תוכן מסטרימינג דו-כיווני, חשוב לנהל את הקצב שבו הנציג מעבד את הנתונים הנכנסים. אם הסוכן צורך נתונים לאט מדי, זה עלול לגרום לבעיות כמו זמן אחזור ארוך יותר או עומס על הזיכרון בצד השרת. הטמיעו מנגנונים לשליפת נתונים באופן פעיל כשהנציג מוכן לעבד אותם, והימנעו מפעולות חסימה שעלולות לעצור את צריכת התוכן.
הגבלת קצב יצירת התוכן: אם אתם נתקלים בבעיות של לחץ נגדי (מצב שבו המפיק יוצר נתונים מהר יותר מהקצב שבו הצרכן יכול לעבד אותם), כדאי להגביל את קצב יצירת התוכן. כך אפשר למנוע גלישת מאגר נתונים (buffer overflow) וליהנות מחוויית סטרימינג חלקה.

בדיקת השיטה של שאילתת סטרימינג דו-כיוונית

כדי לבדוק את השאילתה של הסטרימינג הדו-כיווני באופן מקומי, אפשר להפעיל את השיטה bidi_stream_query ולעבור על התוצאות:

import asyncio
import pprint
import time

request_queue = asyncio.Queue()

async def generate_input():
    # This is just an illustration, you're free to use any appropriate input generator.
    request_queue.put_nowait(
        {"input": "What is the exchange rate from US dolloars to Swedish currency"}
    )
    time.sleep(5)
    request_queue.put_nowait(
        {"input": "What is the exchange rate from US dolloars to Euro currency"}
    )
    time.sleep(5)
    request_queue.put_nowait("END")

async def print_query_result():
    async for chunk in agent.bidi_stream_query(request_queue):
        pprint.pprint(chunk, depth=1)

input_task = asyncio.create_task(generate_input())
output_task = asyncio.create_task(print_query_result())

await asyncio.gather(input_task, output_task, return_exceptions=True)

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

{'actions': [...], 'messages': [...]}
{'messages': [...], 'steps': [...]}
{'messages': [...], 'output': 'The exchange rate from US dollars to Swedish currency is 1 USD to 10.5751 SEK. \n'}
{'actions': [...], 'messages': [...]}
{'messages': [...], 'steps': [...]}
{'messages': [...], 'output': 'The exchange rate from US dollars to Euro currency is 1 USD to 0.86 EUR. \n'}

אופציונלי: רישום שיטות בהתאמה אישית

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

from typing import AsyncIterable, Iterable

class CustomAgent(BidiStreamingAgent):

    # ... same get_state and get_state_history function definition.

    async def get_state_bidi_mode(
        self,
        request_queue: asyncio.Queue[Any]
    ) -> AsyncIterable[Any]:
        while True:
            request = await request_queue.get()
            if request == "END":
                break
            yield self.graph.get_state(request)._asdict()

    def register_operations(self):
        return {
            # The list of synchrounous operations to be registered
            "": ["query", "get_state"]
            # The list of streaming operations to be registered
            "stream": ["stream_query", "get_state_history"]
            # The list of bidi streaming operations to be registered
            "bidi_stream": ["bidi_stream_query", "get_state_bidi_mode"]
        }

פריסת סוכן

אחרי שמפתחים את הנציג בתור live_agent, אפשר לפרוס אותו ב-Agent Platform על ידי יצירת מכונה של Agent Platform.

חשוב לדעת: כשיוצרים את המכונה של פלטפורמת הסוכן באמצעות GenAI SDK, כל הגדרות הפריסה (חבילות נוספות ואמצעי בקרה מותאמים אישית על המשאבים) מוקצות כערך של config.

מאתחלים את הלקוח של GenAI:

import agentplatform
from agentplatform import types as agentplatform_types

client = agentplatform.Client(project=PROJECT, location=LOCATION)

פורסים את הנציג ב-Agent Platform. שימו לב שצריך להגדיר את EXPERIMENTAL agent_server_mode לנציג שתומך בסטרימינג דו-כיווני:

remote_live_agent = client.runtimes.create(
    agent=live_agent,
    config={
        "staging_bucket": STAGING_BUCKET,
        "requirements": [
            "google-cloud-agentplatform[runtimes,adk]==2.0.1",
            "cloudpickle==3.0",
            "websockets"
        ],
        "agent_server_mode": agentplatform_types.AgentServerMode.EXPERIMENTAL,
    },
)

במאמר יצירת מכונה של Agent Runtime מוסבר מה קורה ברקע במהלך הפריסה.

משיגים את מזהה המשאב של הסוכן:

remote_live_agent.api_resource.name

שימוש בסוכן

אם הגדרתם פעולת bidi_stream_query כשפיתחתם את הנציג, אתם יכולים לשלוח שאילתות לנציג בסטרימינג דו-כיווני באופן אסינכרוני באמצעות GenAI SDK for Python.

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

async with client.aio.live.runtimes.connect(
        runtime=remote_live_agent.api_resource.name,
        config={"class_method": "bidi_stream_query"}
        ) as connection:
    while True:
        #
        input_str = input("Enter your question: ")
        if input_str == "exit":
            break
        await connection.send({"input": input_str})

        while True:
            response = await connection.receive()
            print(response)
            if response["bidiStreamOutput"]["output"] == "end of turn":
                break

‫Agent Runtime מעביר תשובות בסטרימינג כרצף של אובייקטים שנוצרים באופן איטרטיבי. לדוגמה, קבוצה של שתי תשובות בסבב הראשון עשויה להיראות כך:

Enter your next question: Weather in San Diego?
{'bidiStreamOutput': {'output': "FunctionCall: {'name': 'get_current_weather', 'args': {'location': 'San Diego'}}\n"}}
{'bidiStreamOutput': {'output': 'end of turn'}}

Enter your next question: exit

שימוש בסוכן של ערכה לפיתוח סוכנים (ADK)

אם פיתחתם את הנציג שלכם באמצעות ערכת פיתוח הנציגים (ADK), אתם יכולים להשתמש בסטרימינג דו-כיווני כדי לקיים אינטראקציה עם Gemini Live API.

בדוגמה הבאה נוצר סוכן בממשק שיחה שמקבל שאלות טקסט מהמשתמש ומקבל נתוני אודיו בתגובה מ-Gemini Live API:

import numpy as np
from google.adk.agents.live_request_queue import LiveRequest
from google.adk.events import Event
from google.genai import types

def prepare_live_request(input_text: str) -> LiveRequest:
    part = types.Part.from_text(text=input_text)
    content = types.Content(parts=[part])
    return LiveRequest(content=content)

async with client.aio.live.runtimes.connect(
        runtime=remote_live_agent.api_resource.name,
        config={
            "class_method": "bidi_stream_query",
            "input": {"input_str": "hello"},
        }) as connection:
    first_req = True
    while True:
        input_text = input("Enter your question: ")
        if input_text == "exit":
            break
        if first_req:
            await connection.send({
                "user_id": USER_ID,
                "live_request": prepare_live_request(input_text).dict()
            })
            first_req = False
        else:
            await connection.send(prepare_live_request(input_text).dict())
        audio_data = []
        while True:
            async def receive():
                return await connection.receive()

            receiving = asyncio.Task(receive())
            done, _ = await asyncio.wait([receiving])
            if receiving not in done:
                receiving.cancel()
                break
            event = Event.model_validate(receiving.result()["bidiStreamOutput"])
            part = event.content and event.content.parts and event.content.parts[0]

            if part.inline_data and part.inline_data.data:
                chunk_data = part.inline_data.data
                data = np.frombuffer(chunk_data, dtype=np.int16)
                audio_data.append(data)
            else:
                print(part)
        if audio_data:
            concatenated_audio = np.concatenate(audio_data)
            display(Audio(concatenated_audio, rate=24000, autoplay=True))