Google verwendet KI-Technologie, um Inhalte in Ihre bevorzugte Sprache zu übersetzen. KI-Übersetzungen können Fehler enthalten.

Bidirektionales Streaming

Bidirektionales Streaming in Agent Runtime ermöglicht eine dauerhafte, bidirektionale Kommunikation zwischen Ihrer Anwendung und einem Agenten, die über herkömmliche Anfrage-Antwort-Muster hinausgeht. In diesem Dokument wird beschrieben, wie Sie bidirektionale Streaming-Agenten für Echtzeitanwendungsfälle wie Audio- oder Videointeraktionen entwickeln, testen und bereitstellen.

Übersicht

Das bidirektionale Streaming bietet einen dauerhaften, bidirektionalen Kommunikationskanal zwischen Ihrer Anwendung und dem Agenten, sodass Sie über rundenbasierte Anfrage-Antwort-Muster hinausgehen können. Bidirektionales Streaming eignet sich für Anwendungsfälle, in denen Ihr Agent Informationen verarbeiten und kontinuierlich antworten muss, z. B. bei der Interaktion mit Audio- oder Videoeingaben mit niedriger Latenz.

Bidirektionales Streaming mit Agent Runtime unterstützt interaktive Echtzeit-Anwendungsfälle für Agenten und den Datenaustausch für multimodale Live APIs. Bidirektionales Streaming wird für alle Frameworks unterstützt. Benutzerdefinierte bidirektionale Streamingmethoden sind über die Registrierung benutzerdefinierter Methoden verfügbar. Sie können bidirektionales Streaming verwenden, um direkt oder über das Agent Development Kit (ADK) auf der Agent Platform mit der Gemini Live API zu interagieren.

Das Google Gen AI SDK unterstützt die Bereitstellung eines Remote-Agenten mit bidirektionalen Abfragemethoden. Wenn Sie einen bidirektionalen Agenten bereitstellen möchten, legen Sie den Agentenservermodus EXPERIMENTAL fest, wenn Sie das SDK verwenden oder die Agent Platform API aufrufen.

KI-Agenten entwickeln

Führen Sie die folgenden Schritte aus, um bidirektionales Streaming zu implementieren, während Sie einen Agenten entwickeln:

Bidirektionale Streaming-Abfragemethode definieren
Bidirektionale Streaming-Abfragemethode testen
Benutzerdefinierte Methoden registrieren (optional)

Bidirektionale Streaming-Abfragemethode definieren

Damit Ihr Agent „bidi-fähig“ ist, müssen Sie eine bidi_stream_query-Methode definieren, die Streamanfragen asynchron als Eingabe akzeptiert und Streamingantworten ausgibt. Die folgende Vorlage erweitert beispielsweise die einfache Vorlage, um Anfragen und Antworten zu streamen, und kann auf der Gemini Enterprise Agent Platform bereitgestellt werden:

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

Beachten Sie bei der Verwendung der bidirektionalen Streaming API Folgendes:

asyncio.Queue: Sie können jeden Datentyp in diese Anfragewarteschlange stellen, um auf das Senden an die Modell-API zu warten.
Maximale Zeitüberschreitung: Die maximale Zeitüberschreitung für bidirektionale Streamingabfragen beträgt 10 Minuten. Wenn Ihr Agent längere Verarbeitungszeiten benötigt, sollten Sie die Aufgabe in kleinere Teile aufteilen und den Status mithilfe von Sitzungen oder gemerkten Informationen beibehalten.
Inhaltsnutzung drosseln: Wenn Sie Inhalte aus einem bidirektionalen Stream nutzen, ist es wichtig, die Rate zu verwalten, mit der Ihr Agent eingehende Daten verarbeitet. Wenn Ihr Agent Daten zu langsam verarbeitet, kann dies zu Problemen wie erhöhter Latenz oder Arbeitsspeicherauslastung auf der Serverseite führen. Implementieren Sie Mechanismen, um Daten aktiv abzurufen, wenn Ihr Agent bereit ist, sie zu verarbeiten. Vermeiden Sie blockierende Vorgänge, die die Nutzung von Inhalten unterbrechen könnten.
Inhaltserstellung drosseln: Wenn Probleme mit Backpressure auftreten (der Producer generiert Daten schneller, als der Consumer sie verarbeiten kann), sollten Sie die Rate der Inhaltserstellung drosseln. So können Pufferüberläufe verhindert und ein reibungsloses Streaming ermöglicht werden.

Methode für bidirektionale Streamingabfragen testen

Sie können die bidirektionale Streamingabfrage lokal testen, indem Sie die Methode bidi_stream_query aufrufen und die Ergebnisse durchlaufen:

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)

Über dieselbe bidirektionale Abfrageverbindung können mehrere Anfragen und Antworten verarbeitet werden. Im folgenden Beispiel wird für jede neue Anfrage aus der Warteschlange ein Stream von Chunks mit verschiedenen Informationen zur Antwort generiert:

{'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'}

Optional: Benutzerdefinierte Methoden registrieren

Vorgänge können als Standardvorgänge (dargestellt durch einen leeren String ""), Streamingvorgänge (stream) oder bidirektionale Streamingvorgänge (bidi_stream) registriert werden.

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"]
        }

KI-Agenten bereitstellen

Nachdem Sie Ihren Agenten als live_agent entwickelt haben, können Sie ihn auf der Agent Platform bereitstellen, indem Sie eine Agent Platform-Instanz erstellen.

Beachten Sie, dass beim GenAI SDK alle Bereitstellungskonfigurationen (zusätzliche Pakete und benutzerdefinierte Ressourcenkontrollen) beim Erstellen der Agent Platform-Instanz als Wert von config zugewiesen werden.

Initialisieren Sie den GenAI-Client:

import agentplatform
from agentplatform import types as agentplatform_types

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

Stellen Sie den Agenten auf der Agent Platform bereit. Beachten Sie, dass der EXPERIMENTAL agent_server_mode für einen Agenten erforderlich ist, der bidirektionales Streaming unterstützt:

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,
    },
)

Informationen zu den Schritten, die während der Bereitstellung im Hintergrund ausgeführt werden, finden Sie unter Agent Runtime-Instanz erstellen.

Rufen Sie die Ressourcen-ID des KI-Agenten ab:

remote_live_agent.api_resource.name

KI-Agent verwenden

Wenn Sie beim Entwickeln Ihres KI-Agenten einen bidi_stream_query-Vorgang definiert haben, können Sie den KI-Agenten mit dem GenAI SDK for Python asynchron bidirektional streamen.

Sie können das folgende Beispiel mit allen Daten ändern, die von Ihrem Agenten erkannt werden, und dabei eine beliebige Beendigungslogik für den Eingabe- und Ausgabestream verwenden:

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 streamt Antworten als Sequenz iterativ generierter Objekte. Ein Set aus zwei Antworten in der ersten Runde könnte beispielsweise so aussehen:

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

Agent Development Kit-Agenten verwenden

Wenn Sie Ihren Agenten mit dem Agent Development Kit (ADK) entwickelt haben, können Sie bidirektionales Streaming verwenden, um mit der Gemini Live API zu interagieren.

Im folgenden Beispiel wird ein Konversations-Agent erstellt, der Textfragen von Nutzern entgegennimmt und Audiodaten als API-Antwort von der Gemini Live API erhält:

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