Methods to Design a Gemini-Powered Self-Correcting Multi-Agent AI System with Semantic Routing, Symbolic Guardrails, and Reflexive Orchestration

import os
import json
import time
import typing
from dataclasses import dataclass, asdict
from google import genai
from google.genai import varieties


API_KEY = os.environ.get("GEMINI_API_KEY", "API Key")
consumer = genai.Consumer(api_key=API_KEY)


@dataclass
class AgentMessage:
   supply: str
   goal: str
   content material: str
   metadata: dict
   timestamp: float = time.time()

class CognitiveEngine:
   @staticmethod
   def generate(immediate: str, system_instruction: str, json_mode: bool = False) -> str:
       config = varieties.GenerateContentConfig(
           temperature=0.1,
           response_mime_type="utility/json" if json_mode else "textual content/plain"
       )
       strive:
           response = consumer.fashions.generate_content(
               mannequin="gemini-2.0-flash",
               contents=immediate,
               config=config
           )
           return response.textual content
       besides Exception as e:
           increase ConnectionError(f"Gemini API Error: {e}")


class SemanticRouter:
   def __init__(self, agents_registry: dict):
       self.registry = agents_registry


   def route(self, user_query: str) -> str:
       immediate = f"""
       You're a Grasp Dispatcher. Analyze the consumer request and map it to the ONE greatest agent.
       AVAILABLE AGENTS:
       {json.dumps(self.registry, indent=2)}
       USER REQUEST: "{user_query}"
       Return ONLY a JSON object: {{"selected_agent": "agent_name", "reasoning": "temporary motive"}}
       """
       response_text = CognitiveEngine.generate(immediate, "You're a routing system.", json_mode=True)
       strive:
           determination = json.hundreds(response_text)
           print(f"   [Router] Chosen: {determination['selected_agent']} (Cause: {determination['reasoning']})")
           return determination['selected_agent']
       besides:
           return "general_agent"

class Agent:
   def __init__(self, identify: str, instruction: str):
       self.identify = identify
       self.instruction = instruction


   def execute(self, message: AgentMessage) -> str:
       return CognitiveEngine.generate(
           immediate=f"Enter: {message.content material}",
           system_instruction=self.instruction
       )


class Orchestrator:
   def __init__(self):
       self.agents_info = {
           "analyst_bot": "Analyzes knowledge, logic, and math. Returns structured JSON summaries.",
           "creative_bot": "Writes poems, tales, and inventive textual content. Returns plain textual content.",
           "coder_bot": "Writes Python code snippets."
       }
       self.employees = {
           "analyst_bot": Agent("analyst_bot", "You're a Information Analyst. output strict JSON."),
           "creative_bot": Agent("creative_bot", "You're a Inventive Author."),
           "coder_bot": Agent("coder_bot", "You're a Python Professional. Return solely code.")
       }
       self.router = SemanticRouter(self.agents_info)

 def validate_constraint(self, content material: str, constraint_type: str) -> tuple[bool, str]:
       if constraint_type == "json_only":
           strive:
               json.hundreds(content material)
               return True, "Legitimate JSON"
           besides:
               return False, "Output was not legitimate JSON."
       if constraint_type == "no_markdown":
           if "```" in content material:
               return False, "Output incorporates Markdown code blocks, that are forbidden."
           return True, "Legitimate Textual content"
       return True, "Cross"


   def run_task(self, user_input: str, constraint: str = None, max_retries: int = 2):
       print(f"n--- New Process: {user_input} ---")
       target_name = self.router.route(user_input)
       employee = self.employees.get(target_name)
       current_input = user_input
       historical past = []
       for try in vary(max_retries + 1):
           strive:
               msg = AgentMessage(supply="Consumer", goal=target_name, content material=current_input, metadata={})
               print(f"   [Exec] {employee.identify} working... (Try {try+1})")
               consequence = employee.execute(msg)
               if constraint:
                   is_valid, error_msg = self.validate_constraint(consequence, constraint)
                   if not is_valid:
                       print(f"   [Guardrail] VIOLATION: {error_msg}")
                       current_input = f"Your earlier reply failed a test.nOriginal Request: {user_input}nYour Reply: {consequence}nError: {error_msg}nFIX IT instantly."
                       proceed
               print(f"   [Success] Remaining Output:n{consequence[:100]}...")
               return consequence
           besides Exception as e:
               print(f"   [System Error] {e}")
               time.sleep(1)
       print("   [Failed] Max retries reached or self-correction failed.")
       return None

if __name__ == "__main__":
   orchestrator = Orchestrator()
   orchestrator.run_task(
       "Examine the GDP of France and Germany in 2023.",
       constraint="json_only"
   )
   orchestrator.run_task(
       "Write a Python operate for Fibonacci numbers.",
       constraint="no_markdown"
   )