Streaming dua arah di Agent Runtime memungkinkan komunikasi dua arah yang persisten antara aplikasi Anda dan agen, melampaui pola permintaan-respons konvensional. Dokumen ini menjelaskan cara mengembangkan, menguji, dan men-deploy agen streaming dua arah untuk kasus penggunaan real-time seperti interaksi audio atau video.
Ringkasan
Streaming dua arah menyediakan saluran komunikasi dua arah yang persisten antara aplikasi dan agen, sehingga Anda dapat beralih dari pola permintaan-respons berbasis giliran. Streaming dua arah berfungsi untuk kasus penggunaan saat agen Anda perlu memproses informasi dan merespons secara terus-menerus, seperti berinteraksi dengan input audio atau video dengan latensi rendah.
Streaming dua arah dengan Agent Runtime mendukung kasus penggunaan agen interaktif dan real-time serta pertukaran data untuk API live multimodal. Streaming dua arah didukung untuk semua framework, dan metode streaming dua arah kustom tersedia melalui pendaftaran metode kustom. Anda dapat menggunakan streaming dua arah untuk berinteraksi dengan Gemini Live API secara langsung atau menggunakan Agent Development Kit (ADK) di Agent Platform.
Men-deploy agen jarak jauh dengan metode kueri dua arah didukung dengan baik oleh Google Gen AI SDK. Untuk men-deploy agen yang mendukung komunikasi dua arah, tetapkan mode server agen EXPERIMENTAL saat menggunakan SDK atau memanggil Agent Platform API.
Mengembangkan agen
Saat mengembangkan agen, gunakan langkah-langkah berikut untuk menerapkan streaming dua arah:
Mendaftarkan metode kustom (opsional)
Menentukan metode kueri streaming dua arah
Agar agen Anda "mendukung bidi", Anda perlu menentukan metode bidi_stream_query yang secara asinkron mengambil permintaan streaming sebagai input dan menghasilkan respons streaming. Sebagai contoh, template berikut memperluas template dasar untuk men-streaming permintaan dan respons serta dapat di-deploy di 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()
Perhatikan hal-hal berikut saat menggunakan API streaming dua arah:
asyncio.Queue: Anda dapat memasukkan jenis data apa pun ke dalam antrean permintaan ini untuk menunggu dikirim ke API model.Waktu tunggu maksimum: Waktu tunggu maksimum untuk kueri streaming dua arah adalah 10 menit. Jika agen Anda memerlukan waktu pemrosesan yang lebih lama, pertimbangkan untuk membagi tugas menjadi bagian-bagian yang lebih kecil dan gunakan sesi atau memori untuk mempertahankan status.
Membatasi konsumsi konten: Saat menggunakan konten dari streaming dua arah, penting untuk mengelola kecepatan agen Anda dalam memproses data yang masuk. Jika agen Anda menggunakan data terlalu lambat, hal ini dapat menyebabkan masalah seperti peningkatan latensi atau tekanan memori di sisi server. Terapkan mekanisme untuk menarik data secara aktif saat agen Anda siap memprosesnya, dan hindari operasi pemblokiran yang dapat menghentikan konsumsi konten.
Membatasi pembuatan konten: Jika Anda mengalami masalah tekanan balik (ketika produsen menghasilkan data lebih cepat daripada yang dapat diproses oleh konsumen), Anda harus membatasi kecepatan pembuatan konten. Hal ini dapat membantu mencegah buffer overflow dan memastikan pengalaman streaming yang lancar.
Menguji metode kueri streaming dua arah
Anda dapat menguji kueri streaming dua arah secara lokal dengan memanggil metode bidi_stream_query dan melakukan iterasi melalui hasilnya:
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)
Koneksi kueri dua arah yang sama dapat menangani beberapa permintaan dan respons. Untuk setiap permintaan baru dari antrean, contoh berikut menghasilkan aliran potongan yang berisi berbagai informasi tentang respons:
{'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'}
Opsional: Mendaftarkan metode kustom
Operasi dapat didaftarkan sebagai mode eksekusi standar (diwakili oleh string kosong ""), streaming (stream), atau streaming dua arah (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"]
}
Men-deploy agen
Setelah mengembangkan agen sebagai live_agent, Anda dapat men-deploy agen ke Agent Platform dengan membuat instance Agent Platform.
Perhatikan bahwa dengan GenAI SDK, semua konfigurasi deployment (paket tambahan
dan kontrol resource yang disesuaikan) ditetapkan sebagai nilai config saat
membuat instance Agent Platform.
Melakukan inisialisasi klien GenAI:
import agentplatform
from agentplatform import types as agentplatform_types
client = agentplatform.Client(project=PROJECT, location=LOCATION)
Men-deploy agen ke Agent Platform. Perhatikan bahwa EXPERIMENTAL
agent_server_mode diperlukan untuk agen yang mendukung streaming dua arah:
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,
},
)
Untuk mengetahui informasi tentang langkah-langkah yang terjadi di latar belakang selama deployment, lihat Membuat instance Agent Runtime.
Dapatkan ID resource agen:
remote_live_agent.api_resource.name
Menggunakan agen
Jika Anda menentukan operasi bidi_stream_query saat mengembangkan agen, Anda dapat melakukan streaming kueri dua arah ke agen secara asinkron menggunakan GenAI SDK for Python.
Anda dapat mengubah contoh berikut dengan data apa pun yang dapat dikenali oleh agen Anda, menggunakan logika penghentian yang berlaku untuk aliran input dan aliran 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 menstreaming respons sebagai urutan objek yang dihasilkan secara iteratif. Misalnya, kumpulan dua respons dalam giliran pertama mungkin terlihat seperti berikut:
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
Menggunakan agen Agent Development Kit
Jika Anda mengembangkan agen menggunakan Agent Development Kit (ADK), Anda dapat menggunakan streaming dua arah untuk berinteraksi dengan Gemini Live API.
Contoh berikut membuat agen percakapan yang menerima pertanyaan teks pengguna dan menerima data audio respons 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))