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

En este tutorial, exploramos cómo diseñamos y ejecutamos un canal de orquestación de IA agente completo impulsado por enrutamiento semántico, barreras de seguridad simbólicas y bucles de autocorrección utilizando Gemini. Explicamos cómo estructuramos agentes, despachamos tareas, aplicamos restricciones y refinamos los resultados utilizando una arquitectura limpia y modular. A medida que avanzamos en cada fragmento, vemos cómo el sistema elige inteligentemente al agente adecuado, valida su resultado y se mejora a través de una reflexión iterativa. Consulte los códigos completos aquí.

importar os importar json importar tiempo importar escribir desde clases de datos importar clase de datos, asdict desde google importar genai desde google.genai tipos de importación API_KEY = os.environ.get(“GEMINI_API_KEY”, “API Key”) cliente = genai.Client(api_key=API_KEY) @dataclass clase AgentMessage: fuente: str destino: str contenido: str metadatos: dict marca de tiempo: float = time.time()

Configuramos nuestro entorno central importando bibliotecas esenciales, definiendo la clave API e inicializando el cliente Gemini. También establecemos la estructura AgentMessage, que actúa como formato de comunicación compartida entre agentes. Consulte los códigos completos aquí.

clase CognitiveEngine: @staticmethod def generar (mensaje: str, system_instruction: str, json_mode: bool = False) -> str: config = tipos.GenerateContentConfig( temperatura = 0.1, respuesta_mime_type=”application/json” si json_mode else “text/plain” ) intente: respuesta = client.models.generate_content( model=”gemini-2.0-flash”, contenidos=prompt, config=config ) devuelve respuesta.text excepto Excepción como e: elevar ConnectionError(f”Gemini API Error: {e}”) clase SemanticRouter: def __init__(self, agentes_registry: dict): self.registry = agentes_registry def ruta(self, user_query: str) -> str: Prompt = f””” Eres un maestro Dispatcher. Analice la solicitud del usuario y asígnela al UNO mejor agente DISPONIBLE: {json.dumps(self.registry, indent=2)} SOLICITUD DEL USUARIO: “{user_query}” Devuelve SÓLO un objeto JSON: {{“selected_agent”: “agent_name”, “reasoning”: “breef Reason”}} “”” Response_text = CognitiveEngine.generate(prompt, “You are a). sistema de enrutamiento.”, json_mode=True) intente: decisión = json.loads(response_text) print(f” [Router] Seleccionado: {decisión[‘selected_agent’]} (Razón: {decisión[‘reasoning’]})”) decisión de devolución[‘selected_agent’]
excepto: devolver “agente_general”

Construimos la capa cognitiva usando Gemini, lo que nos permite generar salidas de texto y JSON según la instrucción. También implementamos el enrutador semántico, que analiza las consultas y selecciona el agente más adecuado. Consulte los códigos completos aquí.

clase Agente: def __init__(self, nombre: str, instrucción: str): self.name = nombre self.instruction = instrucción def ejecutar(self, mensaje: AgentMessage) -> str: return CognitiveEngine.generate( Prompt=f”Entrada: {message.content}”, system_instruction=self.instruction ) class Orchestrator: def __init__(self): self.agents_info = { “analyst_bot”: “Analiza datos, lógica y matemáticas. Devuelve resúmenes JSON estructurados.”, “creative_bot”: “Escribe poemas, historias y texto creativo. Devuelve texto sin formato.”, “coder_bot”: “Escribe fragmentos de código Python”. } self.workers = { “analyst_bot”: Agent(“analyst_bot”, “Eres un analista de datos. Genera JSON estricto.”), “creative_bot”: Agent(“creative_bot”, “Eres un escritor creativo.”), “coder_bot”: Agent(“coder_bot”, “Eres un experto en Python. Devuelve solo código”) } self.router = SemanticRouter(self.agents_info)

Construimos los agentes trabajadores y el orquestador central. Cada agente recibe un rol claro, analista, creativo o codificador, y configuramos el orquestador para administrarlos. Al revisar esta sección, vemos cómo definimos el ecosistema de agentes y lo preparamos para la delegación inteligente de tareas. Consulte los códigos completos aquí.

def validar_constraint(self, contenido: str, constraint_type: str) -> tupla[bool, str]: si constraint_type == “json_only”: intente: json.loads(content) devuelve Verdadero, “JSON válido” excepto: devuelve Falso, “La salida no fue JSON válido”. if constraint_type == “no_markdown”: si ““`” en el contenido: devuelve False, “La salida contiene bloques de código Markdown, que están prohibidos”. return True, “Texto válido” return True, “Pass” def run_task(self, user_input: str, constraint: str = Ninguna, max_retries: int = 2): print(f”\n— Nueva tarea: {user_input} —“) target_name = self.router.route(user_input) trabajador = self.workers.get(target_name) current_input = user_input historial = []
para intento en el rango (max_retries + 1): intente: msg = AgentMessage(source=”User”, target=target_name, content=current_input, metadata={}) print(f” [Exec] {worker.name} trabajando… (Intento {intento+1})”) resultado = trabajador.execute(msg) si la restricción: es_válida, error_msg = self.validate_constraint(resultado, restricción) si no es_válida: print(f” [Guardrail] VIOLACIÓN: {error_msg}”) current_input = f”Su respuesta anterior no pasó la verificación.\nSolicitud original: {user_input}\nSu respuesta: {resultado}\nError: {error_msg}\nSOLUCIÓN inmediatamente.” continuar print(f” [Success] Salida final:\n{resultado[:100]}…”) devuelve resultado excepto Excepción como e: print(f” [System Error] {e}”) tiempo.dormir(1) imprimir(” [Failed] Se alcanzó el número máximo de reintentos o la autocorrección falló.”) return Ninguno

Implementamos barreras de seguridad simbólicas y un bucle de autocorrección para imponer restricciones como JSON estricto o sin Markdown. Realizamos un refinamiento iterativo cada vez que los resultados violan los requisitos, lo que permite a nuestros agentes corregir sus propios errores. Consulte los códigos completos aquí.

if __name__ == “__main__”: Orchestrator = Orchestrator() Orchestrator.run_task( “Compare el PIB de Francia y Alemania en 2023.”, constraint=”json_only” ) Orchestrator.run_task( “Escriba una función de Python para los números de Fibonacci.”, constraint=”no_markdown” )

Ejecutamos dos escenarios completos, mostrando el enrutamiento, la ejecución del agente y la validación de restricciones en acción. Ejecutamos una tarea analítica aplicada en JSON y una tarea de codificación con restricciones de Markdown para observar el comportamiento reflexivo.

En conclusión, ahora vemos cómo múltiples componentes, enrutamiento, agentes trabajadores, barreras de seguridad y autocorrección, se unen para crear un sistema de agentes inteligente y confiable. Somos testigos de cómo cada parte contribuye a una ejecución sólida de las tareas, garantizando que los resultados sigan siendo precisos, alineados y conscientes de las limitaciones. Al reflexionar sobre la arquitectura, reconocemos con qué facilidad podemos expandirla con nuevos agentes, restricciones más ricas o estrategias de razonamiento más avanzadas.

Consulte los códigos completos aquí. No dude en consultar nuestra página de GitHub para tutoriales, códigos y cuadernos. Además, no dude en seguirnos en Twitter y no olvide unirse a nuestro SubReddit de más de 100.000 ML y suscribirse a nuestro boletín. ¡Esperar! estas en telegrama? Ahora también puedes unirte a nosotros en Telegram.

Asif Razzaq es el director ejecutivo de Marktechpost Media Inc.. Como empresario e ingeniero visionario, Asif está comprometido a aprovechar el potencial de la inteligencia artificial para el bien social. Su esfuerzo más reciente es el lanzamiento de una plataforma de medios de inteligencia artificial, Marktechpost, que se destaca por su cobertura en profundidad del aprendizaje automático y las noticias sobre aprendizaje profundo que es técnicamente sólida y fácilmente comprensible para una amplia audiencia. La plataforma cuenta con más de 2 millones de visitas mensuales, lo que ilustra su popularidad entre el público.