Langlebiger KI-Agent mit Gemini und Temporal

In dieser Anleitung erfahren Sie, wie Sie eine ReAct-ähnliche agentische Schleife erstellen, die die Gemini API für die Schlussfolgerung und Temporal für die Dauerhaftigkeit verwendet. Der vollständige Quellcode für diese Anleitung ist auf GitHub verfügbar.

Der Agent kann Tools aufrufen, z. B. um Wetterwarnungen zu suchen oder eine IP-Adresse zu geolokalisieren. Er wird so lange wiederholt, bis er genügend Informationen für eine Antwort hat.

Was sich von einer typischen Agent-Demo unterscheidet, ist die Dauerhaftigkeit. Jeder LLM-Aufruf, jeder Tool-Aufruf und jeder Schritt der agentischen Schleife wird von Temporal gespeichert. Wenn der Prozess abstürzt, die Netzwerkverbindung unterbrochen wird oder ein API-Aufruf eine Zeitüberschreitung verursacht, wiederholt Temporal den Vorgang automatisch und setzt ihn beim letzten abgeschlossenen Schritt fort. Der Unterhaltungsverlauf geht nicht verloren und Tool-Aufrufe werden nicht fälschlicherweise wiederholt.

Architektur

Die Architektur besteht aus drei Teilen:

• Workflow:Die agentische Schleife, die die Ausführungslogik orchestriert.
• Aktivitäten:Einzelne Arbeitseinheiten (LLM-Aufrufe, Tool-Aufrufe), die von Temporal dauerhaft gemacht werden.
• Worker:Der Prozess, der die Workflows und Aktivitäten ausführt.

In diesem Beispiel werden alle drei Teile in einer einzigen Datei (durable_agent_worker.py) platziert. In einer realen Implementierung würden Sie sie trennen, um verschiedene Vorteile bei der Bereitstellung und Skalierbarkeit zu nutzen. Der Code, der dem Agent einen Prompt liefert, wird in einer zweiten Datei (start_workflow.py) platziert.

Vorbereitung

Für diese Anleitung benötigen Sie Folgendes:

• Einen Gemini API-Schlüssel. Sie können einen kostenlos in Google AI Studio erstellen.
• Python-Version 3.10 oder höher.
• Die Temporal CLI zum Ausführen eines lokalen Entwicklungsservers.

Einrichtung

Bevor Sie beginnen, muss ein Temporal-Entwicklungsserver lokal ausgeführt werden:

temporal server start-dev

Installieren Sie als Nächstes die erforderlichen Abhängigkeiten:

pip install temporalio google-genai httpx pydantic python-dotenv

Erstellen Sie in Ihrem Projektverzeichnis eine .env-Datei mit Ihrem Gemini API-Schlüssel. Sie können einen API-Schlüssel von Google AI Studioabrufen.

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

Implementierung

Im weiteren Verlauf dieser Anleitung wird durable_agent_worker.py von oben nach unten durchgegangen und der Agent Schritt für Schritt erstellt. Erstellen Sie die Datei und folgen Sie der Anleitung.

Importe und Sandbox-Einrichtung

Beginnen Sie mit den Importen, die im Voraus definiert werden müssen. Der Block workflow.unsafe.imports_passed_through() weist die Workflow-Sandbox von Temporal an, bestimmte Module ohne Einschränkung durchzulassen. Das ist erforderlich, da mehrere Bibliotheken (insbesondere httpx, das von urllib.request.Request abgeleitet wird) Muster verwenden, die von der Sandbox andernfalls blockiert würden.

from temporalio import workflow

with workflow.unsafe.imports_passed_through():
    import pydantic_core  # noqa: F401
    import annotated_types  # noqa: F401

    import httpx
    from pydantic import BaseModel, Field
    from google import genai
    from google.genai import types

Systemanweisungen

Definieren Sie als Nächstes die Persönlichkeit des Agenten. Die Systemanweisungen geben an, wie sich das Modell verhalten soll. Dieser Agent wird angewiesen, in Haikus zu antworten, wenn keine Tools erforderlich sind.

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.
"""

Tooldefinitionen

Definieren Sie nun die Tools, die der Agent verwenden kann. Jedes Tool ist eine asynchrone Funktion mit einem beschreibenden Docstring. Tools, die Parameter verwenden, haben ein Pydantic-Modell als einziges Argument. Das ist eine Best Practice von Temporal, mit der die Aktivitätssignaturen stabil bleiben, wenn Sie im Laufe der Zeit optionale Felder hinzufügen.

import json

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

class GetWeatherAlertsRequest(BaseModel):
    """Request model for getting weather alerts."""

    state: str = Field(description="Two-letter US state code (e.g. CA, NY)")

async def get_weather_alerts(request: GetWeatherAlertsRequest) -> str:
    """Get weather alerts for a US state.

    Args:
        request: The request object containing:
            - 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/{request.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())

Definieren Sie als Nächstes Tools für die Geolokalisierung von IP-Adressen:

class GetLocationRequest(BaseModel):
    """Request model for getting location info from an IP address."""

    ipaddress: str = Field(description="An IP address")

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()

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

    Args:
        request: The request object containing:
            - ipaddress: An IP address to look up
    """
    async with httpx.AsyncClient() as client:
        response = await