Lo streaming bidirezionale su Agent Runtime consente una comunicazione persistente e bidirezionale tra la tua applicazione e un agente, andando oltre i modelli di richiesta-risposta convenzionali. Questo documento spiega come sviluppare, testare ed eseguire il deployment di agenti di streaming bidirezionale per casi d'uso in tempo reale come l'interazione audio o video.
Panoramica
Lo streaming bidirezionale fornisce un canale di comunicazione bidirezionale persistente tra la tua applicazione e l'agente, consentendoti di andare oltre i pattern di richiesta-risposta basati sui turni. Lo streaming bidirezionale è utile per i casi d'uso in cui l'agente deve elaborare le informazioni e rispondere continuamente, ad esempio interagendo con input audio o video con bassa latenza.
Lo streaming bidirezionale con Agent Runtime supporta casi d'uso interattivi degli agenti in tempo reale e lo scambio di dati per le API Multimodal Live. Lo streaming bidirezionale è supportato per tutti i framework e i metodi di streaming bidirezionale personalizzati sono disponibili tramite la registrazione di metodi personalizzati. Puoi utilizzare lo streaming bidirezionale per interagire con l'API Gemini Live direttamente o utilizzando l'Agent Development Kit (ADK) su Agent Platform.
Il deployment di un agente remoto con metodi di query bidirezionali è ben supportato dall'SDK Google Gen AI. Per eseguire il deployment di un agente con funzionalità bidirezionali, imposta la modalità server dell'agente EXPERIMENTAL quando utilizzi l'SDK o chiami l'API Agent Platform.
Sviluppa un agente
Durante lo sviluppo di un agente, segui questi passaggi per implementare lo streaming bidirezionale:
Definisci un metodo di query di flussi di dati bidirezionali
Registra i metodi personalizzati (facoltativo)
Definisci un metodo di query di streaming bidirezionale
Per rendere l'agente "compatibile con la comunicazione bidirezionale", devi definire un metodo bidi_stream_query che accetti in modo asincrono le richieste di stream come input e restituisca risposte in streaming. Ad esempio, il seguente template estende il template di base per trasmettere in streaming richieste e risposte ed è implementabile su 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()
Quando utilizzi l'API di streaming bidirezionale, tieni presente quanto segue:
asyncio.Queue: puoi inserire qualsiasi tipo di dati in questa coda di richieste per attendere l'invio all'API del modello.Timeout massimo: il timeout massimo per la query di flussi di dati bidirezionali è di 10 minuti. Se l'agente richiede tempi di elaborazione più lunghi, valuta la possibilità di suddividere l'attività in parti più piccole e di utilizzare la sessione o la memoria per mantenere lo stato persistente.
Limitare il consumo di contenuti: quando utilizzi contenuti di uno stream bidirezionale, è importante gestire la velocità con cui l'agente elabora i dati in entrata. Se l'agente consuma i dati troppo lentamente, possono verificarsi problemi come un aumento della latenza o della pressione sulla memoria lato server. Implementa meccanismi per estrarre attivamente i dati quando l'agente è pronto a elaborarli ed evita di bloccare le operazioni che potrebbero interrompere la fruizione dei contenuti.
Limita la generazione di contenuti: se riscontri problemi di contropressione (in cui il producer genera dati più velocemente di quanto il consumer possa elaborarli), devi limitare la frequenza di generazione dei contenuti. In questo modo, puoi evitare l'overflow del buffer e garantire un'esperienza di streaming senza interruzioni.
Testa il metodo di query di streaming bidirezionale
Puoi testare la query di streaming bidirezionale localmente chiamando il metodo bidi_stream_query e ripetendo i risultati:
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)
La stessa connessione di query bidirezionale può gestire più richieste e risposte. Per ogni nuova richiesta dalla coda, l'esempio seguente genera uno stream di blocchi contenenti informazioni diverse sulla risposta:
{'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'}
(Facoltativo) Registra i metodi personalizzati
Le operazioni possono essere registrate come modalità di esecuzione standard (rappresentata da una stringa vuota ""), in streaming (stream) o in streaming bidirezionale (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"]
}
Esegui il deployment di un agente
Dopo aver sviluppato l'agente come live_agent, puoi eseguirne il deployment su Agent Platform creando un'istanza di Agent Platform.
Tieni presente che con l'SDK GenAI, tutte le configurazioni di deployment (pacchetti aggiuntivi e controlli delle risorse personalizzati) vengono assegnate come valore di config durante la creazione dell'istanza di Agent Platform.
Inizializza il client GenAI:
import agentplatform
from agentplatform import types as agentplatform_types
client = agentplatform.Client(project=PROJECT, location=LOCATION)
Esegui il deployment dell'agente su Agent Platform. Tieni presente che EXPERIMENTAL
agent_server_mode è obbligatorio per un agente che supporta lo streaming bidirezionale:
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,
},
)
Per informazioni sui passaggi che si verificano in background durante il deployment, consulta Crea un'istanza di runtime dell'agente.
Recupera l'ID risorsa dell'agente:
remote_live_agent.api_resource.name
Utilizza un agente
Se hai definito un'operazione bidi_stream_query durante lo sviluppo dell'agente, puoi eseguire query sull'agente in modalità di streaming bidirezionale in modo asincrono utilizzando l'SDK GenAI per Python.
Puoi modificare l'esempio seguente con qualsiasi dato riconoscibile dall'agente, utilizzando qualsiasi logica di terminazione applicabile per lo stream di input e lo stream di output:
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 trasmette le risposte come una sequenza di oggetti generati in modo iterativo. Ad esempio, un insieme di due risposte nel primo turno potrebbe essere il seguente:
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
Utilizza un agente Agent Development Kit
Se hai sviluppato l'agente utilizzando Agent Development Kit (ADK), puoi utilizzare lo streaming bidirezionale per interagire con l'API Gemini Live.
L'esempio seguente crea un agente conversazionale che accetta domande di testo dell'utente e riceve i dati audio della risposta dell'API Gemini Live:
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))