The right way 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 sorts


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 = sorts.GenerateContentConfig(
           temperature=0.1,
           response_mime_type="software/json" if json_mode else "textual content/plain"
       )
       attempt:
           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 finest agent.
       AVAILABLE AGENTS:
       {json.dumps(self.registry, indent=2)}
       USER REQUEST: "{user_query}"
       Return ONLY a JSON object: {{"selected_agent": "agent_name", "reasoning": "transient purpose"}}
       """
       response_text = CognitiveEngine.generate(immediate, "You're a routing system.", json_mode=True)
       attempt:
           determination = json.masses(response_text)
           print(f"   [Router] Chosen: {determination['selected_agent']} (Motive: {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 artistic textual content. Returns plain textual content.",
           "coder_bot": "Writes Python code snippets."
       }
       self.employees = {
           "analyst_bot": Agent("analyst_bot", "You're a Knowledge Analyst. output strict JSON."),
           "creative_bot": Agent("creative_bot", "You're a Inventive Author."),
           "coder_bot": Agent("coder_bot", "You're a Python Skilled. 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":
           attempt:
               json.masses(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):
           attempt:
               msg = AgentMessage(supply="Consumer", goal=target_name, content material=current_input, metadata={})
               print(f"   [Exec] {employee.identify} working... (Try {try+1})")
               end result = employee.execute(msg)
               if constraint:
                   is_valid, error_msg = self.validate_constraint(end result, 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: {end result}nError: {error_msg}nFIX IT instantly."
                       proceed
               print(f"   [Success] Closing Output:n{end result[:100]}...")
               return end result
           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(
       "Evaluate the GDP of France and Germany in 2023.",
       constraint="json_only"
   )
   orchestrator.run_task(
       "Write a Python perform for Fibonacci numbers.",
       constraint="no_markdown"
   )