Core Concept: Factory
What is a Factory?
A Factory is the central registry in aiNXT that manages object creation. It combines Loaders (which find constructors), Parsers (which transform configurations), and Decorators (which modify objects) into a single, unified interface.
Real-world analogy: Think of a Factory as a smart workshop manager: - It has a catalog of what can be built (registered constructors from Loaders) - It knows how to read blueprints (configuration via Parsers) - It can apply modifications after construction (Decorators) - It coordinates everything to deliver the final product
Source: ainxt/factory/factory.py
Why Do We Need Factories?
Without Factory
# Manual object creation - hard-coded and inflexible
if config["dataset_name"] == "imagenet":
dataset = ImageNetDataset(path=config["path"])
elif config["dataset_name"] == "cifar":
dataset = CIFARDataset(path=config["path"])
# ... hundreds of if-elif statements
if config.get("normalize"):
dataset = normalize_dataset(dataset)
if config.get("augment"):
dataset = augment_dataset(dataset)
With Factory
# config.yaml
dataset:
task: classification
name: imagenet
path: /data
normalize: true
augment: true
# Automatic object creation from config
dataset = DATASETS.build_from_config(config["dataset"])
# Factory handles everything: creation, normalization, augmentation!
How Factories Work
The BuilderKey System
Factories use (task, name) tuples as keys:
# Key format: (task, name)
key = ("classification", "resnet") # Specific model
key = (None, "resnet") # Any task with name "resnet" (wildcard)
key = ("classification", None) # Any name for classification (wildcard)
None acts as a wildcard and matches anything.
Factory Components
┌─────────────────────────────────────────┐
│ Factory │
│ │
│ ┌────────────────────────────────┐ │
│ │ Registry (BuilderKeys) │ │
│ │ ("classification", "resnet") │ │
│ │ ("detection", "yolo") │ │
│ │ (None, "adam") │ │
│ └────────────────────────────────┘ │
│ │
│ ┌────────────────────────────────┐ │
│ │ Builders (from Loaders) │ │
│ │ - Model Loader │ │
│ │ - Dataset Loader │ │
│ └────────────────────────────────┘ │
│ │
│ ┌────────────────────────────────┐ │
│ │ Decorators │ │
│ │ - normalize() │ │
│ │ - augment() │ │
│ └────────────────────────────────┘ │
└─────────────────────────────────────────┘
Creating and Using Factories
Basic Factory Creation
from ainxt.factory import Factory
# Create empty factory
factory = Factory()
# Register constructors manually
factory.register(task="classification", name="resnet", constructor=ResNetModel)
factory.register(task=None, name="default", constructor=DefaultModel)
# Build objects
model = factory.build(task="classification", name="resnet", num_classes=10)
Factory with Loader
from ainxt.factory import Factory, Loader
from ainxt.models import Model
# Create loader to find models automatically
loader = Loader(
template="myapp.models.{task}",
tasks=["classification", "detection"],
kind=Model
)
# Create factory with loader
MODELS = Factory(loader)
# All models from loader are now available!
model = MODELS.build(task="classification", name="resnet", num_classes=10)
Factory with Decorators
# Create decorator factory
decorators = Factory()
decorators.register(None, "normalize", normalize_function)
decorators.register(None, "augment", augment_function)
# Create main factory with decorators
DATASETS = Factory(dataset_loader, decorator=decorators)
# Build with automatic decoration
dataset = DATASETS.build(
task="classification",
name="imagenet",
path="/data",
normalize={"mean": [0.5, 0.5, 0.5]}, # Triggers normalize decorator!
augment=True # Triggers augment decorator!
)
Key Factory Methods
1. register()
Register a constructor with the factory:
factory.register(task="classification", name="resnet", constructor=ResNetModel)
factory.register(task=None, name="adam", constructor=AdamOptimizer)
2. build()
Build an object from task and name:
model = factory.build(
task="classification",
name="resnet",
num_classes=1000, # Constructor arguments
pretrained=True
)
3. build_from_config()
Build directly from a configuration dictionary:
config = {
"task": "classification",
"name": "resnet",
"num_classes": 1000,
"pretrained": True
}
model = factory.build_from_config(config)
4. register_builder()
Add another builder/loader to the factory:
5. register_decorator()
Add decorators to the factory:
Combining Factories
Factories can be combined using the + operator:
# Core aiNXT factories
from ainxt.serving import DATASETS as CORE_DATASETS
# Your custom factories
from myapp.serving import create_dataset_factory
MY_DATASETS = create_dataset_factory("myapp", tasks)
# Combine them!
ALL_DATASETS = CORE_DATASETS + MY_DATASETS
# Now has access to both core and custom datasets
dataset = ALL_DATASETS.build(task="classification", name="my_custom_dataset")
This is exactly how DigitalNXT.Vision extends aiNXT:
# DigitalNXT.Vision/vision/serving/singletons.py
from ainxt.serving import DATASETS as _DATASETS
from vision.constants import PACKAGE
# Extend core with vision-specific datasets
DATASETS = _DATASETS + create_dataset_factory(PACKAGE, Task)
Decorator System
How Decorators Work
When you build an object, the Factory checks if any kwargs match registered decorators:
# Decorators registered
decorators = Factory()
decorators.register(None, "normalize", normalize_fn)
decorators.register(None, "cache", add_cache_fn)
# Main factory with decorators
factory = Factory(loader, decorator=decorators)
# Build with decorator kwargs
dataset = factory.build(
task="classification",
name="imagenet",
path="/data",
normalize={"mean": [0.5]}, # Matches "normalize" decorator!
cache=100 # Matches "cache" decorator!
)
# Flow:
# 1. Create base dataset: ImageNetDataset(path="/data")
# 2. Apply normalize: normalize_fn(dataset, mean=[0.5])
# 3. Apply cache: add_cache_fn(dataset, 100)
# 4. Return decorated dataset
Decorator Example
# Define decorator function
def add_caching(dataset, cache_size):
"""Add caching to dataset."""
dataset._cache = {}
dataset._cache_size = cache_size
print(f"Added cache of size {cache_size}")
return dataset
# Register decorator
DECORATORS = Factory()
DECORATORS.register(None, "cache", add_caching)
# Use in factory
DATASETS = Factory(dataset_loader, decorator=DECORATORS)
# Trigger via config
dataset = DATASETS.build(
task="classification",
name="imagenet",
path="/data",
cache=1000 # Triggers decorator!
)
Wildcard Matching and Resolution
Wildcards with None
# Register with wildcards
factory.register(None, "default", DefaultModel) # Matches ANY task
factory.register("classification", None, GenericClassifier) # Matches ANY name
# Query with wildcards
model = factory.build(task="any_task", name="default") # Finds (None, "default")
model = factory.build(task="classification", name="anything") # Finds ("classification", None)
Resolution Priority
When multiple keys match, Factory prioritizes more specific matches:
factory.register(None, "resnet", GenericResNet) # Less specific
factory.register("classification", "resnet", ClassificationResNet) # More specific
# Resolves to ClassificationResNet (more specific)
model = factory.build(task="classification", name="resnet")
Real-World Usage
Example 1: Standard Setup (ainxt)
# ainxt/serving/singletons.py
from ainxt.serving import create_dataset_factory, create_model_factory
DATASETS = create_dataset_factory(package="ainxt", tasks=["classification", "embedding"])
MODELS = create_model_factory(package="ainxt", tasks=["classification", "embedding"])
# Usage
dataset = DATASETS.build(task="classification", name="csv_dataset", path="/data.csv")
model = MODELS.build(task="classification", name="sklearn_classifier", model_type="random_forest")
Example 2: Extended Setup (DigitalNXT.Vision)
# DigitalNXT.Vision/vision/serving/singletons.py
from ainxt.serving import (
DATASETS as _DATASETS,
MODELS as _MODELS,
create_dataset_factory,
create_model_factory
)
from vision.constants import PACKAGE
from vision.task import Task
# Extend core factories with vision-specific components
DATASETS = _DATASETS + create_dataset_factory(PACKAGE, Task)
MODELS = _MODELS + create_model_factory(PACKAGE, Task)
# Register additional decorators
from vision.parsers.augmenter import DATASET_DECORATORS
DATASETS.register_decorator(DATASET_DECORATORS)
# Usage - has access to both core and vision components
dataset = DATASETS.build(
task="classification",
name="imagenet", # Vision-specific
augment={"flip": True} # Decorator
)
Helper Functions
aiNXT provides helper functions that create Factories with Loaders automatically:
create_dataset_factory()
Source: ainxt/serving/singletons.py:26-57
from ainxt.serving import create_dataset_factory
DATASETS = create_dataset_factory(
package="myapp",
tasks=["classification", "detection"]
)
# Equivalent to:
# dataset_loader = Loader(template="myapp.data.datasets.{task}", tasks=tasks, kind=Dataset)
# decorator_loader = Loader(template="myapp.data.datasets.{task}.decorators", tasks=tasks)
# DATASETS = Factory(dataset_loader, decorator=decorator_loader)
create_model_factory()
Source: ainxt/serving/singletons.py:60-91
from ainxt.serving import create_model_factory
MODELS = create_model_factory(
package="myapp",
tasks=["classification", "detection"]
)
Best Practices
1. Use Helper Functions
# GOOD - concise and standard
from ainxt.serving import create_dataset_factory
DATASETS = create_dataset_factory("myapp", tasks)
# VERBOSE - manual setup
loader = Loader(...)
decorators = Loader(...)
DATASETS = Factory(loader, decorator=decorators)
2. Extend, Don't Replace
# GOOD - extend core factories
from ainxt.serving import DATASETS as CORE_DATASETS
DATASETS = CORE_DATASETS + MY_DATASETS
# BAD - lose core functionality
DATASETS = MY_DATASETS # No access to core datasets!
3. Register Decorators Properly
# GOOD - decorators registered with factory
DATASETS.register_decorator(DECORATORS)
# BAD - decorators separate, won't work
DECORATORS = Factory() # Not connected to DATASETS!
4. Consistent Naming
# GOOD - clear task/name pairs
factory.register("classification", "resnet50", ResNet50)
factory.register("detection", "yolo_v5", YOLOv5)
# BAD - inconsistent or unclear
factory.register("cls", "r50", ResNet50)
factory.register(None, None, SomeModel) # Too generic!
Troubleshooting
Issue 1: "No constructor found"
Problem: KeyError: No constructor found for (task, name)
Solutions:
# 1. Check what's registered
print(list(factory.keys())) # See all registered keys
# 2. Check task/name spelling
factory.build(task="classificaiton", name="resnet") # Typo!
# 3. Try with None (wildcard)
factory.build(task=None, name="resnet") # More permissive
Issue 2: Decorators Not Applied
Problem: Decorator kwargs ignored
Solutions:
# 1. Check decorator is registered
print(list(decorators.keys()))
# 2. Check kwargs match decorator names exactly
factory.build(..., normalize=True) # Must match registered name "normalize"
# 3. Ensure decorators added to factory
factory.register_decorator(decorators)
Issue 3: Wrong Constructor Used
Problem: Factory picks unexpected constructor
Solutions:
# 1. Use more specific task/name
factory.build(task="classification", name="resnet50") # Not just "resnet"
# 2. Check registration priority
# Later registrations override earlier ones with same key
# 3. Use factory.search() to see matches
matches = list(factory.search(task="classification", name="resnet"))
print(matches)
Summary
- Factory is the central registry for object creation
- Combines Loaders, Parsers, and Decorators
- BuilderKeys use
(task, name)tuples with wildcard support - Decorators modify objects based on kwargs
- Factories combine using the
+operator - Helper functions simplify common patterns
Factories provide a clean, configuration-driven interface for creating complex objects while maintaining flexibility and extensibility.
See Also
- Loaders - Automatic constructor discovery
- Parsers - Configuration transformation
- Context - Factory orchestration
- Dataset Decorators - Special decorator system