En este tutorial avanzado, nuestro objetivo es Construir un sistema de automatización de tareas de múltiples agentes usando el Primisai Nexus Marco, que está completamente integrado con la API de OpenAI. Nuestro objetivo principal es demostrar cómo la supervisión jerárquica, la utilización de herramientas inteligentes y las salidas estructuradas pueden facilitar la coordinación de múltiples agentes de IA para realizar tareas complejas, que van desde la planificación y el desarrollo hasta la garantía de calidad y el análisis de datos. Mientras caminamos por cada fase, no solo construimos agentes individuales; Arquitectamos un ecosistema colaborativo donde cada agente tiene un papel claro, responsabilidades y herramientas inteligentes para realizar la tarea.
!pip install primisai openai nest-asyncio
import os
import nest_asyncio
from primisai.nexus.core import AI, Agent, Supervisor
from primisai.nexus.utils.debugger import Debugger
import json
nest_asyncio.apply()Comenzamos instalando las dependencias principales: Primisai para la orquestación de agentes, OpenAI para el acceso LLM y Nest_asyncio para manejar las peculiaridades de bucle de eventos de Colab. Después de aplicar Nest_asyncio, nos aseguramos de que el cuaderno esté listo para ejecutar tareas asíncronas sin problemas, un requisito clave para la ejecución de múltiples agentes.
print("🚀 PrimisAI Nexus Advanced Tutorial with OpenAI API")
print("=" * 55)
os.environ["OPENAI_API_KEY"] = "Use Your Own API Key Here5"
# llm_config = {
# "api_key": os.environ["OPENAI_API_KEY"],
# "model": "gpt-4o-mini",
# "base_url": "https://api.openai.com/v1",
# "temperature": 0.7
# }
llm_config = {
"api_key": os.environ["OPENAI_API_KEY"],
"model": "gpt-3.5-turbo",
"base_url": "https://api.openai.com/v1",
"temperature": 0.7
}
print("📋 API Configuration:")
print(f"• Model: {llm_config['model']}")
print(f"• Base URL: {llm_config['base_url']}")
print("• Note: OpenAI has limited free tokens through April 2025")
print("• Alternative: Consider Puter.js for unlimited free access")Para alimentar a nuestros agentes, nos conectamos a los modelos de OpenAI, comenzando con GPT-3.5-Turbo para tareas rentables. Almacenamos nuestra clave API en las variables de entorno y construimos un diccionario de configuración que especifica el modelo, la temperatura y la URL base. Esta sección nos permite cambiar de manera flexible entre modelos, como GPT-4O-Mini o GPT-4O, dependiendo de la complejidad y el costo de la tarea.
code_schema = {
"type": "object",
"properties": {
"description": {"type": "string", "description": "Code explanation"},
"code": {"type": "string", "description": "Python code implementation"},
"language": {"type": "string", "description": "Programming language"},
"complexity": {"type": "string", "enum": ["beginner", "intermediate", "advanced"]},
"test_cases": {"type": "array", "items": {"type": "string"}, "description": "Example usage"}
},
"required": ["description", "code", "language"]
}
analysis_schema = {
"type": "object",
"properties": {
"summary": {"type": "string", "description": "Brief analysis summary"},
"insights": {"type": "array", "items": {"type": "string"}, "description": "Key insights"},
"recommendations": {"type": "array", "items": {"type": "string"}, "description": "Action items"},
"confidence": {"type": "number", "minimum": 0, "maximum": 1},
"methodology": {"type": "string", "description": "Analysis approach used"}
},
"required": ["summary", "insights", "confidence"]
}
planning_schema = {
"type": "object",
"properties": {
"tasks": {"type": "array", "items": {"type": "string"}, "description": "List of tasks to complete"},
"priority": {"type": "string", "enum": ["low", "medium", "high"]},
"estimated_time": {"type": "string", "description": "Time estimate"},
"dependencies": {"type": "array", "items": {"type": "string"}, "description": "Task dependencies"}
},
"required": ["tasks", "priority"]
}
Definimos esquemas JSON para tres tipos de agentes: Codewriter, Analyst de datos y planificador de proyectos. Estos esquemas imponen la estructura en las respuestas del agente, lo que hace que la máquina salga sea legible y predecible. Nos ayuda a garantizar que el sistema devuelva datos consistentes, como bloques de código, ideas o plazos del proyecto, incluso cuando diferentes LLM están detrás de escena.
def calculate_metrics(data_str):
"""Calculate comprehensive statistics for numerical data"""
try:
data = json.loads(data_str) if isinstance(data_str, str) else data_str
if isinstance(data, list) and all(isinstance(x, (int, float)) for x in data):
import statistics
return {
"mean": statistics.mean(data),
"median": statistics.median(data),
"mode": statistics.mode(data) if len(set(data)) < len(data) else "No mode",
"std_dev": statistics.stdev(data) if len(data) > 1 else 0,
"max": max(data),
"min": min(data),
"count": len(data),
"sum": sum(data)
}
return {"error": "Invalid data format - expecting array of numbers"}
except Exception as e:
return {"error": f"Could not parse data: {str(e)}"}
def validate_code(code):
"""Advanced code validation with syntax and basic security checks"""
try:
dangerous_imports = ['os', 'subprocess', 'eval', 'exec', '__import__']
security_warnings = []
for danger in dangerous_imports:
if danger in code:
security_warnings.append(f"Potentially dangerous: {danger}")
compile(code, '<string>', 'exec')
return {
"valid": True,
"message": "Code syntax is valid",
"security_warnings": security_warnings,
"lines": len(code.split('\n'))
}
except SyntaxError as e:
return {
"valid": False,
"message": f"Syntax error: {e}",
"line": getattr(e, 'lineno', 'unknown'),
"security_warnings": []
}
def search_documentation(query):
"""Simulate searching documentation (placeholder function)"""
docs = {
"python": "Python is a high-level programming language",
"list": "Lists are ordered, mutable collections in Python",
"function": "Functions are reusable blocks of code",
"class": "Classes define objects with attributes and methods"
}
results = []
for key, value in docs.items():
if query.lower() in key.lower():
results.append(f"{key}: {value}")
return {
"query": query,
"results": results if results else ["No documentation found"],
"total_results": len(results)
}A continuación, agregamos herramientas personalizadas que los agentes podrían llamar, como calculado_metrics para resúmenes estadísticos, validate_code para la sintaxis y las verificaciones de seguridad, y search_documentation para ayuda de programación simulada. Estas herramientas extienden las habilidades de los agentes, convirtiéndolas de chatbots simples en trabajadores interactivos impulsados por servicios públicos capaces de razonamiento y validación autónomos.
print("\n📋 Setting up Multi-Agent Hierarchy with OpenAI")
main_supervisor = Supervisor(
name="ProjectManager",
llm_config=llm_config,
system_message="You are a senior project manager coordinating development and analysis tasks. Delegate appropriately, provide clear summaries, and ensure quality delivery. Always consider time estimates and dependencies."
)
dev_supervisor = Supervisor(
name="DevManager",
llm_config=llm_config,
is_assistant=True,
system_message="You manage development tasks. Coordinate between coding, testing, and code review. Ensure best practices and security."
)
analysis_supervisor = Supervisor(
name="AnalysisManager",
llm_config=llm_config,
is_assistant=True,
system_message="You manage data analysis and research tasks. Ensure thorough analysis, statistical rigor, and actionable insights."
)
qa_supervisor = Supervisor(
name="QAManager",
llm_config=llm_config,
is_assistant=True,
system_message="You manage quality assurance and testing. Ensure thorough validation and documentation."
)
Para simular una estructura de gestión del mundo real, creamos una jerarquía de niveles múltiples. Un ProjectManager sirve como supervisor de raíz, supervisando a tres supervisores asistentes (Devmanager, AnalysisManager y Qamanager), cada uno a cargo de agentes específicos del dominio. Esta jerarquía modular permite que las tareas fluyan desde la estrategia de alto nivel hasta la ejecución granular.
code_agent = Agent(
name="CodeWriter",
llm_config=llm_config,
system_message="You are an expert Python developer. Write clean, efficient, well-documented code with proper error handling. Always include test cases and follow PEP 8 standards.",
output_schema=code_schema,
tools=[{
"metadata": {
"function": {
"name": "validate_code",
"description": "Validates Python code syntax and checks for security issues",
"parameters": {
"type": "object",
"properties": {
"code": {"type": "string", "description": "Python code to validate"}
},
"required": ["code"]
}
}
},
"tool": validate_code
}, {
"metadata": {
"function": {
"name": "search_documentation",
"description": "Search for programming documentation and examples",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "Documentation topic to search for"}
},
"required": ["query"]
}
}
},
"tool": search_documentation
}],
use_tools=True
)
review_agent = Agent(
name="CodeReviewer",
llm_config=llm_config,
system_message="You are a senior code reviewer. Analyze code for best practices, efficiency, security, maintainability, and potential issues. Provide constructive feedback and suggestions.",
keep_history=True,
tools=[{
"metadata": {
"function": {
"name": "validate_code",
"description": "Validates code syntax and security",
"parameters": {
"type": "object",
"properties": {
"code": {"type": "string", "description": "Code to validate"}
},
"required": ["code"]
}
}
},
"tool": validate_code
}],
use_tools=True
)
analyst_agent = Agent(
name="DataAnalyst",
llm_config=llm_config,
system_message="You are a data scientist specializing in statistical analysis and insights generation. Provide thorough analysis with confidence metrics and actionable recommendations.",
output_schema=analysis_schema,
tools=[{
"metadata": {
"function": {
"name": "calculate_metrics",
"description": "Calculates comprehensive statistics for numerical data",
"parameters": {
"type": "object",
"properties": {
"data_str": {"type": "string", "description": "JSON string of numerical data array"}
},
"required": ["data_str"]
}
}
},
"tool": calculate_metrics
}],
use_tools=True
)
planner_agent = Agent(
name="ProjectPlanner",
llm_config=llm_config,
system_message="You are a project planning specialist. Break down complex projects into manageable tasks with realistic time estimates and clear dependencies.",
output_schema=planning_schema
)
tester_agent = Agent(
name="QATester",
llm_config=llm_config,
system_message="You are a QA specialist focused on comprehensive testing strategies, edge cases, and quality assurance.",
tools=[{
"metadata": {
"function": {
"name": "validate_code",
"description": "Validates code for testing",
"parameters": {
"type": "object",
"properties": {
"code": {"type": "string", "description": "Code to test"}
},
"required": ["code"]
}
}
},
"tool": validate_code
}],
use_tools=True
)
Luego creamos un conjunto diverso de agentes especializados: Codewriter para generar código de Python, CodeReviewer para revisar la lógica y la seguridad, DataAnalyst para realizar análisis de datos estructurados, proyectar para el desglose de tareas y Qatester para verificaciones de calidad. Cada agente tiene herramientas específicas de dominio, esquemas de salida e instrucciones del sistema adaptadas a su papel.
dev_supervisor.register_agent(code_agent)
dev_supervisor.register_agent(review_agent)
analysis_supervisor.register_agent(analyst_agent)
qa_supervisor.register_agent(tester_agent)
main_supervisor.register_agent(dev_supervisor)
main_supervisor.register_agent(analysis_supervisor)
main_supervisor.register_agent(qa_supervisor)
main_supervisor.register_agent(planner_agent)Todos los agentes están registrados bajo sus respectivos supervisores, y los supervisores asistentes están, a su vez, registrados con el supervisor principal. Esta configuración crea un ecosistema de agente totalmente vinculado, donde las instrucciones podrían en cascada desde el agente de nivel superior a cualquier agente especializado en la red.
print("\n🌳 Agent Hierarchy:")
main_supervisor.display_agent_graph()
print("\n🧪 Testing Full Multi-Agent Communication")
print("-" * 45)
try:
test_response = main_supervisor.chat("Hello! Please introduce your team and explain how you coordinate complex projects.")
print(f"✅ Supervisor communication test successful!")
print(f"Response preview: {test_response[:200]}...")
except Exception as e:
print(f"❌ Supervisor test failed: {str(e)}")
print("Falling back to direct agent testing...")Visualizamos toda la jerarquía usando display_agent_graph () para confirmar nuestra estructura. Ofrece una visión clara de cómo cada agente está conectado dentro del flujo de gestión de tareas más amplio, un diagnóstico útil antes del despliegue.
print("\n🎯 Complex Multi-Agent Task Execution")
print("-" * 40)
complex_task = """Create a Python function that implements a binary search algorithm,
have it reviewed for optimization, tested thoroughly, and provide a project plan
for integrating it into a larger search system."""
print(f"Complex Task: {complex_task}")
try:
complex_response = main_supervisor.chat(complex_task)
print(f"✅ Complex task completed")
print(f"Response: {complex_response[:300]}...")
except Exception as e:
print(f"❌ Complex task failed: {str(e)}")Le damos al sistema completo una tarea del mundo real: crear una función de búsqueda binaria, revisarla, probarlo y planificar su integración en un proyecto más grande. El ProjectManager coordina a los agentes sin problemas en el desarrollo, el control de calidad y la planificación, lo que demuestra el verdadero poder de la orquestación jerárquica de agente basada en herramientas.
print("\n🔧 Tool Integration & Structured Outputs")
print("-" * 43)
print("Testing Code Agent with tools...")
try:
code_response = code_agent.chat("Create a function to calculate fibonacci numbers with memoization")
print(f"✅ Code Agent with tools: Working")
print(f"Response type: {type(code_response)}")
if isinstance(code_response, str) and code_response.strip().startswith('{'):
code_data = json.loads(code_response)
print(f" - Description: {code_data.get('description', 'N/A')[:50]}...")
print(f" - Language: {code_data.get('language', 'N/A')}")
print(f" - Complexity: {code_data.get('complexity', 'N/A')}")
else:
print(f" - Raw response: {code_response[:100]}...")
except Exception as e:
print(f"❌ Code Agent error: {str(e)}")
print("\nTesting Analyst Agent with tools...")
try:
analysis_response = analyst_agent.chat("Analyze this sales data: [100, 150, 120, 180, 200, 175, 160, 190, 220, 185]. What trends do you see?")
print(f"✅ Analyst Agent with tools: Working")
if isinstance(analysis_response, str) and analysis_response.strip().startswith('{'):
analysis_data = json.loads(analysis_response)
print(f" - Summary: {analysis_data.get('summary', 'N/A')[:50]}...")
print(f" - Confidence: {analysis_data.get('confidence', 'N/A')}")
print(f" - Insights count: {len(analysis_data.get('insights', []))}")
else:
print(f" - Raw response: {analysis_response[:100]}...")
except Exception as e:
print(f"❌ Analyst Agent error: {str(e)}")Probamos directamente las capacidades de dos agentes especializados que usan indicaciones reales. Primero le pedimos al agente Codewriter que genere una función Fibonacci con memoización y validara que devuelve la salida estructurada que contiene una descripción del código, el lenguaje y el nivel de complejidad. Luego, evaluamos el agente DataAnalyst alimentando sus datos de ventas para extraer tendencias.
print("\n🔨 Manual Tool Usage")
print("-" * 22)
# Test all tools manually
sample_data = "[95, 87, 92, 88, 91, 89, 94, 90, 86, 93]"
metrics_result = calculate_metrics(sample_data)
print(f"Statistics for {sample_data}:")
for key, value in metrics_result.items():
print(f" {key}: {value}")
print("\nCode validation test:")
test_code = """
def binary_search(arr, target):
left, right = 0, len(arr) - 1
while left <= right:
mid = (left + right) // 2
if arr[mid] == target:
return mid
elif arr[mid] < target:
left = mid + 1
else:
right = mid - 1
return -1
"""
validation_result = validate_code(test_code)
print(f"Validation result: {validation_result}")
print("\nDocumentation search test:")
doc_result = search_documentation("python function")
print(f"Search results: {doc_result}")Salimos del marco del agente para probar cada herramienta directamente. Primero, utilizamos la herramienta Calcullate_Metrics en un conjunto de datos de diez números, lo que confirma que las estadísticas devueltas correctamente como media, mediana, modo y desviación estándar. A continuación, ejecutamos la herramienta Validate_Code en una función de búsqueda binaria de muestra, que confirma tanto la corrección sintáctica como los indicadores sin advertencias de seguridad. Finalmente, probamos la herramienta Search_Documation con la consulta “Función Python” y recibimos fragmentos de documentación relevantes, verificando su capacidad para simular eficientemente la búsqueda contextual.
print("\n🚀 Advanced Multi-Agent Workflow")
print("-" * 35)
workflow_stages = [
("Planning", "Create a project plan for building a web scraper for news articles"),
("Development", "Implement the web scraper with error handling and rate limiting"),
("Review", "Review the web scraper code for security and efficiency"),
("Testing", "Create comprehensive test cases for the web scraper"),
("Analysis", "Analyze sample scraped data: [45, 67, 23, 89, 12, 56, 78, 34, 91, 43]")
]
workflow_results = {}
for stage, task in workflow_stages:
print(f"\n{stage} Stage: {task}")
try:
if stage == "Planning":
response = planner_agent.chat(task)
elif stage == "Development":
response = code_agent.chat(task)
elif stage == "Review":
response = review_agent.chat(task)
elif stage == "Testing":
response = tester_agent.chat(task)
elif stage == "Analysis":
response = analyst_agent.chat(task)
workflow_results[stage] = response
print(f"✅ {stage} completed: {response[:80]}...")
except Exception as e:
print(f"❌ {stage} failed: {str(e)}")
workflow_results[stage] = f"Error: {str(e)}"Simulamos un ciclo de vida del proyecto de cinco etapas: planificación, desarrollo, revisión, pruebas y análisis. Cada tarea se pasa al agente más relevante, y las respuestas se recopilan para evaluar el rendimiento. Esto demuestra la capacidad del marco para administrar flujos de trabajo de extremo a extremo sin intervención manual.
print("\n📊 System Monitoring & Performance")
print("-" * 37)
debugger = Debugger(name="OpenAITutorialDebugger")
debugger.log("Advanced OpenAI tutorial execution completed successfully")
print(f"Main Supervisor ID: {main_supervisor.workflow_id}")Activamos la herramienta de depuración para rastrear el rendimiento de nuestros eventos del sistema de sesión y registro. También imprimimos el Flow_ID de trabajo del supervisor principal como un identificador rastreable, útil al administrar múltiples flujos de trabajo en la producción.
En conclusión, hemos construido con éxito un sistema multi-agente multipatible totalmente automatizado y compatible con OpenAI utilizando Primisai Nexus. Cada agente opera con claridad, precisión y autonomía, ya sea escribir código, validar la lógica, analizar datos o desglosar flujos de trabajo complejos. Nuestra estructura jerárquica permite la delegación de tareas sin problemas y la escalabilidad modular. Primisai Nexus Framework establece una base sólida para automatizar tareas del mundo real, ya sea en desarrollo de software, investigación, planificación o operaciones de datos, a través de una colaboración inteligente entre agentes especializados.
Mira el Codos. Todo el crédito por esta investigación va a los investigadores de este proyecto. Además, siéntete libre de seguirnos Gorjeo, YouTube y Spotify Y no olvides unirte a nuestro Subreddit de 100k+ ml y suscribirse a Nuestro boletín.
Asif Razzaq es el CEO de MarktechPost Media Inc .. Como empresario e ingeniero visionario, ASIF se compromete 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 profunda de noticias de aprendizaje automático y de aprendizaje profundo que es técnicamente sólido y fácilmente comprensible por una audiencia amplia. La plataforma cuenta con más de 2 millones de vistas mensuales, ilustrando su popularidad entre el público.