Esta página descreve como usar o streaming bidirecional com o tempo de execução do Vertex AI Agent Engine.
Vista geral
O streaming bidirecional oferece um canal de comunicação bidirecional persistente entre a sua aplicação e o agente, o que lhe permite ir além dos padrões de pedido-resposta baseados em turnos. O streaming bidirecional funciona para exemplos de utilização em que o seu agente precisa de processar informações e responder continuamente, como interagir com entradas de áudio ou vídeo com baixa latência.
O streaming bidirecional com o tempo de execução do Vertex AI Agent Engine suporta exemplos de utilização de agentes interativos e em tempo real, bem como a troca de dados para APIs em direto multimodais. O streaming bidirecional é suportado para todas as estruturas, e os métodos de streaming bidirecional personalizados estão disponíveis através do registo de métodos personalizados. Pode usar o streaming bidirecional para interagir com a API Gemini Live através do Agent Development Kit (ADK) no Vertex AI Agent Engine.
A implementação de um agente remoto com métodos de consulta bidirecionais só é suportada através do SDK de IA gen da Google. Quando são detetados métodos de consulta bidirecionais, o SDK de IA gen define automaticamente o modo de servidor do agente quando chama a API REST.EXPERIMENTAL
Desenvolva um agente
Ao desenvolver um agente, use os seguintes passos para implementar o streaming bidirecional:
Registe métodos personalizados (opcional)
Defina um método de consulta de streaming bidirecional
Pode definir um método bidi_stream_query que recebe pedidos de streams de forma assíncrona e produz respostas de streaming. Por exemplo, o seguinte modelo expande o modelo básico para transmitir pedidos e respostas, e é implementável no Agent Engine:
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()
Tenha em atenção o seguinte quando usar a API de streaming bidirecional:
asyncio.Queue: pode colocar qualquer tipo de dados nesta fila de pedidos para aguardar o envio para a API Model.Tempo limite máximo: o tempo limite máximo para a consulta de streaming bidirecional é de 10 minutos. Se o seu agente precisar de tempos de processamento mais longos, considere dividir a tarefa em partes mais pequenas e usar a sessão ou a memória para manter os estados.
Restrinja o consumo de conteúdo: quando consome conteúdo a partir de uma stream bidirecional, é importante gerir a taxa à qual o seu agente processa os dados recebidos. Se o seu agente consumir dados demasiado lentamente, pode originar problemas como o aumento da latência ou a pressão da memória do lado do servidor. Implemente mecanismos para obter ativamente dados quando o seu agente estiver pronto para os processar e evite bloquear operações que possam interromper o consumo de conteúdo.
Limitar a geração de conteúdo: se encontrar problemas de contrapressão (em que o produtor gera dados mais rapidamente do que o consumidor os consegue processar), deve limitar a taxa de geração de conteúdo. Isto pode ajudar a evitar o excesso de memória intermédia e garantir uma experiência de streaming sem problemas.
Teste o método de consulta de streaming bidirecional
Pode testar a consulta de streaming bidirecional localmente chamando o método bidi_stream_query e iterando os resultados:
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)
A mesma ligação de consulta bidirecional pode processar vários pedidos e respostas. Para cada novo pedido da fila, o exemplo seguinte gera uma stream de fragmentos com informações diferentes sobre a resposta:
{'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'}
(Opcional) Registe métodos personalizados
As operações podem ser registadas como modos de execução padrão (representado por uma string vazia ""), de streaming (stream) ou de streaming bidirecional (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"]
}
Implemente um agente
Depois de desenvolver o seu agente como live_agent, pode implementá-lo no Agent Engine criando uma instância do Agent Engine.
Tenha em atenção que, com o SDK de IA gen., todas as configurações de implementação (pacotes adicionais e controlos de recursos personalizados) são atribuídas como um valor de config quando cria a instância do Agent Engine.
Inicialize o cliente de IA gen:
import vertexai
client = vertexai.Client(project=PROJECT, location=LOCATION)
Implemente o agente no motor de agentes:
remote_live_agent = client.agent_engines.create(
agent=live_agent,
config={
"staging_bucket": STAGING_BUCKET,
"requirements": [
"google-cloud-aiplatform[agent_engines,adk]==1.88.0",
"cloudpickle==3.0",
"websockets"
],
},
)
Para obter informações sobre os passos que ocorrem em segundo plano durante a implementação, consulte o artigo Crie uma instância do AgentEngine.
Obtenha o ID de recurso do agente:
remote_live_agent.api_resource.name
Use um agente
Se definiu uma operação bidi_stream_query ao desenvolver o seu agente, pode consultar o agente de forma bidirecional e assíncrona através do SDK de IA gen para Python.
Pode modificar o exemplo seguinte com quaisquer dados reconhecíveis pelo seu agente, usando qualquer lógica de terminação aplicável para a stream de entrada e a stream de saída:
async with client.aio.live.agent_engines.connect(
agent_engine=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
O tempo de execução do Vertex AI Agent Engine transmite respostas como uma sequência de objetos gerados iterativamente. Por exemplo, um conjunto de duas respostas no primeiro turno pode ter o seguinte aspeto:
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
Use um agente do Agent Development Kit
Se desenvolveu o seu agente com o Agent Development Kit (ADK), pode usar o streaming bidirecional para interagir com a API Gemini Live.
O exemplo seguinte cria um agente de conversa que recebe perguntas de texto do utilizador e recebe dados de áudio de resposta da API Gemini Live:
import numpy as np
from google.adk.agents.live_request_queue improt 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.agent_engines.connect(
agent_engine=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))
O que se segue?
- Saiba mais sobre a API Gemini Live.