Gemini와 Temporal로 지속형 AI 에이전트 구축

Gemini와 Temporal로 지속형 AI 에이전트 구축

이 튜토리얼은 추론에 Gemini API를 사용하고 지속성(durability)에 Temporal을 사용하는 지속형 AI 에이전트를 만드는 과정을 안내해요. Temporal의 내장 Gemini SDK 통합을 사용해요.

이 에이전트는 날씨 경보 조회나 IP 주소 지오로케이션 같은 도구를 호출할 수 있고, 응답할 충분한 정보를 얻을 때까지 루프를 돌아요.

이것이 일반 에이전트 데모와 다른 점은 지속성이에요. 모든 LLM 호출과 모든 도구 호출이 Temporal에 의해 영구 저장돼요. 프로세스가 충돌하거나, 네트워크가 끊기거나, API가 시간 초과되면 Temporal이 자동으로 마지막 완료 단계부터 재시도하고 재개해요. 대화 히스토리가 손실되지 않고, 도구 호출이 잘못 반복되지 않아요.

출처: 원문

본문

아키텍처

아키텍처는 세 부분으로 구성돼요.

• Workflow: 단일 generate_content 호출. Gemini SDK의 자동 함수 호출(AFC) 루프가 Workflow 내부에서 실행되고, Temporal이 그 모든 단계를 지속형으로 만들어요.
• Activities: Temporal이 지속형으로 만드는 개별 작업 단위. Gemini API 호출은 자동으로 Activities가 돼요.
• Worker: Workflow와 Activities를 실행하는 프로세스이며, API 키가 존재하는 유일한 곳이에요.

이 예시에서는 이 세 부분을 모두 단일 파일(durable_agent_worker.py)에 넣어요. 실제 구현에서는 다양한 배포 및 확장성 이점을 위해 분리할 거예요. Temporal CLI로 에이전트에 프롬프트를 제공하므로 작성할 클라이언트 코드는 없어요.

사전 요구사항

이 가이드를 완료하려면 다음이 필요해요.

• Gemini API 키. Google AI Studio에서 무료로 만들 수 있어요.
• 버전 3.10 이상의 Python.
• 의존성 관리를 위한 uv.
• 로컬 개발 서버 실행과 Workflow 시작을 위한 Temporal CLI.

설정

시작하기 전에 Temporal 개발 서버가 로컬에서 실행 중인지 확인하세요.

temporal server start-dev

다음으로 프로젝트를 만들고 필요한 의존성을 설치하세요.

uv init durable-gemini-agent
cd durable-gemini-agent
uv add "temporalio[google-genai]" httpx python-dotenv

uv가 가상 환경을 만들고 관리해 주므로, 이 튜토리얼의 이후 모든 Python 명령은 uv run으로 실행돼요.

Gemini API 키로 프로젝트 디렉토리에 .env 파일을 만드세요. API 키는 Google AI Studio에서 얻을 수 있어요.

echo "GOOGLE_API_KEY=your-api-key-here" > .env

구현

튜토리얼의 나머지는 durable_agent_worker.py를 위에서 아래로 살펴보며 에이전트를 조각조각 구성해요. 파일을 만들고 따라가 보세요.

임포트 및 샌드박스 설정

먼저 사전에 정의해야 하는 임포트부터 시작해요. workflow.unsafe.imports_passed_through() 블록은 Temporal의 Workflow 샌드박스가 httpx를 제한 없이 통과시키도록 지시해요. httpx를 임포트하면 class _CookieCompatRequest(urllib.request.Request)를 실행하는데, 샌드박스는 그 stdlib 클래스의 서브클래싱을 차단해요.

도구는 httpx를 사용하고, activity_as_tool()은 Workflow가 해당 도구 함수를 임포트해 Gemini가 시그니처에서 스키마를 파생할 수 있어야 해요. 따라서 파일을 어떻게 나누든 httpx는 샌드박스에 도달해요. 도구를 별도 모듈로 옮겨도 피할 수 없어요.

from temporalio import workflow

with workflow.unsafe.imports_passed_through():
    import httpx

여기에 google.genai를 나열할 필요는 없어요. 나중에 구성하는 Temporal 플러그인이 pydantic_core와 annotated_types와 함께 샌드박스 통과 세트에 추가해 줘요.

시스템 지침

다음으로 에이전트의 성격을 정의해요. 시스템 지침은 모델이 어떻게 행동할지 알려줘요. 이 에이전트는 도구가 필요하지 않을 때 하이쿠로 응답하도록 지시받아요.

SYSTEM_INSTRUCTIONS = """
You are a helpful agent that can use tools to help the user.
You will be given an input from the user and a list of tools to use.
You may or may not need to use the tools to satisfy the user ask.
If no tools are needed, respond in haikus.
"""

도구 정의

이제 에이전트가 사용할 도구를 정의해요. 각 도구는 일반 Temporal Activity예요. @activity.defn으로 데코레이트된 비동기 함수로, 타입 주석이 달린 매개변수와 설명적인 docstring을 가져요. Gemini는 그 시그니처와 docstring에서 함수 선언을 만들므로 각 매개변수를 Args 섹션에 문서화하세요.

import json

from temporalio import activity

NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-app/1.0"

@activity.defn
async def get_weather_alerts(state: str) -> str:
    """Get weather alerts for a US state.

    Args:
        state: Two-letter US state code (e.g. CA, NY)
    """
    headers = {"User-Agent": USER_AGENT, "Accept": "application/geo+json"}
    url = f"{NWS_API_BASE}/alerts/active/area/{state}"

    async with httpx.AsyncClient() as client:
        response = await client.get(url, headers=headers, timeout=5.0)
        response.raise_for_status()
        return json.dumps(response.json())

다음으로 IP 주소 지오로케이션용 도구를 정의해요.

@activity.defn
async def get_ip_address() -> str:
    """Get the public IP address of the current machine."""
    async with httpx.AsyncClient() as client:
        response = await client.get("https://icanhazip.com")
        response.raise_for_status()
        return response.text.strip()

@activity.defn
async def get_location_info(ipaddress: str) -> str:
    """Get the location information for an IP address including city, state, and country.

    Args:
        ipaddress: An IP address to look up
    """
    async with httpx.AsyncClient() as client:
        response = await client.get(f"http://ip-api.com/json/{ipaddress}")
        response.raise_for_status()
        result = response.json()
        return f"{result['city']}, {result['regionName']}, {result['country']}"

이것이 전체 도구 계층이에요. 도구 레지스트리도, FunctionDeclaration 구성도, 디스패치 테이블도 없어요. 다음 섹션에서 이 Activities를 activity_as_tool()로 래핑하는데, 이는 각 매개변수를 Activity에 위치적으로 전달해요. 매개변수가 0개, 1개 또는 여러 개인 도구 모두 동작해요.

에이전트 Workflow

이제 에이전트를 완성하는 데 필요한 모든 조각이 있어요. AgentWorkflow 클래스는 하나의 generate_content 호출을 만들어요. TemporalAsyncClient는 모든 API 호출이 Temporal Activity로 실행되는 드롭인 AsyncClient이고, activity_as_tool()은 각 Activity를 Gemini 도구로 바꿔요.

모델이 도구를 요청하면 SDK의 AFC 루프 — Workflow 내부에서 실행 — 가 workflow.execute_activity를 통해 디스패치하고, 결과를 대화에 추가한 뒤 모델을 다시 호출해요. 그 루프가 에이전트이며, 각 단계가 Temporal의 이벤트 히스토리에 기록된 Activity이므로 지속형이에요.

from datetime import timedelta

from google.genai import types
from temporalio.contrib.google_genai import TemporalAsyncClient, activity_as_tool
from temporalio.workflow import ActivityConfig

TOOL_CONFIG = ActivityConfig(start_to_close_timeout=timedelta(seconds=30))

@workflow.defn
class AgentWorkflow:
    """Agent workflow that uses Gemini for LLM calls and executes tools."""

    @workflow.run
    async def run(self, prompt: str) -> str:
        client = TemporalAsyncClient()

        response = await client.models.generate_content(
            model="gemini-3.8-flash",
            contents=prompt,
            config=types.GenerateContentConfig(
                system_instruction=SYSTEM_INSTRUCTIONS,
                tools=[
                    activity_as_tool(get_weather_alerts, activity_config=TOOL_CONFIG),
                    activity_as_tool(get_ip_address, activity_config=TOOL_CONFIG),
                    activity_as_tool(get_location_info, activity_config=TOOL_CONFIG),
                ],
            ),
        )

        # Leave this in place. You will un-comment it during a durability
        # test later on.
        # await workflow.sleep(timedelta(seconds=10))

        return response.text or ""

몇 가지 주의할 점이 있어요.

• TemporalAsyncClient를 Workflow 내부에서 생성하세요. 자격증명을 가지지 않으며, API 호출을 Activity 호출로 바꾸는 방법만 알아요.
• activity_config는 start_to_close_timeout 또는 schedule_to_close_timeout을 설정해야 해요. Temporal은 타임아웃을 요구하며 도구 Activity에 대한 기본값이 없어요.
• Gemini API Activities는 기본적으로 60초 start_to_close_timeout을 사용해요. 모델 호출이 더 오래 걸려야 한다면 TemporalAsyncClient(activity_config=...)로 재정의하세요.

에이전트는 완전히 지속형이에요. 몇 턴 후에 워커가 충돌하면 Temporal은 이미 실행된 LLM 호출이나 도구 호출을 다시 호출하지 않고 정확히 중단된 지점부터 이어받아요.

재시도

Temporal이 재시도를 소유하므로 Gemini SDK 자체의 재시도 루프를 활성화하지 마세요. Activity 구성의 retry_policy로 재시도 동작을 설정하세요.

from temporalio.common import RetryPolicy

TOOL_CONFIG = ActivityConfig(
    start_to_close_timeout=timedelta(seconds=30),
    retry_policy=RetryPolicy(maximum_attempts=3),
)

API 실패도 분류되어 제공돼요. 일시적 상태(408, 429, 5xx)는 재시도 가능하게 유지되어 Activity의 재시도 정책이 적용돼요. 다른 상태(예: 잘못된 요청의 400)는 재시도 불가능하므로, 해결되지 않을 오류에 시도를 낭비하는 대신 Workflow가 빠르게 실패해요.

그 분류를 확장할 수 있어요. 통합은 각 API 실패를 유형이 Gemini 예외 클래스 이름 — 4xx는 ClientError, 5xx는 ServerError — 인 ApplicationError로 표면화하므로, non_retryable_error_types에 이름을 나열하면 일시적 집합에서 빠져나와요. 예를 들어 Gemini 측 장애 재시도를 중단하고 첫 5xx에서 Workflow를 실패시키려면 TemporalAsyncClient를 통해 Gemini API Activities에 정책을 적용해요.

from temporalio.common import RetryPolicy

client = TemporalAsyncClient(
    activity_config=ActivityConfig(
        start_to_close_timeout=timedelta(seconds=60),
        retry_policy=RetryPolicy(
            maximum_attempts=5,
            non_retryable_error_types=["ServerError"],
        ),
    ),
)

워커 시작

마지막으로 모든 것을 연결해요. Temporal 워커는 Temporal 서비스에 연결되고 Workflow 및 Activity 작업의 스케줄러 역할을 해요.

여기서 API 키를 가진 실제 genai.Client가 생성돼요. GoogleGenAIPlugin은 해당 클라이언트를 받아 Gemini API Activities를 등록하고, Pydantic 데이터 변환기를 설치하며, Workflow 샌드박스를 구성해요.

import asyncio
import os

from dotenv import load_dotenv
from google import genai
from temporalio.client import Client
from temporalio.contrib.google_genai import GoogleGenAIPlugin
from temporalio.envconfig import ClientConfig
from temporalio.worker import Worker

async def main():
    gemini = genai.Client(api_key=os.environ["GOOGLE_API_KEY"])
    plugin = GoogleGenAIPlugin(gemini)

    config = ClientConfig.load_client_connect_config()
    config.setdefault("target_host", "localhost:7233")
    client = await Client.connect(**config, plugins=[plugin])

    worker = Worker(
        client,
        task_queue="gemini-agent",
        workflows=[
            AgentWorkflow,
        ],
        activities=[
            get_weather_alerts,
            get_ip_address,
            get_location_info,
        ],
    )
    await worker.run()

if __name__ == "__main__":
    load_dotenv()
    asyncio.run(main())

플러그인은 여기서 필요할 세 가지 보일러플레이트를 제거해요.

• data_converter=pydantic_data_converter 없음 — 플러그인이 Pydantic 페이로드 변환기를 직접 설치해요.
• activity_executor=ThreadPoolExecutor 없음 — 모든 Activity가 비동기예요.
• activities 목록의 Gemini Activities 없음 — 플러그인이 등록해요. 자체 도구만 등록하면 돼요.

에이전트 실행

그것이 전체 에이전트예요. 클라이언트를 작성할 필요 없이 Temporal CLI가 Workflow를 시작할 수 있어요.

아직 하지 않았다면 Temporal 개발 서버를 시작하세요.

temporal server start-dev

새 터미널 창에서 에이전트 워커를 시작하세요.

uv run durable_agent_worker.py

세 번째 터미널 창에서 에이전트에 쿼리를 제출하세요.

temporal workflow execute --type AgentWorkflow --task-queue gemini-agent \
    --input '"are there any weather alerts for where I am?"'

태스크 큐에 주목하세요. 워커가 폴링하는 것과 동일해요. Workflow를 시작하면 사용자 프롬프트를 담은 Workflow 태스크가 해당 큐로 디스패치되고, 그것이 에이전트를 시작해요. execute는 Workflow가 완료될 때까지 차단하고 결과를 출력해요. 기다리기 싫다면 명시적 --workflow-id와 함께 temporal workflow start를 사용하고, 나중에 temporal workflow result -w your-workflow-id로 결과를 수집하세요. --workflow-id를 생략하면 Temporal이 Workflow ID를 생성해 줘요.

--input은 JSON을 받으므로, 단일 문자열 프롬프트는 셸 따옴표 안에 자체 따옴표가 필요해요. CLI는 Gemini API 키가 필요 없고 데이터 변환기 구성도 필요 없어요. Workflow의 인자와 반환 값이 모두 일반 문자열이라 기본 JSON 페이로드 변환기가 처리해요.

http://localhost:8233/namespaces/default/workflows에서 Temporal UI를 열어 에이전트 루프가 펼쳐지는 것을 지켜봐요. 각 모델 턴마다 하나씩 gemini_api_client_async_request Activity가 도구 호출당 하나의 Activity와 번갈아 나타나는 것을 볼 수 있어요. 각각 tool_call 요약으로 라벨링돼요. 그 번갈아 나타남이 바로 AFC 루프를 지속적이고 관찰 가능하게 만든 것이에요.

몇 가지 다른 프롬프트를 시도해 에이전트가 추론하고 도구를 호출하는 것을 보세요. 각 명령은 위와 동일하며 새 --input을 가져요.

temporal workflow execute --type AgentWorkflow --task-queue gemini-agent \
    --input '"are there any weather alerts for New York?"'
temporal workflow execute --type AgentWorkflow --task-queue gemini-agent \
    --input '"where am I?"'
temporal workflow execute --type AgentWorkflow --task-queue gemini-agent \
    --input '"what is my ip address?"'
temporal workflow execute --type AgentWorkflow --task-queue gemini-agent \
    --input '"tell me a joke"'

마지막 프롬프트는 도구가 필요하지 않으므로 에이전트는 SYSTEM_INSTRUCTIONS에 따라 하이쿠로 응답해요.

지속성 테스트

Temporal을 기반으로 구축하면 에이전트가 장애를 매끄럽게 견디게 돼요. 두 가지 뚜렷한 실험으로 이것을 테스트할 수 있어요.

네트워크 장애 시뮬레이션

이 테스트에서는 컴퓨터의 인터넷 연결을 일시적으로 비활성화하고, Workflow를 제출하고, Temporal이 자동으로 재시도하는 것을 지켜본 뒤 네트워크를 복원해 회복을 확인해요.

1. 기기를 인터넷에서 분리하세요(예: Wi-Fi 끄기).
2. Workflow 제출: temporal workflow execute --type AgentWorkflow --task-queue gemini-agent --input '"tell me a joke"'
3. Temporal UI(http://localhost:8233)를 확인하세요. Gemini API Activity가 실패하고 Temporal이 백그라운드에서 재시도를 자동 관리하는 것을 볼 수 있어요.
4. 인터넷에 다시 연결하세요.
5. 다음 자동 재시도가 Gemini API에 성공적으로 도달하고, 터미널이 최종 결과를 출력할 거예요.

워커 충돌 생존

이 테스트에서는 실행 중에 워커를 죽이고 다시 시작해요. Temporal은 Workflow 히스토리(이벤트 소싱)를 재생하고 마지막 완료 Activity부터 재개해요. 이미 완료된 LLM 호출과 도구 호출은 반복되지 않아요.

1. 워커를 죽일 시간을 갖기 위해 durable_agent_worker.py를 열고 AgentWorkflow.run에서 지속형 타이머를 주석 해제하세요: await workflow.sleep(timedelta(seconds=10)). workflow.sleep은 로컬 타이머가 아니라 Temporal 타이머예요. 히스토리에 기록되어 재시작을 견디므로 이 테스트를 안정적으로 만들어 줘요.
2. 워커를 다시 시작하세요: uv run durable_agent_worker.py
3. 여러 도구를 트리거하는 쿼리를 제출하세요: temporal workflow execute --type AgentWorkflow --task-queue gemini-agent --input '"are there any weather alerts where I am?"'
4. 도구 호출이 완료되고 타이머가 실행 중일 때 워커 프로세스를 죽이세요(워커 터미널에서 Ctrl-C, 또는 백그라운드 실행 시 kill %1).
5. 워커를 다시 시작하세요: uv run durable_agent_worker.py

Temporal이 Workflow 히스토리를 재생해요. 이미 완료된 LLM 호출과 도구 호출은 다시 실행되지 않아요 — 결과가 히스토리(이벤트 로그)에서 즉시 재생되고, 타이머는 재개되며, Workflow가 성공적으로 끝나요.

더 나아가기

이 통합은 이 튜토리얼이 다루는 것보다 더 많은 것을 지원해요. 자세한 내용은 플러그인 문서를 참조하세요.

• Streaming. 평소처럼 generate_content_stream을 사용하세요. 외부 소비자(채팅 UI)가 Workflow가 지속적으로 실행되는 동안 청크를 실시간으로 관찰하도록 하려면 TemporalAsyncClient(streaming_topic=...)을 설정하고 Workflow에서 WorkflowStream을 호스팅하세요.
• MCP. GoogleGenAIPlugin(mcp_servers={...})으로 워커에 클라이언트 측 MCP 서버를 등록하고, Workflow에서 TemporalMcpClientSession으로 이름으로 참조하세요. 도구 발견과 호출은 풀링된 워커 측 연결에 대해 Activities로 실행돼요.
• Vertex AI. 워커 측 genai.Client와 Workflow 측 TemporalAsyncClient 양쪽에 vertexai=True를 전달하고, 재생이 결정적으로 유지되도록 Workflow 측에서 project와 location을 명시적으로 설정하세요.

추가 리소스

• Temporal 문서
• Temporal Python SDK
• Temporal Gemini SDK 통합
• Google GenAI SDK

더 알아보기 (Learn more)

• Temporal 문서
• Google GenAI SDK
• Robotics cookbook