Repository: Cinnamon/kotaemon
File: libs/ktem/ktem/tags/pipelines.py (Python)

Code before:
from .crud import ChunkTagIndexCRUD, TagCRUD


class MetaIndexPipeline(IndexDocumentPipeline):
    """Meta index pipeline for tagging documents.
    We reuse the logic from IndexDocumentPipeline and add custom
    tagging procedure with .stream() method.
    """

    llm: BaseLLM
    tags: list[str]
    DEFAULT_CHUNK_SIZE = 20

    @property
    def chunk_tag_index_crud(self) -> ChunkTagIndexCRUD:
        return ChunkTagIndexCRUD(engine)

    @property
    def tag_crud(self) -> TagCRUD:
        return TagCRUD(engine)

    @classmethod
    def resolve_tag_names(cls, tag_str: str) -> list[str]:
        tag_str = tag_str.strip()
        tags = []
        if tag_str != "":
            if "," in tag_str:
                tags = [tag.strip() for tag in tag_str.split(",")]
            else:
                tags = [tag_str]

        return tags

    @classmethod
    def get_pipeline(
        cls, user_settings: dict, index_settings: dict[str, str]
    ) -> BaseFileIndexIndexing:
        """Get custom settings (tag_ids) from index settings"""
        llm = llms.get(index_settings.get("llm", llms.get_default_name()), None)
        tags = index_settings.get("tags", "")
        obj = super().get_pipeline(user_settings, index_settings)

        # assign custom settings for the pipeline
        obj.llm = llm
        obj.tags = cls.resolve_tag_names(tags)
        return obj

    def route(self, file_path: Path) -> IndexPipeline:
        """Decide the pipeline based on the file type
        Can subclass this method for a more elaborate pipeline routing strategy.
        """

Review comment (bug):
we could fail fast in this case
```python
if tag_str.strip() == '':
    return []
return list(map(str.strip, tag_str.split(',')))
```

Code after:
from .crud import ChunkTagIndexCRUD, TagCRUD


class MetaIndexPipeline(IndexDocumentPipeline):
    """Meta index pipeline for tagging documents.
    We reuse the logic from IndexDocumentPipeline and add custom
    tagging procedure with .stream() method.
    """

    llm: BaseLLM
    tags: list[str]
    DEFAULT_CHUNK_SIZE = 20

    @property
    def chunk_tag_index_crud(self) -> ChunkTagIndexCRUD:
        return ChunkTagIndexCRUD(engine)

    @property
    def tag_crud(self) -> TagCRUD:
        return TagCRUD(engine)

    @classmethod
    def resolve_tag_names(cls, tag_str: str) -> list[str]:
        tag_str = tag_str.strip()
        tags = []
        if tag_str.strip() == "":
            return []

        if "," in tag_str:
            tags = [tag.strip() for tag in tag_str.split(",")]
        else:
            tags = [tag_str]

        return tags

    @classmethod
    def get_pipeline(
        cls, user_settings: dict, index_settings: dict[str, str]
    ) -> BaseFileIndexIndexing:
        """Get custom settings (tag_ids) from index settings"""
        llm = llms.get(index_settings.get("llm", llms.get_default_name()), None)
        tags = index_settings.get("tags", "")
        obj = super().get_pipeline(user_settings, index_settings)

        # assign custom settings for the pipeline
        obj.llm = llm
        obj.tags = cls.resolve_tag_names(tags)
        return obj

    def route(self, file_path: Path) -> IndexPipeline:
        """Decide the pipeline based on the file type