Надежный ИИ-агент с технологиями 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-адресу: