Add retry logic for AGI commands

agi_cli.py

@@ -91,35 +91,100 @@ def plan_task(self, goal: str) -> str:
         ]
         return "\n".join(plan_steps)
 
-async def run_command(command: str, *args):
-    """Run an AGI command"""
+# 1) Move the help text to a module‐level constant
+HELP_TEXT = """
+🤖 AGI Command Line Interface
+Available commands:
+  reason <query>              - Perform AGI reasoning on a query
+  file <filename> [operation] - Process file (analyze, optimize, transform, summarize)
+  code <language> <description> - Generate code in specified language
+  plan <goal>                 - Create and execute a plan for a goal
+  help                        - Show this help message
+Examples:
+  python agi_cli.py reason "How does machine learning work?"
+  python agi_cli.py file myfile.txt analyze
+  python agi_cli.py code python "Create a web scraper"
+  python agi_cli.py plan "Build a recommendation system"
+"""
+
+# 2) Extract retry logic into a decorator
+def retry_async(max_retries: int = 3, retry_delay: float = 1.0):
+    def decorator(fn):
+        async def wrapper(*args, **kwargs):
+            for attempt in range(1, max_retries + 1):
+                try:
+                    return await fn(*args, **kwargs)
+                except Exception as e:
+                    if attempt == max_retries:
+                        print(f"❌ Error after {max_retries} attempts: {e}")
+                        return False
+                    print(f"⚠️  Attempt {attempt} failed: {e}. Retrying in {retry_delay} seconds…")
+                    await asyncio.sleep(retry_delay)
+        return wrapper
+    return decorator
+
+# 3) Define per‐command handlers, then dispatch via a map
+async def _handle_reason(agi, args):
+    query = " ".join(args) if args else "What is artificial general intelligence?"
+    fn = agi.kernel.get_function("agi_cli", "reason")
+    return await fn.invoke(agi.kernel, query=query)
+
+# … similarly define _handle_file, _handle_code, _handle_plan …
+
+COMMAND_HANDLERS = {
+    "reason": _handle_reason,
+    "file":   _handle_file,
+    "code":   _handle_code,
+    "plan":   _handle_plan,
+    "help":   lambda agi, args: HELP_TEXT,
+}
+
+# 4) Simplify run_command
+@retry_async()
+async def run_command(command: str, *args) -> bool:
+    agi = AGICommandLine()
+    handler = COMMAND_HANDLERS.get(command)
+    if not handler:
+        print(f"❌ Unknown command: {command}. Use 'help' for available commands.")
+        return False
+
+    result = await handler(agi, args)
+    print(result)
+    return True
+    command: str,
+    *args,
+    max_retries: int = 3,
+    retry_delay: float = 1.0,
+) -> bool:
+    """Run an AGI command with basic retry logic."""
     agi = AGICommandLine()
 
-    try:
-        if command == "reason":
-            query = " ".join(args) if args else "What is artificial general intelligence?"
-            function = agi.kernel.get_function("agi_cli", "reason")
-            result = await function.invoke(agi.kernel, query=query)
-
-        elif command == "file":
-            filename = args[0] if args else "example.txt"
-            operation = args[1] if len(args) > 1 else "analyze"
-            function = agi.kernel.get_function("agi_cli", "process_file")
-            result = await function.invoke(agi.kernel, filename=filename, operation=operation)
-
-        elif command == "code":
-            language = args[0] if args else "python"
-            description = " ".join(args[1:]) if len(args) > 1 else "Simple example function"
-            function = agi.kernel.get_function("agi_cli", "generate_code")
-            result = await function.invoke(agi.kernel, language=language, description=description)
-
-        elif command == "plan":
-            goal = " ".join(args) if args else "Complete project tasks"
-            function = agi.kernel.get_function("agi_cli", "plan_task")
-            result = await function.invoke(agi.kernel, goal=goal)
-
-        elif command == "help":
-            result = """
+    for attempt in range(1, max_retries + 1):
+        try:
+            if command == "reason":
+                query = " ".join(args) if args else "What is artificial general intelligence?"
+                function = agi.kernel.get_function("agi_cli", "reason")
+                result = await function.invoke(agi.kernel, query=query)
+
+            elif command == "file":
+                filename = args[0] if args else "example.txt"
+                operation = args[1] if len(args) > 1 else "analyze"
+                function = agi.kernel.get_function("agi_cli", "process_file")
+                result = await function.invoke(agi.kernel, filename=filename, operation=operation)
+
+            elif command == "code":
+                language = args[0] if args else "python"
+                description = " ".join(args[1:]) if len(args) > 1 else "Simple example function"
+                function = agi.kernel.get_function("agi_cli", "generate_code")
+                result = await function.invoke(agi.kernel, language=language, description=description)
+
+            elif command == "plan":
+                goal = " ".join(args) if args else "Complete project tasks"
+                function = agi.kernel.get_function("agi_cli", "plan_task")
+                result = await function.invoke(agi.kernel, goal=goal)
+
+            elif command == "help":
+                result = """
 🤖 AGI Command Line Interface
 
 Available commands:
@@ -134,16 +199,19 @@ async def run_command(command: str, *args):
   python agi_cli.py file myfile.txt analyze
   python agi_cli.py code python "Create a web scraper"
   python agi_cli.py plan "Build a recommendation system"
-"""
-        else:
-            result = f"❌ Unknown command: {command}. Use 'help' for available commands."
-
-        print(result)
-        return True
-
-    except Exception as e:
-        print(f"❌ Error executing command: {e}")
-        return False
+            """
+            else:
+                result = f"❌ Unknown command: {command}. Use 'help' for available commands."
+
+            print(result)
+            return True
+
+        except Exception as e:
+            if attempt == max_retries:
+                print(f"❌ Error executing command after {max_retries} attempts: {e}")
+                return False
+            print(f"⚠️  Attempt {attempt} failed: {e}. Retrying in {retry_delay} seconds...")
+            await asyncio.sleep(retry_delay)
 
 def main():
     """Main CLI function"""