Refactor/improve architecture (Pipelex#144)

thomashebrard · lchoquel · web-flow · commit c43ad687a75b · 2025-07-09T11:54:42.000+02:00
* Better singleton

* Fixed major bug in identifying structure of implicit concept

---------

Co-authored-by: Louis Choquel &lt;louis@pipelex.com&gt;
diff --git a/pipelex/core/concept.py b/pipelex/core/concept.py
@@ -106,7 +106,7 @@ def extract_domain_and_concept_from_str(cls, concept_str: str) -> Tuple[str, str
         if "." in concept_str:
             domain_code, concept_code = concept_str.split(".")
             return domain_code, concept_code
-        raise ConceptError(f"No extraction of domain and concept from concept code '{concept_str}'")
+        raise ConceptError(f"Could not extract domain and concept from concept code '{concept_str}'")
 
     @classmethod
     def extract_concept_name_from_str(cls, concept_str: str) -> str:
diff --git a/pipelex/core/concept_factory.py b/pipelex/core/concept_factory.py
@@ -82,26 +82,33 @@ def make_from_details_dict(
         return the_concept
 
     @classmethod
-    def make_concept_from_definition(
+    def make_concept_from_definition_str(
         cls,
         domain_code: str,
-        code: str,
+        concept_str: str,
         definition: str,
     ) -> Concept:
         structure_class_name: str
-        if Concept.is_valid_structure_class(structure_class_name=code):
+        refines: List[str]
+        if Concept.concept_str_contains_domain(concept_str=concept_str):
+            concept_name = Concept.extract_concept_name_from_str(concept_str=concept_str)
+        else:
+            concept_name = concept_str
+        if Concept.is_valid_structure_class(structure_class_name=concept_name):
             # structure is set implicitly, by the concept's code
-            structure_class_name = code
+            structure_class_name = concept_name
+            refines = []
         else:
             structure_class_name = TextContent.__name__
+            refines = [NativeConcept.TEXT.code]
 
         try:
             the_concept = Concept(
-                code=ConceptCodeFactory.make_concept_code(domain_code, code),
+                code=ConceptCodeFactory.make_concept_code(domain_code, concept_name),
                 domain=domain_code,
                 definition=definition,
                 structure_class_name=structure_class_name,
-                refines=[NativeConcept.TEXT.code],
+                refines=refines,
             )
             return Concept.model_validate(the_concept)
         except ValidationError as e:
diff --git a/pipelex/core/concept_library.py b/pipelex/core/concept_library.py
@@ -106,13 +106,22 @@ def is_compatible(self, tested_concept: Concept, wanted_concept: Concept) -> boo
     @override
     def is_compatible_by_concept_code(self, tested_concept_code: str, wanted_concept_code: str) -> bool:
         if wanted_concept_code == NativeConcept.ANYTHING.code:
+            log.debug(
+                f"Concept '{tested_concept_code}' is compatible with '{wanted_concept_code}' "
+                f"because '{wanted_concept_code}' is '{NativeConcept.ANYTHING.code}'"
+            )
             return True
         tested_concept = self.get_required_concept(concept_code=tested_concept_code)
         wanted_concept = self.get_required_concept(concept_code=wanted_concept_code)
         if tested_concept.code == wanted_concept.code:
+            log.debug(f"Concept '{tested_concept_code}' is compatible with '{wanted_concept_code}' because they have the same code")
             return True
         for inherited_concept_code in tested_concept.refines:
             if self.is_compatible_by_concept_code(inherited_concept_code, wanted_concept_code):
+                log.debug(
+                    f"Concept '{tested_concept_code}' is compatible with '{wanted_concept_code}' "
+                    f"because '{tested_concept_code}' refines '{inherited_concept_code}' which is compatible with '{wanted_concept_code}'"
+                )
                 return True
         return False
 
@@ -134,9 +143,9 @@ def get_required_concept(self, concept_code: str) -> Concept:
             if self.is_concept_implicit(concept_code=concept_code):
                 # The implicit concept is obviously coming with a domain (the one it is used in)
                 # TODO: replace this with a concept factory method make_implicit_concept
-                return ConceptFactory.make_concept_from_definition(
+                return ConceptFactory.make_concept_from_definition_str(
                     domain_code="implicit",
-                    code=Concept.extract_domain_and_concept_from_str(concept_str=concept_code)[1],
+                    concept_str=Concept.extract_domain_and_concept_from_str(concept_str=concept_code)[1],
                     definition=concept_code,
                 )
             else:
diff --git a/pipelex/libraries/library_manager.py b/pipelex/libraries/library_manager.py
@@ -256,24 +256,29 @@ def _load_library_components_from_recursive_dict(
                 continue
 
     def _load_concepts(self, domain_code: str, obj_dict: Dict[str, Any]):
-        for concept_code, concept_obj in obj_dict.items():
+        for concept_str, concept_obj in obj_dict.items():
             if isinstance(concept_obj, str):
                 # we only have a definition
-                concept_from_def = ConceptFactory.make_concept_from_definition(domain_code=domain_code, code=concept_code, definition=concept_obj)
+                definition = concept_obj
+                concept_from_def = ConceptFactory.make_concept_from_definition_str(
+                    domain_code=domain_code,
+                    concept_str=concept_str,
+                    definition=definition,
+                )
                 self.concept_library.add_new_concept(concept=concept_from_def)
             elif isinstance(concept_obj, dict):
                 # blueprint dict definition
                 concept_obj_dict: Dict[str, Any] = concept_obj
                 try:
                     concept_from_dict = ConceptFactory.make_from_details_dict(
-                        domain_code=domain_code, code=concept_code, details_dict=concept_obj_dict
+                        domain_code=domain_code, code=concept_str, details_dict=concept_obj_dict
                     )
                 except ValidationError as exc:
                     error_msg = format_pydantic_validation_error(exc)
-                    raise ConceptLibraryError(f"Error loading concept '{concept_code}' because of: {error_msg}") from exc
+                    raise ConceptLibraryError(f"Error loading concept '{concept_str}' because of: {error_msg}") from exc
                 self.concept_library.add_new_concept(concept=concept_from_dict)
             else:
-                raise ConceptLibraryError(f"Unexpected type for concept_code '{concept_code}' in domain '{domain_code}': {type(concept_obj)}")
+                raise ConceptLibraryError(f"Unexpected type for concept_code '{concept_str}' in domain '{domain_code}': {type(concept_obj)}")
 
     def _load_pipes(self, domain_code: str, obj_dict: Dict[str, Any]):
         for pipe_code, pipe_obj in obj_dict.items():
diff --git a/pipelex/pipelex.py b/pipelex/pipelex.py
@@ -1,10 +1,11 @@
 from importlib.metadata import metadata
-from typing import Any, ClassVar, List, Optional, Type
+from typing import Any, List, Optional, Type, cast
 
 from dotenv import load_dotenv
 from kajson.class_registry import ClassRegistry
 from kajson.class_registry_abstract import ClassRegistryAbstract
 from kajson.kajson_manager import KajsonManager
+from kajson.singleton import MetaSingleton
 from pydantic import ValidationError
 from rich import print
 from typing_extensions import Self
@@ -55,39 +56,9 @@
 PACKAGE_VERSION = metadata(PACKAGE_NAME)["Version"]
 
 
-class Pipelex:
-    _pipelex_instance: ClassVar[Optional[Self]] = None
-
-    def __new__(
-        cls,
-        pipelex_cls: Optional[Type[Self]] = None,
-        pipelex_hub: Optional[PipelexHub] = None,
-        config_cls: Optional[Type[ConfigRoot]] = None,
-        ready_made_config: Optional[ConfigRoot] = None,
-        class_registry: Optional[ClassRegistryAbstract] = None,
-        template_provider: Optional[TemplateLibrary] = None,
-        llm_model_provider: Optional[LLMModelLibrary] = None,
-        inference_manager: Optional[InferenceManager] = None,
-        pipeline_manager: Optional[PipelineManager] = None,
-        pipeline_tracker: Optional[PipelineTracker] = None,
-        activity_manager: Optional[ActivityManagerProtocol] = None,
-        reporting_delegate: Optional[ReportingProtocol] = None,
-    ) -> Self:
-        if cls._pipelex_instance is not None:
-            raise RuntimeError(
-                "Pipelex is a singleton, it is instantiated only once. Its instance is private. All you need is accesible through the hub."
-            )
-        if pipelex_cls is None:
-            pipelex_cls = cls
-
-        if not issubclass(pipelex_cls, cls):
-            raise TypeError(f"{pipelex_cls!r} is not a subclass of {cls.__name__}")
-
-        return super().__new__(pipelex_cls)
-
+class Pipelex(metaclass=MetaSingleton):
     def __init__(
         self,
-        pipelex_cls: Optional[Type[Self]] = None,
         pipelex_hub: Optional[PipelexHub] = None,
         config_cls: Optional[Type[ConfigRoot]] = None,
         ready_made_config: Optional[ConfigRoot] = None,
@@ -153,7 +124,11 @@ def __init__(
         self.pipelex_hub.set_domain_provider(domain_provider=domain_library)
         self.pipelex_hub.set_concept_provider(concept_provider=concept_library)
         self.pipelex_hub.set_pipe_provider(pipe_provider=pipe_library)
-        self.library_manager = LibraryManager(domain_library=domain_library, concept_library=concept_library, pipe_library=pipe_library)
+        self.library_manager = LibraryManager(
+            domain_library=domain_library,
+            concept_library=concept_library,
+            pipe_library=pipe_library,
+        )
         self.library_manager.setup()
         self.pipelex_hub.set_library_manager(library_manager=self.library_manager)
 
@@ -178,7 +153,6 @@ def __init__(
             self.activity_manager = ActivityManagerNoOp()
         self.pipelex_hub.set_activity_manager(activity_manager=self.activity_manager)
 
-        Pipelex._pipelex_instance = self
         log.debug(f"{PACKAGE_NAME} version {PACKAGE_VERSION} init done")
 
     def setup(
@@ -257,10 +231,12 @@ def teardown(self):
         self.class_registry.teardown()
         func_registry.teardown()
 
-        Pipelex._pipelex_instance = None
         project_name = get_config().project_name
         log.debug(f"{PACKAGE_NAME} version {PACKAGE_VERSION} teardown done for {get_config().project_name} (except config & logs)")
         self.pipelex_hub.reset_config()
+        # Clear the singleton instance from metaclass
+        if self.__class__ in MetaSingleton.instances:
+            del MetaSingleton.instances[self.__class__]
         print(f"{PACKAGE_NAME} version {PACKAGE_VERSION} config reset done for {project_name}")
 
     # TODO: add kwargs to make() so that subclasses can employ specific parameters
@@ -274,10 +250,12 @@ def make(cls, structure_classes: Optional[List[Type[Any]]] = None) -> Self:
 
     @classmethod
     def get_optional_instance(cls) -> Optional[Self]:
-        return cls._pipelex_instance
+        instance = MetaSingleton.instances.get(cls)
+        return cast(Optional[Self], instance)
 
     @classmethod
     def get_instance(cls) -> Self:
-        if cls._pipelex_instance is None:
+        instance = MetaSingleton.instances.get(cls)
+        if instance is None:
             raise RuntimeError("Pipelex is not initialized")
-        return cls._pipelex_instance
+        return cast(Self, instance)
diff --git a/pyproject.toml b/pyproject.toml
@@ -24,7 +24,7 @@ dependencies = [
     "instructor>=1.8.3",
     "jinja2>=3.1.4",
     "json2html>=1.3.0",
-    "kajson==0.2.3",
+    "kajson @ git+https://github.com/Pipelex/kajson.git@refactor/Proper-Singleton",
     "markdown>=3.6",
     "networkx>=3.4.2",
     "openai>=1.60.1",
@@ -284,3 +284,6 @@ packages = ["pipelex"]
 
 [tool.hatch.build.targets.sdist.force-include]
 "pyproject.toml" = "pipelex/pyproject.toml"
+
+[tool.hatch.metadata]
+allow-direct-references = true
diff --git a/uv.lock b/uv.lock
