عامل هوش مصنوعی بادوام با Gemini و Temporal

این آموزش شما را در ساخت یک عامل هوش مصنوعی بادوام که از Gemini API برای استدلال و Temporal برای دوام استفاده می‌کند، راهنمایی می‌کند. این عامل از ادغام داخلی Gemini SDK Temporal بهره می‌برد.

عامل می‌تواند ابزارهایی مانند جستجوی هشدارهای آب و هوا یا تعیین موقعیت مکانی یک آدرس IP را فراخوانی کند و تا زمانی که اطلاعات کافی برای پاسخ دادن نداشته باشد، به صورت حلقه‌ای عمل کند.

چیزی که این را از یک دموی عامل معمولی متمایز می‌کند، پایداری آن است. هر فراخوانی LLM و هر فراخوانی ابزار توسط Temporal حفظ می‌شود. اگر فرآیند از کار بیفتد، شبکه از کار بیفتد یا یک API دچار وقفه شود، Temporal به طور خودکار دوباره تلاش می‌کند و از آخرین مرحله تکمیل شده ادامه می‌دهد. هیچ سابقه مکالمه‌ای از بین نمی‌رود و هیچ فراخوانی ابزاری به اشتباه تکرار نمی‌شود.

معماری

معماری از سه بخش تشکیل شده است:

  • گردش کار: یک فراخوانی generate_content واحد. حلقه فراخوانی خودکار تابع (AFC) در Gemini SDK درون گردش کار اجرا می‌شود و Temporal هر مرحله از آن را بادوام می‌کند.
  • فعالیت‌ها: واحدهای کاری مجزا که Temporal آنها را بادوام می‌کند. فراخوانی‌های Gemini API به طور خودکار به فعالیت‌ها تبدیل می‌شوند.
  • Worker: فرآیندی که گردش‌های کاری و فعالیت‌ها را اجرا می‌کند و تنها جایی است که کلید 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 محیط مجازی را برای شما ایجاد و مدیریت می‌کند، بنابراین هر دستور پایتون که بعداً در این آموزش ارائه می‌شود، از طریق uv run اجرا می‌شود.

یک فایل .env در دایرکتوری پروژه خود با کلید API مربوط به Gemini خود ایجاد کنید. می‌توانید کلید API را از Google AI Studio دریافت کنید.

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

پیاده‌سازی

بقیه این آموزش، فایل durable_agent_worker.py را از بالا به پایین بررسی می‌کند و عامل را قطعه قطعه می‌سازد. فایل را ایجاد کنید و مراحل را دنبال کنید.

واردات و راه‌اندازی جعبه شنی

با importهایی شروع کنید که باید از قبل تعریف شوند. بلوک workflow.unsafe.imports_passed_through() به محیط sandbox مربوط به Workflow در Temporal می‌گوید که به httpx اجازه عبور بدون محدودیت را بدهد. import کردن httpx باعث اجرای class _CookieCompatRequest(urllib.request.Request) می‌شود و محیط sandbox از ایجاد subclasses از آن کلاس stdlib جلوگیری می‌کند.

ابزارهای شما از httpx استفاده می‌کنند و activity_as_tool() به Workflow نیاز دارد تا توابع آن ابزار را وارد کند تا Gemini بتواند طرحواره‌های آنها را از امضاها استخراج کند. بنابراین httpx صرف نظر از نحوه تقسیم فایل‌ها، به sandbox می‌رسد - انتقال ابزارها به ماژول خودشان از آن جلوگیری نمی‌کند.

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 معمولی است: یک تابع async که با @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 و هیچ جدول dispatch وجود ندارد - بخش بعدی این فعالیت‌ها را با activity_as_tool() پوشش می‌دهد، که هر پارامتر را به صورت موقعیتی به Activity ارسال می‌کند. ابزارهایی با صفر، یک یا چند پارامتر، همگی کار می‌کنند.

گردش کار عامل

حالا شما تمام قطعات لازم برای ساخت عامل را دارید. کلاس AgentWorkflow یک فراخوانی generate_content انجام می‌دهد. TemporalAsyncClient یک AsyncClient اضافه است که هر فراخوانی API آن به عنوان یک Temporal Activity اجرا می‌شود و activity_as_tool() هر یک از Activityهای شما را به یک ابزار Gemini تبدیل می‌کند.

وقتی مدل ابزاری را درخواست می‌کند، حلقه AFC مربوط به SDK - که درون Workflow اجرا می‌شود - آن را از طریق workflow.execute_activity ارسال می‌کند، نتیجه را به مکالمه اضافه می‌کند و دوباره مدل را فراخوانی می‌کند. آن حلقه، عامل (agent) است و به دلیل اینکه هر مرحله، یک Activity ثبت شده در تاریخچه رویدادهای Temporal است، ماندگار (Durable) است.

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 به یک timeout نیاز دارد و هیچ پیش‌فرضی برای ابزار Activities وجود ندارد.
  • فعالیت‌های API مربوط به Gemini به صورت پیش‌فرض روی start_to_close_timeout 60 ثانیه تنظیم شده‌اند. اگر فراخوانی‌های مدل شما به زمان بیشتری نیاز دارند، آن را با TemporalAsyncClient(activity_config=...) لغو کنید.

این عامل کاملاً بادوام است. اگر کارگر پس از چندین نوبت از کار بیفتد، Temporal دقیقاً از همان جایی که متوقف شده بود، بدون فراخوانی مجدد فراخوانی‌های LLM یا فراخوانی‌های ابزار که قبلاً اجرا شده‌اند، ادامه می‌دهد.

تلاش‌های مجدد

Temporal مالک تلاش‌های مجدد است، بنابراین حلقه‌ی تلاش مجدد Gemini SDK را فعال نکنید. در عوض، رفتار تلاش مجدد را با 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 برای یک درخواست ناقص) غیرقابل تلاش مجدد هستند، بنابراین گردش کار به جای اینکه تلاش‌های مربوط به خطایی که حل نمی‌شود را از بین ببرد، به سرعت با شکست مواجه می‌شود.

شما می‌توانید این طبقه‌بندی را گسترش دهید. این ادغام، هر خطای API را به عنوان یک ApplicationError که نوع آن نام کلاس استثنای Gemini است، نشان می‌دهد ClientError برای 4xx، ServerError برای 5xx - بنابراین قرار دادن یک نام در non_retryable_error_types آن را از مجموعه گذرا خارج می‌کند. به عنوان مثال، برای متوقف کردن تلاش مجدد برای قطع شدن‌های سمت Gemini و عدم موفقیت Workflow در 5xx اول، این سیاست را از طریق TemporalAsyncClient به فعالیت‌های API Gemini اعمال کنید: 

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 عمل می‌کند.

اینجاست که genai.Client واقعی با کلید API شما ایجاد می‌شود. GoogleGenAIPlugin آن کلاینت را می‌گیرد و فعالیت‌های Gemini API را ثبت می‌کند، مبدل داده 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 ناهمگام است.
  • هیچ فعالیت Gemini در لیست activities وجود ندارد - افزونه آنها را ثبت می‌کند. شما فقط ابزارهای خودتان را ثبت می‌کنید.

عامل را اجرا کنید

این کل agent است. نیازی به نوشتن کلاینت ندارید - 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?"'

به صف وظایف توجه کنید: این همان صفی است که worker از آن نظرسنجی می‌کند. شروع Workflow یک وظیفه Workflow را که حاوی اعلان کاربر است به آن صف ارسال می‌کند، که همان چیزی است که بلوک‌های agent.exe execute تا زمان تکمیل Workflow و چاپ نتیجه، آغاز می‌کند. اگر ترجیح می‌دهید منتظر نمانید، از temporal workflow start با یک --workflow-id صریح استفاده کنید، سپس نتیجه را بعداً با temporal workflow result -w your-workflow-id جمع‌آوری کنید. Temporal وقتی --workflow-id حذف می‌کنید، Workflow ID را برای شما تولید می‌کند.

--input JSON را دریافت می‌کند، بنابراین یک اعلان رشته‌ای ساده به نقل قول‌های مخصوص به خود در داخل نقل قول‌های پوسته نیاز دارد. رابط خط فرمان (CLI) به هیچ کلید API Gemini و هیچ پیکربندی مبدل داده‌ای نیز نیاز ندارد: آرگومان و مقدار بازگشتی Workflow هر دو رشته‌های ساده هستند که مبدل پیش‌فرض JSON آنها را مدیریت می‌کند.

رابط کاربری Temporal را در http://localhost:8233/namespaces/default/workflows باز کنید تا شاهد روند حلقه agentic باشید. خواهید دید که فعالیت‌های 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. اتصال دستگاه خود را به اینترنت قطع کنید (برای مثال، وای‌فای خود را خاموش کنید).

  2. ارسال گردش کار: 

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

  3. رابط کاربری Temporal را بررسی کنید ( http://localhost:8233 ). خواهید دید که فعالیت API Gemini با شکست مواجه می‌شود و Temporal به طور خودکار تلاش‌های مجدد را در پس‌زمینه مدیریت می‌کند.

  4. دوباره به اینترنت وصل شوید.

  5. تلاش مجدد خودکار بعدی با موفقیت به API Gemini خواهد رسید و ترمینال شما نتیجه نهایی را چاپ خواهد کرد.

زنده ماندن در تصادف کارگری

در این تست، شما worker را در اواسط اجرا از بین می‌برید و آن را مجدداً راه‌اندازی می‌کنید. Temporal تاریخچه گردش کار (منبع رویداد) را دوباره اجرا می‌کند و از آخرین فعالیت تکمیل‌شده ادامه می‌دهد - فراخوانی‌های LLM و فراخوانی‌های ابزار که قبلاً تکمیل شده‌اند، تکرار نمی‌شوند.

  1. برای اینکه به خودتان زمان بدهید تا worker را از بین ببرید، 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 سمت کلاینت را روی worker با GoogleGenAIPlugin(mcp_servers={...}) ثبت کنید و آن را با نام در Workflow با TemporalMcpClientSession ارجاع دهید. کشف ابزار و فراخوانی‌ها به عنوان Activityها در برابر یک اتصال worker-side pooled اجرا می‌شوند.
  • Vertex AI. مقدار vertexai=True را هم به genai.Client سمت worker و هم به TemporalAsyncClient سمت Workflow ارسال کنید، project و location را به طور صریح در سمت Workflow تنظیم کنید تا replay قطعی باقی بماند.

منابع بیشتر