import google.generativeai as genaiimport jsonimport timefrom typing import Dict, List, Any, Optionalfrom dataclasses import dataclass, asdictfrom enum import Enumclass TaskStatus(Enum):   PENDING = "pending"   IN_PROGRESS = "in_progress"   COMPLETED = "completed"   FAILED = "failed"

@dataclassclass Task:   id: str   description: str   priority: int   status: TaskStatus = TaskStatus.PENDING   dependencies: List[str] = None   result: Optional[str] = None     def __post_init__(self):       if self.dependencies is None:           self.dependencies = []class SAGEAgent:   """Self-Adaptive Goal-oriented Execution AI Agent"""     def __init__(self, api_key: str, model_name: str = "gemini-1.5-flash"):       genai.configure(api_key=api_key)       self.model = genai.GenerativeModel(model_name)       self.memory = []       self.tasks = {}       self.context = {}       self.iteration_count = 0         def self_assess(self, goal: str, context: Dict[str, Any]) -> Dict[str, Any]:       """S: Self-Assessment - Evaluate current state and capabilities"""       assessment_prompt = f"""       You are an AI agent conducting self-assessment. Respond ONLY with valid JSON, no additional text.       GOAL: {goal}       CONTEXT: {json.dumps(context, indent=2)}       TASKS_PROCESSED: {len(self.tasks)}             Provide assessment as JSON with these exact keys:       {{           "progress_score": <number 0-100>,           "resources": ["list of available resources"],           "gaps": ["list of knowledge gaps"],           "risks": ["list of potential risks"],           "recommendations": ["list of next steps"]       }}       """             response = self.model.generate_content(assessment_prompt)       try:           text = response.text.strip()           if text.startswith('```'):               text = text.split('```')[1]               if text.startswith('json'):                   text = text[4:]           text = text.strip()           return json.loads(text)       except Exception as e:           print(f"Assessment parsing error: {e}")           return {               "progress_score": 25,               "resources": ["AI capabilities", "Internet knowledge"],               "gaps": ["Specific domain expertise", "Real-time data"],               "risks": ["Information accuracy", "Scope complexity"],               "recommendations": ["Break down into smaller tasks", "Focus on research first"]           }     def adaptive_plan(self, goal: str, assessment: Dict[str, Any]) -> List[Task]:       """A: Adaptive Planning - Create dynamic, context-aware task decomposition"""       planning_prompt = f"""       You are an AI task planner. Respond ONLY with valid JSON array, no additional text.       MAIN_GOAL: {goal}       ASSESSMENT: {json.dumps(assessment, indent=2)}             Create 3-4 actionable tasks as JSON array:       [           {{               "id": "task_1",               "description": "Clear, specific task description",               "priority": 5,               "dependencies": []           }},           {{               "id": "task_2",               "description": "Another specific task",               "priority": 4,               "dependencies": ["task_1"]           }}       ]             Each task must have: id (string), description (string), priority (1-5), dependencies (array of strings)       """             response = self.model.generate_content(planning_prompt)       try:           text = response.text.strip()           if text.startswith('```'):               text = text.split('```')[1]               if text.startswith('json'):                   text = text[4:]           text = text.strip()                     task_data = json.loads(text)           tasks = []           for i, task_info in enumerate(task_data):               task = Task(                   id=task_info.get('id', f'task_{i+1}'),                   description=task_info.get('description', 'Undefined task'),                   priority=task_info.get('priority', 3),                   dependencies=task_info.get('dependencies', [])               )               tasks.append(task)           return tasks       except Exception as e:           print(f"Planning parsing error: {e}")           return [               Task(id="research_1", description="Research sustainable urban gardening basics", priority=5),               Task(id="research_2", description="Identify space-efficient growing methods", priority=4),               Task(id="compile_1", description="Organize findings into structured guide", priority=3, dependencies=["research_1", "research_2"])           ]     def execute_goal_oriented(self, task: Task) -> str:       """G: Goal-oriented Execution - Execute specific task with focused attention"""       execution_prompt = f"""       GOAL-ORIENTED EXECUTION:       Task: {task.description}       Priority: {task.priority}       Context: {json.dumps(self.context, indent=2)}             Execute this task step-by-step:       1. Break down the task into concrete actions       2. Execute each action methodically       3. Validate results at each step       4. Provide comprehensive output             Focus on practical, actionable results. Be specific and thorough.       """             response = self.model.generate_content(execution_prompt)       return response.text.strip()     def integrate_experience(self, task: Task, result: str, success: bool) -> Dict[str, Any]:       """E: Experience Integration - Learn from outcomes and update knowledge"""       integration_prompt = f"""       You are learning from task execution. Respond ONLY with valid JSON, no additional text.       TASK: {task.description}       RESULT: {result[:200]}...       SUCCESS: {success}             Provide learning insights as JSON:       {{           "learnings": ["key insight 1", "key insight 2"],           "patterns": ["pattern observed 1", "pattern observed 2"],           "adjustments": ["adjustment for future 1", "adjustment for future 2"],           "confidence_boost": <number -10 to 10>       }}       """             response = self.model.generate_content(integration_prompt)       try:           text = response.text.strip()           if text.startswith('```'):               text = text.split('```')[1]               if text.startswith('json'):                   text = text[4:]           text = text.strip()                     experience = json.loads(text)           experience['task_id'] = task.id           experience['timestamp'] = time.time()           self.memory.append(experience)           return experience       except Exception as e:           print(f"Experience parsing error: {e}")           experience = {               "learnings": [f"Completed task: {task.description}"],               "patterns": ["Task execution follows planned approach"],               "adjustments": ["Continue systematic approach"],               "confidence_boost": 5 if success else -2,               "task_id": task.id,               "timestamp": time.time()           }           self.memory.append(experience)           return experience     def execute_sage_cycle(self, goal: str, max_iterations: int = 3) -> Dict[str, Any]:       """Execute complete SAGE cycle for goal achievement"""       print(f" Starting SAGE cycle for goal: {goal}")       results = {"goal": goal, "iterations": [], "final_status": "unknown"}             for iteration in range(max_iterations):           self.iteration_count += 1           print(f"\n SAGE Iteration {iteration + 1}")                     print(" Self-Assessment...")           assessment = self.self_assess(goal, self.context)           print(f"Progress Score: {assessment.get('progress_score', 0)}/100")                     print("  Adaptive Planning...")           tasks = self.adaptive_plan(goal, assessment)           print(f"Generated {len(tasks)} tasks")                     print(" Goal-oriented Execution...")           iteration_results = []                     for task in sorted(tasks, key=lambda x: x.priority, reverse=True):               if self._dependencies_met(task):                   print(f"  Executing: {task.description}")                   task.status = TaskStatus.IN_PROGRESS                                     try:                       result = self.execute_goal_oriented(task)                       task.result = result                       task.status = TaskStatus.COMPLETED                       success = True                       print(f"   Completed: {task.id}")                   except Exception as e:                       task.status = TaskStatus.FAILED                       task.result = f"Error: {str(e)}"                       success = False                       print(f"   Failed: {task.id}")                                     experience = self.integrate_experience(task, task.result, success)                                     self.tasks[task.id] = task                   iteration_results.append({                       "task": asdict(task),                       "experience": experience                   })                     self._update_context(iteration_results)                     results["iterations"].append({               "iteration": iteration + 1,               "assessment": assessment,               "tasks_generated": len(tasks),               "tasks_completed": len([r for r in iteration_results if r["task"]["status"] == "completed"]),               "results": iteration_results           })                     if assessment.get('progress_score', 0) >= 90:               results["final_status"] = "achieved"               print(" Goal achieved!")               break             if results["final_status"] == "unknown":           results["final_status"] = "in_progress"             return results     def _dependencies_met(self, task: Task) -> bool:       """Check if task dependencies are satisfied"""       for dep_id in task.dependencies:           if dep_id not in self.tasks or self.tasks[dep_id].status != TaskStatus.COMPLETED:               return False       return True     def _update_context(self, results: List[Dict[str, Any]]):       """Update agent context based on execution results"""       completed_tasks = [r for r in results if r["task"]["status"] == "completed"]       self.context.update({           "completed_tasks": len(completed_tasks),           "total_tasks": len(self.tasks),           "success_rate": len(completed_tasks) / len(results) if results else 0,           "last_update": time.time()       })

if __name__ == "__main__":   API_KEY = "Use Your Own API Key Here"      try:       agent = SAGEAgent(API_KEY, model_name="gemini-1.5-flash")             goal = "Research and create a comprehensive guide on sustainable urban gardening practices"             results = agent.execute_sage_cycle(goal, max_iterations=2)             print("\n" + "="*50)       print(" SAGE EXECUTION SUMMARY")       print("="*50)       print(f"Goal: {results['goal']}")       print(f"Status: {results['final_status']}")       print(f"Iterations: {len(results['iterations'])}")             for i, iteration in enumerate(results['iterations'], 1):           print(f"\nIteration {i}:")           print(f"  Assessment Score: {iteration['assessment'].get('progress_score', 0)}/100")           print(f"  Tasks Generated: {iteration['tasks_generated']}")           print(f"  Tasks Completed: {iteration['tasks_completed']}")             print("\n Agent Memory Entries:", len(agent.memory))       print(" Total Tasks Processed:", len(agent.tasks))         except Exception as e:       print(f"Demo requires valid Gemini API key. Error: {e}")       print("Get your free API key from: https://makersuite.google.com/app/apikey")