Repository: langgenius/dify
File: api/services/rag_pipeline/rag_pipeline_transform_service.py (Python)

Code before:
logger = logging.getLogger(__name__)


class RagPipelineTransformService:
    def transform_dataset(self, dataset_id: str):
        dataset = db.session.query(Dataset).where(Dataset.id == dataset_id).first()
        if not dataset:
            raise ValueError("Dataset not found")
        if dataset.pipeline_id and dataset.runtime_mode == "rag_pipeline":
            return {
                "pipeline_id": dataset.pipeline_id,
                "dataset_id": dataset_id,
                "status": "success",
            }
        if dataset.provider != "vendor":
            raise ValueError("External dataset is not supported")
        datasource_type = dataset.data_source_type
        indexing_technique = dataset.indexing_technique

        if not datasource_type and not indexing_technique:
            return self._transform_to_empty_pipeline(dataset)

        doc_form = dataset.doc_form
        if not doc_form:
            return self._transform_to_empty_pipeline(dataset)
        retrieval_model = dataset.retrieval_model
        pipeline_yaml = self._get_transform_yaml(doc_form, datasource_type, indexing_technique)
        # deal dependencies
        self._deal_dependencies(pipeline_yaml, dataset.tenant_id)
        # Extract app data
        workflow_data = pipeline_yaml.get("workflow")
        if not workflow_data:
            raise ValueError("Missing workflow data for rag pipeline")
        graph = workflow_data.get("graph", {})
        nodes = graph.get("nodes", [])
        new_nodes = []

        for node in nodes:
            if (
                node.get("data", {}).get("type") == "datasource"
                and node.get("data", {}).get("provider_type") == "local_file"
            ):
                node = self._deal_file_extensions(node)
            if node.get("data", {}).get("type") == "knowledge-index":
                node = self._deal_knowledge_index(dataset, doc_form, indexing_technique, retrieval_model, node)
            new_nodes.append(node)
        if new_nodes:
            graph["nodes"] = new_nodes
            workflow_data["graph"] = graph
            pipeline_yaml["workflow"] = workflow_data
        # create pipeline

Review comment (suggestion):
rename this valuable, we have RetrievalMethod <-> retrieval_method

So this be RetrievalSetting <-> retrieval_setting


Code after:
logger = logging.getLogger(__name__)


class RagPipelineTransformService:
    def transform_dataset(self, dataset_id: str):
        dataset = db.session.query(Dataset).where(Dataset.id == dataset_id).first()
        if not dataset:
            raise ValueError("Dataset not found")
        if dataset.pipeline_id and dataset.runtime_mode == "rag_pipeline":
            return {
                "pipeline_id": dataset.pipeline_id,
                "dataset_id": dataset_id,
                "status": "success",
            }
        if dataset.provider != "vendor":
            raise ValueError("External dataset is not supported")
        datasource_type = dataset.data_source_type
        indexing_technique = dataset.indexing_technique

        if not datasource_type and not indexing_technique:
            return self._transform_to_empty_pipeline(dataset)

        doc_form = dataset.doc_form
        if not doc_form:
            return self._transform_to_empty_pipeline(dataset)
        retrieval_model = RetrievalSetting.model_validate(dataset.retrieval_model) if dataset.retrieval_model else None
        pipeline_yaml = self._get_transform_yaml(doc_form, datasource_type, indexing_technique)
        # deal dependencies
        self._deal_dependencies(pipeline_yaml, dataset.tenant_id)
        # Extract app data
        workflow_data = pipeline_yaml.get("workflow")
        if not workflow_data:
            raise ValueError("Missing workflow data for rag pipeline")
        graph = workflow_data.get("graph", {})
        nodes = graph.get("nodes", [])
        new_nodes = []

        for node in nodes:
            if (
                node.get("data", {}).get("type") == "datasource"
                and node.get("data", {}).get("provider_type") == "local_file"
            ):
                node = self._deal_file_extensions(node)
            if node.get("data", {}).get("type") == "knowledge-index":
                node = self._deal_knowledge_index(dataset, doc_form, indexing_technique, retrieval_model, node)
            new_nodes.append(node)
        if new_nodes:
            graph["nodes"] = new_nodes
            workflow_data["graph"] = graph
            pipeline_yaml["workflow"] = workflow_data
        # create pipeline