Vídeo com IA: Claude Code, Codex e ElevenLabs · Capítulo 23 de 25 · 26 min
Orquestrar o pipeline com etapas idempotentes e teto de gasto
Escreva um orquestrador em Python que reaproveita etapas pelo hash do insumo, enfileira cada API, repete só o que é seguro e para antes de passar do teto.
Este capítulo faz parte do curso gratuito Vídeo com IA: Claude Code, Codex e ElevenLabs.
Um pipeline de vídeo com IA chama vários serviços pagos em sequência: o Claude escreve a narração, o GPT-6.1 Sol revisa, a ElevenLabs narra, o Gemini Omni 1.1 Flash cria o b-roll gerado e o FFmpeg monta. Quando a montagem cai, rodar o script de novo do zero paga outra vez tudo o que já deu certo. Pior: um timeout no meio de uma geração de vídeo pode virar duas cobranças pelo mesmo plano.
Não resolver isso custa na fatura e no prazo, porque cada falha vira uma rodada inteira e ninguém sabe dizer quanto o vídeo custou. Nesta aula você escreve um orquestrador em Python com etapas idempotentes, isto é, etapas que podem ser repetidas sem duplicar o efeito, como apertar duas vezes o botão do elevador. Cada etapa ganha uma chave de cache, uma fila por fornecedor, repetição só onde é segura e uma reserva de orçamento que para tudo antes de passar do teto.
Se a figura ultrapassar a área visível, deslize para os lados. Pelo teclado, foque a figura e use as setas.
Separar direção, execução e registro
O Claude Code escreve e coordena o código; quem executa em produção é um script previsível, que roda igual às três da manhã sem ninguém olhando. A divisão de papéis segue a aula sobre o desenho do pipeline: cada etapa recebe arquivos, devolve arquivos e deixa um comprovante. Aqui esse comprovante vira o manifesto de estado, um JSON que guarda, por etapa, o status, a chave, o ID remoto, os hashes das saídas e o custo.
A chave de cache é um hash, a impressão digital do conteúdo: o SHA-256 dos parâmetros, de cada arquivo de entrada e da versão do adaptador, tudo junto. Mudou uma vírgula no prompt ou um byte no áudio, a chave muda e a etapa roda; nada mudou e a saída está íntegra, o terminal mostra CACHE e o gasto é zero. A tabela abaixo mostra as etapas do exemplo, com os IDs e as tarifas de 12/10/2026.
Etapas do exemplo e como repetir cada uma com segurança
| Etapa | Fornecedor e modelo | Repetição segura |
|---|---|---|
| Narração escrita | claude-opus-5-5, US$ 4 e US$ 20 por milhão de tokens | Sim em 429 e 5xx: texto não cria artefato remoto |
| Revisão | gpt-6.1-sol, US$ 2 e US$ 10 por milhão de tokens | Sim em 429 e 5xx, com o mesmo cuidado |
| Voz | eleven_v4 pela Text to Dialogue, US$ 0,08 por mil caracteres, preço de lista desde 13/10/2026 (a promoção de US$ 0,022 terminou em 12/10/2026) | Só em 429, quando o pedido foi recusado |
| B-roll gerado | gemini-omni-1.1-flash na Gemini API, cerca de US$ 0,10 por segundo em 720p, cobrado por token; rascunho em 360p por cerca de um terço | Nunca a criação; a consulta pelo ID sim |
| Montagem | FFmpeg local | Sim, sem custo de API |
Acima de 272 mil tokens de entrada, o 6.1 Sol cobra US$ 4 e US$ 15. O Astra fica para o problema em que o Sol falhou duas vezes, como na aula 12. Para tarefa trivial, como conferir formato ou contar palavras, o claude-haiku-5-5 custa US$ 0,10 e US$ 0,50 por milhão de tokens desde 07/10/2026. A assinatura do ChatGPT não inclui créditos de API.
Dois limites do ecossistema não substituem os seus. O teto do Agent SDK (max_budget_usd em Python) e o limite de 20 subagentes simultâneos, vistos na aula de subagentes, controlam agentes, não a ElevenLabs nem a Agent Platform. Cada fornecedor precisa de fila e de orçamento próprios no seu código.
Escrever o núcleo com cache, fila e teto de gasto
O núcleo usa só a biblioteca padrão do Python. O estado é gravado num arquivo temporário e trocado de uma vez com os.replace, para nunca sobrar meio JSON se a máquina desligar. O orçamento reserva a estimativa antes da chamada e acerta pelo custo real depois. Cada fornecedor tem um semáforo, o contador de atendentes do balcão: dois para a Anthropic, um para o Google. A variável TETO_USD e a leitura dos erros 401, 429 e 5xx seguem a aula da bancada.
"""scripts/nucleo.py: estado, cache por hash, teto de gasto, filas e retries.
Só biblioteca padrão (Python 3.12+). Importado por adaptadores.py e pipeline.py."""
import asyncio
import hashlib
import json
import os
import random
from pathlib import Path
ARQ_ESTADO = Path("estado/manifesto-estado.json")
ARQ_ESTADO.parent.mkdir(parents=True, exist_ok=True)
def sha256_arquivo(caminho):
h = hashlib.sha256()
with open(caminho, "rb") as f:
for bloco in iter(lambda: f.read(1024 * 1024), b""):
h.update(bloco)
return h.hexdigest()
def chave_da_etapa(etapa, versao_adaptador):
"""Parâmetros + conteúdo dos insumos + versão do adaptador, tudo num hash só."""
assinatura = {
"etapa": etapa["id"],
"tipo": etapa["tipo"],
"params": etapa.get("params", {}),
"insumos": {p: sha256_arquivo(p) for p in etapa.get("insumos", [])},
"adaptador": versao_adaptador,
}
texto = json.dumps(assinatura, sort_keys=True, ensure_ascii=False)
return hashlib.sha256(texto.encode("utf-8")).hexdigest()
class Estado:
"""Manifesto de estado: um registro por etapa, gravado a cada mudança."""
def __init__(self):
if ARQ_ESTADO.exists():
self.dados = json.loads(ARQ_ESTADO.read_text(encoding="utf-8"))
else:
self.dados = {"etapas": {}, "execucoes": []}
def etapa(self, id_etapa):
return self.dados["etapas"].setdefault(id_etapa, {"status": "pendente"})
def gravar(self):
tmp = ARQ_ESTADO.with_suffix(".tmp")
tmp.write_text(json.dumps(self.dados, ensure_ascii=False, indent=2), encoding="utf-8")
os.replace(tmp, ARQ_ESTADO) # troca atômica: nunca fica meio arquivo
class TetoExcedido(Exception):
pass
class ErroTransitorio(Exception):
"""Falha que pode ser repetida sem duplicar efeito (pedido recusado ou consulta)."""
def __init__(self, msg, retry_after=None):
super().__init__(msg)
self.retry_after = retry_after
class EstadoIncerto(Exception):
"""A criação paga pode ter acontecido sem que o ID voltasse. Não repetir."""
class BloqueadoPorCota(Exception):
"""429 repetido sem Retry-After: cota zerada no painel ou limite do projeto. Não insistir."""
class Orcamento:
"""Teto por execução: reserva a estimativa antes, acerta pelo custo real depois."""
def __init__(self, teto_usd):
self.teto, self.gasto, self.reservado = teto_usd, 0.0, 0.0
self._lock = asyncio.Lock()
async def reservar(self, estimativa, rotulo):
async with self._lock:
if self.gasto + self.reservado + estimativa > self.teto:
raise TetoExcedido(
f"{rotulo}: reserva de US$ {estimativa:.2f} passa o teto de US$ {self.teto:.2f}")
self.reservado += estimativa
async def acertar(self, estimativa, custo_real):
async with self._lock:
self.reservado -= estimativa
self.gasto += custo_real
async def com_retry(funcao, tentativas=5, base=2.0, espera_max=60.0, teto_sem_aviso=3):
"""Backoff exponencial com variação aleatória. Use só em operação segura.
Sem Retry-After, desiste na tentativa teto_sem_aviso: insistir não resolve cota zerada."""
for n in range(1, tentativas + 1):
try:
return await funcao()
except ErroTransitorio as erro:
if erro.retry_after is None and n >= teto_sem_aviso:
raise BloqueadoPorCota(f"{erro}: {n} tentativas sem Retry-After") from erro
if n == tentativas:
raise
# Retry-After do servidor vence; senão até 2, 4... segundos com sorteio.
espera = erro.retry_after or min(espera_max, base ** n) * random.uniform(0.5, 1.0)
print(f" tentativa {n + 1}/{tentativas} em {espera:.1f} s: {erro}")
await asyncio.sleep(espera)
# Uma fila por fornecedor: o semáforo é o número de atendentes no balcão.
FILAS = {
"anthropic": asyncio.Semaphore(2),
"openai": asyncio.Semaphore(1),
"elevenlabs": asyncio.Semaphore(2),
"google": asyncio.Semaphore(1),
"local": asyncio.Semaphore(2),
}Os adaptadores seguem um contrato só. Recebem a etapa, o registro dela e a função que grava o estado; devolvem o custo real. Texto e voz repetem quando o fornecedor recusou o pedido, com 429, porque nada foi criado. O adaptador do b-roll gerado grava a marca de envio antes do POST e o nome do arquivo remoto logo depois: se o processo cair entre os dois, a próxima rodada encontra a marca sem ID e para com estado incerto em vez de gerar outro plano. A espera pelo vídeo é o polling da aula de geração por API, aqui no recurso Files a cada 5 s, como pede a página do Omni, e retomável pelo nome do arquivo. O custo usa a duração medida no arquivo baixado; como o Omni cobra por token, a fatura confirma o valor.
"""scripts/adaptadores.py: uma função por fornecedor, com o contrato do curso.
pip install anthropic openai elevenlabs google-genai
Ambiente: ANTHROPIC_API_KEY, OPENAI_API_KEY, ELEVENLABS_API_KEY e GEMINI_API_KEY.
Nenhuma chave no código."""
import asyncio
import subprocess
from pathlib import Path
from nucleo import ErroTransitorio, EstadoIncerto, com_retry
VERSAO = "2026-10-12" # entra na chave de cache: mudou o adaptador, o cache expira
# Tarifas de referência em 12/10/2026, em US$. Atualize quando a tabela mudar.
PRECO = {
"claude-opus-5-5": (4 / 1e6, 20 / 1e6), # por token de entrada e de saída
"gpt-6.1-sol": (2 / 1e6, 10 / 1e6), # padrão do curso (aula 12), faixa até 272 mil tokens
"eleven_v4": 0.08 / 1000, # por caractere, preço de lista desde 13/10/2026
# gemini-omni-1.1-flash: cerca de US$ 0,10/s em 720p, cobrado por token; 360p por cerca de um terço.
"omni_por_segundo": {"720p": 0.10, "360p": 0.10 / 3},
}
def _duracao(caminho):
"""Segundos medidos no arquivo baixado, pelo ffprobe."""
return float(subprocess.check_output(
["ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "csv=p=0", caminho],
text=True).strip())
def _ler(caminho):
return Path(caminho).read_text(encoding="utf-8")
def _gravar_texto(caminho, texto):
Path(caminho).parent.mkdir(parents=True, exist_ok=True)
Path(caminho).write_text(texto, encoding="utf-8")
async def narracao_claude(etapa, registro, gravar):
import anthropic
cliente = anthropic.Anthropic() # lê ANTHROPIC_API_KEY
def chamar():
try:
return cliente.messages.create(
model="claude-opus-5-5",
max_tokens=4000,
messages=[{"role": "user",
"content": etapa["params"]["instrucao"] + "\n\n" + _ler(etapa["insumos"][0])}],
)
except anthropic.RateLimitError as e:
raise ErroTransitorio("429 na Anthropic",
float(e.response.headers.get("retry-after", 0)) or None)
except anthropic.InternalServerError as e:
raise ErroTransitorio(f"{e.status_code} na Anthropic")
r = await com_retry(lambda: asyncio.to_thread(chamar))
_gravar_texto(etapa["saidas"][0], "".join(b.text for b in r.content if b.type == "text"))
ent, sai = PRECO["claude-opus-5-5"]
return {"custo_usd": r.usage.input_tokens * ent + r.usage.output_tokens * sai,
"remoto": {"id": r.id}}
async def revisao_sol(etapa, registro, gravar):
from openai import InternalServerError, OpenAI, RateLimitError
cliente = OpenAI() # lê OPENAI_API_KEY
def chamar():
try:
return cliente.responses.create(
model="gpt-6.1-sol",
max_output_tokens=4000,
input=etapa["params"]["instrucao"] + "\n\n" + _ler(etapa["insumos"][0]),
)
except (RateLimitError, InternalServerError) as e:
raise ErroTransitorio(f"{e.status_code} na OpenAI")
r = await com_retry(lambda: asyncio.to_thread(chamar))
_gravar_texto(etapa["saidas"][0], r.output_text)
ent, sai = PRECO["gpt-6.1-sol"]
return {"custo_usd": r.usage.input_tokens * ent + r.usage.output_tokens * sai,
"remoto": {"id": r.id}}
async def voz_eleven(etapa, registro, gravar):
from elevenlabs.client import ElevenLabs
from elevenlabs.core.api_error import ApiError
cliente = ElevenLabs() # lê ELEVENLABS_API_KEY
texto = _ler(etapa["insumos"][0]).strip()
if len(texto) > 2_000:
raise ValueError("A Text to Dialogue aceita até 2.000 caracteres por pedido: divida o texto (aula 15).")
def chamar():
try:
audio = cliente.text_to_dialogue.convert(
model_id="eleven_v4",
language_code="pt",
inputs=[{"text": texto, "voice_id": etapa["params"]["voice_id"]}],
output_format="mp3_44100_128",
)
return b"".join(audio)
except ApiError as e:
if e.status_code == 429: # recusado: nada foi gerado nem cobrado
raise ErroTransitorio("429 na ElevenLabs")
raise
Path(etapa["saidas"][0]).write_bytes(await com_retry(lambda: asyncio.to_thread(chamar)))
return {"custo_usd": len(texto) * PRECO["eleven_v4"], "remoto": {}}
async def broll_omni(etapa, registro, gravar):
"""gemini-omni-1.1-flash na Gemini API. O pedido devolve um arquivo no recurso Files,
consultado até ACTIVE ou FAILED. Confira os nomes dos campos na página do Omni
(ai.google.dev/gemini-api/docs/omni) para a versão do SDK que você instalou."""
from google import genai
cliente = genai.Client() # lê GEMINI_API_KEY
nome = registro.get("remoto", {}).get("arquivo")
if nome:
print(" retomando o arquivo existente:", nome) # não cria outra geração
else:
if registro.get("envio") == "iniciado":
raise EstadoIncerto("b-roll enviado sem ID gravado. Reconcilie na lista de arquivos do projeto.")
registro["envio"] = "iniciado"
gravar() # a marca vai para o disco ANTES do POST pago
# Proporção 9:16 e resolução (params["resolucao"]: 360p no rascunho, 720p no take)
# entram nos campos que a página do Omni documenta; acrescente-os a este config.
resposta = await asyncio.to_thread(
cliente.models.generate_content,
model="gemini-omni-1.1-flash",
contents=etapa["params"]["prompt"],
config={"response_format": {"type": "video", "delivery": "uri"}},
)
nome = resposta.output_video.uri # id do arquivo no recurso Files
registro["remoto"] = {"arquivo": nome}
gravar() # o ID vai para o disco ANTES de qualquer espera
async def consultar():
try:
return await asyncio.to_thread(cliente.files.get, name=nome)
except Exception as e: # consulta não cria nada: pode repetir
raise ErroTransitorio(f"consulta falhou: {e}")
arquivo = await com_retry(consultar)
while arquivo.state.name not in ("ACTIVE", "FAILED"):
await asyncio.sleep(5) # intervalo sugerido pela página do Omni
arquivo = await com_retry(consultar)
if arquivo.state.name == "FAILED":
raise RuntimeError(f"Gemini API marcou o arquivo como FAILED: {nome}")
dados = await asyncio.to_thread(cliente.files.download, file=arquivo)
Path(etapa["saidas"][0]).write_bytes(dados)
tarifa = PRECO["omni_por_segundo"][etapa["params"]["resolucao"]]
return {"custo_usd": _duracao(etapa["saidas"][0]) * tarifa, "remoto": registro["remoto"]}
async def montagem_ffmpeg(etapa, registro, gravar):
broll, voz = etapa["insumos"]
proc = await asyncio.create_subprocess_exec(
"ffmpeg", "-nostdin", "-v", "error", "-y", "-stream_loop", "-1", "-i", broll,
"-i", voz, "-map", "0:v", "-map", "1:a", "-shortest", "-c:v", "libx264",
"-pix_fmt", "yuv420p", "-c:a", "aac", "-movflags", "+faststart", etapa["saidas"][0])
if await proc.wait() != 0:
raise RuntimeError("FFmpeg falhou na montagem; rode o comando à mão para ver o erro.")
return {"custo_usd": 0.0, "remoto": {}}
def _caracteres(etapa):
return len(_ler(etapa["insumos"][0]))
# tipo -> (função, fila, estimativa em US$ para a reserva de orçamento)
ADAPTADORES = {
"narracao_claude": (narracao_claude, "anthropic", lambda e: 0.15),
"revisao_sol": (revisao_sol, "openai", lambda e: 0.06),
"voz_eleven": (voz_eleven, "elevenlabs", lambda e: _caracteres(e) * PRECO["eleven_v4"]),
"broll_omni": (broll_omni, "google",
lambda e: e["params"]["segundos"] * PRECO["omni_por_segundo"][e["params"]["resolucao"]]),
"montagem_ffmpeg": (montagem_ffmpeg, "local", lambda e: 0.0),
}O scripts/pipeline.py junta tudo. O manifesto lista as etapas com dependências, insumos, saídas e parâmetros. O executor roda em ondas: tudo o que tem as dependências prontas sai junto, cada etapa na fila do seu fornecedor. Um arquivo de bloqueio impede duas execuções simultâneas, e qualquer falha interrompe as etapas dependentes, nunca as já concluídas.
"""scripts/pipeline.py: executa o manifesto com cache, filas e teto de gasto.
Uso (na raiz video-ia): TETO_USD=3 python scripts/pipeline.py"""
import asyncio
import os
import time
from pathlib import Path
from adaptadores import ADAPTADORES, VERSAO
from nucleo import (FILAS, BloqueadoPorCota, Estado, EstadoIncerto, Orcamento,
chave_da_etapa, sha256_arquivo)
MANIFESTO = [
{"id": "narracao", "tipo": "narracao_claude", "deps": [],
"insumos": ["briefing/receita-da-semana.md"], "saidas": ["roteiro/narracao.txt"],
"params": {"instrucao": "Escreva só o texto da narração em pt-BR, de 140 a 160 "
"palavras, números por extenso, sem marcações."}},
{"id": "revisao", "tipo": "revisao_sol", "deps": ["narracao"],
"insumos": ["roteiro/narracao.txt"], "saidas": ["roteiro/revisao.md"],
"params": {"instrucao": "Revise quantidades, tempos e temperaturas. Não reescreva: "
"liste cada problema com a frase exata."}},
{"id": "voz", "tipo": "voz_eleven", "deps": ["narracao"],
"insumos": ["roteiro/narracao.txt"], "saidas": ["audio/narracao.mp3"],
# voice_id: o campo voice_id de privado/vozes/rafa.json, gravado na aula 16
"params": {"voice_id": "ID_DA_VOZ_AUTORIZADA"}},
{"id": "broll", "tipo": "broll_omni", "deps": [], "insumos": [],
"saidas": ["midia/gerados/broll.mp4"],
"params": {"segundos": 8, "resolucao": "720p", "prompt": (
"SHOT: Close-up, 50mm, vertical 9:16. CAMERA: slow push-in. "
"LIGHT: warm window light from the left. ACTION: hands roll a thin layer of "
"bolo de rolo on a wooden counter. AUDIO: soft kitchen ambience. No music. "
"No on-screen text.")}},
{"id": "montagem", "tipo": "montagem_ffmpeg", "deps": ["voz", "broll"],
"insumos": ["midia/gerados/broll.mp4", "audio/narracao.mp3"],
"saidas": ["renders/rascunho_9x16.mp4"], "params": {}},
]
async def executar(etapa, estado, orcamento):
registro = estado.etapa(etapa["id"])
funcao, fila, estimar = ADAPTADORES[etapa["tipo"]]
for saida in etapa["saidas"]:
Path(saida).parent.mkdir(parents=True, exist_ok=True)
chave = chave_da_etapa(etapa, VERSAO)
integras = all(Path(p).is_file() and registro.get("hashes", {}).get(p) == sha256_arquivo(p)
for p in etapa["saidas"])
if registro.get("chave") == chave and registro["status"] == "ok" and integras:
print("CACHE", etapa["id"])
return
if registro.get("chave") != chave:
# Insumo ou parâmetro novo: registro novo, sem herdar ID remoto antigo.
registro.clear()
registro.update({"status": "pendente", "chave": chave})
if registro["status"] == "incerto":
raise EstadoIncerto(f"{etapa['id']} está incerto. Reconcilie e limpe o registro à mão.")
estimativa, custo = estimar(etapa), 0.0
await orcamento.reservar(estimativa, etapa["id"])
try:
async with FILAS[fila]:
registro["status"] = "executando"
estado.gravar()
inicio = time.monotonic()
resultado = await funcao(etapa, registro, estado.gravar)
custo = resultado["custo_usd"]
faltando = [p for p in etapa["saidas"] if not Path(p).is_file()]
if faltando:
raise RuntimeError(f"{etapa['id']} terminou sem gerar {faltando}")
registro.update({
"status": "ok", "custo_usd": round(custo, 6), "remoto": resultado["remoto"],
"segundos": round(time.monotonic() - inicio, 1),
"hashes": {p: sha256_arquivo(p) for p in etapa["saidas"]},
})
print(f"OK {etapa['id']}: US$ {custo:.4f}")
except EstadoIncerto:
registro["status"] = "incerto"
raise
except BloqueadoPorCota:
registro["status"] = "bloqueado_cota" # nada é reenviado; confira o painel
raise
except Exception:
registro["status"] = "falhou"
raise
finally:
await orcamento.acertar(estimativa, custo)
estado.gravar()
async def principal():
teto = float(os.environ.get("TETO_USD", "3"))
estado, orcamento = Estado(), Orcamento(teto)
etapas = {e["id"]: e for e in MANIFESTO}
feitas, erros = set(), []
while len(feitas) < len(etapas) and not erros:
prontas = [e for i, e in etapas.items() if i not in feitas and set(e["deps"]) <= feitas]
if not prontas:
raise SystemExit("Dependência ausente ou ciclo no manifesto.")
resultados = await asyncio.gather(*(executar(e, estado, orcamento) for e in prontas),
return_exceptions=True)
for e, r in zip(prontas, resultados):
if isinstance(r, BaseException):
erros.append(e["id"])
print(f"PAROU {e['id']}: {type(r).__name__}: {r}")
else:
feitas.add(e["id"])
estado.dados["execucoes"].append({"teto_usd": teto, "gasto_usd": round(orcamento.gasto, 6),
"falhas": erros})
estado.gravar()
print(f"Gasto desta execução: US$ {orcamento.gasto:.4f} de US$ {teto:.2f}")
if erros:
raise SystemExit(1)
if __name__ == "__main__":
lock = Path("estado/pipeline.lock")
lock.parent.mkdir(parents=True, exist_ok=True)
try:
fd = os.open(lock, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
except FileExistsError:
raise SystemExit("Outra execução ou lock pendente. Investigue antes de remover.")
os.close(fd)
try:
asyncio.run(principal())
finally:
lock.unlink(missing_ok=True)O que mudou em outubro de 2026
Trocar o modelo de b-roll antes de 22/10/2026
Os previews do Veo 3.1 saem da Gemini API em 22/10/2026, como registra o mapa de outubro de 2026. A página de descontinuações indica gemini-omni-1.1-flash como substituto, em disponibilidade geral (GA) na Gemini API desde 27/08/2026, e é ele que o adaptador desta aula chama. O ID GA do Veo 3.1 Fast na Agent Platform não foi encontrado em página pública em 12/10/2026, e a página do Veo 3.1 naquela plataforma marca a retirada para 17/11/2026 ou depois. Se preferir ficar no Veo, confira no console o ID que existe depois de 22/10 e registre-o com a data de desligamento no config/modelos.toml da aula do desenho do pipeline.
Quando trocar de modelo, atualize a data de versão no topo de adaptadores.py: a chave de cache muda e nenhum plano antigo é reaproveitado como se fosse do modelo novo. Está pronto quando grep -rn "generate-preview" scripts/ volta vazio e a primeira rodada mostra o b-roll executando, não CACHE.
Se a figura ultrapassar a área visível, deslize para os lados. Pelo teclado, foque a figura e use as setas.
Pôr teto no backoff quando não vem Retry-After
Relatos de 28/09 e 01/10/2026 no fórum oficial da Gemini API mostram projetos pagos recebendo 429 com a cota zerada no painel e sem o cabeçalho Retry-After. Nesse caso, esperar e repetir não resolve. O com_retry acima obedece ao Retry-After quando ele vem; sem ele, desiste na terceira tentativa e lança BloqueadoPorCota, que o pipeline grava como bloqueado_cota sem reenviar nenhuma criação paga. Se a etapa parou assim, confira a cota no painel do fornecedor ou abra chamado; não aumente tentativas. Deu certo quando um 429 simulado sem cabeçalho para na terceira tentativa, com menos de dez segundos de espera somada.
Se a figura ultrapassar a área visível, deslize para os lados. Pelo teclado, foque a figura e use as setas.
Gerar o rascunho barato antes do take caro
Desde 27/08/2026, o Gemini Omni 1.1 Flash gera em 360p por cerca de um terço do custo do 720p. Transforme isso em duas etapas do manifesto: o rascunho em 360p e o take em 720p, com o mesmo prompt. Assista ao rascunho inteiro, aprove quadro e movimento e só então grave o arquivo de aprovação que libera o take. Como são etapas diferentes, cada uma tem a sua chave de cache, e mudar o prompt depois da aprovação invalida as duas. Se o rascunho errou duas vezes no mesmo ponto, reescreva o prompt antes da terceira tentativa, em vez de subir a resolução. Erro comum: aprovar pela miniatura, sem ver o movimento. Deu certo quando o manifesto-estado.json mostra o rascunho concluído antes de qualquer take em 720p e o custo do take aparece uma vez só.
Se a figura ultrapassar a área visível, deslize para os lados. Pelo teclado, foque a figura e use as setas.
# O take só roda depois que qa/broll_aprovado.txt existe: sem ele, sha256_arquivo
# falha e a etapa para antes de qualquer chamada paga. É de propósito.
PROMPT_BROLL = "..." # o mesmo texto nas duas etapas
{"id": "broll_rascunho", "tipo": "broll_omni", "deps": [], "insumos": [],
"saidas": ["midia/gerados/broll_rascunho.mp4"],
"params": {"segundos": 8, "resolucao": "360p", "prompt": PROMPT_BROLL}},
{"id": "broll", "tipo": "broll_omni", "deps": ["broll_rascunho"],
"insumos": ["qa/broll_aprovado.txt"], # escreva o nome do rascunho aprovado
"saidas": ["midia/gerados/broll.mp4"],
"params": {"segundos": 8, "resolucao": "720p", "prompt": PROMPT_BROLL}},Antes de rodar com chaves reais, veja no laboratório qual etapa pesa mais. Com metade do vídeo em b-roll gerado pelo Veo 3.1 Fast, a imagem em movimento passa de 90% do custo, e é nela que o cache e a reserva de orçamento mais economizam.
Quanto custa cada etapa do seu pipeline de vídeo
Troque a ferramenta de cada etapa e o percentual de b-roll e veja qual etapa pesa no custo por minuto e por vídeo.
Simulador de custo de um vídeo feito com IA. Você escolhe a ferramenta de cada etapa (roteiro com Claude Opus 5.5, GPT-6 Astra ou GPT-6.1 Sol; voz com ElevenLabs Eleven v4, a preço de lista, ou Flash v2.5; legenda com Scribe v2, gpt-transcribe ou Whisper local; imagem em movimento com Veo 3.1 Fast ou Standard, Seedance 2.5, MiniMax H3 Max ou Runway Gen-4.5; montagem com FFmpeg, Remotion, Remotion Lambda ou HyperFrames), a parte do vídeo coberta por b-roll gerado e a duração. O painel mostra o custo de cada etapa, a etapa mais cara e o total por minuto e por vídeo, em dólar e em reais, com preços de tabela de 09/10/2026. O simulador não tem Sora, desligado em 24/09/2026, nem o Omni 1.1 Flash; para o Omni, use o Veo 3.1 Fast como referência próxima, um pouco mais cara. A conversão usa um câmbio de referência fixo de R$ 5,40 por US$ 1, não a cotação do dia. Com metade do vídeo em Veo 3.1 Fast, a imagem gerada passa de 90% do custo.
A demonstração interativa carrega quando este trecho entra na tela.
Tratar falhas sem pagar duas vezes
Timeout não significa que o fornecedor deixou de executar a tarefa. A conexão pode cair depois que o servidor aceitou a geração e antes de devolver o ID. Por isso o arquivo de cache, sozinho, não garante idempotência numa API externa: ele protege o que está no seu disco, não o que está na fila do fornecedor. O contrato dos adaptadores fecha essa brecha com quatro regras: gravar a marca de envio antes do POST, gravar o ID remoto assim que ele chega, consultar o ID existente em vez de criar outra tarefa e marcar incerto quando a resposta se perde sem ID.
Repetir às cegas ou repetir a partir do estado
Repetir às cegas
Evitar
- Roda o pipeline inteiro de novo
- Reenvia a criação paga depois do timeout
- Usa o nome do arquivo como chave
- Todos os workers batem na mesma API
Repetir a partir do estado
Recomendado
- Reaproveita etapas com a mesma chave
- Consulta o ID gravado e reconcilia
- Usa o hash do conteúdo e dos parâmetros
- Cada fornecedor tem a sua fila
Se a figura ultrapassar a área visível, deslize para os lados. Pelo teclado, foque a figura e use as setas.
Repita a operação, não o efeito
Só entra em retry automático a operação que não cria nada novo do lado do fornecedor.
- 429: espere o Retry-After, quando vier, e reduza os workers daquele fornecedor
- 429 sem Retry-After: pare na terceira tentativa e marque bloqueado_cota
- 5xx ou timeout numa consulta: backoff de até 2 e 4 s com sorteio
- 5xx ou timeout numa criação paga: pare e reconcilie pelo console
- 401: pare, revise a credencial e não imprima o token no log
O teste: Desligue a rede durante a etapa de b-roll gerado: a próxima rodada deve parar como incerto.
alexandrecaramaschi.com/educacao/frontends-com-vibecoding
Os números da fila e das tentativas são decisões do curso, não cotas dos fornecedores; ajuste pela resposta real de cada API. O curso de design de APIs explica o outro lado desse limite em rate limiting e throttling, e o curso LLM FinOps mostra o teto de gasto que vale ligar esta semana. Para rodar o mesmo pipeline por agendamento, o n8n 2.42.6 desliga por padrão o nó que executa comandos locais, então o render precisa virar um serviço HTTP chamado por webhooks e triggers.
No servidor, a conta do render também entra no custo por etapa. Em 09/10/2026, o Modal cobra US$ 0,0000131 por núcleo físico por segundo e a AWS Lambda x86, US$ 0,0000166667 por GB-segundo mais US$ 0,20 por milhão de requisições, sem armazenamento nem transferência.
Provar que o pipeline é retomável
Rode cada passo e guarde a saída do terminal; ela é a evidência de que o pipeline funciona.
Passo 1: Escreva o manifesto com entradas e saídas por etapa
Deu certo quando: Cada etapa lista insumos, saídas, dependências e parâmetros.
Erro comum: Uma etapa única chamada faça tudo, sem arquivo intermediário.
Passo 2: Rode com TETO_USD=0.5 para testar o limite
Deu certo quando: O terminal mostra TetoExcedido antes de qualquer chamada de vídeo.
Erro comum: Descobrir o teto na fatura, depois do b-roll gerado.
Passo 3: Rode com o teto real e confira o manifesto de estado
Deu certo quando: Cada etapa tem status ok, custo, segundos e hashes.
Erro comum: Gravar só o vídeo final e perder o custo por etapa.
Passo 4: Rode de novo sem mudar nada
Deu certo quando: Todas as etapas aparecem como CACHE e o gasto é US$ 0,0000.
Erro comum: Refazer voz ou b-roll sem necessidade.
Passo 5: Altere só a instrução da revisão e rode outra vez
Deu certo quando: Só a revisão roda; narração, voz, b-roll e montagem saem do cache.
Erro comum: Usar o nome do arquivo como chave e não perceber a mudança.
Passo 6: Interrompa o processo durante o b-roll e rode de novo
Deu certo quando: A rodada retoma a operação pelo nome gravado, sem gerar outro plano.
Erro comum: Apagar o estado para liberar a etapa e pagar a geração duas vezes.
Passo 7: Apague o rascunho final e rode outra vez
Deu certo quando: Só a montagem roda de novo, com custo zero de API.
No fim você tem: Um pipeline que reaproveita o que já pagou, para antes do teto e nunca reenvia criação ambígua.
Guia prático: teto em duas camadas e conciliação com a fatura
O TETO_USD da aula protege o que o seu código chama. Ele não vê uma chave copiada para outro script nem uma tarifa desatualizada no dicionário PRECO. Aqui entram a segunda camada, o limite no painel de cada fornecedor, e a conciliação, a conferência entre o custo que o manifesto anotou e o que o fornecedor cobrou. No fim ficam um workspace só do projeto na Anthropic, o teto mensal no AI Studio e o scripts/conciliar_gasto.py.
Funções, casos de uso e aplicações
Funções
Limites do lado do fornecedor que completam o TETO_USD
| Função ou recurso | O que faz | Quando usar | Desde |
|---|---|---|---|
| Project Spend Caps no Google AI Studio | Teto mensal em dólar por projeto da Gemini API, na aba Spend; leva cerca de dez minutos para valer, e o excedente nesse intervalo é seu | Projeto separado só para o b-roll gerado do video-ia | 16/03/2026 |
| Orçamento do Google Cloud | Alerta por gasto real ou previsto; sozinho não corta o uso, e o corte exige automação por Pub/Sub | Veo pela Agent Platform, cobrado no faturamento do Cloud | Em vigor em 12/10/2026 |
| Workspace com limite de gasto na Anthropic | Teto mensal e alertas por workspace; não vale no Default Workspace e nunca passa do limite da organização | Workspace video-ia com a chave do pipeline | Em vigor em 12/10/2026 |
| Cost API da Anthropic (/v1/organizations/cost_report) | Custo diário em dólar, em centavos, agrupável por workspace; dados em cerca de cinco minutos | Conciliação diária do custo do roteiro com o manifesto | Em vigor em 12/10/2026 |
| POST /v1/organization/usage na Runway | Consulta o uso de créditos da organização | Conciliação semanal das cenas geradas na Runway | Em vigor em 12/10/2026 |
Casos de uso
Produtora de vídeos institucionais em Campinas que atende três faculdades
Educação
Dor
Um script de teste com a mesma chave do pipeline gastou fora do manifesto, e a fatura não batia com o manifesto-estado.json.
Solução
Cria um workspace por cliente na Anthropic, com limite mensal, e roda o conciliar_gasto.py no fim de cada dia de produção.
Resultado
Observar se a linha DIVERGÊNCIA some depois que cada script passa a usar a chave do próprio workspace.
Cenário construído para fins didáticos. Números não correspondem a cliente específico.
Equipe de conteúdo de uma marca de cosméticos em Belém
Marketing digital
Dor
O b-roll gerado no Gemini API não tinha limite fora do código, e um laço com erro poderia gerar a noite toda.
Solução
Põe o projeto do b-roll num teto mensal no AI Studio, com folga sobre o orçamento do mês, e mantém o TETO_USD por execução no pipeline.
Resultado
Acompanhar no painel de custo do AI Studio se o gasto diário segue o somatório do manifesto; a diferença indica tentativa fora do pipeline.
Cenário construído para fins didáticos. Números não correspondem a cliente específico.
Na prática
O script lê o custo do dia no workspace do projeto, pela Cost API, com a chave de Admin, que nunca vai para o .env do pipeline, e compara com o custo_usd das etapas de narração no estado/manifesto-estado.json. Rode no dia da execução, alguns minutos depois de terminar, porque o manifesto guarda o último custo de cada etapa.
"""conciliar_gasto.py: compara o custo do Claude no manifesto com a Cost API da Anthropic.
Uso (na raiz video-ia): ANTHROPIC_ADMIN_KEY=... WORKSPACE_VIDEO=wrkspc_... \
python scripts/conciliar_gasto.py 2026-10-12"""
import datetime as dt
import json
import os
import sys
import urllib.parse
import urllib.request
sys.path.insert(0, "scripts")
from pipeline import MANIFESTO # noqa: E402 (as etapas e seus tipos)
dia = dt.date.fromisoformat(sys.argv[1])
consulta = urllib.parse.urlencode({"starting_at": f"{dia}T00:00:00Z",
"ending_at": f"{dia + dt.timedelta(days=1)}T00:00:00Z",
"group_by[]": "workspace_id"})
pedido = urllib.request.Request(
"https://api.anthropic.com/v1/organizations/cost_report?" + consulta,
headers={"x-api-key": os.environ["ANTHROPIC_ADMIN_KEY"], "anthropic-version": "2023-06-01"})
dados = json.load(urllib.request.urlopen(pedido, timeout=60))
cobrado = sum(float(r["amount"]) / 100 # a API devolve centavos de dólar
for bloco in dados["data"] for r in bloco["results"]
if r.get("workspace_id") == os.environ["WORKSPACE_VIDEO"])
etapas = json.load(open("estado/manifesto-estado.json", encoding="utf-8"))["etapas"]
tipos = {e["id"]: e["tipo"] for e in MANIFESTO}
anotado = sum(r.get("custo_usd", 0.0) for i, r in etapas.items()
if tipos.get(i) == "narracao_claude")
print(f"{dia}: manifesto US$ {anotado:.4f} | Anthropic US$ {cobrado:.4f}")
if cobrado > anotado * 1.10 + 0.01: # 10% de folga para arredondamento
sys.exit("DIVERGÊNCIA: gasto fora do pipeline neste workspace ou tarifa velha em PRECO")Ligar a segunda camada de teto
Passo 1: Crie o workspace video-ia na Anthropic e gere nele a chave do pipeline
Em Settings > Workspaces, abra Spend limits e ponha o teto mensal com alertas.
Deu certo quando: A resposta da API traz anthropic-workspace-id com o id do workspace novo.
Erro comum: Deixar a chave no Default Workspace, que não aceita limite próprio.
Passo 2: Ponha o teto mensal do projeto de b-roll no AI Studio
Deu certo quando: A aba Spend mostra o Monthly spend cap ativo.
Erro comum: Contar com o corte imediato; há cerca de dez minutos de atraso.
Passo 3: Crie o orçamento do Google Cloud para a Agent Platform
Deu certo quando: Os alertas de 50%, 90% e 100% chegam por e-mail no teste.
Erro comum: Achar que o orçamento corta o gasto; ele só avisa.
Passo 4: Rode o conciliar_gasto.py no fim do dia de produção
Deu certo quando: As duas cifras ficam a menos de 10% uma da outra.
No fim você tem: Um teto no código, por execução, e outro no fornecedor, por mês, que não depende de o código estar certo.
Armadilhas
Um teto só ou duas camadas que se conferem
Evitar
Coluna a evitar
- Só o TETO_USD, que não vê outros scripts com a mesma chave
- Orçamento do Cloud tratado como corte automático
- Chave de Admin no .env do pipeline
- Ler a fatura só no fim do mês
Recomendado
Coluna recomendada
- Teto por execução no código e teto mensal no painel
- Alerta no Cloud e, se precisar de corte, automação por Pub/Sub
- Chave de Admin só no ambiente de quem concilia
- Conciliação diária, minutos depois de cada execução
Algumas decisões aparecem já na primeira semana. Se há ID remoto gravado, consulte-o, mesmo depois de uma queda local, e não crie outra geração. Se o estado ficou incerto, interrompa só aquela etapa paga e reconcilie no console do fornecedor antes de limpar o registro. Se a concorrência provoca 429, reduza os workers daquele fornecedor no dicionário FILAS, e não o número de subagentes; desde 07/09/2026, a resposta da API de chaves da ElevenLabs informa o limite de concorrência de TTS, música e dublagem, e esse número define o semáforo dela.
Na Cozinha da Rafa, canal de receitas regionais em Recife com três vídeos por semana, o pipeline semanal produz o vídeo longo 16:9 e os dois cortes 9:16 a partir do mesmo briefing. Quando a exportação de um corte falha, a nova rodada reaproveita narração, voz e b-roll do bolo de rolo e refaz só a montagem, sem pagar a ElevenLabs nem o Google de novo.
Rode os sete passos do guia no seu projeto e guarde o manifesto-estado.json das duas primeiras execuções. Está pronto quando duas execuções iguais reaproveitam tudo, uma mudança invalida só as etapas afetadas, uma falha impede as dependentes e nenhuma criação paga é reenviada depois de um timeout.
Seu caderno neste capítulo
Abrir o caderno completoSelecione um trecho do capítulo para destacar ou anotar. Nos vídeos e áudios, use Anotar este momento. No teclado, selecione com Shift e as setas e use Alt+Shift+D para destacar ou Alt+Shift+N para anotar.
Salvo neste navegador. Entre na sua conta para levar o caderno a outros aparelhos.
Entre na sua conta para compartilhar o que aprendeu e convidar alguém para estudar com você.
Conexões deste capítulo
Veja o que estudar antes e para onde seguir depois, explore os conceitos e compare abordagens em outros cursos. Sugestões marcadas como “sugerido pelo grafo” vêm dos assuntos em comum, não do índice do curso.
Estude antes
- CursoClaude Code CLI: Programação com IA
- CursoAstra profundosugerido pelo grafoA quinta parte do Astra planeja o vídeo e edita o material bruto com o modelo dirigindo a ferramenta. O curso de vídeo com IA aprofunda esse trabalho com Claude Code, Codex e ElevenLabs, do roteiro à publicação. O grafo já sugeria esta ordem pela regra de entrada da trilha; com a ordem anterior curada, ela passa a ser declarada (Etapa 4, wave 3).
- CursoSetup Completo: do zero ao Claude Codesugerido pelo grafo7 capítulos de “Vídeo com IA: Claude Code, Codex e ElevenLabs” usam conceitos apresentados em “Setup Completo: do zero ao Claude Code”.
Depois, siga para
- CursoFrontends com Vibecodingsugerido pelo grafo21 capítulos de “Frontends com Vibecoding” usam conceitos apresentados em “Vídeo com IA: Claude Code, Codex e ElevenLabs”.
- CursoDo Agile ao Agentic Operating Modelsugerido pelo grafo5 capítulos de “Do Agile ao Agentic Operating Model” usam conceitos apresentados em “Vídeo com IA: Claude Code, Codex e ElevenLabs”.
- CursoAgentes de IA Autônomos: Do Conceito à Produçãosugerido pelo grafo7 capítulos de “Agentes de IA Autônomos: Do Conceito à Produção” usam conceitos apresentados em “Vídeo com IA: Claude Code, Codex e ElevenLabs”.
- CursoLLM FinOps: o custo da IA para quem assina o orçamentosugerido pelo grafo4 capítulos de “LLM FinOps: o custo da IA para quem assina o orçamento” usam conceitos apresentados em “Vídeo com IA: Claude Code, Codex e ElevenLabs”.
O mesmo assunto em outros cursos
- Letramento em IA para ExecutivosEscolha e combine modelos com governança e custoExaminar conexões de Escolha e combine modelos com governança e custo
- Agentes de código para executivosComparar Claude e Gemini depois da troca pelo AntigravityExaminar conexões de Comparar Claude e Gemini depois da troca pelo Antigravity
- Astra profundoGovernar a esteira: gerar uma vez e multiplicar com cota zeroExaminar conexões de Governar a esteira: gerar uma vez e multiplicar com cota zero
- Prompt Engineering para ExecutivosEscrever o relatório de uma página para a diretoriaExaminar conexões de Escrever o relatório de uma página para a diretoria
Leve uma ideia para a prática
Escolha uma atividade no painel e volte a este curso quando quiser.
Atividades e evolução ficam reunidas no painel da sua conta.
Laboratórios ao vivo deste curso
Cada laboratório monta a biblioteca real dentro da aula, com controles para você mexer e o código do exemplo para copiar. O texto de reserva descreve a demonstração para quem lê sem JavaScript. O link abre o capítulo já no ponto do laboratório.
MediaRecorder, Canvas 2D e Long Animation Frames (plataforma)
Grave a cena em tempo real e veja os quadros que se perdem
Suba as partículas, force um bloqueio de 300 ms e grave seis segundos; compare com o render quadro a quadro desta aula.
Pacote
Abrir no capítuloMontar vídeo em HTML com HyperFrames e escolher entre ele e o Remotionsem pacoteMotion
Ver agentes em paralelo trocando mensagens numa linha do tempo
Reproduza a cena, percorra o tempo pelo controle e compare o tempo de parede com a soma das barras de cada agente.
Pacote
Abrir no capítuloParalelizar cenas com subagentes sem estourar o limite das APIsmotionReact (sem dependência extra), com as APIs do Veo e do fal.ai
Prompt de vídeo em cinco camadas, pronto para a API
Escolha plano, câmera, luz, ação e áudio e copie o prompt e o JSON para o Veo ou para o fal.ai.
Pacote
Abrir no capítuloEscrever prompt de vídeo em cinco camadas e manter o personagem consistentereactReact (sem dependência extra), com o comando para o FFmpeg
Corte o vídeo apagando palavras da transcrição
Aperte as palavras para removê-las, ligue a remoção de pausas e copie o comando FFmpeg e a legenda SRT gerados.
Pacote
Abrir no capítuloRemover pausas e reenquadrar o vídeo de 16:9 para 9:16 com legenda por palavrareactReact (sem dependência extra)
Quanto custa cada etapa do seu pipeline de vídeo
Troque a ferramenta de cada etapa e o percentual de b-roll e veja qual etapa pesa no custo por minuto e por vídeo.
Pacote
Abrir no capítuloOrquestrar o pipeline com etapas idempotentes e teto de gastoreact