Python + IA: Fundamentos e Projetos Práticos
Quando uma aplicação espera a resposta inteira da inteligência artificial antes de mostrar qualquer coisa, o usuário enxerga apenas uma tela parada. A mesma geração pode parecer muito mais rápida quando o texto começa a chegar em poucos instantes. Essa é a função do streaming: entregar a resposta em eventos menores, conforme o modelo produz o conteúdo.
Nesta continuação da série Python + IA, vamos implementar esse fluxo com a Responses API, o SDK oficial para Python e Server-Sent Events (SSE). O objetivo não é somente imprimir fragmentos no terminal. Vamos organizar os eventos, preservar a resposta final e criar um endpoint que uma interface web possa consumir.
O streaming complementa recursos vistos nas aulas sobre saídas estruturadas, function calling e Web Search. Ele não muda o que o modelo é capaz de fazer; muda como a aplicação apresenta e controla a execução.
Por que streaming muda a experiência, mas não reduz o trabalho do modelo
Sem streaming, o fluxo é simples: seu código envia uma requisição, aguarda a conclusão e recebe um objeto pronto. Com streaming, a conexão permanece aberta e o servidor envia uma sequência de eventos. O SDK oficial implementa esse transporte sobre SSE quando stream=True é usado na criação da resposta.
Na prática, isso reduz a latência percebida. Uma geração que leva oito segundos pode começar a exibir conteúdo no primeiro ou segundo segundo. O tempo total e a quantidade de tokens, porém, podem continuar praticamente iguais. Portanto, streaming é uma melhoria de experiência e de arquitetura de entrega, não um desconto automático de custo.
Ele faz sentido em:
- chats e copilotos;
- geração de textos mais longos;
- explicações de código;
- pesquisas com várias etapas;
- interfaces nas quais o usuário precisa perceber progresso.
Para respostas curtas, tarefas em segundo plano ou automações que só podem agir depois de validar o resultado completo, uma chamada comum pode ser mais simples. Também é importante separar duas ideias: o fluxo pode exibir texto parcial, mas decisões críticas devem aguardar o evento de conclusão e as validações necessárias.
Prepare o projeto sem colocar a chave no código
Crie um ambiente isolado e instale o SDK. Se você acompanhou a aula sobre ambientes com uv, pode usar o mesmo padrão:
uv init streaming-ia
cd streaming-ia
uv add openai python-dotenv fastapi uvicorn
Guarde a credencial em uma variável de ambiente. Um arquivo .env local é conveniente durante o desenvolvimento, desde que esteja no .gitignore:
OPENAI_API_KEY=sua_chave_aqui
OPENAI_MODEL=gpt-5.5
O modelo fica configurável porque nomes e opções disponíveis podem variar entre projetos. O código não precisa ser alterado quando você troca o valor no ambiente.
import os
from dotenv import load_dotenv
from openai import OpenAI
load_dotenv()
client = OpenAI()
MODEL = os.getenv("OPENAI_MODEL", "gpt-5.5")
Nunca envie a chave ao navegador nem a inclua em JavaScript público. A interface deve conversar com seu backend; somente o backend chama o provedor de IA.
Faça o primeiro streaming e filtre os eventos úteis
A implementação mínima usa stream=True e percorre o objeto retornado. O fluxo contém eventos de tipos diferentes: início, deltas de texto, conclusão de itens, término da resposta e possíveis falhas. Para mostrar texto, o evento mais importante é response.output_text.delta.
from openai import OpenAI
client = OpenAI()
stream = client.responses.create(
model=MODEL,
input="Explique em cinco passos como revisar um pull request.",
stream=True,
)
for event in stream:
if event.type == "response.output_text.delta":
print(event.delta, end="", flush=True)
elif event.type == "response.completed":
print("\n\nResposta concluída.")
elif event.type == "response.failed":
print("\nA geração falhou.")
O flush=True evita que o terminal retenha pequenos fragmentos em um buffer. Em uma interface web, o equivalente é encaminhar cada delta imediatamente pela conexão aberta.
Evite imprimir todos os eventos para o usuário. Eles são úteis durante a depuração, mas podem incluir detalhes internos que poluem a interface. Trate explicitamente os tipos necessários e registre o restante de forma controlada.
Acumule o texto e diferencie parcial de concluído
Uma resposta parcial é ótima para leitura, mas não deve ser confundida com o resultado definitivo. A conexão pode cair, o usuário pode cancelar ou a geração pode terminar com erro. Por isso, acumule os deltas enquanto transmite e marque o resultado como concluído somente após receber o evento terminal esperado.
def gerar_resposta(pergunta: str) -> str:
partes: list[str] = []
concluida = False
stream = client.responses.create(
model=MODEL,
input=pergunta,
stream=True,
)
for event in stream:
if event.type == "response.output_text.delta":
partes.append(event.delta)
print(event.delta, end="", flush=True)
elif event.type == "response.completed":
concluida = True
elif event.type == "response.failed":
raise RuntimeError("A API informou falha na resposta")
if not concluida:
raise RuntimeError("O fluxo terminou sem confirmação de conclusão")
return "".join(partes)
Esse cuidado é especialmente importante quando o texto será salvo, enviado por e-mail ou usado como entrada de outra automação. Você pode mostrar o conteúdo enquanto chega, mas só persiste a versão final depois da confirmação.
Também registre informações operacionais sem armazenar conteúdo sensível: duração, tipo do evento terminal, modelo configurado e um identificador interno da requisição. Logs devem ajudar a diagnosticar falhas, não criar uma cópia desnecessária dos dados do usuário.
Exponha o fluxo para uma interface com FastAPI e SSE
O navegador pode consumir eventos enviados pelo seu backend. Neste exemplo, o FastAPI recebe a pergunta, abre o streaming com o SDK e encaminha apenas deltas sanitizados. StreamingResponse mantém a conexão HTTP aberta com o tipo text/event-stream.
import json
import os
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from openai import AsyncOpenAI
from pydantic import BaseModel, Field
app = FastAPI()
client = AsyncOpenAI()
MODEL = os.getenv("OPENAI_MODEL", "gpt-5.5")
class Mensagem(BaseModel):
pergunta: str = Field(min_length=2, max_length=4000)
@app.post("/chat/stream")
async def chat_stream(payload: Mensagem, request: Request):
async def eventos():
stream = await client.responses.create(
model=MODEL,
input=payload.pergunta,
stream=True,
)
async for event in stream:
if await request.is_disconnected():
break
if event.type == "response.output_text.delta":
data = json.dumps(
{"tipo": "delta", "texto": event.delta},
ensure_ascii=False,
)
yield f"data: {data}\n\n"
elif event.type == "response.completed":
yield 'data: {"tipo":"concluida"}\n\n'
elif event.type == "response.failed":
yield 'data: {"tipo":"erro"}\n\n'
return StreamingResponse(
eventos(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"X-Accel-Buffering": "no",
},
)
Inicie o servidor com uv run uvicorn main:app --reload. O cabeçalho X-Accel-Buffering: no ajuda quando há um proxy compatível, mas a configuração real depende da sua infraestrutura. Teste o caminho completo em produção: aplicação, proxy, CDN e navegador. Um intermediário que acumula a resposta em buffer elimina o benefício do streaming.
Para uma aplicação pública, acrescente autenticação, limite por usuário, timeout, moderação adequada ao caso e proteção contra abuso. Também limite o tamanho da pergunta antes de iniciar uma chamada que gera custo.
Checklist de produção e próximos passos
Antes de liberar o recurso, verifique:
- [ ] a chave existe somente no backend e no gerenciador de segredos;
- [ ] a entrada tem limite de tamanho e validação;
- [ ] o frontend diferencia
delta, conclusão e erro; - [ ] texto parcial não dispara ações irreversíveis;
- [ ] desconexões interrompem o trabalho quando possível;
- [ ] proxy e CDN não acumulam o fluxo em buffer;
- [ ] timeouts e tentativas não duplicam uma operação;
- [ ] logs evitam dados pessoais e segredos;
- [ ] métricas acompanham tempo até o primeiro delta e tempo total;
- [ ] a interface oferece cancelar e tentar novamente.
Duas métricas ajudam bastante. Tempo até o primeiro delta mede a sensação de resposta. Tempo até a conclusão mede a duração real. Se o primeiro está baixo e o segundo alto, a interface parece ágil, mas talvez o prompt ou a tarefa ainda precisem ser otimizados.
O próximo passo natural é combinar streaming com ferramentas. Nesse caso, a interface precisa comunicar estados além do texto, como “consultando documentos” ou “verificando pedido”, sem expor argumentos internos nem fingir que uma ação terminou antes da validação.
Comece pelo script de terminal, confirme os eventos que sua aplicação realmente recebe e só então adicione FastAPI e frontend. Essa progressão torna os erros mais fáceis de localizar. Com o fluxo bem tratado, o streaming deixa de ser um efeito visual e vira uma camada confiável de entrega para produtos de IA mais responsivos.
