mirror of
https://github.com/semantica-agi/semantica.git
synced 2026-08-30 04:40:16 +00:00
Update core module structure to match semantic_extract and split patterns
- Add registry.py for method registration system - Add methods.py with reusable orchestration functions - Enhance __init__.py with comprehensive exports and build() convenience function - Update config_manager.py with enhanced documentation - Follow same structure pattern as semantic_extract and split modules
This commit is contained in:
KaifAhmad1
+130
-9
@@ -1,9 +1,25 @@
|
||||
"""
|
||||
Core Orchestration Module
|
||||
|
||||
This module provides the main orchestration capabilities for the Semantica framework,
|
||||
including the primary Semantica class, configuration management, lifecycle management,
|
||||
and plugin system.
|
||||
This module provides comprehensive orchestration capabilities for the Semantica framework,
|
||||
enabling framework initialization, knowledge base construction, pipeline execution, configuration
|
||||
management, lifecycle management, and plugin system integration.
|
||||
|
||||
Key Features:
|
||||
- Framework initialization and lifecycle management
|
||||
- Knowledge base construction from various data sources
|
||||
- Pipeline execution and resource management
|
||||
- Configuration loading, validation, and management
|
||||
- Plugin discovery, loading, and lifecycle management
|
||||
- System health monitoring and status tracking
|
||||
- Method registry for extensible orchestration methods
|
||||
|
||||
Algorithms Used:
|
||||
- Configuration Management: YAML/JSON parsing, environment variable resolution, validation
|
||||
- Lifecycle Management: Priority-based hook execution, state machine transitions
|
||||
- Plugin Management: Dynamic module loading, dependency resolution, version management
|
||||
- Resource Management: Dynamic allocation, cleanup, graceful shutdown
|
||||
- Health Monitoring: Component health checks, status aggregation, error tracking
|
||||
|
||||
Main Components:
|
||||
- Semantica: Main framework class for orchestration and knowledge base building
|
||||
@@ -11,36 +27,141 @@ Main Components:
|
||||
- ConfigManager: Configuration loading, validation, and management
|
||||
- LifecycleManager: System lifecycle management with hooks and health monitoring
|
||||
- PluginRegistry: Dynamic plugin discovery, loading, and management
|
||||
- MethodRegistry: Registry for custom orchestration methods
|
||||
- Orchestration Methods: Reusable functions for common orchestration tasks
|
||||
|
||||
Example Usage:
|
||||
>>> from semantica.core import Semantica, ConfigManager
|
||||
>>> # Initialize framework
|
||||
>>> from semantica.core import Semantica, build
|
||||
>>> # Using main class
|
||||
>>> framework = Semantica()
|
||||
>>> # Load configuration
|
||||
>>> config_manager = ConfigManager()
|
||||
>>> config = config_manager.load_from_file("config.yaml")
|
||||
>>> framework.initialize()
|
||||
>>> result = framework.build_knowledge_base(sources=["doc1.pdf"])
|
||||
>>>
|
||||
>>> # Using convenience function
|
||||
>>> from semantica.core import build
|
||||
>>> result = build(sources=["doc1.pdf", "doc2.docx"], embeddings=True, graph=True)
|
||||
>>>
|
||||
>>> # Using methods directly
|
||||
>>> from semantica.core.methods import build_knowledge_base
|
||||
>>> result = build_knowledge_base(sources=["doc.pdf"], method="default")
|
||||
|
||||
Author: Semantica Contributors
|
||||
License: MIT
|
||||
"""
|
||||
|
||||
from typing import Any, Dict, List, Optional, Union
|
||||
from pathlib import Path
|
||||
|
||||
from .orchestrator import Semantica
|
||||
from .config_manager import Config, ConfigManager
|
||||
from .lifecycle import LifecycleManager, SystemState, HealthStatus
|
||||
from .plugin_registry import PluginRegistry, PluginInfo, LoadedPlugin
|
||||
from .registry import MethodRegistry, method_registry
|
||||
from .methods import (
|
||||
build_knowledge_base,
|
||||
run_pipeline,
|
||||
initialize_framework,
|
||||
get_status,
|
||||
get_orchestration_method,
|
||||
list_available_methods
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
# Main orchestrator
|
||||
"Semantica",
|
||||
|
||||
# Configuration
|
||||
"Config",
|
||||
"ConfigManager",
|
||||
|
||||
# Lifecycle
|
||||
"LifecycleManager",
|
||||
"SystemState",
|
||||
"HealthStatus",
|
||||
|
||||
# Plugins
|
||||
"PluginRegistry",
|
||||
"PluginInfo",
|
||||
"LoadedPlugin",
|
||||
]
|
||||
|
||||
# Registry
|
||||
"MethodRegistry",
|
||||
"method_registry",
|
||||
|
||||
# Methods
|
||||
"build_knowledge_base",
|
||||
"run_pipeline",
|
||||
"initialize_framework",
|
||||
"get_status",
|
||||
"get_orchestration_method",
|
||||
"list_available_methods",
|
||||
|
||||
# Convenience
|
||||
"build",
|
||||
]
|
||||
|
||||
|
||||
def build(
|
||||
sources: Union[str, List[Union[str, Path]]],
|
||||
extract_entities: bool = True,
|
||||
extract_relations: bool = True,
|
||||
embeddings: bool = True,
|
||||
graph: bool = True,
|
||||
**options
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Build knowledge base from sources (module-level convenience function).
|
||||
|
||||
This is a user-friendly wrapper that performs comprehensive knowledge base
|
||||
construction including entity extraction, relation extraction, embeddings,
|
||||
and knowledge graph building.
|
||||
|
||||
Args:
|
||||
sources: Input source or list of sources (files, URLs, streams)
|
||||
extract_entities: Whether to extract named entities (default: True)
|
||||
extract_relations: Whether to extract relationships (default: True)
|
||||
embeddings: Whether to generate embeddings (default: True)
|
||||
graph: Whether to build knowledge graph (default: True)
|
||||
**options: Additional processing options
|
||||
|
||||
Returns:
|
||||
Dictionary containing:
|
||||
- knowledge_graph: Knowledge graph data
|
||||
- embeddings: Embedding vectors
|
||||
- results: Processing results
|
||||
- statistics: Processing statistics
|
||||
- metadata: Processing metadata
|
||||
|
||||
Examples:
|
||||
>>> import semantica
|
||||
>>> result = semantica.core.build(
|
||||
... sources=["doc1.pdf", "doc2.docx"],
|
||||
... extract_entities=True,
|
||||
... extract_relations=True,
|
||||
... embeddings=True,
|
||||
... graph=True
|
||||
... )
|
||||
>>> print(f"Processed {result['statistics']['sources_processed']} sources")
|
||||
"""
|
||||
# Normalize sources to list
|
||||
if isinstance(sources, str):
|
||||
sources = [sources]
|
||||
|
||||
# Build pipeline configuration from options
|
||||
pipeline_config = options.get("pipeline", {})
|
||||
if extract_entities or extract_relations:
|
||||
pipeline_config.setdefault("extract", {})
|
||||
if extract_entities:
|
||||
pipeline_config["extract"]["entities"] = True
|
||||
if extract_relations:
|
||||
pipeline_config["extract"]["relations"] = True
|
||||
|
||||
# Build knowledge base
|
||||
return build_knowledge_base(
|
||||
sources=sources,
|
||||
method=options.get("method", "default"),
|
||||
embeddings=embeddings,
|
||||
graph=graph,
|
||||
pipeline=pipeline_config,
|
||||
**{k: v for k, v in options.items() if k not in ["pipeline", "method"]}
|
||||
)
|
||||
@@ -3,6 +3,23 @@ Configuration Management Module
|
||||
|
||||
This module provides comprehensive configuration management for the Semantica framework,
|
||||
including loading from files, environment variables, validation, and dynamic updates.
|
||||
It supports multiple configuration sources and formats with automatic fallback chains
|
||||
and validation.
|
||||
|
||||
Supported Configuration Sources:
|
||||
- Configuration files: YAML, JSON formats
|
||||
- Environment variables: SEMANTICA_ prefix for automatic loading
|
||||
- Programmatic: Python API for setting configuration values
|
||||
- Dictionary: Direct dictionary-based configuration
|
||||
|
||||
Algorithms Used:
|
||||
- YAML Parsing: YAML parser for configuration file loading
|
||||
- JSON Parsing: JSON parser for configuration file loading
|
||||
- Environment Variable Parsing: OS-level environment variable access with prefix matching
|
||||
- Fallback Chain: Priority-based configuration resolution (file -> env -> defaults)
|
||||
- Dictionary Merging: Deep merge algorithms for configuration updates
|
||||
- Validation: Type checking, range validation, required field checking
|
||||
- Nested Access: Dot notation parsing for nested configuration access
|
||||
|
||||
Key Features:
|
||||
- YAML/JSON configuration file parsing
|
||||
@@ -11,12 +28,24 @@ Key Features:
|
||||
- Dynamic configuration updates at runtime
|
||||
- Configuration inheritance and merging
|
||||
- Nested configuration access via dot notation
|
||||
- Automatic type conversion for environment variables
|
||||
- Progress tracking for configuration loading operations
|
||||
|
||||
Main Classes:
|
||||
- Config: Configuration data class with validation
|
||||
- ConfigManager: Configuration loading, validation, and management
|
||||
|
||||
Example Usage:
|
||||
>>> from semantica.core import ConfigManager
|
||||
>>> manager = ConfigManager()
|
||||
>>> config = manager.load_from_file("config.yaml")
|
||||
>>> batch_size = config.get("processing.batch_size", default=32)
|
||||
>>>
|
||||
>>> # Merge multiple configurations
|
||||
>>> merged = manager.merge_configs(config1, config2, config3)
|
||||
>>>
|
||||
>>> # Load from dictionary
|
||||
>>> config = manager.load_from_dict({"processing": {"batch_size": 64}})
|
||||
|
||||
Author: Semantica Contributors
|
||||
License: MIT
|
||||
@@ -369,41 +398,47 @@ class ConfigManager:
|
||||
file_path = Path(file_path)
|
||||
|
||||
if not file_path.exists():
|
||||
raise ConfigurationError(
|
||||
f"Configuration file not found: {file_path}",
|
||||
config_context={"file_path": str(file_path)}
|
||||
)
|
||||
|
||||
# Detect format from extension
|
||||
suffix = file_path.suffix.lower()
|
||||
|
||||
try:
|
||||
if suffix in (".yaml", ".yml"):
|
||||
with open(file_path, "r", encoding="utf-8") as f:
|
||||
config_dict = yaml.safe_load(f)
|
||||
|
||||
elif suffix == ".json":
|
||||
config_dict = read_json_file(file_path)
|
||||
else:
|
||||
raise ConfigurationError(
|
||||
f"Unsupported configuration file format: {suffix}. "
|
||||
"Supported formats: .yaml, .yml, .json"
|
||||
f"Configuration file not found: {file_path}",
|
||||
config_context={"file_path": str(file_path)}
|
||||
)
|
||||
|
||||
# Create config object from loaded dictionary
|
||||
config = Config(config_dict=config_dict)
|
||||
# Detect format from extension
|
||||
suffix = file_path.suffix.lower()
|
||||
|
||||
# Validate configuration if requested
|
||||
if validate:
|
||||
config.validate()
|
||||
|
||||
# Store config and file path for potential reload
|
||||
self._config = config
|
||||
self._last_file_path = file_path
|
||||
|
||||
self.progress_tracker.stop_tracking(tracking_id, status="completed",
|
||||
message="Configuration loaded successfully")
|
||||
return config
|
||||
try:
|
||||
if suffix in (".yaml", ".yml"):
|
||||
with open(file_path, "r", encoding="utf-8") as f:
|
||||
config_dict = yaml.safe_load(f)
|
||||
|
||||
elif suffix == ".json":
|
||||
config_dict = read_json_file(file_path)
|
||||
else:
|
||||
raise ConfigurationError(
|
||||
f"Unsupported configuration file format: {suffix}. "
|
||||
"Supported formats: .yaml, .yml, .json"
|
||||
)
|
||||
|
||||
# Create config object from loaded dictionary
|
||||
config = Config(config_dict=config_dict)
|
||||
|
||||
# Validate configuration if requested
|
||||
if validate:
|
||||
config.validate()
|
||||
|
||||
# Store config and file path for potential reload
|
||||
self._config = config
|
||||
self._last_file_path = file_path
|
||||
|
||||
self.progress_tracker.stop_tracking(tracking_id, status="completed",
|
||||
message="Configuration loaded successfully")
|
||||
return config
|
||||
except Exception as e:
|
||||
# Re-raise as ConfigurationError if inner try fails
|
||||
raise ConfigurationError(
|
||||
f"Failed to parse configuration file: {str(e)}",
|
||||
config_context={"file_path": str(file_path)}
|
||||
) from e
|
||||
|
||||
except Exception as e:
|
||||
self.progress_tracker.stop_tracking(tracking_id, status="failed", message=str(e))
|
||||
|
||||
@@ -0,0 +1,415 @@
|
||||
"""
|
||||
Orchestration Methods Module
|
||||
|
||||
This module provides all orchestration methods as simple, reusable functions for
|
||||
framework initialization, knowledge base construction, pipeline execution, and
|
||||
system status management. It supports multiple orchestration approaches and
|
||||
integrates with the method registry for extensibility.
|
||||
|
||||
Supported Methods:
|
||||
|
||||
Knowledge Base Construction:
|
||||
- "default": Standard knowledge base construction with embeddings and graph
|
||||
- "minimal": Minimal knowledge base without embeddings or graph
|
||||
- "full": Full knowledge base with all features enabled
|
||||
|
||||
Pipeline Execution:
|
||||
- "default": Standard pipeline execution
|
||||
- "async": Asynchronous pipeline execution
|
||||
- "batch": Batch pipeline execution
|
||||
|
||||
Framework Initialization:
|
||||
- "default": Standard framework initialization
|
||||
- "minimal": Minimal initialization without plugins
|
||||
- "full": Full initialization with all components
|
||||
|
||||
System Status:
|
||||
- "default": Standard status retrieval
|
||||
- "detailed": Detailed status with component health
|
||||
- "summary": Summary status only
|
||||
|
||||
Algorithms Used:
|
||||
|
||||
Knowledge Base Construction:
|
||||
- Source Validation: File system and URL validation
|
||||
- Pipeline Orchestration: Multi-stage processing pipeline execution
|
||||
- Graph Construction: Entity-relationship graph building
|
||||
- Embedding Generation: Vector embedding creation from text
|
||||
|
||||
Pipeline Execution:
|
||||
- Resource Allocation: Dynamic resource management
|
||||
- Error Handling: Graceful error recovery and reporting
|
||||
- Progress Tracking: Real-time progress monitoring
|
||||
- Result Aggregation: Result collection and formatting
|
||||
|
||||
Framework Initialization:
|
||||
- Component Initialization: Ordered component startup
|
||||
- Configuration Validation: Config validation and loading
|
||||
- Plugin Loading: Dynamic plugin discovery and loading
|
||||
- Health Checking: Component health verification
|
||||
|
||||
System Status:
|
||||
- Health Aggregation: Component health status collection
|
||||
- State Tracking: System state monitoring
|
||||
- Module Status: Module availability checking
|
||||
- Plugin Status: Plugin loading status
|
||||
|
||||
Key Features:
|
||||
- Multiple orchestration methods for knowledge base construction
|
||||
- Multiple pipeline execution methods
|
||||
- Framework initialization methods
|
||||
- System status retrieval methods
|
||||
- Method dispatchers with registry support
|
||||
- Custom method registration capability
|
||||
- Consistent interface across all methods
|
||||
|
||||
Main Functions:
|
||||
- build_knowledge_base: Knowledge base construction wrapper
|
||||
- run_pipeline: Pipeline execution wrapper
|
||||
- initialize_framework: Framework initialization wrapper
|
||||
- get_status: System status retrieval wrapper
|
||||
- get_orchestration_method: Get orchestration method by name
|
||||
|
||||
Example Usage:
|
||||
>>> from semantica.core.methods import build_knowledge_base, get_orchestration_method
|
||||
>>> result = build_knowledge_base(sources=["doc1.pdf"], method="default")
|
||||
>>> method = get_orchestration_method("knowledge_base", "custom_method")
|
||||
|
||||
Author: Semantica Contributors
|
||||
License: MIT
|
||||
"""
|
||||
|
||||
from typing import Any, Dict, List, Optional, Union, Callable
|
||||
from pathlib import Path
|
||||
|
||||
from ..utils.logging import get_logger
|
||||
from ..utils.exceptions import ProcessingError, ConfigurationError
|
||||
from .orchestrator import Semantica
|
||||
from .config_manager import Config, ConfigManager
|
||||
from .registry import method_registry
|
||||
|
||||
logger = get_logger("core_methods")
|
||||
|
||||
|
||||
def build_knowledge_base(
|
||||
sources: Union[str, List[Union[str, Path]]],
|
||||
method: str = "default",
|
||||
config: Optional[Union[Config, Dict[str, Any]]] = None,
|
||||
**kwargs
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Build knowledge base from data sources (convenience function).
|
||||
|
||||
This is a user-friendly wrapper that constructs a knowledge base from
|
||||
various data sources using the specified method.
|
||||
|
||||
Args:
|
||||
sources: Single source or list of sources (files, URLs, streams)
|
||||
method: Knowledge base construction method (default: "default")
|
||||
config: Optional configuration object or dictionary
|
||||
**kwargs: Additional options:
|
||||
- embeddings: Whether to generate embeddings (default: True)
|
||||
- graph: Whether to build knowledge graph (default: True)
|
||||
- pipeline: Custom pipeline configuration
|
||||
- fail_fast: Whether to stop on first error (default: False)
|
||||
|
||||
Returns:
|
||||
Dictionary containing:
|
||||
- knowledge_graph: Knowledge graph data
|
||||
- embeddings: Embedding vectors
|
||||
- results: Processing results
|
||||
- statistics: Processing statistics
|
||||
- metadata: Processing metadata
|
||||
|
||||
Examples:
|
||||
>>> from semantica.core.methods import build_knowledge_base
|
||||
>>> result = build_knowledge_base(
|
||||
... sources=["doc1.pdf", "doc2.docx"],
|
||||
... method="default",
|
||||
... embeddings=True,
|
||||
... graph=True
|
||||
... )
|
||||
>>> print(f"Processed {result['statistics']['sources_processed']} sources")
|
||||
"""
|
||||
# Normalize sources to list
|
||||
if isinstance(sources, str):
|
||||
sources = [sources]
|
||||
|
||||
# Check for custom method in registry
|
||||
custom_method = method_registry.get("knowledge_base", method)
|
||||
if custom_method:
|
||||
return custom_method(sources, config=config, **kwargs)
|
||||
|
||||
# Use default Semantica framework
|
||||
framework = Semantica(config=config)
|
||||
framework.initialize()
|
||||
|
||||
try:
|
||||
# Map method to kwargs
|
||||
if method == "minimal":
|
||||
kwargs.setdefault("embeddings", False)
|
||||
kwargs.setdefault("graph", False)
|
||||
elif method == "full":
|
||||
kwargs.setdefault("embeddings", True)
|
||||
kwargs.setdefault("graph", True)
|
||||
else: # default
|
||||
kwargs.setdefault("embeddings", True)
|
||||
kwargs.setdefault("graph", True)
|
||||
|
||||
result = framework.build_knowledge_base(sources, **kwargs)
|
||||
return result
|
||||
finally:
|
||||
framework.shutdown(graceful=True)
|
||||
|
||||
|
||||
def run_pipeline(
|
||||
pipeline: Union[Dict[str, Any], Any],
|
||||
data: Any,
|
||||
method: str = "default",
|
||||
config: Optional[Union[Config, Dict[str, Any]]] = None,
|
||||
**kwargs
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Execute a processing pipeline (convenience function).
|
||||
|
||||
This is a user-friendly wrapper that executes a processing pipeline
|
||||
on input data using the specified method.
|
||||
|
||||
Args:
|
||||
pipeline: Pipeline object or configuration dictionary
|
||||
data: Input data for pipeline
|
||||
method: Pipeline execution method (default: "default")
|
||||
config: Optional configuration object or dictionary
|
||||
**kwargs: Additional pipeline options
|
||||
|
||||
Returns:
|
||||
Dictionary containing:
|
||||
- output: Pipeline output data
|
||||
- metadata: Processing metadata
|
||||
- metrics: Performance metrics
|
||||
|
||||
Examples:
|
||||
>>> from semantica.core.methods import run_pipeline
|
||||
>>> result = run_pipeline(
|
||||
... pipeline={"steps": ["extract", "transform"]},
|
||||
... data="sample text",
|
||||
... method="default"
|
||||
... )
|
||||
"""
|
||||
# Check for custom method in registry
|
||||
custom_method = method_registry.get("pipeline", method)
|
||||
if custom_method:
|
||||
return custom_method(pipeline, data, config=config, **kwargs)
|
||||
|
||||
# Use default Semantica framework
|
||||
framework = Semantica(config=config)
|
||||
framework.initialize()
|
||||
|
||||
try:
|
||||
result = framework.run_pipeline(pipeline, data, **kwargs)
|
||||
return result
|
||||
finally:
|
||||
framework.shutdown(graceful=True)
|
||||
|
||||
|
||||
def initialize_framework(
|
||||
config: Optional[Union[Config, Dict[str, Any]]] = None,
|
||||
method: str = "default",
|
||||
**kwargs
|
||||
) -> Semantica:
|
||||
"""
|
||||
Initialize Semantica framework (convenience function).
|
||||
|
||||
This is a user-friendly wrapper that initializes the framework
|
||||
using the specified method.
|
||||
|
||||
Args:
|
||||
config: Optional configuration object or dictionary
|
||||
method: Initialization method (default: "default")
|
||||
**kwargs: Additional initialization options
|
||||
|
||||
Returns:
|
||||
Initialized Semantica framework instance
|
||||
|
||||
Examples:
|
||||
>>> from semantica.core.methods import initialize_framework
|
||||
>>> framework = initialize_framework(
|
||||
... config={"llm_provider": {"name": "openai"}},
|
||||
... method="default"
|
||||
... )
|
||||
>>> status = framework.get_status()
|
||||
"""
|
||||
# Check for custom method in registry
|
||||
custom_method = method_registry.get("orchestration", method)
|
||||
if custom_method:
|
||||
return custom_method(config=config, **kwargs)
|
||||
|
||||
# Use default initialization
|
||||
framework = Semantica(config=config, **kwargs)
|
||||
|
||||
if method == "minimal":
|
||||
# Minimal initialization - just create instance, don't initialize
|
||||
pass
|
||||
elif method == "full":
|
||||
# Full initialization
|
||||
framework.initialize()
|
||||
else: # default
|
||||
framework.initialize()
|
||||
|
||||
return framework
|
||||
|
||||
|
||||
def get_status(
|
||||
framework: Optional[Semantica] = None,
|
||||
method: str = "default",
|
||||
**kwargs
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Get system status (convenience function).
|
||||
|
||||
This is a user-friendly wrapper that retrieves system status
|
||||
using the specified method.
|
||||
|
||||
Args:
|
||||
framework: Optional Semantica framework instance (creates new if None)
|
||||
method: Status retrieval method (default: "default")
|
||||
**kwargs: Additional options
|
||||
|
||||
Returns:
|
||||
Dictionary containing:
|
||||
- state: System state
|
||||
- health: Health summary
|
||||
- modules: Module status
|
||||
- plugins: Plugin status
|
||||
- config: Configuration status
|
||||
|
||||
Examples:
|
||||
>>> from semantica.core.methods import get_status
|
||||
>>> status = get_status(framework=my_framework, method="detailed")
|
||||
>>> print(f"System state: {status['state']}")
|
||||
"""
|
||||
# Check for custom method in registry
|
||||
custom_method = method_registry.get("orchestration", f"get_status_{method}")
|
||||
if custom_method:
|
||||
return custom_method(framework=framework, **kwargs)
|
||||
|
||||
# Use default status retrieval
|
||||
if framework is None:
|
||||
framework = Semantica()
|
||||
framework.initialize()
|
||||
should_shutdown = True
|
||||
else:
|
||||
should_shutdown = False
|
||||
|
||||
try:
|
||||
status = framework.get_status()
|
||||
|
||||
# Filter based on method
|
||||
if method == "summary":
|
||||
return {
|
||||
"state": status.get("state"),
|
||||
"health": {
|
||||
"is_healthy": status.get("health", {}).get("is_healthy"),
|
||||
"healthy_components": status.get("health", {}).get("healthy_components"),
|
||||
}
|
||||
}
|
||||
elif method == "detailed":
|
||||
return status
|
||||
else: # default
|
||||
return status
|
||||
finally:
|
||||
if should_shutdown:
|
||||
framework.shutdown(graceful=True)
|
||||
|
||||
|
||||
def get_orchestration_method(task: str, name: str) -> Optional[Callable]:
|
||||
"""
|
||||
Get orchestration method by task and name.
|
||||
|
||||
This function retrieves a registered orchestration method from the registry
|
||||
or returns a built-in method if available.
|
||||
|
||||
Args:
|
||||
task: Task type ("pipeline", "knowledge_base", "orchestration", "lifecycle")
|
||||
name: Method name
|
||||
|
||||
Returns:
|
||||
Method function or None if not found
|
||||
|
||||
Examples:
|
||||
>>> from semantica.core.methods import get_orchestration_method
|
||||
>>> method = get_orchestration_method("knowledge_base", "custom_method")
|
||||
>>> if method:
|
||||
... result = method(sources=["doc.pdf"])
|
||||
"""
|
||||
# First check registry
|
||||
method = method_registry.get(task, name)
|
||||
if method:
|
||||
return method
|
||||
|
||||
# Check built-in methods
|
||||
builtin_methods = {
|
||||
"knowledge_base": {
|
||||
"default": build_knowledge_base,
|
||||
"minimal": lambda sources, **kwargs: build_knowledge_base(sources, method="minimal", **kwargs),
|
||||
"full": lambda sources, **kwargs: build_knowledge_base(sources, method="full", **kwargs),
|
||||
},
|
||||
"pipeline": {
|
||||
"default": run_pipeline,
|
||||
},
|
||||
"orchestration": {
|
||||
"default": initialize_framework,
|
||||
"minimal": lambda config=None, **kwargs: initialize_framework(config=config, method="minimal", **kwargs),
|
||||
"full": lambda config=None, **kwargs: initialize_framework(config=config, method="full", **kwargs),
|
||||
},
|
||||
}
|
||||
|
||||
if task in builtin_methods and name in builtin_methods[task]:
|
||||
return builtin_methods[task][name]
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def list_available_methods(task: Optional[str] = None) -> Dict[str, List[str]]:
|
||||
"""
|
||||
List all available orchestration methods.
|
||||
|
||||
Args:
|
||||
task: Optional task type to filter by
|
||||
|
||||
Returns:
|
||||
Dictionary mapping task types to method names
|
||||
|
||||
Examples:
|
||||
>>> from semantica.core.methods import list_available_methods
|
||||
>>> all_methods = list_available_methods()
|
||||
>>> kb_methods = list_available_methods("knowledge_base")
|
||||
"""
|
||||
# Get registered methods
|
||||
registered = method_registry.list_all(task=task)
|
||||
|
||||
# Add built-in methods
|
||||
builtin_methods = {
|
||||
"knowledge_base": ["default", "minimal", "full"],
|
||||
"pipeline": ["default"],
|
||||
"orchestration": ["default", "minimal", "full"],
|
||||
"lifecycle": [],
|
||||
}
|
||||
|
||||
if task:
|
||||
# Merge for specific task
|
||||
result = {task: list(set(registered.get(task, []) + builtin_methods.get(task, [])))}
|
||||
else:
|
||||
# Merge for all tasks
|
||||
result = {}
|
||||
for t in set(list(registered.keys()) + list(builtin_methods.keys())):
|
||||
result[t] = list(set(registered.get(t, []) + builtin_methods.get(t, [])))
|
||||
|
||||
return result
|
||||
|
||||
|
||||
# Register default methods with registry
|
||||
method_registry.register("knowledge_base", "default", build_knowledge_base)
|
||||
method_registry.register("pipeline", "default", run_pipeline)
|
||||
method_registry.register("orchestration", "default", initialize_framework)
|
||||
|
||||
@@ -0,0 +1,129 @@
|
||||
"""
|
||||
Method Registry Module for Core Orchestration
|
||||
|
||||
This module provides a method registry system for registering custom orchestration methods,
|
||||
enabling extensibility and community contributions to the core orchestration toolkit.
|
||||
|
||||
Supported Registration Types:
|
||||
- Method Registry: Register custom orchestration methods for:
|
||||
* "pipeline": Pipeline execution methods
|
||||
* "knowledge_base": Knowledge base construction methods
|
||||
* "orchestration": General orchestration methods
|
||||
* "lifecycle": Lifecycle management methods
|
||||
|
||||
Algorithms Used:
|
||||
- Registry Pattern: Dictionary-based registration and lookup
|
||||
- Dynamic Registration: Runtime function registration
|
||||
- Type Checking: Type validation for registered components
|
||||
- Lookup Algorithms: Hash-based O(1) lookup for methods
|
||||
- Task-based Organization: Hierarchical organization by task type
|
||||
|
||||
Key Features:
|
||||
- Method registry for custom orchestration methods
|
||||
- Task-based method organization (pipeline, knowledge_base, orchestration, lifecycle)
|
||||
- Dynamic registration and unregistration
|
||||
- Easy discovery of available methods
|
||||
- Support for community-contributed extensions
|
||||
|
||||
Main Classes:
|
||||
- MethodRegistry: Registry for custom orchestration methods
|
||||
|
||||
Global Instances:
|
||||
- method_registry: Global method registry instance
|
||||
|
||||
Example Usage:
|
||||
>>> from semantica.core.registry import method_registry
|
||||
>>> method_registry.register("pipeline", "custom_method", custom_pipeline_function)
|
||||
>>> available = method_registry.list_all("pipeline")
|
||||
|
||||
Author: Semantica Contributors
|
||||
License: MIT
|
||||
"""
|
||||
|
||||
from typing import Dict, Callable, Any, List, Optional
|
||||
|
||||
|
||||
class MethodRegistry:
|
||||
"""Registry for custom orchestration methods."""
|
||||
|
||||
_methods: Dict[str, Dict[str, Callable]] = {
|
||||
"pipeline": {},
|
||||
"knowledge_base": {},
|
||||
"orchestration": {},
|
||||
"lifecycle": {},
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def register(cls, task: str, name: str, method_func: Callable):
|
||||
"""
|
||||
Register a custom orchestration method.
|
||||
|
||||
Args:
|
||||
task: Task type ("pipeline", "knowledge_base", "orchestration", "lifecycle")
|
||||
name: Method name
|
||||
method_func: Method function
|
||||
"""
|
||||
if task not in cls._methods:
|
||||
cls._methods[task] = {}
|
||||
cls._methods[task][name] = method_func
|
||||
|
||||
@classmethod
|
||||
def get(cls, task: str, name: str) -> Optional[Callable]:
|
||||
"""
|
||||
Get method by task and name.
|
||||
|
||||
Args:
|
||||
task: Task type ("pipeline", "knowledge_base", "orchestration", "lifecycle")
|
||||
name: Method name
|
||||
|
||||
Returns:
|
||||
Method function or None
|
||||
"""
|
||||
return cls._methods.get(task, {}).get(name)
|
||||
|
||||
@classmethod
|
||||
def list_all(cls, task: Optional[str] = None) -> Dict[str, List[str]]:
|
||||
"""
|
||||
List all registered methods.
|
||||
|
||||
Args:
|
||||
task: Optional task type to filter by
|
||||
|
||||
Returns:
|
||||
Dictionary mapping task types to method names
|
||||
"""
|
||||
if task:
|
||||
return {task: list(cls._methods.get(task, {}).keys())}
|
||||
return {t: list(m.keys()) for t, m in cls._methods.items()}
|
||||
|
||||
@classmethod
|
||||
def unregister(cls, task: str, name: str):
|
||||
"""
|
||||
Unregister a method.
|
||||
|
||||
Args:
|
||||
task: Task type ("pipeline", "knowledge_base", "orchestration", "lifecycle")
|
||||
name: Method name
|
||||
"""
|
||||
if task in cls._methods and name in cls._methods[task]:
|
||||
del cls._methods[task][name]
|
||||
|
||||
@classmethod
|
||||
def clear(cls, task: Optional[str] = None):
|
||||
"""
|
||||
Clear all registered methods for a task or all tasks.
|
||||
|
||||
Args:
|
||||
task: Optional task type to clear (clears all if None)
|
||||
"""
|
||||
if task:
|
||||
if task in cls._methods:
|
||||
cls._methods[task].clear()
|
||||
else:
|
||||
for task_dict in cls._methods.values():
|
||||
task_dict.clear()
|
||||
|
||||
|
||||
# Global registry
|
||||
method_registry = MethodRegistry()
|
||||
|
||||
Reference in New Issue
Block a user