كيفية بناء وكلاء ذكاء اصطناعي طويلة الأمد باستخدام Google Gen AI SDK

ما الذي نبنيه

مع إعلان Google Cloud عن وكلاء طويلة الأمد في Gemini Enterprise، تغيرت قواعد التطوير بشكل رسمي. في هذا الدرس، سنقوم ببناء البنية التحتية الأساسية “الإيقاف والاستئناف” المطلوبة لبناء هؤلاء الوكلاء.

لن نبني مجرد روبوت دردشة. سنبني عامل Python غير متزامن يحتفظ بالحالة وينفذ مهمة متعددة الخطوات، ويتوقف عن عمد عندما يحتاج إلى موافقة بشرية محاكاة (محاكاة لـ Unified Inbox)، ويستأنف عند التأكيد.

الإعداد والتثبيت

سنستخدم Google Gen AI SDK الرسمي و`python-dotenv` لمتغيرات البيئة الخاصة بنا.

pip install google-genai python-dotenv asyncio

قم بتأمين بيانات اعتماد API الخاصة بك في ملف `.env`.

    # .env file
    GEMINI_API_KEY=your_gemini_api_key_here
  

الخطوة ١: تصميم العميل ذو الحالة

على عكس روبوت الدردشة القياسي الذي ينسى البيانات بين الاستفسارات، يجب على الوكيل طويل الأمد الحفاظ على قاموس حالة صارم. نقوم بتهيئة `genai.Client` الرسمي ونقوم بإعداد مدير الحالة الخاص بنا.

import os
import asyncio
from google import genai
from dotenv import load_dotenv

load_dotenv()

class LongRunningAgent:
    def __init__(self):
        # Initialize the official Google Gen AI Client
        self.client = genai.Client(api_key=os.getenv("GEMINI_API_KEY"))
        self.model = "gemini-1.5-pro"
        
        # This dictionary mimics the persistent state stored in a database
        self.state = {
            "status": "idle", # idle, running, awaiting_approval, completed
            "workflow_history": [],
            "pending_approval_request": None
        }

    def log_action(self, action):
        print(f"[AGENT LOG]: {action}")
        self.state["workflow_history"].append(action)
  

الخطوة ٢: بناء منطق “الإيقاف والاستئناف” HITL

الابتكار الأساسي في تحديث Gemini الجديد هو صندوق الوارد البشري في الحلقة (HITL). هنا، نبني المنطق غير المتزامن الذي يسمح للوكيل بإيقاف التنفيذ عندما يصل إلى إجراء مقيد.

    async def request_human_approval(self, task_description):
        """Simulates pushing a task to the Unified Inbox"""
        self.state["status"] = "awaiting_approval"
        self.state["pending_approval_request"] = task_description
        
        self.log_action(f"PAUSED: Awaiting human approval for: {task_description}")
        
        # Simulate waiting for the manager to click "Approve" in the Inbox
        while self.state["status"] == "awaiting_approval":
            await asyncio.sleep(2) # Check database/state every 2 seconds
            
        self.log_action("RESUMED: Human approval granted.")
        return True

    def simulate_manager_approval(self):
        """External function called by your UI/Inbox when a user clicks approve"""
        if self.state["status"] == "awaiting_approval":
            self.state["status"] = "running"
            self.state["pending_approval_request"] = None
            print("n✅ [INBOX]: Manager approved the action.n")
  

الخطوة ٣: تنفيذ سير العمل غير المتزامن

الآن، نربط الأمور معًا. سنستخدم طريقة `client.models.generate_content` لمعالجة البيانات، ولكن نغلفها في حلقة التنفيذ غير المتزامن الخاصة بنا.

    async def run_multi_day_workflow(self, initial_prompt):
        self.state["status"] = "running"
        self.log_action("Starting long-running workflow...")
        
        # Phase 1: Autonomous Processing
        self.log_action("Analyzing request via Gemini API...")
        response = self.client.models.generate_content(
            model=self.model,
            contents=f"Analyze this task and propose a 3-step execution plan: {initial_prompt}"
        )
        self.log_action(f"Plan generated: {response.text[:100]}...")
        
        # Phase 2: Hitting a permission wall (Mimicking the Unified Inbox feature)
        await asyncio.sleep(1) # Simulating heavy compute time
        
        # The agent realizes it needs access to a restricted system (e.g., Google Drive)
        await self.request_human_approval("Access restricted Drive Folder: 'Q3 Financials'")
        
        # Phase 3: Post-Approval Execution
        self.log_action("Finalizing workflow with approved access...")
        final_response = self.client.models.generate_content(
            model=self.model,
            contents="The human approved access. Generate the final summary report."
        )
        
        self.state["status"] = "completed"
        self.log_action("Workflow Completed.")
        return final_response.text
  

اختبار بنية صندوق الوارد الموحد

لتشغيل هذا، سنستخدم `asyncio.gather` في Python لتشغيل الوكيل في الخلفية أثناء محاكاة شخص يتحقق من صندوق الوارد الخاص به.

async def main():
    agent = LongRunningAgent()
    
    # Start the agent as a background task
    agent_task = asyncio.create_task(
        agent.run_multi_day_workflow("Audit the Q3 Marketing Spend")
    )
    
    # Simulate the human manager checking their inbox after 5 seconds
    await asyncio.sleep(5)
    agent.simulate_manager_approval()
    
    # Wait for the agent to finish
    result = await agent_task
    print(f"n[FINAL OUTPUT]:n{result}")

if __name__ == '__main__':
    asyncio.run(main())
  

ما الذي يجب بناؤه بعد ذلك

• استبدل حلقة الانتظار المحاكاة بحفظ حالة الوكيل في قاعدة بيانات PostgreSQL.
• قم ببناء واجهة مستخدم React/Next.js “Unified Inbox” التي تقوم بتشغيل الويب هوك لاستئناف الوكيل.
• قم بتنفيذ تكوينات genai.types.Tool الرسمية للسماح للوكيل بتنفيذ الإجراءات فعليًا بعد الموافقة.