diff --git a/semantica/core/__init__.py b/semantica/core/__init__.py index ba2e3c0c..cb3799d7 100644 --- a/semantica/core/__init__.py +++ b/semantica/core/__init__.py @@ -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", -] \ No newline at end of file + + # 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"]} + ) \ No newline at end of file diff --git a/semantica/core/config_manager.py b/semantica/core/config_manager.py index 66f4580d..4fd4f6c1 100644 --- a/semantica/core/config_manager.py +++ b/semantica/core/config_manager.py @@ -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)) diff --git a/semantica/core/methods.py b/semantica/core/methods.py new file mode 100644 index 00000000..e66abd77 --- /dev/null +++ b/semantica/core/methods.py @@ -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) + diff --git a/semantica/core/registry.py b/semantica/core/registry.py new file mode 100644 index 00000000..5ecc422b --- /dev/null +++ b/semantica/core/registry.py @@ -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() +