Ky tutorial ju udhëzon në ndërtimin e një agjenti të qëndrueshëm të IA-së që përdor Gemini API për arsyetim dhe Temporal për qëndrueshmëri. Ai përdor integrimin e integruar të Gemini SDK të Temporal.
Agjenti mund të thërrasë mjete, si kërkimi i alarmeve të motit ose gjeolokacioni i një adrese IP, dhe do të kryejë cikël derisa të ketë informacion të mjaftueshëm për t'u përgjigjur.
Ajo që e bën këtë të ndryshme nga një demo tipike agjentësh është qëndrueshmëria . Çdo thirrje LLM dhe çdo thirrje mjeti vazhdon nga Temporal. Nëse procesi rrëzohet, rrjeti ndërpritet ose një API skadon, Temporal automatikisht riprovon dhe rifillon nga hapi i fundit i përfunduar. Nuk humbet historiku i bisedave dhe asnjë thirrje mjeti nuk përsëritet gabimisht.
Arkitekturë
Arkitektura përbëhet nga tre pjesë:
- Fluksi i Punës: Një thirrje e vetme
generate_content. Cikli i thirrjes automatike të funksionit (AFC) i Gemini SDK funksionon brenda Fluksit të Punës, dhe Temporal e bën çdo hap të tij të qëndrueshëm. - Aktivitetet: Njësi individuale pune që Temporal i bën të qëndrueshme. Thirrjet e Gemini API bëhen automatikisht Aktivitete.
- Punëtori: Procesi që ekzekuton Rrjedhat e Punës dhe Aktivitetet, dhe i vetmi vend ku ndodhet çelësi juaj API.
Në këtë shembull, do t'i vendosni të tre këto pjesë në një skedar të vetëm ( durable_agent_worker.py ). Në një implementim në botën reale, do t'i ndani ato për të lejuar avantazhe të ndryshme të vendosjes dhe shkallëzueshmërisë. Do t'i jepni agjentit udhëzime me CLI-në Temporale, kështu që nuk ka kod klienti për të shkruar.
Parakushte
Për të përfunduar këtë udhëzues, do t'ju duhet:
- Një çelës API Gemini. Mund të krijoni një falas në Google AI Studio .
- Versioni 3.10 i Python ose më i ri.
- uv për menaxhimin e varësive.
- CLI-ja Temporale për ekzekutimin e një serveri zhvillimi lokal dhe nisjen e rrjedhave të punës.
Konfigurimi
Para se të filloni, sigurohuni që keni një server zhvillimi Temporal që funksionon lokalisht:
temporal server start-devTjetra, krijoni një projekt dhe instaloni varësitë e kërkuara:
uv init durable-gemini-agentcd durable-gemini-agentuv add "temporalio[google-genai]" httpx python-dotenv
uv krijon dhe menaxhon mjedisin virtual për ju, kështu që çdo komandë Python më vonë në këtë tutorial ekzekutohet përmes uv run .
Krijo një skedar .env në direktorinë e projektit tënd me çelësin tënd Gemini API. Mund të marrësh një çelës API nga Google AI Studio .
echo "GOOGLE_API_KEY=your-api-key-here" > .envZbatimi
Pjesa tjetër e këtij tutoriali përshkruan durable_agent_worker.py nga lart poshtë, duke e ndërtuar agjentin pjesë-pjesë. Krijo skedarin dhe ndiqe më tej.
Importet dhe konfigurimi i sandbox-it
Filloni me importet që duhet të përcaktohen paraprakisht. Blloku workflow.unsafe.imports_passed_through() i tregon sandbox-it të Workflow-it të Temporal-it që të lejojë që httpx të kalojë pa kufizim. Importimi i httpx ekzekuton class _CookieCompatRequest(urllib.request.Request) , dhe sandbox-i bllokon nënklasifikimin e asaj klase stdlib.
Mjetet tuaja përdorin httpx , dhe activity_as_tool() ka nevojë për Workflow për të importuar ato funksione të mjeteve në mënyrë që Gemini të mund të nxjerrë skemat e tyre nga nënshkrimet. Pra, httpx arrin në sandbox pavarësisht se si i ndani skedarët - zhvendosja e mjeteve në modulin e tyre nuk e shmang atë.
from temporalio import workflow
with workflow.unsafe.imports_passed_through():
import httpx
Nuk keni nevojë ta listoni google.genai këtu. Shtojca Temporal që konfiguroni më vonë e shton atë—së bashku me pydantic_core dhe annotated_types —në setin e kalimit sandbox për ju.
Udhëzimet e sistemit
Më pas, përcaktoni personalitetin e agjentit. Udhëzimet e sistemit i tregojnë modelit se si të sillet. Këtij agjenti i është dhënë udhëzim të përgjigjet me haiku kur nuk nevojiten mjete.
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.
"""
Përkufizimet e mjeteve
Tani përcaktoni mjetet që agjenti mund të përdorë. Çdo mjet është një Aktivitet i zakonshëm Temporal: një funksion asinkron i dekoruar me @activity.defn , me parametra të shënuar sipas llojit dhe një docstring përshkrues. Gemini ndërton deklaratën e funksionit nga ajo nënshkrim dhe docstring, prandaj dokumentoni çdo parametër në seksionin 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())
Tjetra, përcaktoni mjetet për gjeolokacionin e adresës 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']}"
Kjo është e gjithë shtresa e mjeteve. Nuk ka regjistër mjetesh, nuk ka ndërtim FunctionDeclaration dhe nuk ka tabelë dërgimi - seksioni tjetër i mbështjell këto Aktivitete me activity_as_tool() , e cila ia kalon çdo parametër Activity në mënyrë pozicionale. Mjetet me zero, një ose disa parametra funksionojnë të gjitha.
Fluksi i punës së agjentit
Tani i keni të gjitha pjesët për të përfunduar ndërtimin e agjentit. Klasa AgentWorkflow bën një thirrje generate_content . TemporalAsyncClient është një AsyncClient i integruar, çdo thirrje API e të cilit ekzekutohet si një Aktivitet Temporal, dhe activity_as_tool() i kthen të gjitha Aktivitetet tuaja në një mjet Gemini.
Kur modeli kërkon një mjet, cikli AFC i SDK-së — që funksionon brenda Rrjedhës së Punës — e dërgon atë përmes workflow.execute_activity , ia shton rezultatin bisedës dhe e thërret modelin përsëri. Ky cikli është agjenti dhe është i qëndrueshëm sepse çdo hap është një Aktivitet i regjistruar në historikun e ngjarjeve të 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.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 ""
Disa gjëra për t'u vënë re:
- Ndërtoni
TemporalAsyncClientbrenda Workflow. Ai nuk mbart kredenciale; di vetëm si t'i shndërrojë thirrjet API në thirrje të Aktivitetit. -
activity_configduhet të caktojëstart_to_close_timeoutoseschedule_to_close_timeout. Vlera e përkohshme kërkon një skadencë dhe nuk ka një vlerë të parazgjedhur për Aktivitetet e mjetit. - Aktivitetet e API-t Gemini kanë një
start_to_close_timeoutprej 60 sekondash. Anashkalojeni atë meTemporalAsyncClient(activity_config=...)nëse thirrjet e modelit tuaj kërkojnë më shumë kohë.
Agjenti është plotësisht i qëndrueshëm. Nëse punonjësi rrëzohet pas disa kthesave, Temporal vazhdon saktësisht aty ku e la pa riaktivizuar thirrjet LLM ose thirrjet e mjeteve të ekzekutuara tashmë.
Ripërpjekje
Përkohësisht zotëron ripërpjekjet, prandaj mos e aktivizoni ciklin e ripërpjekjes së vetë Gemini SDK. Vendosni sjelljen e ripërpjekjes me një retry_policy në konfigurimin e Aktivitetit në vend të kësaj:
from temporalio.common import RetryPolicy
TOOL_CONFIG = ActivityConfig(
start_to_close_timeout=timedelta(seconds=30),
retry_policy=RetryPolicy(maximum_attempts=3),
)
Dështimet e API-t klasifikohen gjithashtu për ju. Statuset kalimtare (408, 429, 5xx) mbeten të riprovueshme, kështu që zbatohet politika e riprovimit të Aktivitetit; statuset e tjera (si p.sh. 400 për një kërkesë të keqformuar) nuk mund të riprovohen, kështu që Fluksi i Punës dështon shpejt në vend që të djegë përpjekjet për një gabim që nuk do të zgjidhet.
Ju mund ta zgjeroni atë klasifikim. Integrimi nxjerr në pah çdo dështim të API-t si një ApplicationError , lloji i të cilit është emri i klasës së përjashtimit Gemini— ClientError për 4xx, ServerError për 5xx—kështu që renditja e një emri në non_retryable_error_types e zhvendos atë nga grupi kalimtar. Për shembull, për të ndaluar riprovimin e ndërprerjeve nga ana e Gemini dhe dështimin e Workflow në 5xx-in e parë, zbatoni politikën në Aktivitetet e API-t Gemini përmes 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"],
),
),
)
Startup i punëtorëve
Së fundmi, lidhni gjithçka së bashku. Punëtori i Temporal lidhet me shërbimin Temporal dhe vepron si një planifikues për detyrat e Fluksit të Punës dhe Aktivitetit.
Këtu krijohet genai.Client i vërtetë, me çelësin tuaj API. GoogleGenAIPlugin e merr atë klient dhe regjistron Aktivitetet e API-t Gemini, instalon konvertuesin e të dhënave Pydantic dhe konfiguron sandbox-in e 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())
Shtojca heq tre pjesë të standardit standard që përndryshe do të nevojiteshin këtu:
- No
data_converter=pydantic_data_converter— plugin-i instalon vetë konvertuesin e ngarkesës Pydantic. - No
activity_executor=ThreadPoolExecutor—çdo aktivitet është asinkron. - Nuk ka aktivitete Gemini në listën e
activities—shtojca i regjistron ato. Ju regjistroni vetëm mjetet tuaja.
Ekzekutoni agjentin
Ky është i gjithë agjenti. Nuk keni nevojë të shkruani një klient - CLI Temporal mund të fillojë Rrjedhën e Punës për ju.
Nëse nuk e keni bërë ende, nisni serverin e zhvillimit Temporal:
temporal server start-devNë një dritare të re terminali, nisni agent worker-in:
uv run durable_agent_worker.pyNë një dritare të tretë të terminalit, dërgoni një pyetje te agjenti juaj:
temporal workflow execute --type AgentWorkflow --task-queue gemini-agent \
--input '"are there any weather alerts for where I am?"'Vini re radhën e detyrave: është e njëjta që anketon punëtori. Fillimi i Rrjedhës së Punës dërgon një detyrë të Rrjedhës së Punës që e çon kërkesën e përdoruesit në atë radhë, e cila është ajo që inicion agjentin. execute blloqe derisa Rrjedha e Punës të përfundojë dhe të printojë rezultatin. Nëse nuk dëshironi të prisni, përdorni temporal workflow start me një --workflow-id të qartë, pastaj mbledhni rezultatin më vonë me temporal workflow result -w your-workflow-id . Temporal gjeneron ID-në e Rrjedhës së Punës për ju kur hiqni dorë --workflow-id .
--input merr JSON, kështu që një prompt me varg të thjeshtë ka nevojë për thonjëzat e veta brenda thonjëzave të shell-it. CLI nuk ka nevojë për çelës Gemini API dhe as për konfigurim të konvertuesit të të dhënave: argumenti dhe vlera e kthimit të Workflow-it janë të dy vargje të thjeshta, të cilat i trajton konvertuesi i ngarkesës JSON parazgjedhur.
Hapni ndërfaqen e përdoruesit të përkohshëm në http://localhost:8233/namespaces/default/workflows për të parë zhvillimin e ciklit agentic. Do të shihni Aktivitetet gemini_api_client_async_request - një për çdo kthesë të modelit - të ndërthurura me një Aktivitet për thirrje mjeti, secila e etiketuar me një përmbledhje tool_call . Ky ndërthurje është cikli AFC, i bërë i qëndrueshëm dhe i vëzhgueshëm.
Provo disa kërkesa të ndryshme për të parë arsyen e agjentit dhe mjetet e thirrjes. Çdo komandë është e njëjtë me atë më sipër me një --input të ri:
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"'
Kërkesa e fundit nuk kërkon ndonjë mjet, kështu që agjenti përgjigjet në një haiku bazuar në SYSTEM_INSTRUCTIONS .
Testimi i qëndrueshmërisë
Ndërtimi mbi Temporal siguron që agjenti juaj t'i mbijetojë dështimeve pa probleme. Mund ta testoni këtë duke përdorur dy eksperimente të dallueshme.
Simulimi i një ndërprerjeje të rrjetit
Në këtë test, do ta çaktivizoni përkohësisht lidhjen e internetit të kompjuterit tuaj, do të paraqisni një Rrjedhë Pune, do të shikoni Temporal të provojë përsëri automatikisht dhe më pas do ta rivendosni rrjetin për ta parë atë të rikuperohet.
- Shkëputeni pajisjen tuaj nga interneti (për shembull, fikni Wi-Fi-në).
Dorëzoni një rrjedhë pune:
temporal workflow execute --type AgentWorkflow --task-queue gemini-agent \ --input '"tell me a joke"'Kontrolloni ndërfaqen e përdoruesit Temporal (
http://localhost:8233). Do të shihni që aktiviteti i Gemini API dështon dhe Temporal do të menaxhojë automatikisht ripërpjekjet në sfond.Rilidhu me internetin.
Përpjekja tjetër automatike do të arrijë me sukses në Gemini API dhe terminali juaj do të shtypë rezultatin përfundimtar.
Mbijetesa në një aksident pune
Në këtë test, ju e mbyllni punëtorin në mes të ekzekutimit dhe e ristartoni atë. Temporal riluan historikun e Fluksit të Punës (burimi i ngjarjeve) dhe rifillon nga Aktiviteti i fundit i përfunduar - thirrjet e LLM-së dhe thirrjet e mjeteve që janë tashmë të përfunduara nuk përsëriten.
Për t'i dhënë vetes kohë për të vrarë worker-in, hapni
durable_agent_worker.pydhe ç'komenti i kohëmatësit durable nëAgentWorkflow.run:await workflow.sleep(timedelta(seconds=10))workflow.sleepështë një kohëmatës i përkohshëm, jo lokal. Ai regjistrohet në histori dhe i mbijeton rinisjes, gjë që e bën këtë test të besueshëm.Rinisni punëtorin:
uv run durable_agent_worker.pyDërgoni një pyetje që aktivizon disa mjete:
temporal workflow execute --type AgentWorkflow --task-queue gemini-agent \ --input '"are there any weather alerts where I am?"'Pasi thirrjet e mjetit të kenë përfunduar dhe kohëmatësi të jetë në punë, mbyllni procesin punëtor (
Ctrl-Cnë terminalin punëtor osekill %1nëse po funksionon në sfond).Rinisni punëtorin:
uv run durable_agent_worker.py
Riluan përsëri historikun e Fluksit të Punës. Thirrjet LLM dhe thirrjet e mjeteve që janë përfunduar tashmë nuk riekzekutohen - rezultatet e tyre riprodhohen menjëherë nga historiku (regjistri i ngjarjeve), kohëmatësi rifillon dhe Fluksi i Punës përfundon me sukses.
Duke shkuar më tej
Integrimi mbështet më shumë sesa mbulon ky tutorial. Shihni dokumentacionin e plugin-it për detaje:
- Transmetim. Përdorni
generate_content_streamsi zakonisht. Për të lejuar një konsumator të jashtëm (një ndërfaqe përdoruesi bisede) të vëzhgojë pjesët në kohë reale ndërsa Fluksi i Punës funksionon në mënyrë të qëndrueshme, caktoniTemporalAsyncClient(streaming_topic=...)dhe strehoni njëWorkflowStreamnë Fluksin e Punës. - MCP. Regjistro një server MCP në anën e klientit te punëtori me
GoogleGenAIPlugin(mcp_servers={...})dhe referencoje atë me emër në Rrjedhën e Punës meTemporalMcpClientSession. Zbulimi i mjeteve dhe thirrjet ekzekutohen si Aktivitete kundrejt një lidhjeje të përbashkët në anën e punëtorit. - Vertex AI. Kaloni
vertexai=Truesi përgenai.Clientnë anën e punëtorit ashtu edhe përTemporalAsyncClientnë anën e rrjedhës së punës, duke e vendosurprojectdhelocationnë mënyrë të qartë në anën e rrjedhës së punës në mënyrë që riprodhimi të mbetet determinist.