Cuando se implementan flujos de trabajo de agentes de varios pasos (como la síntesis de documentos a gran escala y la investigación de larga duración), la ejecución de agentes en segundo plano en modelos en tiempo real puede crear una presión innecesaria en la infraestructura y activar errores de agotamiento de recursos (429).
Gemini Enterprise Agent Platform ofrece un nivel diferido, un programador optimizado para el rendimiento diseñado específicamente para cargas de trabajo tolerantes a la latencia. En lugar de tratar las tareas autónomas de larga duración con la misma urgencia inmediata que una consulta de chat en vivo, el programador pone en cola tus flujos de trabajo de agentes complejos de varios pasos para las horas de menor actividad con el objetivo de obtener altas tasas de éxito y un rendimiento general.
Cuando envías una solicitud con el nivel diferido, la API acepta la tarea de forma asíncrona y muestra un ID de interacción de inmediato. El nivel diferido incluye lo siguiente:
Tarifa con descuento: Recibes un descuento del 50% en los precios de inferencia del modelo en comparación con una solicitud estándar, por lo que puedes administrar el costo de tu agente en producción. Para obtener más información, consulta Precios.
Mayor capacidad de procesamiento: El nivel diferido mitiga los errores 429 (restricciones de capacidad del modelo) y los límites de frecuencia trasladando tus cargas de trabajo pesadas y asíncronas fuera de las horas de menor actividad, lo que libera la cuota del nivel estándar para tus necesidades de producción en tiempo real.
Tiempo de espera para la finalización: El nivel diferido tiene como objetivo completar el 95% de las tareas en un plazo de 24 horas. Si una tarea no se completa dentro de este período, vence y pasa a un estado
failed. El tiempo real que se pasa en la cola depende de la capacidad y la demanda actuales del clúster regional.
Casos de uso
El nivel diferido es adecuado para los casos de uso que pueden tolerar horas de tiempo de respuesta, como los siguientes ejemplos:
Finanzas: Investigación de mercado y de acciones diaria o semanal.
Asuntos legales y cumplimiento: Diligencia debida reglamentaria y de fusiones y adquisiciones de varios documentos.
Estrategia: Inteligencia competitiva continua y síntesis de tendencias.
Seguridad: Análisis y corrección de vulnerabilidades de la base de código.
Agentes compatibles
Puedes configurar la programación de agentes autónomos para el agente de investigación profunda.
Crea una tarea diferida
En el siguiente ejemplo, se muestra cómo iniciar una tarea de Deep Research
con
el nivel diferido con 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}")
El método muestra de inmediato status="in_progress" y
service_tier="deferred".
Supervisar el progreso de las tareas
Mientras se espera la capacidad fuera de las horas de menor actividad y se ejecuta de forma activa, el status de la interacción permanece como in_progress. A medida que el agente ejecuta los pasos de planificación, búsqueda y análisis, se agregan elementos nuevos a la lista steps.
Puedes hacer un seguimiento del estado de la tarea de forma programática mediante el sondeo periódico de la interacción o la transmisión de actualizaciones.
Encuestas
Realiza encuestas periódicamente sobre la interacción (por ejemplo, cada 15 a 30 segundos) hasta que esta alcance uno de los estados terminales: completed, failed o 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}.")
Transmisión
Puedes transmitir actualizaciones en tiempo real una vez que la interacción ingresa al estado in_progress si configuras stream=True junto con background=True y store=True.
La transmisión envía eventos como ideas intermedias, deltas de texto y actualizaciones de estado a medida que ocurren.
Si se interrumpe la conexión mientras la tarea aún está in_progress, puedes volver a conectarte a la transmisión con client.interactions.get() con stream=True y pasar el último ID de evento recibido a last_event_id. Si omites last_event_id, la API vuelve a reproducir todos los eventos desde el principio.
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}")
Cancela una tarea
Puedes cancelar una tarea mientras su estado sea queued, in_progress o requires_action. Cuando cancelas una tarea, su estado pasa a cancelled.
Para cancelar una tarea, usa client.interactions.cancel():
client.interactions.cancel(INTERACTION_ID)
Recupera el resultado final y el uso de tokens
Cuando la interacción alcanza el estado completed, la transcripción completa está disponible en la lista steps. La respuesta final es el contenido de texto del último paso que produjo el resultado.
Como la interacción se almacena (store=True), puedes recuperar el resultado en cualquier momento con el ID de interacción de cualquier sesión:
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}"
)