O Futuro da IA em Tempo Real: Comunicação Bidirecional Contínua
Em 2025, a inteligência artificial generativa transcendeu a simples geração de texto. As aplicações modernas demandam muito mais do que respostas transacionais isoladas. Elas exigem capacidades multimodais abrangentes — desde transcrição e tradução de áudio até agentes de voz sofisticados — e, sobretudo, necessitam de comunicação contínua e fluida em tempo real.
O diferencial está em um requisito essencial: os dados precisam fluir simultaneamente em ambas as direções, através de uma única conexão persistente. Imagine um cenário de conversão de fala para texto onde o áudio é transmitido de forma contínua enquanto, ao mesmo tempo, o modelo processa e retorna a transcrição em tempo real, palavra por palavra. Esses casos de uso demandam exatamente essa capacidade bidirecional.
A AWS anunciou a transmissão bidirecional para Amazon SageMaker AI Inference, transformando fundamentalmente o paradigma de inferência — de uma troca transacional isolada para uma conversa contínua e natural. A experiência de fala flui naturalmente quando não há interrupções. Com transmissão bidirecional, a conversão de fala para texto torna-se imediata: o modelo escuta e transcreve simultaneamente, fazendo as palavras aparecerem no exato momento em que são pronunciadas.
Considere o caso prático de um centro de atendimento ao cliente. Conforme o cliente descreve seu problema, a transcrição ao vivo aparece instantaneamente na tela do atendente, fornecendo contexto imediato e permitindo que ele responda sem esperar o cliente terminar de falar. Esse tipo de troca contínua torna as experiências de voz mais fluidas, responsivas e genuinamente humanas.
Como Funciona a Transmissão Bidirecional
O Paradigma Tradicional versus Bidirecional
Na abordagem tradicional de requisições de inferência, existe um padrão bem definido: o cliente envia uma pergunta completa e aguarda, enquanto o modelo processa e retorna uma resposta completa antes que o cliente possa enviar a próxima pergunta. É um ciclo de espera e resposta.
Com transmissão bidirecional, esse fluxo é radicalmente diferente. A pergunta começa a fluir do cliente enquanto o modelo simultaneamente inicia o processamento e começa a retornar a resposta imediatamente. O cliente pode continuar refinando sua entrada enquanto o modelo adapta sua resposta em tempo real. Os resultados aparecem assim que o modelo os gera — palavra por palavra para texto, frame por frame para vídeo, amostra por amostra para áudio.
Esse padrão também oferece benefícios infraestruturais significativos: manter uma única conexão persistente elimina a necessidade de centenas de conexões de curta duração. Isso reduz consideravelmente a sobrecarga de gerenciamento de rede, handshakes TLS e administração de conexões. Além disso, os modelos conseguem manter contexto ao longo de um fluxo contínuo, possibilitando interações multi-turnos sem a necessidade de reenviar o histórico de conversa a cada requisição.
Arquitetura de Três Camadas do SageMaker AI Inference
A implementação técnica da transmissão bidirecional no SageMaker AI Inference combina dois protocolos complementares: HTTP/2 e WebSocket, criando um canal robusto de comunicação bidirecional em tempo real entre cliente e modelo. O fluxo ocorre em três camadas:
Cliente para Roteador SageMaker AI: Sua aplicação se conecta ao endpoint de tempo de execução do Amazon SageMaker AI usando HTTP/2, estabelecendo uma conexão eficiente e multiplexada que suporta transmissão bidirecional.
Roteador para Contêiner do Modelo: O roteador encaminha a requisição para um Sidecar — um proxy leve executando ao lado de seu contêiner de modelo — que estabelece uma conexão WebSocket com o contêiner em ws://localhost:8080/invocations-bidirectional-stream. Uma vez estabelecida, os dados fluem livremente em ambas as direções.
Fluxo de Requisição e Resposta: Sua aplicação envia entrada como uma série de blocos de dados via HTTP/2. A infraestrutura do SageMaker AI converte esses blocos em frames de dados WebSocket — texto (para dados UTF-8) ou binário — e os encaminha ao contêiner. O modelo recebe esses frames em tempo real e começa a processar imediatamente, antes mesmo da chegada da entrada completa. Na direção oposta, o modelo gera saída e a envia como frames WebSocket. O SageMaker AI encapsula cada frame em um payload de resposta e o transmite diretamente à sua aplicação via HTTP/2.
A conexão WebSocket entre o Sidecar e o contêiner permanece aberta pela duração da sessão, com monitoramento de saúde integrado. Para manter a integridade da conexão, o SageMaker AI envia frames de ping a cada 60 segundos para verificar se a conexão está ativa. Seu contêiner responde com pong frames para confirmar que está saudável. Se 5 pings consecutivos não receberem resposta, a conexão é fechada de forma controlada.
Construindo seu Próprio Contêiner com Transmissão Bidirecional
Pré-requisitos e Preparação
Se você deseja utilizar modelos de código aberto ou seus próprios modelos, é possível customizar seu contêiner para suportar transmissão bidirecional. Seu contêiner deve implementar o protocolo WebSocket para lidar com frames de dados recebidos e enviar frames de resposta de volta ao SageMaker AI.
Para começar, você precisará de:
- Uma conta AWS com permissões no SageMaker AI
- A política de acesso gerenciado AmazonSageMakerFullAccess IAM Managed Policy para permitir criação de endpoints
- Permissões de IAM que permitam explicitamente a ação
sagemaker:InvokeEndpoint*para invocação de endpoints - Docker instalado localmente
- Python 3.12 ou superior
- Instalar aws-sdk-python para a API
InvokeEndpointWithBidirectionalStreamdo tempo de execução do SageMaker AI
Construindo e Implantando o Contêiner
O processo começa clonando o repositório de demonstração e configurando seu ambiente conforme definido no README.md. Os passos a seguir criarão uma imagem Docker simples de demonstração e a enviarão para seu repositório de ECR na AWS.
# As variáveis de ambiente devem ser definidas para autenticação AWS
# export AWS_ACCESS_KEY_ID="sua-chave-de-acesso"
# export AWS_SECRET_ACCESS_KEY="sua-chave-secreta"
# export AWS_DEFAULT_REGION="us-west-2"
container_name="sagemaker-bidirectional-streaming"
container_tag="latest"
cd container
account=$(aws sts get-caller-identity --query Account --output text)
region=$(aws configure get region)
region=${region:-us-west-2}
container_image_uri="${account}.dkr.ecr.${region}.amazonaws.com/${container_name}:${container_tag}"
# Se o repositório não existe no ECR, crie-o
aws ecr describe-repositories --repository-names "${container_name}" --region "${region}" > /dev/null 2>&1
if [ $? -ne 0 ]
then
aws ecr create-repository --repository-name "${container_name}" --region "${region}" > /dev/null
fi
# Obtenha o comando de login do ECR e execute-o
aws ecr get-login-password --region ${region} | docker login --username AWS --password-stdin ${account}.dkr.ecr.${region}.amazonaws.com/${container_name}
# Construa a imagem Docker localmente e envie-a para o ECR
docker build --platform linux/amd64 --provenance=false -t ${container_name} .
docker tag ${container_name} ${container_image_uri}
docker push ${container_image_uri}Este processo cria um contêiner com um rótulo Docker indicando ao SageMaker AI que o suporte a transmissão bidirecional está habilitado: com.amazonaws.sagemaker.capabilities.bidirectional-streaming=true.
Após criar a imagem, você pode implantar o contêiner em um endpoint do SageMaker AI através de um script Python que cria o modelo, a configuração do endpoint e finalmente o endpoint propriamente dito.
Invocando o Endpoint com a Nova API
Uma vez que o endpoint do SageMaker AI esteja em estado InService, você pode proceder à invocação do endpoint para testar a funcionalidade de transmissão bidirecional. O cliente Python conecta-se ao endpoint do SageMaker AI e envia dados em chunks, recebendo respostas simultâneas.
#!/usr/bin/env python3
"""
Script Python para Transmissão Bidirecional no SageMaker AI.
Conecta-se a um endpoint do SageMaker AI para comunicação bidirecional.
"""
import argparse
import asyncio
import sys
from aws_sdk_sagemaker_runtime_http2.client import SageMakerRuntimeHTTP2Client
from aws_sdk_sagemaker_runtime_http2.config import Config, HTTPAuthSchemeResolver
from aws_sdk_sagemaker_runtime_http2.models import InvokeEndpointWithBidirectionalStreamInput, RequestStreamEventPayloadPart, RequestPayloadPart
from smithy_aws_core.identity import EnvironmentCredentialsResolver
from smithy_aws_core.auth.sigv4 import SigV4AuthScheme
import logging
def parse_arguments():
"""Analisa argumentos da linha de comando."""
parser = argparse.ArgumentParser(
description="Conecta-se a um endpoint do SageMaker AI para transmissão bidirecional"
)
parser.add_argument(
"ENDPOINT_NAME",
help="Nome do endpoint do SageMaker AI para conectar"
)
return parser.parse_args()
AWS_REGION = "us-west-2"
BIDI_ENDPOINT = f"https://runtime.sagemaker.{AWS_REGION}.amazonaws.com:8443"
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class SimpleClient:
def __init__(self, endpoint_name, region=AWS_REGION):
self.endpoint_name = endpoint_name
self.region = region
self.client = None
self.stream = None
self.response = None
self.is_active = False
def _initialize_client(self):
config = Config(
endpoint_uri=BIDI_ENDPOINT,
region=self.region,
aws_credentials_identity_resolver=EnvironmentCredentialsResolver(),
auth_scheme_resolver=HTTPAuthSchemeResolver(),
auth_schemes={"aws.auth#sigv4": SigV4AuthScheme(service="sagemaker")}
)
self.client = SageMakerRuntimeHTTP2Client(config=config)
async def start_session(self):
if not self.client:
self._initialize_client()
logger.info(f"Iniciando sessão com endpoint: {self.endpoint_name}")
self.stream = await self.client.invoke_endpoint_with_bidirectional_stream(
InvokeEndpointWithBidirectionalStreamInput(endpoint_name=self.endpoint_name)
)
self.is_active = True
self.response = asyncio.create_task(self._process_responses())
async def send_words(self, words):
for i, word in enumerate(words):
logger.info(f"Enviando payload: {word}")
await self.send_event(word.encode('utf-8'))
await asyncio.sleep(1)
async def send_event(self, data_bytes):
payload = RequestPayloadPart(bytes_=data_bytes)
event = RequestStreamEventPayloadPart(value=payload)
await self.stream.input_stream.send(event)
async def end_session(self):
if not self.is_active:
return
await self.stream.input_stream.close()
logger.info("Stream fechado")
async def _process_responses(self):
try:
output = await self.stream.await_output()
output_stream = output[1]
while self.is_active:
result = await output_stream.receive()
if result is None:
logger.info("Sem mais respostas")
break
if result.value and result.value.bytes_:
response_data = result.value.bytes_.decode('utf-8')
logger.info(f"Recebido: {response_data}")
except Exception as e:
logger.error(f"Erro ao processar respostas: {e}")
def main():
"""Função principal para analisar argumentos e executar o cliente de streaming."""
args = parse_arguments()
print("=" * 60)
print("Cliente de Transmissão Bidirecional SageMaker AI")
print("=" * 60)
print(f"Nome do Endpoint: {args.ENDPOINT_NAME}")
print(f"Região AWS: {AWS_REGION}")
print("=" * 60)
async def run_client():
sagemaker_client = SimpleClient(endpoint_name=args.ENDPOINT_NAME)
try:
await sagemaker_client.start_session()
words = ["Preciso de ajuda com", "meu saldo da conta", "Posso ajudar com isso", "e cobranças recentes"]
await sagemaker_client.send_words(words)
await asyncio.sleep(2)
await sagemaker_client.end_session()
sagemaker_client.is_active = False
if sagemaker_client.response and not sagemaker_client.response.done():
sagemaker_client.response.cancel()
logger.info("Sessão encerrada com sucesso")
return 0
except Exception as e:
logger.error(f"Erro no cliente: {e}")
return 1
try:
exit_code = asyncio.run(run_client())
sys.exit(exit_code)
except KeyboardInterrupt:
logger.info("Interrompido pelo usuário")
sys.exit(1)
except Exception as e:
logger.error(f"Erro inesperado: {e}")
sys.exit(1)
if __name__ == "__main__":
main()
Integração com Modelos Deepgram no SageMaker AI
Parceria e Capacidades
A AWS e a Deepgram colaboraram para construir suporte de transmissão bidirecional especificamente para endpoints do SageMaker AI. A Deepgram, parceira de Tier Avançado da AWS, oferece modelos de IA de voz em nível empresarial com precisão e velocidade líderes do setor. Seus modelos alimentam aplicações de transcrição em tempo real, conversão de texto para fala e agentes de voz para contact centers, plataformas de mídia e aplicações de IA conversacional.
Para clientes com requisitos rigorosos de conformidade que exigem que o processamento de áudio nunca saia de sua Nuvem Privada Virtual (VPC) na AWS, as opções tradicionais auto-hospedadas demandavam sobrecarga operacional considerável para configuração e manutenção. A transmissão bidirecional do Amazon SageMaker transforma completamente essa experiência, permitindo que os clientes implantem e escalem aplicações de IA em tempo real com apenas alguns poucos cliques no Console de Gerenciamento da AWS.
O modelo de fala para texto Deepgram Nova-3 está disponível atualmente no AWS Marketplace para implantação como um endpoint do SageMaker AI, com modelos adicionais chegando em breve. As capacidades incluem transcrição multilíngue, desempenho em escala empresarial e reconhecimento específico de domínio. A Deepgram oferece um período de avaliação gratuita de 14 dias no Amazon SageMaker AI para desenvolvedores prototiparem aplicações sem incorrer em custos de licença de software. As cobranças de infraestrutura do tipo de máquina escolhido ainda serão incorridas durante esse período. Para mais detalhes, consulte a documentação de preços do Amazon SageMaker AI.
Configurando um Endpoint Deepgram no SageMaker AI
Para configurar um endpoint Deepgram no SageMaker AI, navegue até a seção de pacotes de modelos do AWS Marketplace dentro do console do Amazon SageMaker AI e procure por Deepgram. Inscreva-se no produto e proceda ao assistente de lançamento na página do produto. Continue fornecendo detalhes no assistente de criação de endpoint de inferência em tempo real do Amazon SageMaker AI. Certifique-se de editar a variante de produção para incluir um tipo de instância válido ao criar sua configuração de endpoint. O botão de edição pode estar oculto até que você role a direita na tabela de variantes de produção. O ml.g6.2xlarge é um tipo de instância preferido para testes iniciais. Consulte a documentação Deepgram para requisitos específicos de hardware e orientação de seleção. Na página de resumo do endpoint, anote o nome do endpoint que você forneceu, pois será necessário na seção seguinte.
Usando o Endpoint Deepgram no SageMaker AI
Uma aplicação TypeScript de exemplo mostra como transmitir um arquivo de áudio para o modelo Deepgram hospedado em um endpoint de inferência em tempo real do SageMaker AI e imprimir uma transcrição transmitida em tempo real. A função cria um fluxo do arquivo WAV abrindo um arquivo de áudio local e enviando-o para o Amazon SageMaker AI Inference em pequenos blocos binários.
import * as fs from "fs";
import * as path from "path";
import { RequestStreamEvent } from '@aws-sdk/client-sagemaker-runtime-http2';
function sleep(ms: number): Promise {
return new Promise(resolve => setTimeout(resolve, ms));
}
async function* streamWavFile(filePath: string): AsyncIterable {
const full = path.resolve(filePath);
if (!fs.existsSync(full)) {
throw new Error(`Arquivo de áudio não encontrado: ${full}`);
}
console.log(`Transmitindo áudio: ${full}`);
const readStream = fs.createReadStream(full, { highWaterMark: 512_000 }); // 512 KB
for await (const chunk of readStream) {
yield { PayloadPart: { Bytes: chunk, DataType: "BINARY" } };
}
// Mantenha o stream ativo para receber respostas de transcrição após todo arquivo ser enviado
console.log("Áudio enviado, aguardando conclusão da transcrição...");
await sleep(15000); // Aguarde 15 segundos para processamento do bloco final de áudio
// Avise ao contêiner que terminamos
yield { PayloadPart: { Bytes: new TextEncoder().encode('{"type":"CloseStream"}'), DataType: "UTF8" } };
} O cliente de tempo de execução do AWS SageMaker AI é configurado especificando a região AWS, o nome do endpoint do SageMaker AI e a rota do modelo Deepgram dentro do contêiner. Você precisa atualizar esses valores conforme necessário.
import { SageMakerRuntimeHTTP2Client, InvokeEndpointWithBidirectionalStreamCommand } from '@aws-sdk/client-sagemaker-runtime-http2';
const region = "us-east-1"; // Região AWS
const endpointName = "REPLACEME"; // Nome do seu endpoint Deepgram SageMaker
const audioFile = "test.wav"; // Arquivo de áudio local
const modelInvocationPath = "v1/listen"; // Caminho WebSocket dentro do contêiner do modelo
const modelQueryString = "model=nova-3";
const client = new SageMakerRuntimeHTTP2Client({ region });O trecho final envia o fluxo de áudio ao endpoint do SageMaker AI e imprime os eventos JSON de transmissão da Deepgram conforme chegam. A aplicação exibe a saída de fala para texto ao vivo sendo gerada.
async function run() {
console.log("Enviando áudio para Deepgram via SageMaker...");
const command = new InvokeEndpointWithBidirectionalStreamCommand({
EndpointName: endpointName,
Body: streamWavFile(audioFile),
ModelInvocationPath: modelInvocationPath,
ModelQueryString: modelQueryString
});
const response = await client.send(command);
if (!response.Body) {
console.log("Nenhuma resposta de streaming recebida.");
return;
}
const decoder = new TextDecoder();
for await (const msg of response.Body) {
if (msg.PayloadPart?.Bytes) {
const text = decoder.decode(msg.PayloadPart.Bytes);
try {
const parsed = JSON.parse(text);
// Extraia e exiba a transcrição
if (parsed.channel?.alternatives?.[0]?.transcript) {
const transcript = parsed.channel.alternatives[0].transcript;
if (transcript.trim()) {
console.log("Transcrição:", transcript);
}
}
console.debug("Deepgram (bruto):", parsed);
} catch {
console.error("Deepgram (erro):", text);
}
}
}
}
console.log("Streaming concluído.");
}
run().catch(console.error);Para informações mais detalhadas e exemplos adicionais, consulte a página de documentação Deepgram. Se precisar de suporte adicional com a configuração, conecte-se com a Comunidade de Desenvolvedores Deepgram.
Conclusão
A transmissão bidirecional no Amazon SageMaker AI Inference representa um avanço significativo na forma como os desenvolvedores podem construir aplicações de IA em tempo real. Ao eliminar as limitações das interações transacionais tradicionais, essa capacidade abre possibilidades para experiências de usuário genuinamente conversacionais e responsivas.
Com suporte para contêineres customizados e integração pronta com modelos parceiros como o Deepgram, a plataforma oferece flexibilidade tanto para quem deseja trazer seus próprios modelos quanto para quem busca soluções prontas para produção. A infraestrutura de três camadas garante eficiência, enquanto o monitoramento automático de conexão assegura confiabilidade.
Developers podem começar a construir aplicações de transmissão bidirecional com Modelos de Linguagem Grande (LLMs) e SageMaker AI hoje mesmo, abrindo novos horizontes para agentes de voz, assistentes conversacionais e qualquer aplicação que se beneficie de comunicação em tempo real verdadeiramente bidirecional.
Fonte
Introducing bidirectional streaming for real-time inference on Amazon SageMaker AI (https://aws.amazon.com/blogs/machine-learning/introducing-bidirectional-streaming-for-real-time-inference-on-amazon-sagemaker-ai/)
Leave a Reply