Find out how to design self-modifying multi-agent AI methods powered by Gemini with semantic routing, symbolic guardrails, and recursive 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")
shopper = 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="software/json" if json_mode else "textual content/plain"
       )
       strive:
           response = shopper.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 person 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 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']} (Purpose: {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 information, 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 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":
           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 accommodates Markdown code blocks, that are forbidden."
           return True, "Legitimate Textual content"
       return True, "Go"


   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="Person", goal=target_name, content material=current_input, metadata={})
               print(f"   [Exec] {employee.identify} working... (Try {try+1})")
               outcome = employee.execute(msg)
               if constraint:
                   is_valid, error_msg = self.validate_constraint(outcome, 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: {outcome}nError: {error_msg}nFIX IT instantly."
                       proceed
               print(f"   [Success] Closing Output:n{outcome[:100]}...")
               return outcome
           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"
   )