대규모 문서 합성 및 장기 실행 연구와 같은 다단계 에이전트 워크플로를 배포할 때 실시간 모델에서 백그라운드 에이전트를 실행하면 불필요한 인프라 압력이 발생하고 리소스 소진(429) 오류가 발생할 수 있습니다.
Gemini Enterprise Agent Platform은 지연 시간에 민감하지 않은 워크로드에 맞게 설계된 처리량 최적화 스케줄러인 지연된 등급을 제공합니다. 스케줄러는 장기 실행되는 자율 태스크를 실시간 채팅 쿼리와 동일한 긴급성으로 처리하는 대신 복잡한 다단계 에이전트 워크플로를 대기열에 추가하여 성공률과 전반적인 처리량을 높입니다.
지연된 등급을 사용하여 요청을 제출하면 API가 태스크를 비동기식으로 수락하고 상호작용 ID를 즉시 반환합니다. 지연된 등급에는 다음과 같은 기능이 있습니다.
할인된 요금: 표준 요청과 비교하여 모델 추론 가격을 50% 할인받으므로 프로덕션에서 에이전트 비용을 관리할 수 있습니다. 자세한 내용은 가격 책정을 참조하세요.
처리량 증가: 지연된 등급은 많은 비동기 워크로드를 사용량이 적은 시간대로 이동하여 429 (모델 용량 제약조건) 및 비율 제한을 완화하고 실시간 프로덕션 요구사항에 맞게 표준 등급 할당량을 확보합니다.
완료 시간 제한: 지연된 등급은 24시간 이내에 태스크의 95% 를 완료하는 것을 목표로 합니다. 이 기간 내에 태스크가 완료되지 않으면 만료되고
failed상태로 전환됩니다. 대기열에서 소요되는 실제 시간은 현재 리전 클러스터 용량 및 수요에 따라 다릅니다.
사용 사례
지연된 등급은 다음과 같은 예와 같이 처리 시간을 몇 시간 동안 허용할 수 있는 사용 사례에 적합합니다.
금융: 일별 또는 주별 주식 및 시장 조사.
법률 및 규정 준수: 다중 문서 규제 및 인수합병 실사.
전략: 지속적인 경쟁 인텔리전스 및 트렌드 합성.
보안: 코드베이스 취약점 스캔 및 수정.
고객 지원 담당자
심층 연구 에이전트에 자율 에이전트 예약을 구성할 수 있습니다.
지연된 태스크 만들기
다음 예에서는 Deep Research
태스크
를 지연된 등급으로 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 목록에 추가됩니다.
상호작용을 주기적으로 폴링하거나 업데이트를 스트리밍하여 프로그래매틱 방식으로 태스크 상태를 추적할 수 있습니다.
폴링
상호작용이 터미널 상태 중 하나(completed, failed, cancelled)에 도달할 때까지 상호작용을 주기적으로(예: 15~30초마다) 폴링합니다.
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}.")
스트리밍
background=True 및 store=True와 함께 stream=True를 설정하여 상호작용이 in_progress 상태로 전환되면 업데이트를 실시간으로 스트리밍할 수 있습니다.
스트림은 중간 생각, 텍스트 델타, 상태 업데이트와 같은 이벤트를 발생할 때 푸시합니다.
태스크가 아직 in_progress 상태인 동안 연결이 끊어지면 stream=True로 client.interactions.get()을 사용하여 스트림에 다시 연결하고 마지막으로 수신된 이벤트 ID를 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) 모든 세션의 상호작용 ID를 사용하여 언제든지 결과를 가져올 수 있습니다.
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}"
)