How 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 types


API_KEY = os.environ.get("GEMINI_API_KEY", "API Key")
client = genai.Client(api_key=API_KEY)


@dataclass
class AgentMessage:
   source: str
   target: str
   content: str
   metadata: dict
   timestamp: float = time.time()

class CognitiveEngine:
   @staticmethod
   def generate(prompt: str, system_instruction: str, json_mode: bool = False) -> str:
       config = types.GenerateContentConfig(
           temperature=0.1,
           response_mime_type="application/json" if json_mode else "text/plain"
       )
       try:
           response = client.models.generate_content(
               model="gemini-2.0-flash",
               contents=prompt,
               config=config
           )
           return response.text
       except Exception as e:
           raise 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:
       prompt = f"""
       You are a Master Dispatcher. Analyze the user request and map it to the ONE best agent.
       AVAILABLE AGENTS:
       {json.dumps(self.registry, indent=2)}
       USER REQUEST: "{user_query}"
       Return ONLY a JSON object: {{"selected_agent": "agent_name", "reasoning": "brief reason"}}
       """
       response_text = CognitiveEngine.generate(prompt, "You are a routing system.", json_mode=True)
       try:
           decision = json.loads(response_text)
           print(f"   [Router] Selected: {decision['selected_agent']} (Reason: {decision['reasoning']})")
           return decision['selected_agent']
       except:
           return "general_agent"

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


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


class Orchestrator:
   def __init__(self):
       self.agents_info = {
           "analyst_bot": "Analyzes data, logic, and math. Returns structured JSON summaries.",
           "creative_bot": "Writes poems, stories, and creative text. Returns plain text.",
           "coder_bot": "Writes Python code snippets."
       }
       self.workers = {
           "analyst_bot": Agent("analyst_bot", "You are a Data Analyst. output strict JSON."),
           "creative_bot": Agent("creative_bot", "You are a Creative Writer."),
           "coder_bot": Agent("coder_bot", "You are a Python Expert. Return only code.")
       }
       self.router = SemanticRouter(self.agents_info)

 def validate_constraint(self, content: str, constraint_type: str) -> tuple[bool, str]:
       if constraint_type == "json_only":
           try:
               json.loads(content)
               return True, "Valid JSON"
           except:
               return False, "Output was not valid JSON."
       if constraint_type == "no_markdown":
           if "```" in content:
               return False, "Output contains Markdown code blocks, which are forbidden."
           return True, "Valid Text"
       return True, "Pass"


   def run_task(self, user_input: str, constraint: str = None, max_retries: int = 2):
       print(f"\n--- New Task: {user_input} ---")
       target_name = self.router.route(user_input)
       worker = self.workers.get(target_name)
       current_input = user_input
       history = []
       for attempt in range(max_retries + 1):
           try:
               msg = AgentMessage(source="User", target=target_name, content=current_input, metadata={})
               print(f"   [Exec] {worker.name} working... (Attempt {attempt+1})")
               result = worker.execute(msg)
               if constraint:
                   is_valid, error_msg = self.validate_constraint(result, constraint)
                   if not is_valid:
                       print(f"   [Guardrail] VIOLATION: {error_msg}")
                       current_input = f"Your previous answer failed a check.\nOriginal Request: {user_input}\nYour Answer: {result}\nError: {error_msg}\nFIX IT immediately."
                       continue
               print(f"   [Success] Final Output:\n{result[:100]}...")
               return result
           except 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(
       "Compare the GDP of France and Germany in 2023.",
       constraint="json_only"
   )
   orchestrator.run_task(
       "Write a Python function for Fibonacci numbers.",
       constraint="no_markdown"
   )