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

Code before:
        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.
        """
        pipeline = super().route(file_path)

        # disable vectorstore for this kind of Index
        pipeline.VS = None
        # disable chunking for this kind of Index
        pipeline.splitter = None

        return pipeline

    def generate_with_llm(self, tag_prompt: str, doc_content: str) -> str:
        # TODO: add LLM logic to automatically tag document content
        return f"{type(self.llm)} {tag_prompt} - {doc_content}"

    def create_tag_content(
        self, tag_id: str, chunk_id: str, tag_prompt: str, doc_content: str

Review comment (suggestion):
if we have `IndexSettings`, things could go like
```python
llm = llms.get(index_settings.llm, None)
```

Code after:
        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
        Can subclass this method for a more elaborate pipeline routing strategy.
        """
        pipeline = super().route(file_path)

        # disable vectorstore for this kind of Index
        pipeline.VS = None
        # disable chunking for this kind of Index
        pipeline.splitter = None

        return pipeline

    def generate_with_llm(self, tag_prompt: str, doc_content: str) -> str:
        # TODO: add LLM logic to automatically tag document content
        return f"{type(self.llm)} {tag_prompt} - {doc_content}"