En este tutorial, construimos un sistema de estrategia de infraestructura y datos agentes utilizando el modelo liviano Qwen2.5-0.5B-Instruct para una ejecución eficiente. Comenzamos creando un marco de agentes LLM flexible y luego desarrollamos agentes especializados que manejan diferentes capas de gestión de datos, desde la ingesta y el análisis de calidad hasta la optimización de la infraestructura. Integramos a estos agentes en un orquestador que coordina sus interacciones, garantizando una colaboración fluida entre múltiples agentes en todo el proceso de datos. A través de ejemplos prácticos como el comercio electrónico y los canales de IoT, exploramos cómo la toma de decisiones autónoma puede optimizar operaciones de datos complejas. Consulta los CÓDIGOS COMPLETOS aquí.
def generar_respuesta(self, mensaje: str, max_tokens: int = 150) -> str: mensajes = [
{“role”: “system”, “content”: f”You are a {self.role} agent in a data infrastructure system.”},
{“role”: “user”, “content”: prompt}
]
texto = self.tokenizer.apply_chat_template(mensajes, tokenize=False, add_generación_prompt=True) model_inputs = self.tokenizer([text]return_tensors=”pt”).to(self.device) con torch.no_grad(): generate_ids = self.model.generate( model_inputs.input_ids, max_new_tokens=max_tokens, temperatura=0.7, do_sample=True, top_p=0.95 ) generate_ids = [output_ids[len(input_ids):] para input_ids, output_ids en zip(model_inputs.input_ids, generate_ids)]respuesta = self.tokenizer.batch_decode(generated_ids, skip_special_tokens=True)[0]
self.conversation_history.append({“prompt”: mensaje, “respuesta”: respuesta}) devolver respuesta
Comenzamos configurando la infraestructura liviana del agente LLM utilizando el modelo Qwen2.5-0.5B-Instruct. Cargamos el modelo y el tokenizador, y definimos una clase de agente base capaz de manejar conversaciones contextuales y generar respuestas inteligentes. Esto constituye la base fundamental sobre la cual nuestros agentes especializados operan de manera eficiente dentro de Colab. Consulta los CÓDIGOS COMPLETOS aquí.
Diseñamos los agentes de ingesta y calidad de datos para centrarnos en el análisis estructurado de los canales de datos. Dejamos que el agente de ingesta determine el mejor enfoque para el flujo de datos, mientras que el agente de calidad evalúa la integridad, la coherencia y los problemas de los datos para proporcionar información útil. Juntos, establecen las dos primeras capas de gestión autónoma de datos. Consulta los CÓDIGOS COMPLETOS aquí.
Desarrollamos el Agente de optimización de infraestructura para analizar continuamente métricas clave como la utilización de CPU, memoria y almacenamiento. Lo utilizamos para generar sugerencias de optimización inteligentes, lo que nos ayuda a mantener un alto rendimiento y eficiencia de recursos. Este agente garantiza que nuestra infraestructura siga siendo receptiva y escalable durante las operaciones de datos. Consulta los CÓDIGOS COMPLETOS aquí.
def Process_data_pipeline(self, pipeline_config: Dict) -> Dict: resultados = {“pipeline_id”: pipeline_config.get(“id”, “unknown”), “start_time”: datetime.now().isoformat(), “etapas”: []} imprimir(“\n[Stage 1] Análisis de ingesta de datos”) ingestion_result = self.ingestion_agent.analyze_data_source(pipeline_config.get(“source”, {})) print(f”Estrategia: {ingestion_result[‘strategy’][:150]}…”) resultados[“stages”].append({“etapa”: “ingestión”, “resultado”: ingestión_result}) print(“\n[Stage 2] Evaluación de la calidad de los datos”) resultado_calidad = self.quality_agent.assess_data_quality(pipeline_config.get(“quality_metrics”, {})) print(f”Evaluación: {resultado_calidad[‘assessment’][:150]}…”) print(f”Severidad: {calidad_resultado[‘severity’]}”) resultados[“stages”].append({“etapa”: “calidad”, “resultado”: calidad_resultado}) print(“\n[Stage 3] Optimización de infraestructura”) optimización_result = self.optimization_agent.optimize_resources(pipeline_config.get(“infrastructure_metrics”, {})) print(f”Recomendaciones: {optimization_result[‘recommendations’][:150]}…”) print(f”Prioridad: {resultado_optimización[‘priority’]}”) resultados[“stages”].append({“etapa”: “optimización”, “resultado”: optimización_resultado}) resultados[“end_time”] = datetime.now().isoformat() resultados[“status”] = “completado” self.execution_log.append(resultados) devuelve resultados def generate_summary_report(self) -> pd.DataFrame: si no, self.execution_log: devuelve pd.DataFrame() resumen_data = []
para iniciar sesión self.execution_log: resumen_data.append({“ID de canalización”: registro[“pipeline_id”]”Hora de inicio”: registro[“start_time”]”Estado”: iniciar sesión[“status”]”Etapas completadas”: len(log[“stages”])}) devolver pd.DataFrame(summary_data)
Construimos un Agentic Data Orchestrator para coordinar a todos los agentes especializados bajo un flujo de trabajo unificado. Lo usamos para gestionar la ejecución de la canalización de un extremo a otro, activando la ingesta, los controles de calidad y la optimización de forma secuencial. Al hacer esto, aportamos estructura, colaboración y automatización a todo el sistema multiagente. Consulta los CÓDIGOS COMPLETOS aquí.
Demostramos nuestro sistema completo a través de dos ejemplos del mundo real: un comercio electrónico y un canal de datos de IoT. Observamos cómo cada agente desempeña su rol de forma autónoma mientras contribuye a un objetivo compartido. Finalmente, generamos un informe resumido que confirma la eficiencia de la orquestación y el poder de la inteligencia agente ligera.
En conclusión, diseñamos y ejecutamos un marco de infraestructura de datos inteligente y multiagente impulsado por un modelo compacto de código abierto. Somos testigos de cómo agentes independientes pero cooperativos pueden analizar, evaluar y optimizar de forma autónoma sistemas de datos del mundo real. Toda la configuración demuestra cómo los LLM livianos pueden manejar de manera eficiente la inteligencia de infraestructura, al mismo tiempo que resalta cómo la orquestación agente transforma los flujos de trabajo de datos tradicionales en sistemas adaptables y autooptimizados listos para aplicaciones empresariales escalables.
Consulta 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 emprendedor 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.
🙌 Siga MARKTECHPOST: agréguenos como fuente preferida en Google.