Una guía de codificación para diseñar un sistema de IA agente utilizando una arquitectura de plano de control para flujos de trabajo de razonamiento basados ​​en herramientas seguros, modulares y escalables

En este tutorial, construimos una IA Agentic avanzada utilizando el patrón de diseño del plano de control y analizamos cada componente paso a paso a medida que lo implementamos. Tratamos el plano de control como el orquestador central que coordina las herramientas, gestiona las reglas de seguridad y estructura el circuito de razonamiento. Además, configuramos un sistema de recuperación en miniatura, definimos herramientas modulares e integramos una capa de razonamiento agente que planifica y ejecuta acciones dinámicamente. Por último, observamos cómo todo el sistema se comporta como una IA disciplinada y consciente de las herramientas, capaz de recuperar conocimientos, evaluar la comprensión, actualizar los perfiles de los alumnos y registrar todas las interacciones a través de una arquitectura unificada y escalable. Consulta los CÓDIGOS COMPLETOS aquí.

importar subproceso importar sys def install_deps(): deps = [‘anthropic’, ‘numpy’, ‘scikit-learn’]
para departamento en departamento: subproceso.check_call([sys.executable, ‘-m’, ‘pip’, ‘install’, ‘-q’, dep]) intente: importar antrópico excepto ImportError: install_deps() importar antrópico importar json importar numpy como np de sklearn.metrics.pairwise importar coseno_similaridad de clases de datos importar clase de datos, asdict de escribir importar Lista, Dict, Cualquiera, Opcional de fecha y hora importar fecha y hora @dataclass clase Documento: id: str contenido: str metadatos: Dict[str, Any]
incrustación: Opcional[np.ndarray] = Ninguna clase SimpleRAGRetriever: def __init__(self): self.documents = self._init_knowledge_base() def _init_knowledge_base(self) -> Lista[Document]: documentos = [
Document(“cs101”, “Python basics: Variables store data. Use x=5 for integers, name=”Alice” for strings. Print with print().”, {“topic”: “python”, “level”: “beginner”}),
Document(“cs102”, “Functions encapsulate reusable code. Define with def func_name(params): and call with func_name(args).”, {“topic”: “python”, “level”: “intermediate”}),
Document(“cs103”, “Object-oriented programming uses classes. class MyClass: defines structure, __init__ initializes instances.”, {“topic”: “python”, “level”: “advanced”}),
Document(“math101”, “Linear algebra: Vectors are ordered lists of numbers. Matrix multiplication combines transformations.”, {“topic”: “math”, “level”: “intermediate”}),
Document(“ml101”, “Machine learning trains models on data to make predictions. Supervised learning uses labeled examples.”, {“topic”: “ml”, “level”: “beginner”}),
Document(“ml102”, “Neural networks are composed of layers. Each layer applies weights and activation functions to transform inputs.”, {“topic”: “ml”, “level”: “advanced”}),
]
para i, doc en enumerar (docs): doc.embedding = np.random.rand(128) doc.embedding[i*20:(i+1)*20] += 2 return docs def retrieve(self, query: str, top_k: int = 2) -> Lista[Document]: query_embedding = np.random.rand(128) puntuaciones = [cosine_similarity([query_embedding], [doc.embedding])[0][0] para doc en self.documents]top_indices = np.argsort(puntuaciones)[-top_k:][::-1]

devolver [self.documents[i] para i en top_indices]

Configuramos todas las dependencias, importamos las bibliotecas en las que confiamos e inicializamos las estructuras de datos para nuestra base de conocimientos. Definimos un recuperador simple y generamos incrustaciones simuladas para simular la búsqueda de similitudes de una manera liviana. A medida que ejecutamos este bloque, preparamos todo lo necesario para el razonamiento basado en la recuperación en los componentes posteriores. Consulta los CÓDIGOS COMPLETOS aquí.

clase ToolRegistry: def __init__(self, recuperador: SimpleRAGRetriever): self.retriever = recuperador self.interaction_log = []
self.user_state = {“nivel”: “principiante”, “temas_cubiertos”: []} def search_knowledge(self, consulta: str, filtros: Opcional[Dict] = Ninguno) -> Dict: docs = self.retriever.retrieve(consulta, top_k=2) si filtros: docs = [d for d in docs if all(d.metadata.get(k) == v for k, v in filters.items())]
return { “herramienta”: “conocimiento_búsqueda”, “resultados”: [{“content”: d.content, “metadata”: d.metadata} for d in docs]”count”: len(docs) } def evalua_understanding(self, tema: str) -> Dict: preguntas = { “python”: [“What keyword defines a function?”, “How do you create a variable?”]”ml”: [“What is supervised learning?”, “Name two types of ML algorithms.”]”matemáticas”: [“What is a vector?”, “Explain matrix multiplication.”]
} return { “herramienta”: “evaluación_comprensión”, “tema”: tema, “preguntas”: preguntas.get(tema, [“General comprehension check.”]) } def update_learner_profile(self, tema: str, nivel: str) -> Dict: si el tema no está en self.user_state[“topics_covered”]: self.usuario_estado[“topics_covered”].append(tema) self.user_state[“level”] = nivel return { “tool”: “update_learner_profile”, “status”: “updated”, “profile”: self.user_state.copy() } def log_interaction(self, event: str, detalles: Dict) -> Dict: log_entry = { “timestamp”: datetime.now().isoformat(), “event”: event, “details”: detalles } self.interaction_log.append(log_entry) return {“tool”: “log_interaction”, “status”: “registrado”, “entry_id”: len(self.interaction_log)}

Construimos el registro de herramientas que nuestro agente utiliza mientras interactúa con el sistema. Definimos herramientas como búsqueda de conocimientos, evaluaciones, actualizaciones de perfiles y registros, y mantenemos un diccionario persistente de estado de usuario. A medida que usamos esta capa, vemos cómo cada herramienta se convierte en una capacidad modular a la que el plano de control puede dirigirse. Consulta los CÓDIGOS COMPLETOS aquí.

clase ControlPlane: def __init__(self, tool_registry: ToolRegistry): self.tools = tool_registry self.safety_rules = { “max_tools_per_request”: 4, “allowed_tools”: [“search_knowledge”, “assess_understanding”,
“update_learner_profile”, “log_interaction”]
} self.execution_log = []

def ejecutar(self, plan: Dict[str, Any]) -> Dictar[str, Any]: si no es self._validate_request(plan): return {“error”: “Error en la validación de seguridad”, “plan”: plan} acción = plan.get(“acción”) params = plan.get(“parámetros”, {}) resultado = self._route_and_execute(action, params) self.execution_log.append({ “timestamp”: datetime.now().isoformat(), “plan”: plan, “resultado”: resultado }) return { “success”: True, “action”: action, “result”: result, “metadata”: { “execution_count”: len(self.execution_log), “safety_checks_passed”: True } } def _validate_request(self, plan: Dict) -> bool: action = plan.get(“action”) si la acción no está en self.safety_rules[“allowed_tools”]: return False si len(self.execution_log) >= 100: return False return True def _route_and_execute(self, action: str, params: Dict) -> Any: tool_map = { “search_knowledge”: self.tools.search_knowledge, “assess_understanding”: self.tools.assess_understanding, “update_learner_profile”: self.tools.update_learner_profile, “log_interaction”: self.tools.log_interaction } tool_func = tool_map.get(action) if tool_func: return tool_func(**params) return {“error”: f”Acción desconocida: {acción}”}

Implementamos el plano de control que organiza la ejecución de herramientas, verifica las reglas de seguridad y gestiona los permisos. Validamos cada solicitud, enviamos acciones a la herramienta adecuada y mantenemos un registro de ejecución para mayor transparencia. A medida que ejecutamos este fragmento, observamos cómo el plano de control se convierte en el sistema de gobierno que garantiza un comportamiento agente predecible y seguro. Consulta los CÓDIGOS COMPLETOS aquí.

clase TutorAgent: def __init__(self, control_plane: ControlPlane, api_key: str): self.control_plane = control_plane self.client = anthropic.Anthropic(api_key=api_key) self.conversation_history = []

def enseñar(self, Student_query: str) -> str: plan = self._plan_actions(student_query) resultados = []
para action_plan en el plan: resultado = self.control_plane.execute(action_plan) resultados.append(resultado) respuesta = self._synthesize_response(student_query, resultados) self.conversation_history.append({ “query”: Student_query, “plan”: plan, “resultados”: resultados, “respuesta”: respuesta }) devolver respuesta def _plan_actions(self, query: str) -> Lista[Dict]: plano = []
query_lower = query.lower() si existe (kw en query_lower para kw en [“what”, “how”, “explain”, “teach”]): plan.append({ “action”: “search_knowledge”, “parameters”: {“query”: query}, “context”: {“intent”: “knowledge_retrieval”} }) si corresponde (kw en query_lower para kw en [“test”, “quiz”, “assess”, “check”]): topic = “python” if “python” in query_lower else “ml” plan.append({ “action”: “assess_understanding”, “parameters”: {“topic”: topic}, “context”: {“intent”: “evaluación”} }) plan.append({ “action”: “log_interaction”, “parameters”: {“event”: “query_processed”, “details”: {“query”: consulta}}, “contexto”: {“intención”: “registro”} }) plan de retorno def _synthesize_response(self, consulta: str, resultados: Lista[Dict]) -> cadena: partes_respuesta = [f”Student Query: {query}\n”]

para obtener resultados en resultados: si result.get(“éxito”) y “resultado” en resultados: tool_result = resultado[“result”]

si resultado[“action”] == “search_knowledge”: respuesta_parts.append(“\n📚 Conocimientos recuperados:”) para el documento en tool_result.get(“resultados”, []): respuesta_partes.append(f” • {doc[‘content’]}”) resultado elif[“action”] == “assess_understanding”: respuesta_parts.append(“\n✅ Preguntas de evaluación:”) para q en tool_result.get(“preguntas”, []): respuesta_parts.append(f” • {q}”) devuelve “\n”.join(response_parts)

Implementamos el TutorAgent, que planifica acciones, se comunica con el plano de control y sintetiza respuestas finales. Analizamos consultas, generamos planes de varios pasos y combinamos los resultados de las herramientas para obtener respuestas significativas para los alumnos. A medida que ejecutamos este fragmento, vemos que el agente se comporta de manera inteligente al coordinar la recuperación, la evaluación y el registro. Consulta los CÓDIGOS COMPLETOS aquí.

def run_demo(): print(“=” * 70) print(“Plano de control como herramienta: demostración del tutor RAG AI”) print(“=” * 70) API_KEY = “tu-clave-api-aquí” retriever = SimpleRAGRetriever() tool_registry = ToolRegistry(retriever) control_plane = ControlPlane(tool_registry) print(“Sistema inicializado”) print(f”Herramientas: {len(control_plane.seguridad_rules[‘allowed_tools’])}”) print(f”Base de conocimientos: {len(retriever.documents)} documentos”) intente: tutor = TutorAgent(control_plane, API_KEY) excepto: print(“Modo simulado habilitado”) tutor = Ninguno demo_queries = [
“Explain Python functions to me”,
“I want to learn about machine learning”,
“Test my understanding of Python basics”
]

para consulta en demo_queries: print(“\n— Consulta —“) if tutor: print(tutor.teach(consulta)) else: plan = [
{“action”: “search_knowledge”, “parameters”: {“query”: query}},
{“action”: “log_interaction”, “parameters”: {“event”: “query”, “details”: {}}}
]
print(consulta) para la acción en el plan: resultado = control_plane.execute(action) print(f”{action[‘action’]}: {result.get(‘success’, False)}”) print(“Resumen”) print(f”Ejecuciones: {len(control_plane.execution_log)}”) print(f”Registros: {len(tool_registry.interaction_log)}”) print(f”Perfil: {tool_registry.user_state}”) if __name__ == “__main__”: run_demo()

Ejecutamos una demostración completa que inicializa todos los componentes, procesa consultas de muestra de los estudiantes e imprime resúmenes del estado del sistema. Observamos al agente pasar por la recuperación y el registro mientras el plano de control aplica las reglas y rastrea el historial de ejecución. Al terminar este bloque, obtenemos una imagen clara de cómo toda la arquitectura funciona en conjunto en un circuito de enseñanza realista.

En conclusión, obtenemos una comprensión clara de cómo el patrón del plano de control simplifica la orquestación, fortalece la seguridad y crea una separación clara entre el razonamiento y la ejecución de la herramienta. Ahora vemos cómo un sistema de recuperación, un registro de herramientas y una capa de planificación agente se unen para formar un tutor de IA coherente que responde de manera inteligente a las consultas de los estudiantes. A medida que experimentamos con la demostración, observamos cómo el sistema enruta tareas, aplica reglas y sintetiza información útil a partir de los resultados de las herramientas, sin dejar de ser modular y extensible.

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 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.

🙌 Siga MARKTECHPOST: agréguenos como fuente preferida en Google.