Ao implantar fluxos de trabalho de agentes com várias etapas (como síntese de documentos em grande escala e pesquisas de longa duração), a execução de agentes em segundo plano em modelos em tempo real pode criar uma pressão desnecessária na infraestrutura e acionar erros de esgotamento de recursos (429).
A Gemini Enterprise Agent Platform oferece um nível adiado, um programador otimizado para capacidade de processamento projetado especificamente para cargas de trabalho tolerantes à latência. Em vez de tratar tarefas autônomas de longa duração com a mesma urgência imediata de uma consulta de chat ao vivo, o programador enfileira seus fluxos de trabalho de agentes complexos de várias etapas para horários de pico, visando altas taxas de sucesso e capacidade de processamento geral.
Quando você envia uma solicitação usando o nível adiado, a API aceita a tarefa de forma assíncrona e retorna um ID de interação imediatamente. O nível adiado tem os seguintes recursos:
Preço com desconto: você recebe um desconto de 50% no preço da inferência do modelo em comparação com uma solicitação padrão, para que possa gerenciar o custo do agente na produção. Para mais informações, consulte Preços.
Maior capacidade de processamento: o nível adiado atenua 429s (restrições de capacidade do modelo ) e limites de taxa movendo suas cargas de trabalho pesadas e assíncronas fora do horário de pico, liberando a cota do nível padrão para suas necessidades de produção em tempo real.
Tempo limite de conclusão: o nível adiado tem como meta concluir 95% das tarefas em 24 horas. Se uma tarefa não for concluída dentro dessa janela, ela vai expirar e fazer a transição para um estado
failed. O tempo real gasto na fila depende da capacidade e da demanda atual do cluster regional.
Casos de uso
O nível adiado é adequado para casos de uso que podem tolerar horas de tempo de resposta, como os exemplos a seguir:
Finanças: pesquisa de mercado e de ações diária ou semanal.
Jurídico e compliance: auditoria regulatória e de fusões e aquisições de vários documentos.
Estratégia: inteligência competitiva contínua e síntese de tendências.
Segurança: verificação e correção de vulnerabilidades de base de código.
Agentes com suporte
É possível configurar o agendamento de agentes autônomos para o Deep Research Agent.
Criar uma tarefa adiada
O exemplo a seguir mostra como iniciar uma tarefa de Deep Research
usando o nível adiado com client.interactions.create():
import time
from google import genai
client = genai.Client(
enterprise=True,
project="PROJECT_ID",
location="global",
)
PROMPT = "Analyze the latest market trends in renewable energy storage."
DEEP_RESEARCH_AGENT = "deep-research-preview-04-2026"
interaction = client.interactions.create(
input=PROMPT,
agent=DEEP_RESEARCH_AGENT, # Agent identifier
service_tier="deferred", # Run on deferred tier for off-peak scheduling
background=True, # Return immediately instead of waiting for the answer
store=True, # Persist interaction state to poll or stream later
stream=False, # `stream` must be set to False during task creation
)
print(f"Interaction ID: {interaction.id}")
print(f"Status: {interaction.status}")
print(f"Service tier: {interaction.service_tier}")
O método retorna imediatamente com status="in_progress" e
service_tier="deferred".
Monitore o progresso das tarefas
Enquanto aguarda a capacidade fora do horário de pico e está em execução ativa, o status da interação permanece in_progress. À medida que o agente executa as etapas de planejamento, pesquisa e análise, novos itens são anexados à lista steps.
É possível acompanhar o status da tarefa de maneira programática, consultando a interação periodicamente ou transmitindo atualizações.
Enquetes
Consulte a interação periodicamente (por exemplo, a cada 15 a 30 segundos) até que ela atinja um dos estados finais: completed, failed ou cancelled.
TERMINAL_STATES = ("completed", "failed", "cancelled")
POLL_INTERVAL_SECONDS = 15
TIMEOUT_MINUTES = 60
started = time.time()
deadline = started + TIMEOUT_MINUTES * 60
while True:
current = client.interactions.get(interaction.id)
elapsed = int(time.time() - started)
steps = getattr(current, "steps", None) or []
print(f"[{elapsed:>4}s] status={current.status} steps={len(steps)}")
if current.status in TERMINAL_STATES:
break
if time.time() >= deadline:
raise TimeoutError(
f"Still {current.status} after {TIMEOUT_MINUTES} min. The interaction "
"continues running server-side; re-run the check to resume polling."
)
time.sleep(POLL_INTERVAL_SECONDS)
print(f"\nFinished in {int(time.time() - started)}s with status={current.status}.")
Streaming
É possível transmitir atualizações em tempo real assim que a interação entrar no status in_progress definindo stream=True ao lado de background=True e store=True.
O stream envia eventos como pensamentos intermediários, deltas de texto e atualizações de status à medida que ocorrem.
Se a conexão cair enquanto a tarefa ainda estiver in_progress, você poderá se reconectar ao stream usando client.interactions.get() com stream=True e transmitir o ID do último evento recebido para last_event_id. Se você omitir last_event_id, a API vai reproduzir todos os eventos desde o início.
INTERACTION_ID = interaction.id # from the create step
MAX_RECONNECTS = 5
STREAM_TIMEOUT = 300 # seconds
print(
f"streaming interaction: {INTERACTION_ID} (status={interaction.status})\n"
)
def render(event):
"""Prints one SSE event. Returns True once the interaction has finished."""
if event.event_type == "step.delta":
delta = event.delta
if delta.type == "text":
print(delta.text, end="", flush=True)
elif delta.type == "thought_summary":
summary = (getattr(delta.content, "text", "") or "").strip()
if summary:
print(f"\n[thinking] {summary[:200]}", flush=True)
elif delta.type.endswith("_call"):
queries = getattr(getattr(delta, "arguments", None), "queries", None)
print(
f"\n[{delta.type}] {', '.join(queries) if queries else ''}",
flush=True,
)
elif event.event_type == "interaction.status_update":
print(f"[status] {event.status}", flush=True)
elif event.event_type == "interaction.completed":
print(f"\n\n[status] {event.interaction.status}", flush=True)
return True
elif event.event_type == "error":
print(f"\n[error] {event.error.message}", flush=True)
return True
return False
last_event_id = None
finished = False
for attempt in range(MAX_RECONNECTS):
try:
# stream=True turns the GET into a live subscription. last_event_id=None on
# the first pass, so the server starts from the beginning of the run.
for event in client.interactions.get(
INTERACTION_ID,
stream=True,
last_event_id=last_event_id,
timeout=STREAM_TIMEOUT,
):
last_event_id = event.event_id or last_event_id
finished = render(event) or finished
except Exception as e: # pylint: disable=broad-except
# A dropped connection loses nothing: the run continues server-side and the
# next iteration reattaches from last_event_id.
print(f"\n[stream dropped: {type(e).__name__}] reattaching...", flush=True)
if finished:
break
# The server also closes the stream when the run ends, without an error.
if (
client.interactions.get(INTERACTION_ID, timeout=STREAM_TIMEOUT).status
!= "in_progress"
):
break
else:
print(f"\n[gave up after {MAX_RECONNECTS} reconnects]")
print(f"\n\nStreamed interaction: {INTERACTION_ID}")
Cancelar uma tarefa
É possível cancelar uma tarefa enquanto o status dela for queued, in_progress ou requires_action. Quando você cancela uma tarefa, o status dela faz a transição para cancelled.
Para cancelar uma tarefa, use client.interactions.cancel():
client.interactions.cancel(INTERACTION_ID)
Recuperar a saída final e o uso de tokens
Quando a interação atinge o estado completed, a transcrição completa fica disponível na lista steps. A resposta final é o conteúdo de texto da última etapa que produziu a saída.
Como a interação é armazenada (store=True), é possível buscar o resultado a qualquer momento usando o ID de interação de qualquer sessão:
def get_final_text(completed_interaction):
"""Returns the text of the last step that produced output."""
for step in reversed(getattr(completed_interaction, "steps", None) or []):
text = "".join(
part.text for part in (getattr(step, "content", None) or [])
if getattr(part, "text", None)
)
if text:
return text
return ""
final = client.interactions.get(interaction.id)
print(f"Status: {final.status}\n")
print(get_final_text(final) or "(No text output)")
if final.usage:
print(
f"\nToken usage:\n"
f" Input tokens: {final.usage.total_input_tokens}\n"
f" Output tokens: {final.usage.total_output_tokens}\n"
f" Total tokens: {final.usage.total_tokens}"
)