Надежный ИИ-агент с технологиями Gemini и Temporal.

В этом руководстве вы узнаете, как создать надежный ИИ-агента, использующего API Gemini для рассуждений и Temporal для обеспечения надежности. Он использует встроенную интеграцию Temporal с Gemini SDK .

Агент может запускать инструменты, например, для поиска предупреждений о погоде или определения местоположения IP-адреса, и будет зацикливаться до тех пор, пока не получит достаточно информации для ответа.

Отличием от типичной демонстрации работы агента является надежность . Каждый вызов LLM и каждый вызов инструмента сохраняются в Temporal. Если процесс завершается с ошибкой, обрывается связь или истекает время ожидания API, Temporal автоматически повторяет попытку и возобновляет работу с последнего завершенного шага. История разговоров не теряется, и вызовы инструментов не повторяются некорректно.

Архитектура

Архитектура состоит из трех частей:

  • Рабочий процесс: Один вызов generate_content . Цикл автоматического вызова функций (AFC) SDK Gemini выполняется внутри рабочего процесса, а Temporal обеспечивает надежность каждого его шага.
  • Действия: Отдельные рабочие единицы, которые Temporal обеспечивает как устойчивые. Вызовы API Gemini автоматически становятся действиями.
  • Рабочий процесс: Процесс, который выполняет рабочие процессы и действия, и единственное место, где хранится ваш API-ключ.

В этом примере вы поместите все три компонента в один файл ( durable_agent_worker.py ). В реальной реализации вы бы разделили их, чтобы обеспечить различные преимущества развертывания и масштабируемости. Вы будете передавать запросы агенту с помощью 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 .

Создайте в каталоге вашего проекта файл .env , содержащий ваш API-ключ Gemini. Вы можете получить API-ключ в Google AI Studio .

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

Выполнение

В оставшейся части этого руководства мы подробно рассмотрим файл durable_agent_worker.py , шаг за шагом создавая агента. Создайте файл и следуйте инструкциям.

Импорт и настройка песочницы

Начнём с импорта, который необходимо определить заранее. Блок workflow.unsafe.imports_passed_through() указывает песочнице Workflow в Temporal разрешить передачу httpx без ограничений. Импорт httpx запускает class _CookieCompatRequest(urllib.request.Request) , и песочница блокирует создание подкласса этого стандартного класса.

Ваши инструменты используют httpx , и activity_as_tool() требует, чтобы рабочий процесс импортировал эти функции инструментов, чтобы 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.
"""

Определения инструментов

Теперь определим инструменты, которые может использовать агент. Каждый инструмент представляет собой обычную временную активность: асинхронную функцию, помеченную как @activity.defn , с параметрами, аннотированными по типу, и описательной строкой документации. Gemini формирует объявление функции на основе этой сигнатуры и строки документации, поэтому задокументируйте каждый параметр в разделе 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 и нет таблицы диспетчеризации — в следующем разделе эти Activity оборачиваются функцией activity_as_tool() , которая передает каждый параметр в Activity позиционно. Инструменты с нулевым, одним или несколькими параметрами — все работают.

Рабочий процесс агента

Теперь у вас есть все необходимые компоненты для завершения создания агента. Класс AgentWorkflow выполняет один вызов generate_content . TemporalAsyncClient — это готовый AsyncClient , каждый вызов API которого выполняется как Temporal Activity, а activity_as_tool() превращает каждую из ваших Activity в инструмент Gemini.

Когда модель запрашивает инструмент, цикл AFC SDK, работающий внутри рабочего процесса, отправляет запрос через workflow.execute_activity , добавляет результат в диалог и снова вызывает модель. Этот цикл является агентом, и он является надежным, поскольку каждый шаг представляет собой действие, записанное в историю событий Temporal.

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.7-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 требуется таймаут, а для инструментов Activities значения по умолчанию отсутствуют.
  • В Gemini API Activities по умолчанию установлено start_to_close_timeout в 60 секунд. Если для вызовов вашей модели требуется более длительное время, переопределите его с помощью TemporalAsyncClient(activity_config=...)

Агент обладает полной отказоустойчивостью. Если рабочий процесс завершается с ошибкой после нескольких ходов, Temporal продолжает работу с того места, где остановился, без повторного вызова уже выполненных вызовов LLM или вспомогательных функций.

Повторные попытки

Функция повторных попыток отвечает за Temporal, поэтому не включайте собственный цикл повторных попыток SDK Gemini. Вместо этого настройте поведение повторных попыток с помощью параметра retry_policy в конфигурации Activity:

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 как ApplicationError , тип которой соответствует имени класса исключений Gemini — ClientError для ошибок 4xx, ServerError для ошибок 5xx, — поэтому указание имени в non_retryable_error_types выводит ошибку из временного набора. Например, чтобы прекратить повторные попытки устранения сбоев на стороне Gemini и прервать рабочий процесс при первой ошибке 5xx, примените политику к действиям API Gemini через TemporalAsyncClient : 

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"],
        ),
    ),
)

Запуск рабочего проекта

Наконец, соедините все компоненты. Временный обработчик подключается к службе временных процессов и выступает в качестве планировщика для задач рабочего процесса и действий.

Здесь создаётся настоящий объект genai.Client с вашим API-ключом. GoogleGenAIPlugin берёт этот клиент, регистрирует действия API Gemini, устанавливает конвертер данных Pydantic и настраивает песочницу рабочего процесса. 

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 может запустить рабочий процесс за вас.

Если вы еще этого не сделали, запустите сервер разработки 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?"'

Обратите внимание на очередь задач: это та же самая очередь, которую опрашивает рабочий процесс. Запуск рабочего процесса отправляет в эту очередь задачу рабочего процесса, содержащую запрос пользователя, которая инициирует работу агента. execute блокируется до тех пор, пока рабочий процесс не завершится и не выведет результат. Если вы не хотите ждать, используйте команду temporal workflow start с явным указанием --workflow-id , а затем получите результат позже с помощью temporal workflow result -w your-workflow-id . Temporal сгенерирует идентификатор рабочего процесса автоматически, если вы не укажете --workflow-id .

Параметр --input принимает JSON, поэтому строковый запрос должен быть заключен в кавычки внутри кавычек оболочки. Для работы CLI не требуется ключ API Gemini и конфигурация преобразователя данных: аргумент и возвращаемое значение Workflow представляют собой обычные строки, которые обрабатываются преобразователем полезной нагрузки JSON по умолчанию.

Откройте интерфейс Temporal UI по http://localhost:8233/namespaces/default/workflows , чтобы наблюдать за развертыванием цикла обработки запросов. Вы увидите действия gemini_api_client_async_request — по одному на каждый цикл модели — чередующиеся с одним действием на каждый вызов инструмента, каждое из которых помечено сводкой 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 гарантирует, что ваш агент будет бесперебойно работать после сбоев. Вы можете проверить это, проведя два отдельных эксперимента.

Имитация сбоя в сети

В этом тесте вам нужно будет временно отключить интернет-соединение на вашем компьютере, запустить рабочий процесс, понаблюдать за автоматической повторной попыткой подключения со стороны Temporal, а затем восстановить сетевое соединение, чтобы увидеть, как оно восстановится.

  1. Отключите компьютер от интернета (например, выключите Wi-Fi).

  2. Отправить рабочий процесс: 

    temporal workflow execute --type AgentWorkflow --task-queue gemini-agent \
    --input '"tell me a joke"'

  3. Проверьте пользовательский интерфейс Temporal ( http://localhost:8233 ). Вы увидите, что операция Gemini API завершается с ошибкой, а Temporal автоматически обрабатывает повторные попытки в фоновом режиме.

  4. Восстановите подключение к интернету.

  5. Следующая автоматическая попытка подключения успешно завершится при попытке связаться с API Gemini, и ваш терминал выведет окончательный результат.

Выжить после аварии на рабочем месте

В этом тесте вы прерываете выполнение рабочего процесса и перезапускаете его. Temporal воспроизводит историю рабочего процесса (использование событий) и возобновляет работу с последнего завершенного действия — уже завершенные вызовы LLM и вызовы инструментов не повторяются.

  1. Чтобы у вас было время для завершения работы воркера, откройте файл durable_agent_worker.py и раскомментируйте таймер durable в AgentWorkflow.run : 

    await workflow.sleep(timedelta(seconds=10))

    workflow.sleep — это временной таймер, а не локальный. Он записывается в историю и сохраняется после перезапуска, что и делает этот тест надежным.

  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 воспроизводит историю рабочего процесса. Вызовы LLM и запуски инструментов, которые уже завершились, не выполняются повторно — их результаты мгновенно воспроизводятся из истории (журнала событий), таймер возобновляется, и рабочий процесс успешно завершается.

Идем дальше

Интеграция поддерживает больше функций, чем описано в этом руководстве. Подробности см. в документации к плагину :

  • Потоковая передача. Используйте generate_content_stream как обычно. Чтобы внешний пользователь (интерфейс чата) мог наблюдать за фрагментами в реальном времени, пока рабочий процесс выполняется постоянно, установите TemporalAsyncClient(streaming_topic=...) и разместите WorkflowStream в рабочем процессе.
  • MCP. Зарегистрируйте клиентский MCP-сервер на рабочем узле с помощью GoogleGenAIPlugin(mcp_servers={...}) и укажите его по имени в рабочем процессе с помощью TemporalMcpClientSession . Обнаружение инструментов и вызовы выполняются как действия (Activity) с использованием пула соединений на стороне рабочего узла.
  • Vertex AI. Передайте vertexai=True как в genai.Client на стороне рабочего процесса, так и TemporalAsyncClient на стороне рабочего процесса, явно задав project и location на стороне рабочего процесса, чтобы воспроизведение оставалось детерминированным.

Дополнительные ресурсы