Cómo diseñar un sistema de inteligencia industrial multiagente autocorrector impulsado por Gemini con enrutamiento semántico, barreras de seguridad simbólicas y orquestación reflexiva

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"
   )