জেমিনি এবং টেম্পোরালের সাথে টেকসই এআই এজেন্ট

এই টিউটোরিয়ালটি আপনাকে একটি টেকসই এআই এজেন্ট তৈরি করার পদ্ধতি ধাপে ধাপে দেখাবে, যা যুক্তির জন্য জেমিনি এপিআই এবং স্থায়িত্বের জন্য টেম্পোরাল ব্যবহার করে। এটি টেম্পোরালের বিল্ট-ইন জেমিনি এসডিকে ইন্টিগ্রেশন ব্যবহার করে।

এজেন্টটি আবহাওয়ার সতর্কতা খোঁজা বা কোনো আইপি অ্যাড্রেসের অবস্থান নির্ণয় করার মতো টুলগুলোকে কল করতে পারে এবং সাড়া দেওয়ার জন্য পর্যাপ্ত তথ্য না পাওয়া পর্যন্ত এই প্রক্রিয়াটি পুনরাবৃত্তি করতে থাকবে।

একটি সাধারণ এজেন্ট ডেমো থেকে এটিকে যা আলাদা করে তা হলো এর স্থায়িত্ব । প্রতিটি LLM কল এবং প্রতিটি টুল ইনভোকেশন Temporal দ্বারা সংরক্ষিত থাকে। যদি প্রসেসটি ক্র্যাশ করে, নেটওয়ার্ক বিচ্ছিন্ন হয়ে যায়, বা কোনো API-এর সময়সীমা শেষ হয়ে যায়, Temporal স্বয়ংক্রিয়ভাবে পুনরায় চেষ্টা করে এবং সর্বশেষ সম্পন্ন হওয়া ধাপ থেকে কাজ পুনরায় শুরু করে। কোনো কথোপকথনের ইতিহাস হারিয়ে যায় না এবং কোনো টুল কল ভুলভাবে পুনরাবৃত্তি হয় না।

স্থাপত্য

স্থাপত্যটি তিনটি অংশ নিয়ে গঠিত:

  • ওয়ার্কফ্লো: একটিমাত্র generate_content কল। জেমিনি এসডিকে-র অটোমেটিক ফাংশন কলিং (AFC) লুপটি ওয়ার্কফ্লো-র ভেতরে চলে, এবং টেম্পোরাল এর প্রতিটি ধাপকে টেকসই করে তোলে।
  • অ্যাক্টিভিটি: কাজের স্বতন্ত্র একক যা টেম্পোরাল স্থায়ী করে তোলে। জেমিনি এপিআই কলগুলো স্বয়ংক্রিয়ভাবে অ্যাক্টিভিটিতে পরিণত হয়।
  • ওয়ার্কার: যে প্রসেসটি ওয়ার্কফ্লো এবং অ্যাক্টিভিটিগুলো সম্পাদন করে, এবং একমাত্র এখানেই আপনার এপিআই কী থাকে।

এই উদাহরণে, আপনি এই তিনটি অংশই একটি ফাইলে ( durable_agent_worker.py ) রাখবেন। বাস্তব প্রয়োগে, বিভিন্ন ডেপ্লয়মেন্ট এবং স্কেলেবিলিটি সুবিধার জন্য আপনি এগুলোকে আলাদা করে রাখবেন। আপনি টেম্পোরাল সিএলআই (Temporal CLI) ব্যবহার করে এজেন্টকে প্রম্পট সরবরাহ করবেন, তাই কোনো ক্লায়েন্ট কোড লেখার প্রয়োজন নেই।

পূর্বশর্ত

এই নির্দেশিকাটি সম্পূর্ণ করতে আপনার প্রয়োজন হবে:

  • একটি জেমিনি এপিআই কী। আপনি গুগল এআই স্টুডিও- তে বিনামূল্যে এটি তৈরি করতে পারেন।
  • পাইথন সংস্করণ ৩.১০ বা তার পরবর্তী সংস্করণ।
  • নির্ভরতা ব্যবস্থাপনার জন্য ইউভি ।
  • স্থানীয় ডেভেলপমেন্ট সার্ভার চালানো এবং ওয়ার্কফ্লো শুরু করার জন্য টেম্পোরাল সিএলআই ।

সেটআপ

শুরু করার আগে, নিশ্চিত করুন যে আপনার লোকালি একটি টেম্পোরাল ডেভেলপমেন্ট সার্ভার চালু আছে:

temporal server start-dev

এরপর, একটি প্রজেক্ট তৈরি করুন এবং প্রয়োজনীয় ডিপেন্ডেন্সিগুলো ইনস্টল করুন:

uv init durable-gemini-agent
cd durable-gemini-agent
uv add "temporalio[google-genai]" httpx python-dotenv

uv আপনার জন্য ভার্চুয়াল এনভায়রনমেন্ট তৈরি ও পরিচালনা করে, তাই এই টিউটোরিয়ালের পরবর্তী প্রতিটি পাইথন কমান্ড uv run এর মাধ্যমে রান হয়।

আপনার প্রজেক্ট ডিরেক্টরিতে আপনার জেমিনি এপিআই কী (Gemini API key) দিয়ে একটি .env ফাইল তৈরি করুন। আপনি গুগল এআই স্টুডিও (Google AI Studio) থেকে একটি এপিআই কী পেতে পারেন।

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

বাস্তবায়ন

এই টিউটোরিয়ালের বাকি অংশে durable_agent_worker.py ফাইলটি শুরু থেকে শেষ পর্যন্ত ধাপে ধাপে তৈরি করার প্রক্রিয়া দেখানো হয়েছে। ফাইলটি তৈরি করুন এবং নির্দেশাবলী অনুসরণ করুন।

আমদানি এবং স্যান্ডবক্স সেটআপ

যে ইম্পোর্টগুলো আগে থেকেই সংজ্ঞায়িত করতে হবে, সেগুলো দিয়ে শুরু করুন। workflow.unsafe.imports_passed_through() ব্লকটি Temporal-এর Workflow স্যান্ডবক্সকে httpx কোনো বাধা ছাড়াই পাস-থ্রু করতে বলে। httpx ইম্পোর্ট করলে class _CookieCompatRequest(urllib.request.Request) এক্সিকিউট হয়, এবং স্যান্ডবক্সটি সেই stdlib ক্লাসটির সাবক্লাসিং ব্লক করে দেয়।

আপনার টুলগুলো httpx ব্যবহার করে, এবং activity_as_tool() ফাংশনটির জন্য ওয়ার্কফ্লো-কে সেই টুল ফাংশনগুলো ইম্পোর্ট করতে হয়, যাতে জেমিনি সিগনেচারগুলো থেকে তাদের স্কিমা তৈরি করতে পারে। সুতরাং, আপনি ফাইলগুলোকে যেভাবে ভাগ করুন না কেন, 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 দ্বারা সজ্জিত একটি অ্যাসিঙ্ক ফাংশন, যার প্যারামিটারগুলো টাইপ-অ্যানোটেড এবং একটি বর্ণনামূলক ডকস্ট্রিং থাকে। জেমিনি সেই সিগনেচার এবং ডকস্ট্রিং থেকে ফাংশন ডিক্লারেশন তৈরি করে, তাই 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())

এরপরে, আইপি অ্যাড্রেস জিওলোকেশনের জন্য টুলগুলো সংজ্ঞায়িত করুন:

@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_as_tool() দিয়ে র‍্যাপ করা হয়, যা প্রতিটি প্যারামিটারকে পজিশনালি অ্যাক্টিভিটিতে পাঠিয়ে দেয়। শূন্য, এক বা একাধিক প্যারামিটারযুক্ত টুল—সবই কাজ করে।

এজেন্ট ওয়ার্কফ্লো

এখন এজেন্টটি তৈরি শেষ করার জন্য আপনার কাছে সমস্ত উপাদান রয়েছে। AgentWorkflow ক্লাসটি একটি generate_content কল করে। TemporalAsyncClient হলো একটি ড্রপ-ইন AsyncClient যার প্রতিটি API কল একটি Temporal Activity হিসেবে চলে, এবং activity_as_tool() আপনার প্রতিটি Activity-কে একটি Gemini টুলে পরিণত করে।

যখন মডেলটি কোনো টুলের জন্য অনুরোধ করে, তখন ওয়ার্কফ্লো-এর ভেতরে চলমান SDK-এর AFC লুপটি workflow.execute_activity এর মাধ্যমে সেটিকে প্রেরণ করে, ফলাফলটি কনভারসেশনে যুক্ত করে এবং মডেলটিকে পুনরায় কল করে। এই লুপটিই হলো এজেন্ট, এবং এটি টেকসই কারণ এর প্রতিটি ধাপ হলো একটি অ্যাক্টিভিটি যা টেম্পোরাল-এর ইভেন্ট হিস্ট্রিতে রেকর্ড করা থাকে।

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

কয়েকটি বিষয় লক্ষণীয়:

  • Workflow-এর ভিতরে TemporalAsyncClient তৈরি করুন। এর কোনো ক্রেডেনশিয়াল থাকে না; এটি শুধু জানে কীভাবে API কলগুলোকে Activity ইনভোকেশনে পরিণত করতে হয়।
  • activity_config অবশ্যই start_to_close_timeout অথবা schedule_to_close_timeout সেট করতে হবে। Temporal-এর জন্য একটি টাইমআউট প্রয়োজন এবং টুল Activities-এর জন্য কোনো ডিফল্ট মান নেই।
  • জেমিনি এপিআই অ্যাক্টিভিটিগুলোর ডিফল্ট start_to_close_timeout ৬০ সেকেন্ডের হয়ে থাকে। আপনার মডেল কলগুলোর জন্য যদি আরও বেশি সময়ের প্রয়োজন হয়, তবে TemporalAsyncClient(activity_config=...) ব্যবহার করে এটিকে ওভাররাইড করুন।

এজেন্টটি সম্পূর্ণরূপে টেকসই। যদি কর্মীটি কয়েকটি টার্নের পর ক্র্যাশ করে, তাহলে টেম্পোরাল ইতিমধ্যে সম্পাদিত এলএলএম কল বা টুল কলগুলোকে পুনরায় আহ্বান না করেই ঠিক যেখান থেকে থেমেছিল সেখান থেকেই আবার শুরু করে।

পুনরায় চেষ্টা

Temporal-এর নিজস্ব রিট্রাই ব্যবস্থা রয়েছে, তাই Gemini SDK-এর নিজস্ব রিট্রাই লুপ চালু করবেন না। এর পরিবর্তে, Activity কনফিগে একটি retry_policy ব্যবহার করে রিট্রাই আচরণ সেট করুন:

from temporalio.common import RetryPolicy

TOOL_CONFIG = ActivityConfig(
    start_to_close_timeout=timedelta(seconds=30),
    retry_policy=RetryPolicy(maximum_attempts=3),
)

এপিআই ব্যর্থতাগুলোও আপনার জন্য শ্রেণীবদ্ধ করা হয়। ক্ষণস্থায়ী স্ট্যাটাসগুলো (৪০৮, ৪২৯, ৫xx) পুনরায় চেষ্টাযোগ্য থাকে, ফলে অ্যাক্টিভিটির রিট্রাই পলিসি প্রযোজ্য হয়; অন্যান্য স্ট্যাটাসগুলো (যেমন একটি ত্রুটিপূর্ণ অনুরোধের জন্য ৪০০) পুনরায় চেষ্টাযোগ্য নয়, তাই যে ত্রুটি সমাধান হবে না তার জন্য চেষ্টা নষ্ট না করে ওয়ার্কফ্লোটি দ্রুত ব্যর্থ হয়।

আপনি সেই শ্রেণিবিন্যাসটি প্রসারিত করতে পারেন। ইন্টিগ্রেশন প্রতিটি API ব্যর্থতাকে একটি ApplicationError হিসাবে চিহ্নিত করে, যার টাইপ হলো Gemini এক্সেপশন ক্লাসের নাম— 4xx-এর জন্য ClientError , 5xx-এর জন্য ServerError ফলে non_retryable_error_types এ কোনো নাম তালিকাভুক্ত করলে তা ট্রানজিয়েন্ট সেট থেকে বেরিয়ে যায়। উদাহরণস্বরূপ, Gemini-সাইডের বিভ্রাট পুনরায় চেষ্টা করা বন্ধ করতে এবং প্রথম 5xx-এর ক্ষেত্রে ওয়ার্কফ্লো ব্যর্থ করতে, TemporalAsyncClient এর মাধ্যমে Gemini API Activities-গুলিতে পলিসিটি প্রয়োগ করুন: 

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 তৈরি করা হয়। GoogleGenAIPlugin সেই ক্লায়েন্টটিকে নিয়ে Gemini API Activities রেজিস্টার করে, 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 ছাড়া —প্লাগইনটি নিজেই পাইড্যান্টিক পেলোড কনভার্টারটি ইনস্টল করে।
  • activity_executor=ThreadPoolExecutor ব্যবহার করা যাবে না — প্রতিটি Activity অ্যাসিঙ্ক।
  • activities তালিকায় কোনো জেমিনি অ্যাক্টিভিটি নেই—প্লাগইন সেগুলো রেজিস্টার করে। আপনি শুধু আপনার নিজের টুলগুলোই রেজিস্টার করেন।

এজেন্টটি চালান

এটাই সম্পূর্ণ এজেন্ট। আপনার কোনো ক্লায়েন্ট লেখার প্রয়োজন নেই—টেম্পোরাল সিএলআই আপনার জন্য ওয়ার্কফ্লো শুরু করে দিতে পারে।

যদি ইতিমধ্যে না করে থাকেন, তাহলে টেম্পোরাল ডেভেলপমেন্ট সার্ভারটি চালু করুন: 

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 ব্লক হয়ে থাকে। আপনি যদি অপেক্ষা করতে না চান, তাহলে একটি সুস্পষ্ট --workflow-id সহ temporal workflow start ব্যবহার করুন, এবং পরে temporal workflow result -w your-workflow-id দিয়ে ফলাফল সংগ্রহ করুন। আপনি --workflow-id বাদ দিলে Temporal আপনার জন্য ওয়ার্কফ্লো আইডি তৈরি করে দেয়।

--input JSON গ্রহণ করে, তাই একটি সাধারণ স্ট্রিং প্রম্পটের জন্য শেল কোটেশনের ভিতরে নিজস্ব কোটেশনের প্রয়োজন হয়। CLI-এর জন্য কোনো Gemini API কী বা ডেটা কনভার্টার কনফিগারেশনের প্রয়োজন নেই: Workflow-এর আর্গুমেন্ট এবং রিটার্ন ভ্যালু উভয়ই সাধারণ স্ট্রিং, যা ডিফল্ট JSON পেলোড কনভার্টার দ্বারা পরিচালিত হয়।

এজেন্টিক লুপের প্রক্রিয়াটি দেখতে http://localhost:8233/namespaces/default/workflows এ টেম্পোরাল UI খুলুন। আপনি দেখবেন, প্রতিটি মডেল টার্নের জন্য একটি করে 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 উপর ভিত্তি করে একটি হাইকুতে উত্তর দেয়।

স্থায়িত্ব পরীক্ষা করুন

টেম্পোরাল-এর উপর ভিত্তি করে তৈরি করা হলে আপনার এজেন্ট ব্যর্থতা সত্ত্বেও নির্বিঘ্নে টিকে থাকতে পারে। আপনি দুটি স্বতন্ত্র পরীক্ষার মাধ্যমে এটি যাচাই করতে পারেন।

নেটওয়ার্ক বিভ্রাট অনুকরণ করা

এই পরীক্ষায়, আপনি সাময়িকভাবে আপনার কম্পিউটারের ইন্টারনেট সংযোগ বন্ধ করবেন, একটি ওয়ার্কফ্লো জমা দেবেন, দেখবেন টেম্পোরাল স্বয়ংক্রিয়ভাবে পুনরায় চেষ্টা করছে, এবং তারপর নেটওয়ার্কটি পুনরুদ্ধার করে দেখবেন যে এটি ঠিক হয়ে গেছে।

  1. আপনার মেশিনটিকে ইন্টারনেট থেকে সংযোগ বিচ্ছিন্ন করুন (উদাহরণস্বরূপ, আপনার ওয়াই-ফাই বন্ধ করুন)।

  2. একটি ওয়ার্কফ্লো জমা দিন: 

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

  3. টেম্পোরাল UI ( http://localhost:8233 ) চেক করুন। আপনি দেখতে পাবেন যে জেমিনি এপিআই অ্যাক্টিভিটিটি ফেইল করছে এবং টেম্পোরাল স্বয়ংক্রিয়ভাবে ব্যাকগ্রাউন্ডে রিট্রাইগুলো পরিচালনা করছে।

  4. ইন্টারনেটে পুনরায় সংযোগ করুন।

  5. পরবর্তী স্বয়ংক্রিয় পুনঃপ্রচেষ্টাটি সফলভাবে জেমিনি এপিআই-তে পৌঁছাবে এবং আপনার টার্মিনালে চূড়ান্ত ফলাফলটি প্রিন্ট হবে।

শ্রমিক দুর্ঘটনা থেকে বেঁচে যাওয়া

এই পরীক্ষায়, আপনি ওয়ার্কারটিকে চলমান অবস্থায় বন্ধ করে আবার চালু করেন। টেম্পোরাল ওয়ার্কফ্লো হিস্ট্রি (ইভেন্ট সোর্সিং) রিপ্লে করে এবং সর্বশেষ সম্পন্ন হওয়া অ্যাক্টিভিটি থেকে পুনরায় শুরু করে—ইতিমধ্যে সম্পন্ন হওয়া এলএলএম ইনভোকেশন এবং টুল কলগুলোর পুনরাবৃত্তি হয় না।

  1. ওয়ার্কারটিকে কিল করার জন্য সময় পেতে, durable_agent_worker.py ফাইলটি খুলুন এবং AgentWorkflow.run এ থাকা durable timer-টির কমেন্ট তুলে দিন: 

    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

টেম্পোরাল ওয়ার্কফ্লো হিস্ট্রি পুনরায় চালায়। যে LLM কল এবং টুল ইনভোকেশনগুলো ইতিমধ্যে সম্পন্ন হয়েছে, সেগুলো পুনরায় চালানো হয় না —সেগুলোর ফলাফল হিস্ট্রি (ইভেন্ট লগ) থেকে তাৎক্ষণিকভাবে পুনরায় চালানো হয়, টাইমার আবার চালু হয় এবং ওয়ার্কফ্লোটি সফলভাবে শেষ হয়।

আরও এগিয়ে যাওয়া

এই টিউটোরিয়ালে যা আলোচনা করা হয়েছে, ইন্টিগ্রেশনটি তার চেয়েও বেশি কিছু সমর্থন করে। বিস্তারিত জানতে প্লাগইনের ডকুমেন্টেশন দেখুন:

  • স্ট্রিমিং। যথারীতি generate_content_stream ব্যবহার করুন। ওয়ার্কফ্লোটি স্থায়ীভাবে চলার সময় কোনো বাহ্যিক গ্রাহককে (একটি চ্যাট UI) রিয়েল টাইমে খণ্ডাংশ পর্যবেক্ষণ করার সুযোগ দিতে, TemporalAsyncClient(streaming_topic=...) সেট করুন এবং ওয়ার্কফ্লোটির মধ্যে একটি WorkflowStream হোস্ট করুন।
  • MCP। GoogleGenAIPlugin(mcp_servers={...}) ব্যবহার করে ওয়ার্কারে একটি ক্লায়েন্ট-সাইড MCP সার্ভার রেজিস্টার করুন এবং Workflow-তে TemporalMcpClientSession দিয়ে নাম উল্লেখ করে এটিকে রেফারেন্স করুন। টুল ডিসকভারি এবং কলগুলো একটি পুলড ওয়ার্কার-সাইড কানেকশনের বিপরীতে অ্যাক্টিভিটি হিসেবে রান করে।
  • ভার্টেক্স এআই। ওয়ার্কার-সাইডের genai.Client এবং ওয়ার্কফ্লো-সাইডের TemporalAsyncClient উভয় ক্ষেত্রেই vertexai=True পাস করুন এবং ওয়ার্কফ্লো সাইডে project ও location স্পষ্টভাবে সেট করুন, যাতে রিপ্লে ডিটারমিনিস্টিক থাকে।

আরও তথ্যসূত্র