dev to release/v0.5.1 (Pipelex#146)

* Refactor/improve architecture (Pipelex#144)

* Better singleton

* Fixed major bug in identifying structure of implicit concept

---------

Co-authored-by: Louis Choquel <louis@pipelex.com>

* Fix/ut concept library (Pipelex#145)

* add changelog v0.5.1

---------

Co-authored-by: Louis Choquel <louis@pipelex.com>

Author: thomashebrard

CHANGELOG.md

@@ -1,5 +1,17 @@
 # Changelog
 
+## [v0.5.1] - 2025-07-09
+
+## Fixed
+
+- Fixed the `ConceptFactory.make_from_blueprint` method: Concepts defined in single-line format no longer automatically refine `TextContent` when a structure class with the same name exists
+- `ConceptFactory.make_concept_from_definition` is now `ConceptFactory.make_concept_from_definition_str`
+
+## Added
+
+- Bumped `kajson` to `v0.3.0`: Introducing `MetaSingleton` for better singleton management
+- Unit tests for `ConceptLibrary.is_compatible_by_concept_code`
+
 ## [v0.5.0] - 2025-07-01
 
 ### Highlight: Vibe Coding an AI workflow becomes a reality

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:

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:

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:

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():

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)

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "pipelex"
-version = "0.5.0"
+version = "0.5.1"
 description = "Pipelex is an open-source dev tool based on a simple declarative language that lets you define replicable, structured, composable LLM pipelines."
 authors = [{ name = "Evotis S.A.S.", email = "evotis@pipelex.com" }]
 maintainers = [{ name = "Pipelex staff", email = "oss@pipelex.com" }]
@@ -24,7 +24,7 @@ dependencies = [
     "instructor>=1.8.3",
     "jinja2>=3.1.4",
     "json2html>=1.3.0",
-    "kajson==0.2.3",
+    "kajson==0.3.0",
     "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

tests/conftest.py

@@ -9,6 +9,8 @@
 import pipelex.pipelex
 from pipelex import log
 from pipelex.config import get_config
+from pipelex.core.concept_provider_abstract import ConceptProviderAbstract
+from pipelex.hub import get_concept_provider
 from tests.cases.registry import Fruit
 
 pytest_plugins = [
@@ -66,3 +68,9 @@ def cherry() -> Fruit:
 def blueberry() -> Fruit:
     """Blueberry fruit fixture."""
     return Fruit(name="blueberry", color="blue")
+
+
+@pytest.fixture(scope="module")
+def concept_provider() -> ConceptProviderAbstract:
+    """Concept provider fixture for testing concept compatibility."""
+    return get_concept_provider()

tests/test_pipelines/concept_library/is_compatible_concept_code.py

@@ -0,0 +1,45 @@
+from typing import Dict, List, Optional
+
+from pydantic import Field
+
+from pipelex.core.stuff_content import StructuredContent
+
+
+class FundamentalsDoc(StructuredContent):
+    project_overview: Optional[str] = Field(
+        None,
+        description="Mission, key features, architecture diagram, demo links",
+    )
+    core_concepts: Optional[Dict[str, str]] = Field(
+        None,
+        description=(
+            "Names and definitions for project-specific terms, acronyms, data model names, background knowledge, business rules, domain entities"
+        ),
+    )
+    repository_map: Optional[str] = Field(
+        None,
+        description="Directory layout explanation and purpose of each folder",
+    )
+
+
+class DocumentationConcept(StructuredContent):
+    """A specialized documentation concept that extends FundamentalsDoc."""
+
+    title: str = Field(..., description="Title of the documentation")
+    sections: List[str] = Field(default_factory=list, description="List of section names")
+    last_updated: Optional[str] = Field(None, description="Last update timestamp")
+
+
+class MultiMediaConcept(StructuredContent):
+    """A concept that combines text and images."""
+
+    text_content: str = Field(..., description="The text content")
+    image_urls: List[str] = Field(default_factory=list, description="List of image URLs")
+    caption: Optional[str] = Field(None, description="Optional caption for the content")
+
+
+class IndependentConcept(StructuredContent):
+    """An independent concept with its own structure."""
+
+    unique_field: str = Field(..., description="A unique field for this concept")
+    metadata: Dict[str, str] = Field(default_factory=dict, description="Metadata dictionary")

tests/test_pipelines/concept_library/is_compatible_concept_code.toml

@@ -0,0 +1,46 @@
+domain = "concept_library_tests"
+
+[concept]
+# Simple concept with no structure - should default to Text
+SimpleTextConcept = "A simple concept that should default to Text"
+
+# Concept with explicit structure class
+FundamentalsDoc = "A comprehensive overview of the fundamental concepts and principles of software engineering."
+
+# Concept that explicitly refines Text
+[concept.ExplicitTextConcept]
+Concept = "A concept that explicitly refines Text"
+refines = ["Text"]
+
+# Concept that refines Image
+[concept.ImageBasedConcept]
+Concept = "A concept based on images"
+refines = ["Image"]
+
+# Concept that refines FundamentalsDoc
+[concept.DocumentationConcept]
+Concept = "A specialized documentation concept"
+structure = "DocumentationConcept"
+refines = ["FundamentalsDoc"]
+
+# Concept that refines both Text and Image (multiple inheritance)
+[concept.MultiMediaConcept]
+Concept = "A concept that combines text and images"
+structure = "MultiMediaConcept"
+refines = ["Text", "Image"]
+
+# Concept with custom structure that doesn't refine anything
+[concept.IndependentConcept]
+Concept = "An independent concept with custom structure"
+structure = "IndependentConcept"
+
+# Concept that refines a non-native concept
+[concept.SpecializedDoc]
+Concept = "A specialized document that builds on FundamentalsDoc"
+refines = ["FundamentalsDoc"]
+
+# Chain of inheritance: Text -> ExplicitTextConcept -> DerivedTextConcept
+[concept.DerivedTextConcept]
+Concept = "A concept derived from ExplicitTextConcept"
+refines = ["ExplicitTextConcept"]
+