תזמון אוטונומי של סוכנים

כשפורסים תהליכי עבודה מורכבים של סוכנים (כמו סינתזה של מסמכים בהיקף גדול ומחקרים ארוכי טווח), הפעלת סוכנים ברקע במודלים בזמן אמת עלולה ליצור עומס מיותר על התשתית ולהפעיל שגיאות של מיצוי משאבים (429).

‫Gemini Enterprise Agent Platform מציעה רמת דחייה, מתזמן עם אופטימיזציה של קצב העברת הנתונים שנועד במיוחד לעומסי עבודה עם סבילות לזמן אחזור. במקום להתייחס למשימות אוטונומיות ארוכות טווח באותה דחיפות מיידית כמו לשאילתה בצ'אט חי, המתזמן מכניס את תהליכי העבודה המורכבים של הסוכנים מרובי-השלבים לתור לשעות שבהן העומס נמוך, כדי להשיג שיעורי הצלחה גבוהים וקצב נתונים כולל.

כששולחים בקשה באמצעות רמת השירות 'השהיה', ה-API מקבל את המשימה באופן אסינכרוני ומחזיר מזהה אינטראקציה באופן מיידי. המאפיינים של רמת השירות עם ההשהיה:

  • תעריף מוזל: אתם מקבלים 50% הנחה על התמחור של הסקת מסקנות מהמודל בהשוואה לבקשה רגילה, כך שתוכלו לנהל את עלות הסוכן בסביבת הייצור. מידע נוסף מפורט במאמר בנושא תמחור.

  • תפוקה גבוהה יותר: רמת השירות Deferred מפחיתה את מספר השגיאות מסוג 429 (מגבלות על קיבולת המודל) ואת מגבלות הקצב, כי היא מעבירה את עומסי העבודה הכבדים והאסינכרוניים שלכם לשעות השפל, וכך מפנה את המכסה של רמת השירות Standard לצרכים שלכם בייצור בזמן אמת.

  • זמן קצוב לתפוגה להשלמה: ברמת הדחייה, המטרה היא להשלים 95% מהמשימות תוך 24 שעות. אם משימה לא תושלם במהלך פרק הזמן הזה, התוקף שלה יפוג והיא תעבור למצב failed. משך הזמן בפועל בהמתנה בתור תלוי בקיבולת ובביקוש הנוכחיים של האשכול האזורי.

תרחישים לדוגמה

המסלול עם ההמתנה מתאים לתרחישי שימוש שבהם אפשר להמתין כמה שעות עד לקבלת התוצאה, כמו הדוגמאות הבאות:

  • פיננסים: מחקרי שוק ומניות יומיים או שבועיים.

  • משפטי ועמידה בדרישות: בדיקת נאותות רגולטורית ומיזוגים ורכישות במספר מסמכים.

  • אסטרטגיה: מודיעין תחרותי רציף וסינתזה של מגמות.

  • אבטחה: סריקה ותיקון של נקודות חולשה בבסיס הקוד.

סוכנים נתמכים

אפשר להגדיר תזמון של סוכנים אוטונומיים עבור סוכן המחקר המעמיק.

יצירת משימה שנדחית

בדוגמה הבאה מוצג איך להתחיל משימת Deep Research באמצעות רמת התמחור Deferred עם 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}")

השיטה מחזירה באופן מיידי את הערכים status="in_progress" ו-service_tier="deferred".

מעקב אחרי התקדמות המשימות

בזמן ההמתנה לקיבולת מחוץ לשעות העומס ובזמן שהאינטראקציה פועלת באופן פעיל, הסטטוס שלה status נשאר in_progress. במהלך התהליך של התכנון, החיפוש והניתוח, פריטים חדשים מתווספים לרשימה steps.

אפשר לעקוב אחרי סטטוס המשימה באופן פרוגרמטי על ידי שליחת שאילתות לגבי האינטראקציה באופן תקופתי או על ידי סטרימינג של עדכונים.

סקרים

מבצעים בדיקה תקופתית של האינטראקציה (למשל כל 15 עד 30 שניות) עד שהאינטראקציה מגיעה לאחד ממצבי הסיום: completed, failed או 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}.")

סטרימינג

אפשר להגדיר סטרימינג של עדכונים בזמן אמת ברגע שהאינטראקציה עוברת לסטטוס in_progress על ידי הגדרת stream=True לצד background=True ו-store=True. הזרם דוחף אירועים כמו מחשבות ביניים, שינויים בטקסט ועדכוני סטטוס בזמן שהם מתרחשים.

אם החיבור נותק בזמן שהמשימה עדיין in_progress, אפשר להתחבר מחדש לזרם באמצעות client.interactions.get() עם stream=True ולהעביר את מזהה האירוע האחרון שהתקבל אל last_event_id. אם לא מציינים את last_event_id, ה-API מפעיל מחדש כל אירוע מההתחלה.

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}")

ביטול משימה

אפשר לבטל משימה כשהסטטוס שלה הוא queued, in_progress או requires_action. כשמבטלים משימה, הסטטוס שלה משתנה לcancelled.

כדי לבטל משימה, משתמשים ב-client.interactions.cancel():

client.interactions.cancel(INTERACTION_ID)

אחזור הפלט הסופי ונתוני השימוש בטוקנים

כשהאינטראקציה מגיעה למצב completed, התמליל המלא זמין ברשימה steps. התשובה הסופית היא תוכן הטקסט של השלב האחרון שיצר פלט.

מכיוון שהאינטראקציה מאוחסנת (store=True), אפשר לאחזר את התוצאה בכל שלב באמצעות מזהה האינטראקציה מכל סשן:

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}"
  )