source-code for model version v0.10_20250318_214952149_UTC- retrain-pipelines 0.1.1

v0.10_20250318_214952149_UTC/requirements.txt

@@ -0,0 +1,193 @@
+absl-py==2.1.0
+accelerate==1.1.1
+aiohappyeyeballs==2.4.3
+aiohttp==3.10.10
+aiosignal==1.3.1
+airportsdata==20241001
+annotated-types==0.7.0
+anyio==4.8.0
+asttokens==2.4.1
+async-timeout==4.0.3
+attrs==24.2.0
+bitsandbytes==0.44.1
+boto3==1.35.58
+botocore==1.35.58
+certifi==2024.8.30
+charset-normalizer==3.4.0
+click==8.1.7
+cloudpickle==3.1.0
+colorama==0.4.6
+comm==0.2.2
+contourpy==1.3.1
+cuda-python==12.6.2
+cudf-polars-cu12==24.10.1
+cycler==0.12.1
+datasets==3.1.0
+debugpy==1.8.8
+decorator==5.1.1
+dill==0.3.8
+diskcache==5.6.3
+docker==7.1.0
+docker-pycreds==0.4.0
+docstring_parser==0.16
+exceptiongroup==1.2.2
+executing==2.1.0
+fastapi==0.115.8
+fastjsonschema==2.20.0
+filelock==3.16.1
+fonttools==4.54.1
+frozenlist==1.5.0
+fsspec==2024.9.0
+gitdb==4.0.11
+GitPython==3.1.43
+graphviz==0.20.3
+grpcio==1.68.1
+h11==0.14.0
+hf_transfer==0.1.8
+httptools==0.6.4
+huggingface-hub==0.27.1
+idna==3.10
+iniconfig==2.0.0
+interegular==0.3.3
+ipykernel==6.29.5
+ipython==8.29.0
+ipywidgets==8.1.5
+jedi==0.19.2
+Jinja2==3.1.4
+jmespath==1.0.1
+joblib==1.4.2
+jsonschema==4.23.0
+jsonschema-specifications==2024.10.1
+jupyter_client==8.6.3
+jupyter_core==5.7.2
+jupyterlab_widgets==3.0.13
+kiwisolver==1.4.7
+lark==1.2.2
+libcudf-cu12==24.10.1
+litserve==0.2.6
+llvmlite==0.43.0
+lxml==5.3.0
+Markdown==3.7
+markdown-it-py==3.0.0
+MarkupSafe==3.0.2
+matplotlib==3.9.2
+matplotlib-inline==0.1.7
+mdurl==0.1.2
+metaflow==2.10.0
+metaflow-card-html==1.0.2
+mpmath==1.3.0
+multidict==6.1.0
+multiprocess==0.70.16
+nbformat==5.10.4
+nest-asyncio==1.6.0
+networkx==3.2.1
+numba==0.60.0
+numpy==1.26.4
+nvidia-cublas-cu11==11.11.3.6
+nvidia-cublas-cu12==12.4.5.8
+nvidia-cuda-cupti-cu11==11.8.87
+nvidia-cuda-cupti-cu12==12.4.127
+nvidia-cuda-nvrtc-cu11==11.8.89
+nvidia-cuda-nvrtc-cu12==12.4.127
+nvidia-cuda-runtime-cu11==11.8.89
+nvidia-cuda-runtime-cu12==12.4.127
+nvidia-cudnn-cu11==9.1.0.70
+nvidia-cudnn-cu12==9.1.0.70
+nvidia-cufft-cu11==10.9.0.58
+nvidia-cufft-cu12==11.2.1.3
+nvidia-curand-cu11==10.3.0.86
+nvidia-curand-cu12==10.3.5.147
+nvidia-cusolver-cu11==11.4.1.48
+nvidia-cusolver-cu12==11.6.1.9
+nvidia-cusparse-cu11==11.7.5.86
+nvidia-cusparse-cu12==12.3.1.170
+nvidia-nccl-cu11==2.21.5
+nvidia-nccl-cu12==2.21.5
+nvidia-nvjitlink-cu12==12.4.127
+nvidia-nvtx-cu11==11.8.86
+nvidia-nvtx-cu12==12.4.127
+nvtx==0.2.10
+outlines==0.1.3
+outlines_core==0.1.14
+packaging==24.2
+pandas==2.2.3
+parso==0.8.4
+peft==0.13.2
+pexpect==4.9.0
+pillow==11.0.0
+platformdirs==4.3.6
+plotly==5.24.0
+pluggy==1.5.0
+polars==1.8.2
+prompt_toolkit==3.0.48
+propcache==0.2.0
+protobuf==3.20.3
+psutil==6.1.0
+ptyprocess==0.7.0
+pure_eval==0.2.3
+pyarrow==17.0.0
+pycountry==24.6.1
+pydantic==2.9.2
+pydantic_core==2.23.4
+pydot==1.4.2
+Pygments==2.18.0
+pylibcudf-cu12==24.10.1
+pyparsing==3.2.0
+pytest==8.3.3
+python-dateutil==2.9.0.post0
+python-dotenv==1.0.1
+python-multipart==0.0.20
+pytz==2024.2
+PyYAML==6.0.2
+pyzmq==26.2.0
+referencing==0.35.1
+regex==2024.11.6
+requests==2.32.3
+-e git+https://github.com/aurelienmorgan/retrain-pipelines.git@9bbdca8a19b421b90a2640a250d8549680898f9b#egg=retrain_pipelines&subdirectory=pkg_src
+rich==13.9.4
+rmm-cu12==24.10.0
+rpds-py==0.21.0
+s3transfer==0.10.3
+safetensors==0.4.5
+scikit-learn==1.5.1
+scipy==1.14.1
+sentencepiece==0.2.0
+sentry-sdk==2.19.0
+setproctitle==1.3.4
+shtab==1.7.1
+six==1.16.0
+smmap==5.0.1
+sniffio==1.3.1
+stack-data==0.6.3
+starlette==0.45.3
+sympy==1.13.1
+tenacity==9.0.0
+tensorboard==2.18.0
+tensorboard-data-server==0.7.2
+threadpoolctl==3.5.0
+tokenizers==0.20.3
+tomli==2.1.0
+torch==2.5.0
+tornado==6.4.1
+tqdm==4.67.0
+traitlets==5.14.3
+transformers==4.46.2
+triton==3.1.0
+trl==0.12.0
+typing_extensions==4.12.2
+tyro==0.8.14
+tzdata==2024.2
+unsloth==2024.11.5
+unsloth_zoo==2024.11.4
+urllib3==2.2.3
+uvicorn==0.34.0
+uvloop==0.21.0
+wandb==0.18.7
+watchfiles==1.0.4
+wcwidth==0.2.13
+websockets==15.0
+Werkzeug==3.1.3
+widgetsnbextension==4.0.13
+xformers==0.0.28.post2
+xxhash==3.5.0
+yarl==1.17.1

v0.10_20250318_214952149_UTC/retraining_pipeline.py

@@ -0,0 +1,2179 @@
+
+from unsloth import FastLanguageModel, \
+    is_bfloat16_supported, UnslothTrainer, \
+    UnslothTrainingArguments
+
+import torch
+
+import os
+import sys
+
+import gc
+import json
+import time
+import shutil
+import logging
+import traceback
+import subprocess
+import importlib.util
+from enum import Enum
+from io import StringIO
+from textwrap import dedent
+from datetime import datetime
+from contextlib import redirect_stdout
+
+import numpy as np
+import pandas as pd
+
+import polars as pl
+from polars.exceptions import ComputeError
+
+import matplotlib
+import matplotlib.pyplot as plt
+
+from jinja2 import Environment, FileSystemLoader
+
+from metaflow import FlowSpec, step, Parameter, JSONType, \
+    IncludeFile, current, metaflow_config as mf_config, \
+    resources, Flow, Task, card
+from metaflow.current import Current
+from metaflow.cards import Image, Table, Markdown, \
+    Artifact, get_cards
+
+from datasets import load_dataset, Dataset, DatasetDict
+from datasets.config import HF_DATASETS_CACHE, HF_CACHE_HOME
+from huggingface_hub import list_repo_commits
+from transformers import AutoTokenizer
+from transformers.utils import logging as hf_logging
+
+from retrain_pipelines import __version__
+from retrain_pipelines.dataset.hf_utils import get_lazy_df, \
+    get_column_info, iterable_dataset_multi_buffer_sampler, \
+    push_dataset_version_to_hub
+from retrain_pipelines.dataset.tool_calls import \
+    get_unique_tools, count_tool_occurrences, \
+    plot_tools_occurences, column_words_stats, \
+    plot_words_count
+from retrain_pipelines.utils.hf_utils import \
+    get_new_repo_minor_version, push_files_to_hub_repo_branch
+from retrain_pipelines.utils import create_requirements
+
+ 
+class LocalServeReadinessEnum(Enum):
+    """
+    tracking local-serve (infra-validation)
+    status using a "3+"-states enum :
+        - "-1" for "not applicable"
+          (i.e. "model version not blessed"),
+        - "0/1" bool for failure/success.
+    """
+    NOT_APPLICABLE = -1
+    FAILURE = 0
+    FAILURE_NO_DOCKER = 2
+    SUCCESS = 1
+
+
+class UnslothFuncCallFlow(FlowSpec):
+    """
+    Training pipeline
+    """
+    # @see https://github.com/unslothai/unsloth/wiki
+
+    #--- flow parameters -------------------------------------------------------
+
+    RETRAIN_PIPELINE_TYPE = "mf_unsloth_func_call_litserve"
+    # in order to share the config across subprocesses
+    os.environ["retrain_pipeline_type"] = RETRAIN_PIPELINE_TYPE
+
+    hf_dataset = Parameter(
+        "hf_dataset",
+        help="dict with 'repo_id' and 'commit_hash' keys. " + \
+             "if 'commit_hash is None, falls back to latest version " +\
+             "of the dataset available in parquet format.\n" +
+             "Note that there are 3 required 'attributes' of type " + \
+             "str, list[str], list[str]",
+        type=JSONType,
+        default=dedent("""{
+            "repo_id": "Salesforce/xlam-function-calling-60k",
+            "config_name": "",
+            "commit_hash": "",
+            "attributes": {
+                "query_attr": "query",
+                "answers_attr": "answers",
+                "tools_attr": "tools"
+            }
+        }""").replace("'", '"').strip('"')
+    )
+
+    augmentation_rate = Parameter(
+        "augmentation_rate",
+        type=float,
+        default=.05,
+        help="proportion of records to be augmented "+\
+             "(x% of original dataset is created"+\
+             " as additional augmented datapoints), i.e. "+\
+             "truncated queries to serve as negative examples, "+\
+             "meaning they trigger no tool call "+\
+             "due to info incompleteness."
+    )
+
+    hf_enrich_dataset = Parameter(
+        "hf_enrich_dataset",
+        help="dict with 'repo_id', 'config_name' and 'commit_hash', "+\
+             "query_attribute' and 'query_attribute_handler' keys. "+\
+             "if 'commit_hash is None, falls back to latest version "+\
+             "of the dataset available in parquet format."+\
+             "'query_attribute' depicts the dataset attribute "+\
+             "from which 'queries' are to be sampled."+\
+             "'query_attribute_handler' serves for attributes "+\
+             "that have complex structure, "+\
+             "other than 'string' datatype.",
+        type=JSONType,
+        # @see https://huggingface.co/datasets/google-research-datasets/natural_questions
+        default=dedent("""{
+            "repo_id": "lighteval/natural_questions_clean",
+            "config_name": "",
+            "commit_hash": "",
+            "query_attribute": "question",
+            "query_attribute_handler": "lambda x: x"
+        }""").replace("'", '"').strip('"')
+    )
+
+    enrichment_rate = Parameter(
+        "enrichment_rate",
+        type=float,
+        default=.1,
+        help="proportion of records "+\
+             "to be added from the 'hf_enrich_dataset'"+\
+             "(x% of original dataset is sampled and"+\
+             " added as enriching datapoints), i.e. "+\
+             "queries to serve as negative examples, "+\
+             "due to their complete disconnexion "+\
+             "to tool calling situations."
+    )
+
+    dataset_repo_id = Parameter(
+        "dataset_repo_id",
+        type=str,
+        default="retrain-pipelines/func_calls",
+        help="The 'repo_id' to be used " + \
+             "for the Hugging Face dataset version push " + \
+             "(will be created at runtime" + \
+             " if doesn't already exist)."
+    )
+
+    hf_base_model = Parameter(
+        "hf_base_model",
+        help="dict with 'repo_id' and 'commit_hash' keys."+\
+             "if 'commit_hash is None, falls back "+\
+             "to latest available version of the model.",
+        type=JSONType,
+        default=dedent("""{
+            "repo_id": "unsloth/Qwen2.5-1.5B",
+            "commit_hash": ""
+        }""").replace("'", '"').strip('"')
+    )
+
+    cpt_training_args = Parameter(
+        "cpt_training_args",
+        help="dict with `TrainingArguments` params "+\
+             "for the CPT job.",
+        type=JSONType,
+        default=dedent("""{
+            "warmup_ratio": 0.1,
+            "num_train_epochs": 1
+        }""").replace("'", '"').strip('"')
+    )
+
+    sft_training_args = Parameter(
+        "sft_training_args",
+        help="dict with `TrainingArguments` params "+\
+             "for the SFT job.",
+        type=JSONType,
+        default=dedent("""{
+            "warmup_ratio": 0.1,
+            "num_train_epochs": 1
+        }""").replace("'", '"').strip('"')
+    )
+
+    model_repo_id = Parameter(
+        "model_repo_id",
+        type=str,
+        default="retrain-pipelines/function_caller",
+        help="The 'repo_id' to be used " + \
+             "for the Hugging Face model version push " + \
+             "(will be created at runtime" + \
+             " if doesn't already exist)."
+    )
+
+    default_pipeline_card_module_dir = \
+        os.path.dirname(
+            importlib.util.find_spec(
+                f"retrain_pipelines.pipeline_card."+
+                f"{RETRAIN_PIPELINE_TYPE}"
+            ).origin)
+    pipeline_card_artifacts_path = Parameter(
+        "pipeline_card_artifacts_path",
+        type=str,
+        default=default_pipeline_card_module_dir,
+        help="pipeline_card artifacts location "+\
+             "(i.e. dir hosting your optional " + \
+             " custom documentation files :" + \
+             " 'pipeline_card.py' and/or 'template.html'"+\
+             " and/or 'model_readme.py'"+\
+             " and/or 'model_readme_template.md'," +\
+             " and/or 'dataset_readme.py'"+\
+             " and/or 'dataset_readme_template.md' file), " +\
+             "if different from default."
+    )
+    @staticmethod
+    def copy_default_dataset_readme_module(
+        target_dir: str,
+        exists_ok: bool = False
+    ) -> None:
+        os.makedirs(target_dir, exist_ok=True)
+        if (
+            not exists_ok and
+            os.path.exists(os.path.join(target_dir, "dataset_readme.py"))
+        ):
+            print("File already exists. Skipping copy.")
+        else:
+            filefullname = os.path.join(
+                    UnslothFuncCallFlow.default_pipeline_card_module_dir,
+                    "dataset_readme.py"
+                )
+            shutil.copy(filefullname, target_dir)
+            print(filefullname)
+    @staticmethod
+    def copy_default_dataset_readme_template(
+        target_dir: str,
+        exists_ok: bool = False
+    ) -> None:
+        os.makedirs(target_dir, exist_ok=True)
+        if (
+            not exists_ok and
+            os.path.exists(os.path.join(target_dir,
+                                        "dataset_readme_template.md"))
+        ):
+            print("File already exists. Skipping copy.")
+        else:
+            filefullname = os.path.join(
+                    UnslothFuncCallFlow.default_pipeline_card_module_dir,
+                    "dataset_readme_template.md")
+            shutil.copy(filefullname, target_dir)
+            print(filefullname)
+    @staticmethod
+    def copy_default_model_readme_module(
+        target_dir: str,
+        exists_ok: bool = False
+    ) -> None:
+        os.makedirs(target_dir, exist_ok=True)
+        if (
+            not exists_ok and
+            os.path.exists(os.path.join(target_dir, "model_readme.py"))
+        ):
+            print("File already exists. Skipping copy.")
+        else:
+            filefullname = os.path.join(
+                    UnslothFuncCallFlow.default_pipeline_card_module_dir,
+                    "model_readme.py"
+                )
+            shutil.copy(filefullname, target_dir)
+            print(filefullname)
+    @staticmethod
+    def copy_default_model_readme_template(
+        target_dir: str,
+        exists_ok: bool = False
+    ) -> None:
+        os.makedirs(target_dir, exist_ok=True)
+        if (
+            not exists_ok and
+            os.path.exists(os.path.join(target_dir,
+                                        "model_readme_template.md"))
+        ):
+            print("File already exists. Skipping copy.")
+        else:
+            filefullname = os.path.join(
+                    UnslothFuncCallFlow.default_pipeline_card_module_dir,
+                    "model_readme_template.md")
+            shutil.copy(filefullname, target_dir)
+            print(filefullname)
+    @staticmethod
+    def copy_default_pipeline_card_module(
+        target_dir: str,
+        exists_ok: bool = False
+    ) -> None:
+        os.makedirs(target_dir, exist_ok=True)
+        if (
+            not exists_ok and
+            os.path.exists(os.path.join(target_dir, "pipeline_card.py"))
+        ):
+            print("File already exists. Skipping copy.")
+        else:
+            filefullname = os.path.join(
+                    UnslothFuncCallFlow.default_pipeline_card_module_dir,
+                    "pipeline_card.py"
+                )
+            shutil.copy(filefullname, target_dir)
+            print(filefullname)
+    @staticmethod
+    def copy_default_pipeline_card_html_template(
+        target_dir: str,
+        exists_ok: bool = False
+    ) -> None:
+        os.makedirs(target_dir, exist_ok=True)
+        if (
+            not exists_ok and
+            os.path.exists(os.path.join(target_dir, "template.html"))
+        ):
+            print("File already exists. Skipping copy.")
+        else:
+            filefullname = os.path.join(
+                    UnslothFuncCallFlow.default_pipeline_card_module_dir,
+                    "template.html")
+            shutil.copy(filefullname, target_dir)
+            print(filefullname)
+
+    del RETRAIN_PIPELINE_TYPE
+
+    #---------------------------------------------------------------------------
+
+    @step
+    def start(self):
+        print(f"{current.flow_name} - {current.run_id}")
+
+        # GPU availability
+        print(torch.cuda.get_device_name(0))
+        print(torch.__version__)
+        self.engine = "gpu" if torch.cuda.is_available() else "cpu"
+
+        # hf_dataset
+        hf_dataset_dict = \
+            get_lazy_df(
+                repo_id=self.hf_dataset["repo_id"],
+                commit_hash=self.hf_dataset["commit_hash"],
+                files_filter=(
+                    self.hf_dataset['config_name']+"/.*\\.parquet"
+                    if (
+                        self.hf_dataset["config_name"] and
+                        "" < self.hf_dataset["config_name"]
+                    ) else ".*\\.parquet"
+                ),
+                hf_token=os.getenv("HF_TOKEN", None)
+            )
+        try:
+            print(hf_dataset_dict["repo_id"], ", ",
+                  hf_dataset_dict["commit_hash"], "  -  ",
+                  hf_dataset_dict["commit_datetime"], "\n",
+                  hf_dataset_dict["lazy_df"].explain())
+        except ComputeError as ex:
+            if "HF_TOKEN" not in os.environ:
+                print("Does the Hugging Face-hosted dataset " +
+                      "require authentication ?",
+                      file=sys.stderr, flush=True)
+            raise ex
+        self.hf_dataset_dict = hf_dataset_dict
+
+        # hf_enrich_dataset
+        print(self.hf_enrich_dataset)
+        hf_enrich_dataset_dict = \
+            get_lazy_df(
+                repo_id=self.hf_enrich_dataset["repo_id"],
+                commit_hash=self.hf_enrich_dataset["commit_hash"],
+                files_filter=(
+                    self.hf_enrich_dataset['config_name']+"/.*\\.parquet"
+                    if (
+                        self.hf_enrich_dataset["config_name"] and
+                        "" < self.hf_enrich_dataset["config_name"]
+                    ) else ".*\\.parquet"
+                ),
+                hf_token=os.getenv("HF_TOKEN", None)
+            )
+        print(' ; '.join(f"{k}: {hf_enrich_dataset_dict[k]}"
+                                 for k in ['commit_hash',
+                                           'commit_datetime']))
+        self.hf_enrich_dataset_dict = hf_enrich_dataset_dict
+
+        # hf_base_model
+        hf_base_model_commits = list_repo_commits(
+            repo_id=self.hf_base_model["repo_id"],
+            revision=(
+                None if (rev_commit_hash:=self.hf_base_model["commit_hash"]) == ""
+                else rev_commit_hash
+            ),
+            repo_type="model",
+            token=os.getenv("HF_TOKEN", None))
+        self.hf_base_model_dict = {
+            "repo_id": self.hf_base_model["repo_id"],
+            "commit_hash": hf_base_model_commits[0].commit_id,
+            "commit_datetime": \
+                hf_base_model_commits[0].created_at
+        }
+
+        self.model_version_blessed = False
+        self.current_blessed_run = None
+        self.current_blessed_version_dict = None
+        current.run.remove_tag("model_version_blessed")
+
+        self.retrain_pipelines = f"retrain-pipelines {__version__}"
+        self.retrain_pipeline_type = os.environ["retrain_pipeline_type"]
+
+        self.serving_artifacts_local_folder = \
+            os.path.realpath(os.path.join(
+                os.path.dirname(__file__),
+                '..', '..', 'serving_artifacts',
+                os.path.sep.join(current.run.path_components)
+        ))
+
+        if not os.path.exists(self.serving_artifacts_local_folder):
+            os.makedirs(self.serving_artifacts_local_folder)
+
+        self.unsloth_dir = os.path.join(
+            self.serving_artifacts_local_folder,
+            "Unsloth"
+        )
+        print(f"unsloth_dir : {self.unsloth_dir}")
+        self.cpt_model_dir = os.path.join(
+                self.unsloth_dir, "cpt_model")
+        self.sft_model_dir = os.path.join(
+                self.unsloth_dir, "sft_model")
+
+        self.next(self.eda)
+
+
+    @step
+    def eda(self):
+        """
+        exploratory data analysis.
+        """
+
+        ############################
+        #    features and label    #
+        #       basic counts       #
+        ############################
+        self.records_count = self.hf_dataset_dict["lazy_df"] \
+            .select(pl.len()).collect(engine=self.engine).item()
+        self.data_schema = get_column_info(
+            self.hf_dataset_dict["lazy_df"], engine=self.engine)
+        ############################
+
+        ############################
+        #          Answers         #
+        #        tools count       #
+        ############################
+        struct_schema = pl.Struct([
+            pl.Field("name",
+                     pl.String
+                    ),
+            pl.Field("arguments",
+                     pl.List(pl.String)  # we retrieve list of args names
+                                         # (without assigned values)
+                    )
+        ])
+        tool_answer_occurrences_df = \
+            count_tool_occurrences(
+                self.hf_dataset_dict["lazy_df"],
+                self.hf_dataset["attributes"]["answers_attr"],
+                struct_schema) \
+            .collect(engine=self.engine)
+        print(f"{tool_answer_occurrences_df['occurrences'].sum():,} " +
+              f"query/tool-calls pairs")
+        fig = plot_tools_occurences(tool_answer_occurrences_df,
+                                    title_prefix="Dataset answers - ")
+        self.answers_tools_count_fig = fig
+        ############################
+
+        ############################
+        #           Query          #
+        #        words count       #
+        ############################
+        queries_max_length = self.hf_dataset_dict["lazy_df"].select(
+            pl.col(
+                self.hf_dataset["attributes"]["query_attr"]
+            ).str.len_chars().max().alias("max_query_length")
+        ).collect(engine=self.engine)
+        print(f"longuest query counts " +
+              f"{queries_max_length['max_query_length'][0]:,} characters")
+
+        # queries length quartiles
+        self.query_words_stats = \
+            column_words_stats(
+                self.hf_dataset_dict["lazy_df"],
+                self.hf_dataset["attributes"]["query_attr"]
+            ).collect(engine=self.engine)
+        print(self.query_words_stats.to_pandas().to_string(index=False))
+        print("Two thirds of the records have a query with less than " +
+              f"{self.query_words_stats['q3'][0]} words.")
+
+        fig = plot_words_count(
+                self.hf_dataset_dict["lazy_df"],
+                column_name=self.hf_dataset["attributes"]["query_attr"],
+                engine=self.engine)
+        self.words_count_fig = fig
+        ############################
+
+        ############################
+        #     hf_enrich_dataset    #
+        #    Query words count     #
+        ############################
+        enrich_question_words_stats = \
+            column_words_stats(
+                self.hf_enrich_dataset_dict['lazy_df'],
+                self.hf_enrich_dataset["query_attribute"],
+                column_attr_handler=eval(
+                    self.hf_enrich_dataset["query_attribute_handler"])
+            ).collect(engine=self.engine)
+        print(enrich_question_words_stats.to_pandas()
+                .to_string(index=False))
+        del enrich_question_words_stats
+        ############################
+
+        self.next(self.augment_data)
+
+
+    @step
+    def augment_data(self):
+        """
+        Add 'negative' examples, where
+        queries do not trigger any tool call.
+        To achieve that, we sample long user queries,
+        truncate at half words count, and
+        associate this to an empty list of tool-calls.
+        """
+        """
+        We only consider :
+          - records with longuest queries,
+            i.e. queries in the last quartile
+            of "queries with most word-counts"
+            (this is to avoid that 'truncated' queries
+             get really short)
+          - records with answers consisting
+            in a single tool-call
+            (in order to minimize the risk
+             that truncating actually gives
+             a valid answer with
+             one tool-call [or more])
+
+        Note on flow 'augmentation_rate' :
+            we add that many records (at most),
+            as quartiles size permits.
+        """
+
+        print("Sampling within the population with more than " +
+              str(self.query_words_stats['q3'][0]) +
+              " words (longest queries quartile) =>")
+
+        samples_count = \
+            int(self.records_count * self.augmentation_rate)
+        print(f"would represent {samples_count:,.0f} " +
+              f"records to be sampled")
+
+        eligible_records_df = \
+            self.hf_dataset_dict["lazy_df"].filter(
+                pl.col(
+                    self.hf_dataset["attributes"]["query_attr"]
+                )
+                .str.extract_all(r"\w+")
+                .map_elements(
+                    lambda arr: len(arr),
+                    return_dtype=pl.Int16)
+                .gt(self.query_words_stats['q3'][0])
+                & pl.col("answers")
+                .map_elements(
+                    lambda x: len(json.loads(x)) == 1
+                              if isinstance(x, str)
+                              else False,
+                    return_dtype=pl.Boolean)  
+            ) \
+            .collect(engine=self.engine)
+        eligible_records_count = \
+            eligible_records_df.select(pl.len())["len"][0]
+        print(f"eligible_records_count : " +
+              f"{eligible_records_count:,.0f}")
+        samples_count = min(samples_count, eligible_records_count)
+        self.actual_augmentation_rate = \
+            samples_count / self.records_count
+        print("actual augmentation rate : " +
+              f"{self.actual_augmentation_rate:.1%}")
+        sampled_records_df = eligible_records_df.sample(
+            n=samples_count
+        )
+
+        self.augmented_records_df = \
+            sampled_records_df.with_columns(
+                pl.col("query")
+                .map_elements(
+                    lambda query:
+                        " ".join(
+                            query.split()[
+                                :len(query.split()) // 2]),
+                    return_dtype=pl.Utf8)
+                .alias("truncated_query")
+            ).select([
+                pl.col("truncated_query").alias("query"),
+                pl.lit("[]").alias("answers")
+            ])
+        print(self.augmented_records_df.height,
+              self.augmented_records_df.columns)
+
+        self.next(self.enrich_data)
+
+
+    @step
+    def enrich_data(self):
+        """
+        Further enrich our dataset with 'negative' records from
+        another dataset (can be general-purpose text dataset)
+        as specified by the the flow 'hf_enrich_dataset' argument.
+        """
+        """
+        Note : we here use the Hugging Face `datasets` library
+        in 'streaming' mode for records sampling.
+        """
+
+        hf_enrich_ds = load_dataset(
+            path=self.hf_enrich_dataset["repo_id"],
+            name=self.hf_enrich_dataset["config_name"],
+            revision=self.hf_enrich_dataset_dict["commit_hash"],
+            streaming=True)
+        print(hf_enrich_ds["train"])
+
+        samples_count = \
+            int(self.records_count * self.enrichment_rate)
+        print(f"Samplig {samples_count:,.0f} records")
+
+        query_attribute_handler = \
+            eval(self.hf_enrich_dataset["query_attribute_handler"])
+        samples_iterator = iterable_dataset_multi_buffer_sampler(
+                hf_enrich_ds["train"],
+                total_samples=samples_count,
+                attributes_selector=\
+                    (lambda x:query_attribute_handler(
+                        x[self.hf_enrich_dataset["query_attribute"]])),
+                buffer_size=3_000,
+                num_passes=3,
+                seed=None
+            )
+        # Capitalize and add end punctuation if missing
+        start_time = time.time()
+        print("Starting sample enriching records, " +
+              "this may take some time if the source dataset " +
+              "has a complex structure..")
+        samples_list = [
+            s.capitalize() + ("" if s[-1] in ".!?" else "?")
+            for s in samples_iterator]
+        elapsed_time = time.time() - start_time
+        print(f".. sampling completed " +
+              f"({int(elapsed_time // 3_600)}h:" +
+               f"{int((elapsed_time % 3_600) // 60)}m:" +
+               f"{int(elapsed_time % 60)}s).")
+        enriched_records_df = pl.DataFrame(
+                {"query": samples_list,
+                 "answers": \
+                     ["[]"] * \
+                     len(samples_list)}
+            )
+        self.enriched_records_df = enriched_records_df
+
+        self.next(self.dataset_to_hub)
+
+
+    @step
+    def dataset_to_hub(self):
+        """
+        Push to hub dataset version
+        - continued pre-training dataset
+        - training and validation splits of the
+        augmented and enriched
+        supervised finetuning dataset
+        - readme with versioning info
+        """
+
+        #############################
+        #  case of user-provided    #
+        # documentation artifact(s) #
+        #############################
+        # note that user can provide either
+        # 'pipeline_card.py' or 'template.html'
+        # or 'dataset_readme.py'
+        # or 'dataset_readme_template.md'
+        # or 'model_readme.py'
+        # or 'model_readme_template.md'
+        # or any combination of those
+        # when specifying custom
+        # 'pipeline_card_artifacts_path'
+        if (
+            "dataset_readme_template.md" in
+                os.listdir(self.pipeline_card_artifacts_path)
+        ):
+            template_dir = self.pipeline_card_artifacts_path
+        else:
+            template_dir = os.path.dirname(
+                importlib.util.find_spec(
+                    f"retrain_pipelines.pipeline_card."+
+                    f"{os.getenv('retrain_pipeline_type')}"
+                ).origin)
+        print(f"template_dir : '{template_dir}'")
+        #############################
+        if "dataset_readme.py" in os.listdir(
+                self.pipeline_card_artifacts_path):
+            from retrain_pipelines.utils import \
+                get_get_dataset_readme_content
+            get_dataset_readme_content = \
+                get_get_dataset_readme_content(
+                    self.pipeline_card_artifacts_path)
+        else:
+            from retrain_pipelines.pipeline_card import \
+                    get_dataset_readme_content
+        #############################
+    
+
+        #############################
+        #    augmented & enriched   #
+        #     finetuning dataset    #
+        #############################
+        merged_df = pl.concat([
+                # dataset
+                self.hf_dataset_dict["lazy_df"].select([
+                        self.hf_dataset["attributes"]["query_attr"],
+                        self.hf_dataset["attributes"]["answers_attr"]
+                    ]).collect(engine=self.engine),
+                # truncated queries augmentation
+                self.augmented_records_df,
+                # enriching dataset
+                self.enriched_records_df
+            ]).sample(
+                # shuffling
+                fraction=1,
+                shuffle=True,
+                with_replacement=False
+            )
+        merged_df = merged_df.sample(fraction=1, shuffle=True)
+        merged_df.rechunk()
+        print(("merged_df", f"{merged_df.shape[0]:,.0F}",
+              merged_df.columns))
+
+        pandas_df = merged_df.to_pandas()
+        train_size = int(0.8 * len(pandas_df))
+        print(f"validation : {len(pandas_df) - train_size}")
+        sft_dataset = DatasetDict({
+            "train": Dataset.from_pandas(pandas_df[:train_size]),
+            "validation": Dataset.from_pandas(pandas_df[train_size:])
+        })
+        #############################
+
+        #############################
+        #   continued pre-training  #
+        #          dataset          #
+        #############################
+        struct_schema = pl.Struct([
+            pl.Field("name", pl.String),
+            pl.Field("description", pl.String),
+            pl.Field(
+                "parameters",
+                pl.String  # Use String to allow 
+                           # for varying structures
+                           # (different tools indeed having
+                           #  different sets of parameters
+                           #  i.e. different parameters counts,
+                           #  datatypes and names)
+                           # so parsing must be tolerant.
+            )
+        ])
+        unique_tools_df = get_unique_tools(
+                self.hf_dataset_dict["lazy_df"],
+                tools_attr_name=\
+                    self.hf_dataset["attributes"]["tools_attr"],
+                struct_schema=struct_schema
+            ).collect(engine=self.engine)
+        unique_tools_arrow_table = unique_tools_df.to_arrow()
+        self.unique_tools_dataset = \
+            Dataset(unique_tools_arrow_table)
+        print(self.unique_tools_dataset)
+        #############################
+
+        #############################
+        #        DatasetDict        #
+        #    with multiple tables   #
+        #############################
+        dataset_dict = DatasetDict({
+            "continued_pre_training": \
+                self.unique_tools_dataset,
+            "supervised_finetuning": sft_dataset
+        })
+        print(dataset_dict, flush=True)
+        #############################
+
+        #############################
+        #       dataset README      #
+        #       from template       #
+        #############################
+        commit_datetime = datetime.utcnow()
+        new_dataset_version_label = get_new_repo_minor_version(
+            repo_id=self.dataset_repo_id,
+            repo_type="dataset",
+            hf_token=os.getenv("HF_TOKEN", None))
+        readme_content = get_dataset_readme_content(
+            template_folder=template_dir,
+
+            hf_dataset_dict=self.hf_dataset_dict,
+            hf_enrich_dataset_dict=self.hf_enrich_dataset_dict,
+            dataset_dict=dataset_dict,
+
+            augmentation_rate=self.actual_augmentation_rate,
+            enrichment_rate=self.enrichment_rate,
+
+            version_label=new_dataset_version_label,
+            commit_datetime=commit_datetime,
+
+            mf_flow_name=current.flow_name,
+            mf_run_id=current.run.id,
+            engine=self.engine
+        )
+        #############################
+
+        dataset_commit_hash = push_dataset_version_to_hub(
+            repo_id=self.dataset_repo_id,
+            version_label=new_dataset_version_label,
+            timestamp_str=commit_datetime.strftime(
+                "%Y-%m-%d %H:%M:%S UTC"),
+            dataset_dict=dataset_dict,
+            dataset_readme_content=readme_content,
+            hf_token=os.getenv("HF_TOKEN", None)
+        )
+        if not dataset_commit_hash:
+            raise Exception(
+                "Failed to publish dataset version.")
+        print(f"https://huggingface.co/datasets/{self.dataset_repo_id}" +
+              f"/blob/{dataset_commit_hash}/README.md")
+        self.dataset_commit_dict = {
+            "repo_id": self.dataset_repo_id,
+            "commit_hash": dataset_commit_hash,
+            "version_label": new_dataset_version_label,
+            "commit_datetime": commit_datetime,
+        }
+
+        self.next(self.continued_pre_training)
+
+
+    @step
+    def continued_pre_training(self):
+        """
+        Gives the base model some additional intrinsic knowkledge
+        through continued pre-training.
+        See unsloth.ai/blog/contpretraining
+        """
+        from retrain_pipelines.model.hf_utils import \
+            plot_log_history
+
+        #######################################
+        # base-model and associated tokenizer #
+        #      from Hub (or local cache)      #
+        #######################################
+        self.max_seq_length = 2048
+        model, tokenizer = FastLanguageModel.from_pretrained(
+            model_name=self.hf_base_model_dict["repo_id"],
+            revision=self.hf_base_model_dict["commit_hash"],
+            max_seq_length=self.max_seq_length,
+            dtype=None,
+            load_in_4bit=False,
+            # case of a gated or private base-model
+            token=os.getenv("HF_TOKEN", None)
+        )
+        #######################################
+
+        #######################################
+        #   dataset prompt_template mapping   #
+        #######################################
+        tools_dataset = DatasetDict(
+            {"train": self.unique_tools_dataset})
+        print(tools_dataset)
+        tool_prompt_template = "tool: {}"
+        def formatting_prompts_func(tools_batch):
+            tools_batch = tools_batch["tool"]
+            outputs = []
+            for tool in tools_batch:
+                # Must add EOS_TOKEN,
+                # otherwise generation will go on forever!
+                text = tool_prompt_template.format(tool) + \
+                       tokenizer.eos_token
+                outputs.append(text)
+            return { "tools" : outputs, }
+        cpt_dataset = tools_dataset["train"].map(
+            formatting_prompts_func, batched=True,)
+        #######################################
+
+        #######################################
+        #             PEFT adapter            #
+        #      for continued pre-training     #
+        #######################################
+        model = FastLanguageModel.get_peft_model(
+            model,
+            r = 128, # any number >0 ; 8, 16, 32, 64, 128, 256
+            target_modules = ["q_proj", "k_proj", "v_proj", "o_proj",
+                              "gate_proj", "up_proj", "down_proj",
+                              # Add for continued pretraining
+                              "embed_tokens", "lm_head",],
+            lora_alpha = 32,
+            lora_dropout = 0,    # Supports any, 0 is optimized
+            bias = "none",       # Supports any, "none" is optimized
+            # True or "unsloth" for very long context
+            use_gradient_checkpointing = "unsloth",
+            use_rslora = True,   # rank-stabilized LoRA
+            loftq_config = None, # LoftQ
+            #random_state = 3407,
+        )
+        #######################################
+
+        #######################################
+        #             cpt_trainer             #
+        #######################################
+        if (
+            "records_cap" in self.cpt_training_args and
+            self.cpt_training_args["records_cap"] is not None and
+            isinstance(self.cpt_training_args["records_cap"], int)
+        ):
+            cpt_dataset = cpt_dataset.take(
+                self.cpt_training_args["records_cap"])
+            print(f"cpt_dataset : {cpt_dataset}")
+
+        train_args = UnslothTrainingArguments(
+            # https://huggingface.co/docs/transformers/main_classes/trainer#transformers.TrainingArguments.save_strategy
+            per_device_train_batch_size=2,
+            gradient_accumulation_steps=8,
+
+            **{k: v for k, v in self.cpt_training_args.items()
+                    if k != "records_cap"},
+
+            # 2 to 10x smaller learning rate
+            # for the embedding matrices
+            learning_rate=5e-5,
+            embedding_learning_rate=1e-5,
+
+            fp16=not is_bfloat16_supported(),
+            bf16=is_bfloat16_supported(),
+            logging_steps=1,
+            optim="adamw_8bit",
+            weight_decay=0.01,
+            lr_scheduler_type="linear",
+            #seed=3407,
+
+            output_dir=os.path.join(
+                self.unsloth_dir, "outputs", "cpt"),
+            save_total_limit = 2,
+
+            report_to="tensorboard",
+            logging_dir=os.path.join(
+                self.sft_model_dir,
+                "runs", "cpt")
+        )
+
+        self.cpt_traces_file_fullname = os.path.join(
+            self.unsloth_dir, "cpt_trainer_traces.txt")
+        print("Training started. " +
+              f"Check {self.cpt_traces_file_fullname} for live traces.",
+              flush=True)
+
+        trainer = UnslothTrainer(
+            model=model, tokenizer=tokenizer,
+            train_dataset=cpt_dataset,
+            dataset_text_field="tools",
+            max_seq_length=self.max_seq_length,
+            dataset_num_proc=2,
+            args=train_args,
+        )
+        #######################################
+
+        #######################################
+        #      Show current memory stats      #
+        #######################################
+        torch.cuda.ipc_collect()
+        torch.cuda.empty_cache()
+        gc.collect()
+
+        gpu_stats = torch.cuda.get_device_properties(0)
+        self.start_gpu_memory = \
+            round(torch.cuda.max_memory_reserved()
+                  / 1024 / 1024 / 1024, 3)
+        self.max_memory = \
+            round(gpu_stats.total_memory
+                  / 1024 / 1024 / 1024, 3)
+        print(f"GPU = {gpu_stats.name}. " +
+              f"Max memory = {self.max_memory} GB.")
+        print(f"{self.start_gpu_memory} GB of memory reserved.")
+        #######################################
+
+        with open(self.cpt_traces_file_fullname, 'w') as f:
+            with redirect_stdout(f):
+                hf_logging.set_verbosity_error()
+                hf_logging.disable_progress_bar()
+                trainer_stats = trainer.train()
+        hf_logging.set_verbosity_info()
+        hf_logging.enable_progress_bar()
+        print(f"{trainer_stats.metrics['train_runtime']} " +
+              f"seconds used for training " +
+              f"({round(trainer_stats.metrics['train_runtime']/60, 2)}" +
+              f" minutes).")
+
+        self.cpt_log_history = trainer.state.log_history
+        # print(self.cpt_log_history)
+        self.cpt_log_history_fig = \
+            plot_log_history(
+                self.cpt_log_history,
+                title="Continued pretraining loss"
+            )
+
+        model.save_pretrained_merged(
+            save_directory=self.cpt_model_dir,
+            tokenizer=tokenizer,
+            save_method="lora"
+        )
+        print(f"cpt_model_dir : {self.cpt_model_dir}\n")
+
+        self.next(self.supervised_finetuning)
+
+
+    @step
+    def supervised_finetuning(self):
+        """
+        Trains the model on tool-calling
+        task specialization.
+        """
+        from retrain_pipelines.model.hf_utils import \
+            plot_log_history
+
+        torch.cuda.ipc_collect()
+        torch.cuda.empty_cache()
+        gc.collect()
+
+        model, tokenizer = FastLanguageModel.from_pretrained(
+            model_name=self.cpt_model_dir,
+            max_seq_length=self.max_seq_length,
+            dtype=None,
+            load_in_4bit=False,
+        )
+        # !!!! bug fix BEGIN !!!!
+        # otherwise, 'embed_tokens' and 'lm_head'
+        # trained during CPT are "ignored",
+        # i.e. not saved after SFT
+        # (note that, alternatively, we could also 
+        #  do this fix after sft-training and
+        #  just before saving ;
+        #  which would be equivalent to
+        #  freezing embeddings during finetuning
+        #  for better pretrained knowledge retention)
+        # @see https://www.reddit.com/r/unsloth/comments/1dtzcd6/fastlanguagemodelpatch_peft_model_changing/
+        model.model.model.embed_tokens.modules_to_save.default.to(
+            device="cuda:0",
+            dtype=torch.float32,
+            non_blocking=True)
+        model.model.model.embed_tokens.modules_to_save.default \
+            .requires_grad_(True)
+        model.model.lm_head.modules_to_save.default.to(
+            device="cuda:0",
+            dtype=torch.float32,
+            non_blocking=True)
+        model.model.lm_head.modules_to_save.default \
+            .requires_grad_(True)
+        # !!!! bug fix END !!!!
+
+        #######################################
+        #   dataset prompt_template mapping   #
+        #######################################
+        # download from Hub (or get from local cache)
+        queries_dataset = load_dataset(
+            path=self.dataset_commit_dict["repo_id"],
+            name="supervised_finetuning",
+            revision=self.dataset_commit_dict["commit_hash"],
+            token=os.getenv("HF_TOKEN", None))
+        print(f"HF_DATASETS_CACHE : {HF_DATASETS_CACHE}") # HF_CACHE_HOME
+        self.sft_prompt_template = dedent("""
+        You specialize in generating tool calls. Given a query, your task is to return a list of tool calls based on your knowledge of known tools.
+
+        Rules:
+        1. You can only use tools you know. Do not create new tools under any circumstances.
+        2. If a query does not match any known tool, return an empty list ([]).
+        3. If information is missing to use a known tool, do not attempt to use it.
+        4. Your response must always be a valid JSON array, and nothing else.
+
+        Be precise and do not guess.
+
+        # query:
+            {}
+        # response:
+            {}
+        """).strip()
+        tokenizer.chat_template = self.sft_prompt_template
+
+        EOS_TOKEN = tokenizer.eos_token
+        def formatting_prompts_func(records):
+            query = records["query"]
+            tools  = records["answers"]
+            outputs = []
+            for query, tools in zip(query, tools):
+                # Must add EOS_TOKEN,
+                # otherwise your generation will go on forever
+                text = self.sft_prompt_template.format(query, tools) \
+                       + EOS_TOKEN
+                outputs.append(text)
+            return { "text" : outputs, }
+        sft_train_dataset = queries_dataset["train"].map(
+            formatting_prompts_func, batched=True)
+        sft_valid_dataset = queries_dataset["validation"].map(
+            formatting_prompts_func, batched=True,)
+        #######################################
+
+        #######################################
+        #             PEFT adapter            #
+        #      for supervised finetuning      #
+        #######################################
+        # for cases where CPT has been merged into overall model
+        # otherwize, keep on training current LoRa adapter
+        # model = FastLanguageModel.get_peft_model(
+            # model,
+            # r = 128, # any number >0 ; 8, 16, 32, 64, 128, 256
+            # target_modules = ["q_proj", "k_proj", "v_proj", "o_proj",
+                              # "gate_proj", "up_proj", "down_proj"],
+            # lora_alpha = 32,
+            # lora_dropout = 0, # Supports any, but = 0 is optimized
+            # bias = "none",    # Supports any, but = "none" is optimized
+            # # True or "unsloth" for very long context
+            # use_gradient_checkpointing = "unsloth",
+            # random_state = 3407,
+            # use_rslora = True,   # rank stabilized LoRA
+            # loftq_config = None, # LoftQ
+        # )
+        #######################################
+
+        #######################################
+        #             sft_trainer             #
+        #######################################
+        split = sft_train_dataset.train_test_split(
+            test_size=1000,
+            #seed=42
+        )
+        train_dataset = split['train']
+        eval_dataset = split['test']
+        if (
+            "records_cap" in self.sft_training_args and
+            self.sft_training_args["records_cap"] is not None and
+            isinstance(self.sft_training_args["records_cap"], int)
+        ):
+            train_dataset = train_dataset.take(
+                self.sft_training_args["records_cap"])
+            eval_dataset = eval_dataset.take(
+                self.sft_training_args["records_cap"])
+            print(f"train_dataset : {train_dataset}")
+            print(f"eval_dataset :  {eval_dataset}")
+
+        train_args = UnslothTrainingArguments(
+            per_device_train_batch_size=2,
+            gradient_accumulation_steps=8,
+
+            **{k: v for k, v in self.sft_training_args.items()
+                    if k != "records_cap"},
+
+            per_device_eval_batch_size=2,
+            eval_steps=200,
+            eval_strategy="steps",
+            do_eval=True,
+
+            learning_rate=5e-5,
+            # embedding_learning_rate=1e-5, # Optionally here
+
+            fp16=not is_bfloat16_supported(),
+            bf16=is_bfloat16_supported(),
+
+            optim="adamw_8bit",
+            weight_decay=0.00,
+            lr_scheduler_type="linear",
+            #seed=3407,
+
+            output_dir=os.path.join(
+                self.unsloth_dir, "outputs", "sft"),
+            save_total_limit=2,
+
+            logging_steps=1,
+            report_to="tensorboard",
+            logging_dir=os.path.join(
+                self.sft_model_dir,
+                "runs", "sft")
+        )
+
+        self.sft_traces_file_fullname = os.path.join(
+            self.unsloth_dir, "sft_trainer_traces.txt")
+        print("Training started. " +
+              f"Check {self.sft_traces_file_fullname} for live traces.",
+              flush=True)
+
+        trainer = UnslothTrainer(
+            model=model, tokenizer=tokenizer,
+            train_dataset=train_dataset,
+            dataset_text_field="text",
+            eval_dataset=eval_dataset,
+            max_seq_length=self.max_seq_length,
+            dataset_num_proc=8,
+            args=train_args
+        )
+        trainer.can_return_loss = True
+        #######################################
+
+        #######################################
+        #      Show current memory stats      #
+        #######################################
+        torch.cuda.ipc_collect()
+        torch.cuda.empty_cache()
+        gc.collect()
+
+        used_memory = \
+            round(torch.cuda.max_memory_reserved()
+                  /1024/1024/1024, 3)
+        used_memory_for_lora = \
+            round(used_memory-self.start_gpu_memory, 3)
+        used_percentage = \
+            round(used_memory/self.max_memory*100, 3)
+        lora_percentage = \
+            round(used_memory_for_lora/self.max_memory*100,
+                  3)
+        print(f"Peak reserved memory = " +
+              f"{used_memory} GB.")
+        print(f"Peak reserved memory for " +
+              f"training = {used_memory_for_lora} " +
+              f"GB.")
+        print(f"Peak reserved memory % of " +
+              f"max memory = {used_percentage} %.")
+        print(f"Peak reserved memory for training " +
+              f"% of max memory = {lora_percentage} %.")
+        #######################################
+
+        with open(self.sft_traces_file_fullname, 'w') as f:
+            with redirect_stdout(f):
+                hf_logging.set_verbosity_error()
+                hf_logging.disable_progress_bar()
+                trainer_stats = trainer.train()
+        hf_logging.set_verbosity_info()
+        hf_logging.enable_progress_bar()
+        print(f"{trainer_stats.metrics['train_runtime']} " +
+              f"seconds used for training " +
+              f"({round(trainer_stats.metrics['train_runtime']/60, 2)}" +
+              f" minutes).")
+
+        self.sft_log_history = trainer.state.log_history
+        self.sft_log_history_fig = \
+            plot_log_history(
+                self.sft_log_history,
+                title="Supervised finetuning loss"
+            )
+
+        model.save_pretrained_merged(
+            self.sft_model_dir, tokenizer,
+            save_method = "lora"
+        )
+        print(f"sft_model_dir : {self.sft_model_dir}\n")
+
+        self.next(self.evaluate_model)
+
+
+    @step
+    def evaluate_model(self):
+        """
+        Batch inference on the SFT validation dataset.
+        """
+        from retrain_pipelines.model import \
+            infer_validation, compute_counts_n_metrics, \
+            plot_validation_completions
+
+        torch.cuda.ipc_collect()
+        torch.cuda.empty_cache()
+        gc.collect()
+
+
+        ######################################################
+        #              loading trained adapter              #
+        ######################################################
+        # Unsloth (if loading both model & tokenizer at once #
+        # same as we did in prior tasks, but now             #
+        # with tokenizer.chat_template being set             #
+        # in tokenizer.config) is forcing on us some kind of #
+        # chat_template format hard-requirements             #
+        # coming from their dream-fantasmagorical world..    #
+        ######################################################
+        # load base from cache
+        # (with base tokenizer, which we ignore)
+        model, _ = FastLanguageModel.from_pretrained(
+            model_name=self.hf_base_model_dict["repo_id"],
+            revision=self.hf_base_model_dict["commit_hash"],
+            max_seq_length=self.max_seq_length,
+            dtype=None,
+            load_in_4bit=False,
+            # case of a gated or private base-model
+            token=os.getenv("HF_TOKEN", None)
+        )
+        model = FastLanguageModel.for_inference(model)
+        # load our CPT+SFT trained & locally-saved adapter
+        model.load_adapter(peft_model_id=self.sft_model_dir)
+        # Separately load our (potentially trained &)
+        # locally-saved adapter-tokenizer
+        # (loading it below via HF and not Unsloth)
+        tokenizer = AutoTokenizer.from_pretrained(
+            pretrained_model_name_or_path=self.sft_model_dir
+        )
+        ######################################################
+
+        ######################################################
+        #                 validation dataset                 #
+        ######################################################
+        # download from Hub (or get from local cache)
+        queries_dataset = load_dataset(
+            path=self.dataset_commit_dict["repo_id"],
+            name="supervised_finetuning",
+            revision=self.dataset_commit_dict["commit_hash"],
+            token=os.getenv("HF_TOKEN", None))
+        if (
+            "records_cap" in self.sft_training_args and
+            self.sft_training_args["records_cap"] is not None and
+            isinstance(self.sft_training_args["records_cap"], int)
+        ):
+            validation_data = queries_dataset["validation"].take(
+                self.sft_training_args["records_cap"])
+        else:
+            validation_data = queries_dataset["validation"]
+        print(validation_data, flush=True)
+        ######################################################
+
+        self.max_new_tokens = 400
+        start_time = time.time()
+        validation_results = infer_validation(
+            tokenizer=tokenizer,
+            model=model,
+            validation_data=validation_data,
+            prompt_template=tokenizer.chat_template,
+            batch_size=32, # 64,
+            queries_attr_name=\
+                self.hf_dataset["attributes"]["query_attr"],
+            answers_attr_name=\
+                self.hf_dataset["attributes"]["answers_attr"],
+            max_new_tokens=self.max_new_tokens,
+            device="cuda"
+        )
+        print("infer_validation -   Elapsed time: " +
+              f"{(time.time() - start_time):.2f} seconds")
+        self.validation_results = validation_results #  <= to artifacts store
+
+        eval_df  = pl.LazyFrame(validation_results)
+
+        records = eval_df.with_columns(
+            (pl.col("answer") == pl.col("completion")) \
+                .alias("is_ground_truth_identical")
+        ).collect() #engine=self.engine)
+        print("perfect characters-match accuracy : " +
+              str(records['is_ground_truth_identical'].mean()))
+
+        eval_metrics_df = compute_counts_n_metrics(
+            eval_df, is_format_fault_tolerant=True)
+        overall_metrics_df = eval_metrics_df.select([
+                pl.col("precision").mean(),
+                pl.col("recall").mean(), 
+                pl.col("f1").mean(),
+                pl.col("jaccard").mean()
+            ]).collect() #engine=self.engine)
+        self.perf_metrics = overall_metrics_df.row(0, named=True)
+        print(self.perf_metrics)
+
+        self.validation_completions_fig = \
+            plot_validation_completions(
+                eval_metrics_df, engine=self.engine)
+
+        del model
+        del tokenizer
+        torch.cuda.ipc_collect()
+        torch.cuda.empty_cache()
+        gc.collect()
+
+        self.next(self.model_version_blessing)
+
+
+    @step
+    def model_version_blessing(self):
+        """
+        Comparing newly-retrained model version
+        against best-performing predecessor.
+        """
+        """
+        Note: for Hugging Face integrated pipelines,
+        we compare against lastest commit of main branch
+        of the model repository there.
+        When it comes to local "mf_run_id" of the pipeline run
+        having generated that best prior model version
+        (retrieved from model card metadata from HF yaml section),
+        we check against records of the herein ML-framework instance,
+        as "prior best version" of the model here beign retrained
+        may have been originated from another one
+        than the one executing the current retraining
+        (in which case, we simply don't includ a "local" hyperlink
+        in the model version pipeline_cards that will be
+        produced later in the herein pipeline run).
+        """
+        from retrain_pipelines.model.hf_utils import \
+            current_blessed_model_version_dict
+
+        main_perf_metric_name = "jaccard"
+
+        current_blessed_version_dict = \
+            current_blessed_model_version_dict(
+                repo_id=self.model_repo_id,
+                hf_token=os.getenv("HF_TOKEN", None)
+            )
+        print("current_blessed_version_dict : " +
+              str(current_blessed_version_dict))
+
+        if current_blessed_version_dict is None:
+            print("case 'no prior blessed model version found"
+                  " => blessing.'")
+            self.model_version_blessed = True
+
+        elif (
+            main_perf_metric_name in
+                current_blessed_version_dict["perf_metrics"]
+        ):
+            current_blessed_run_id = \
+                current_blessed_version_dict["mf_run_id"]
+            current_blessed_metric_value = \
+                current_blessed_version_dict[
+                    "perf_metrics"][main_perf_metric_name]
+
+            self.model_version_blessed = (
+                self.perf_metrics[main_perf_metric_name] >=
+                current_blessed_metric_value
+            )
+
+            if not self.model_version_blessed:
+                self.current_blessed_version_dict = \
+                    current_blessed_version_dict
+                for run in Flow(self.__class__.__name__):
+                    if str(run.id) == current_blessed_run_id:
+                        last_run_step = next(iter(run.steps()))
+                        last_task = next(iter(last_run_step.tasks()))
+                        # further filtering on successful runs that are
+                        # retraining of a prior version of the same model
+                        # (to minimize the risk that this was obtained
+                        #  on another ML-framework instance)
+                        if (
+                            last_task.successful and
+                            hasattr(last_task.artifacts, 'model_version_blessed') and
+                            last_task.artifacts.model_version_blessed.data and
+                            hasattr(last_task.artifacts, 'model_repo_id') and
+                            last_task.artifacts.model_repo_id.data == self.model_repo_id
+                        ):
+                            self.current_blessed_run = run
+                        break
+
+                if not self.current_blessed_run:
+                    print(
+                        "Couldn't find blessed run " +
+                        f"{current_blessed_run_id} !\n" +
+                        "It seems that prior blessed run was " +
+                        "executed on another ML framework instance.",
+                        file=sys.stderr, flush=True)
+
+            print("new : " +
+                    str(self.perf_metrics[main_perf_metric_name]) +
+                  " - previous best : " +
+                    str(current_blessed_metric_value) +
+                  " - model_version_blessing : " +
+                    str(self.model_version_blessed))
+
+        else:
+            raise Exception(
+                "Performance metric '" +
+                main_perf_metric_name +
+                "' can't be found in eval results " +
+                "from blessed run " +
+                str(current_blessed_version_dict[
+                    "mf_run_id"]) + " !")
+
+        # self.model_version_blessed = True ### DEBUG - DELETE ###
+
+        self.next(self.model_to_hub)
+
+
+    @step
+    def model_to_hub(self):
+        """
+        Push to hub model version, including
+        readme with versioning info.
+        """
+
+        #############################
+        #  case of user-provided    #
+        # documentation artifact(s) #
+        #############################
+        # note that user can provide either
+        # 'pipeline_card.py' or 'template.html'
+        # or 'dataset_readme.py'
+        # or 'dataset_readme_template.md'
+        # or 'model_readme.py'
+        # or 'model_readme_template.md'
+        # or any combination of those
+        # when specifying custom
+        # 'pipeline_card_artifacts_path'
+        if (
+            "model_readme_template.md" in
+                os.listdir(self.pipeline_card_artifacts_path)
+        ):
+            template_dir = self.pipeline_card_artifacts_path
+        else:
+            template_dir = os.path.dirname(
+                importlib.util.find_spec(
+                    f"retrain_pipelines.pipeline_card."+
+                    f"{os.getenv('retrain_pipeline_type')}"
+                ).origin)
+        print(f"template_dir : '{template_dir}'")
+        #############################
+        if "model_readme.py" in os.listdir(
+                self.pipeline_card_artifacts_path):
+            from retrain_pipelines.utils import \
+                get_get_model_readme_content
+            get_model_readme_content = \
+                get_get_model_readme_content(
+                    self.pipeline_card_artifacts_path)
+        else:
+            from retrain_pipelines.pipeline_card import \
+                    get_model_readme_content
+        #############################
+        from retrain_pipelines.model.hf_utils import \
+            push_model_version_to_hub
+
+        #############################
+        #        model README       #
+        #       from template       #
+        #############################
+        commit_datetime = datetime.utcnow()
+        new_model_version_label = get_new_repo_minor_version(
+            repo_id=self.model_repo_id,
+            repo_type="model",
+            hf_token=os.getenv("HF_TOKEN", None))
+        readme_content = get_model_readme_content(
+            template_folder=template_dir,
+
+            model_repo_id=self.model_repo_id,
+
+            base_model_dict=self.hf_base_model_dict,
+            training_dataset_dict=self.dataset_commit_dict,
+
+            version_label=new_model_version_label,
+            commit_datetime=commit_datetime,
+            perf_metrics=self.perf_metrics,
+
+            mf_flow_name=current.flow_name,
+            mf_run_id=current.run.id
+        )
+        #############################
+
+        print("Pushing model version to HF hub " +
+              ("(blessed). " if self.model_version_blessed
+               else "(not blessed). ") +
+              "May take a while..",
+              flush=True)
+        model_commit_hash = push_model_version_to_hub(
+            repo_id=self.model_repo_id,
+            model_version_blessed=\
+                self.model_version_blessed,
+            version_label=new_model_version_label,
+            timestamp_str=commit_datetime.strftime(
+                "%Y-%m-%d %H:%M:%S UTC"),
+            model_dir=self.sft_model_dir,
+            model_readme_content=readme_content,
+            hf_token=os.getenv("HF_TOKEN", None)
+        )
+        if not model_commit_hash:
+            raise Exception(
+                "Failed to publish model version.")
+        print("Push of model version to HF hub completed.",
+              flush=True)
+        print(f"https://huggingface.co/{self.model_repo_id}" +
+              f"/blob/{model_commit_hash}/README.md")
+
+        self.model_commit_dict = {
+            "repo_id": self.model_repo_id,
+            "commit_hash": model_commit_hash,
+            "version_label": new_model_version_label,
+            "commit_datetime": commit_datetime,
+        }
+
+        self.next(self.infra_validator)
+
+
+    @step
+    def infra_validator(self):
+        """
+        If the trained model version is blessed,
+        validate serving.
+        """
+        """
+        Note that using isolated virtual env
+        (using @conda task decorator)
+        is advisable to not embark the whole
+        pipeline dependencies into the local server.
+        We don't for educational purpose,
+        keep things "simple" to grasp
+        as well as to avoid forcing conda
+        (for instance miniconda) as
+        a virtual environment management mean
+        to the user.
+        """
+        """
+        Note : We load base model from HF-cache
+        (mounted as /huggingface_hub_cache
+        docker volume) and adapter from local dir
+        (mounted as /FuncCallAdater docker volume.
+        """
+
+        self.local_serve_is_ready = LocalServeReadinessEnum.NOT_APPLICABLE
+
+        if self.model_version_blessed:
+            from retrain_pipelines.utils.docker import \
+                env_has_docker
+
+            if env_has_docker():
+                model_module_dir = \
+                    os.path.dirname(
+                        importlib.util.find_spec(
+                            "retrain_pipelines.model." +
+                            os.getenv('retrain_pipeline_type')
+                        ).origin)
+
+                # server & data-model & server-config modules artifacts
+                files_to_copy = [
+                    "litserve_server.py",
+                    "litserve_datamodel.py",
+                    "litserve_serverconfig.py",
+                    ".dockerignore" # docker context loading
+                                    # at image-build time,
+                                    # exclude model weights
+                ]
+                for filename in files_to_copy:
+                    shutil.copy(
+                        os.path.join(model_module_dir, "litserve",
+                                     filename),
+                        os.path.join(self.serving_artifacts_local_folder,
+                                     filename)
+                    )
+
+                # save dependencies as artifact
+                create_requirements(self.serving_artifacts_local_folder,
+                                    exclude=["cudf-polars-.*", "cuda-python",
+                                             "nvidia-.*", "(py)?libcudf-.*",
+                                             "nvtx", "rmm-.*", "litserve",
+                                             ".*retrain-pipelines.*"]
+                )
+
+                # server config yaml
+                env = Environment(loader=FileSystemLoader(
+                    os.path.join(model_module_dir, "litserve")))
+                template = env.get_template(
+                    "litserve_serverconfig_template.yaml")
+                server_config_data = {
+                    "port": "8000",
+                    "max_seq_length": self.max_seq_length,
+                    "max_new_token": self.max_new_tokens,
+                    "base_model": {
+                        "repo_id": self.hf_base_model_dict["repo_id"],
+                        "revision": self.hf_base_model_dict["commit_hash"]
+                    },
+                    "adapters": [
+                        {
+                            "name": "func_caller",
+                            "path": "/FuncCallAdapter"
+                        }
+                    ]
+                }
+                server_config_yaml = template.render(server_config_data)
+                print(server_config_yaml)
+                with open(os.path.join(
+                    self.serving_artifacts_local_folder,
+                    "litserve_serverconfig.yaml"), 'w'
+                ) as output_file:
+                    output_file.write(server_config_yaml)
+
+                # Dockerfile
+                env = Environment(loader=FileSystemLoader(
+                    os.path.join(model_module_dir)))
+                template = env.get_template(
+                    "Dockerfile.litserve_template")
+                # Change CUDA version here from available list
+                # @see https://hub.docker.com/r/nvidia/cuda/tags
+                dockerfile_content = template.render(
+                    {"cuda_version": "12.0.0"})
+                with open(os.path.join(
+                    self.serving_artifacts_local_folder,
+                    "Dockerfile.litserve"), 'w'
+                ) as output_file:
+                    output_file.write(dockerfile_content)
+
+                os.environ["no_proxy"] = "localhost,127.0.0.1,0.0.0.0"
+
+                ############################################
+                #   actually deploy the inference service  #
+                ############################################
+                start_time = time.time()
+                from retrain_pipelines.utils.docker import \
+                    build_and_run_docker, print_container_log_tail, \
+                    cleanup_docker
+                from retrain_pipelines.model.litserve import \
+                    endpoint_started, endpoint_is_ready
+
+                self.port = 8765
+                HF_HUB_CACHE = os.path.realpath(os.path.expanduser(
+                    os.getenv(
+                        "HF_HUB_CACHE",
+                        os.path.join(os.getenv("HF_HOME",
+                                               "~/.cache/huggingface"),
+                                     "hub")
+                    )))
+                print(f"HF_HUB_CACHE : {HF_HUB_CACHE}")
+                image_name = container_name = "litserve-model"
+
+                serving_container = build_and_run_docker(
+                    image_name=image_name, image_tag="1.0",
+                    build_path=self.serving_artifacts_local_folder,
+                    dockerfile="Dockerfile.litserve",
+                    ports_publish_dict={'8000/tcp': self.port},
+                    env_vars_dict={
+                        "HF_HUB_CACHE": "/huggingface_hub_cache",
+                        "HF_TOKEN": os.getenv("HF_TOKEN")
+                    },
+                    volumes_dict={
+                        self.sft_model_dir:
+                            {"bind": "/FuncCallAdapter",
+                             "mode": "ro"},
+                        HF_HUB_CACHE:
+                            {"bind": "/huggingface_hub_cache",
+                             "mode": "ro"}
+                    }
+                )
+
+                if not serving_container:
+                    print("failed spinning the LitServe container",
+                          file=sys.stderr)
+                    self.local_serve_is_ready = \
+                        LocalServeReadinessEnum.FAILURE
+                    try:
+                        cleanup_docker(
+                            container_name=container_name,
+                            image_name=f"{image_name}:1.0",
+                            no_pruning=True # for intermediate layers recycling
+                                            # (during later re-runs)
+                                            # to avoid long rebuild time
+                                            # of exactly the same.
+                        )
+                    except Exception as cleanup_ex:
+                        # fail silently
+                        pass
+                else:
+                    print("Awaiting endpoint launch..")
+                    start_time = time.time()
+                    if not endpoint_started(
+                        container_name, port=self.port, timeout=10*60
+                    ):
+                        print(
+                            f"The endpoint '{container_name}' " +
+                            f"did not start.")
+                        self.local_serve_is_ready = \
+                            LocalServeReadinessEnum.FAILURE
+                    # health check on the spun-up endpoint
+                    elif endpoint_is_ready(port=self.port):
+                        self.local_serve_is_ready = \
+                            LocalServeReadinessEnum.SUCCESS
+                elapsed_time = time.time() - start_time
+                print("deploy_local -   Elapsed time: " +
+                      f"{elapsed_time:.2f} seconds")
+                ############################################
+            else:
+                # env doesn't have docker
+                self.local_serve_is_ready = \
+                    LocalServeReadinessEnum.FAILURE_NO_DOCKER
+
+            if LocalServeReadinessEnum.SUCCESS == self.local_serve_is_ready:
+                from retrain_pipelines.model.litserve.litserve_datamodel \
+                    import Response
+
+                import requests
+
+                url = f"http://localhost:{self.port}/predict"
+                headers = {"accept": "application/x-www-form-urlencoded"}
+
+                try:
+                    start_time = time.time()
+                    data = {
+                        "adapter_name": "func_caller",
+                        "queries": '["Hello.", "Is 49 a perfect square?"]'
+                    }
+                    print(f"inference test - data: {data}")
+                    response = requests.post(url, headers=headers, data=data)
+                    parsed_response = Response(**{"output": response.json()})
+                    elapsed_time = time.time() - start_time
+                    print("parsed_response ('func_caller' adapter ON) :" +
+                          str(parsed_response) +
+                          f"\t-\tElapsed time: {elapsed_time:.2f} seconds")
+
+                    start_time = time.time()
+                    data = {
+                        "queries": '["Hello.", "Is 49 a perfect square?"]'
+                    }
+                    print(f"inference test - data: {data}")
+                    response = requests.post(url, headers=headers, data=data)
+                    parsed_response = Response(**{"output": response.json()})
+                    elapsed_time = time.time() - start_time
+                    print(f"parsed_response (no adapter) : {parsed_response}" +
+                          f"\t-\tElapsed time: {elapsed_time:.2f} seconds")
+
+                except Exception as ex:
+                    print(ex, file=sys.stderr)
+                    traceback.print_tb(ex.__traceback__, file=sys.stderr)
+                    self.local_serve_is_ready = \
+                        LocalServeReadinessEnum.FAILURE
+                    pass
+
+            try:
+                cleanup_docker(
+                    container_name=container_name,
+                    image_name=f"{image_name}:1.0",
+                    no_pruning=True # for intermediate layers recycling
+                                    # (during later re-runs)
+                                    # to avoid long rebuild time
+                                    # of exactly the same.
+                )
+            except Exception as cleanup_ex:
+                # fail silently
+                pass
+
+        self.next(self.pipeline_card)
+
+
+    @card(id='default')
+    @card(type='html', id='custom')
+    @step
+    def pipeline_card(self):
+        import re
+        import datetime
+        import importlib.metadata
+
+        #############################
+        #   case of user-provided   #
+        # documentation artifact(s) #
+        #############################
+        # note that user can provide either
+        # 'pipeline_card.py' or 'template.html'
+        # or 'dataset_readme.py'
+        # or 'dataset_readme_template.md'
+        # or 'model_readme.py'
+        # or 'model_readme_template.md'
+        # or any combination of those
+        # when specifying custom
+        # 'pipeline_card_artifacts_path'
+        if "template.html" in os.listdir(
+                                self.pipeline_card_artifacts_path
+        ):
+            template_dir = self.pipeline_card_artifacts_path
+        else:
+            template_dir = os.path.dirname(
+                importlib.util.find_spec(
+                    f"retrain_pipelines.pipeline_card."+
+                    f"{os.getenv('retrain_pipeline_type')}"
+                ).origin)
+        #############################
+        if "pipeline_card.py" in os.listdir(
+                                    self.pipeline_card_artifacts_path
+        ):
+            from retrain_pipelines.utils import get_get_html
+            get_html = \
+                get_get_html(self.pipeline_card_artifacts_path)
+        else:
+            from retrain_pipelines.pipeline_card import \
+                    get_html
+        from retrain_pipelines.pipeline_card.helpers import \
+                mf_dag_svg
+        #############################
+
+
+        #############################
+        ##      "default" card     ##
+        #############################
+        self.metadata = {
+            "name": "TabNet Model",
+            "version": "1.0",
+            "retrain_pipelines": f"retrain-pipelines {__version__}",
+            "retrain_pipeline_type": os.environ["retrain_pipeline_type"],
+            "description": "A PyTorch TabNet model retrained",
+            "authors": [current.username],
+            "tags": ["classification", "tabnet"],
+            "license": "MIT License",
+            "data_augmentation": [
+                {
+                    "name": "Augmentation",
+                    "description": "Truncating queries and " + \
+                                   "associate those to " + \
+                                   "no tool-call answers. " + \
+                                   "Intent being to instruct on " + \
+                                   "not hallucinating missing " + \
+                                   "tool-calls parameters values."
+                },
+                {
+                    "name": "Enrichment",
+                    "description": "Addition of records " + \
+                                   "from an external data-source. " + \
+                                   "Here to instruct on no tool-call."
+                }
+            ],
+            "references": [
+                {
+                    "title": "Base model",
+                    "link": f"https://hf.co/{self.hf_base_model_dict['repo_id']}"
+                },
+                {
+                    "title": "Function-calling dataset",
+                    "link": f"https://hf.co/{self.hf_dataset_dict['repo_id']}"
+                },
+                {
+                    "title": "Data-enrichment dataset",
+                    "link": f"https://hf.co/{self.hf_enrich_dataset_dict['repo_id']}"
+                },
+                {
+                    "title": "Unsloth",
+                    "link": "https://unsloth.ai/blog/contpretraining"
+                }
+            ]
+        }
+
+        current.card['default'].append(Markdown(
+            "model_version_blessed : **%s**" % str(self.model_version_blessed)))
+        current.card['default'].append(Artifact(
+            {"model_version_blessed": self.model_version_blessed}))
+
+        current.card['default'].append(
+            Image.from_matplotlib(self.sft_log_history_fig))
+        current.card['default'].append(
+            Image.from_matplotlib(self.validation_completions_fig))
+        #############################
+
+        #############################
+        ##    html "custom" card   ##
+        #############################
+        dt = datetime.datetime.now(tz=datetime.timezone.utc)
+        formatted_dt = dt.strftime("%A %b %d %Y %I:%M:%S %p %Z")
+        task_obj_python_cmd = f"metaflow.Task(" + \
+            f"\"{current.pathspec}\", " + \
+            f"attempt={str(current.retry_count)})"
+        params={
+            'template_dir': template_dir,
+            'title': f"{current.flow_name}",
+            "subtitle": f"(flow run # {len(list(current.run.parent.runs()))}," + \
+                        f" run_id: {str(current.run.id)}  -  {formatted_dt})",
+            'model_version_blessed': self.model_version_blessed,
+            # 'current_blessed_run': self.current_blessed_run,
+            'current_blessed_model_commit_hash': (
+                self.current_blessed_version_dict["commit_hash"]
+                if self.current_blessed_version_dict
+                else None
+            ),
+            'LocalServeReadinessEnum': LocalServeReadinessEnum,
+            'local_serve_is_ready': self.local_serve_is_ready,
+            # EDA
+            'main_dataset_repo_id': self.hf_dataset['repo_id'],
+            'main_dataset_commit_hash': self.hf_dataset_dict['commit_hash'],
+            'main_dataset_commit_datetime': \
+                self.hf_dataset_dict['commit_datetime'],
+
+            'records_count': self.records_count,
+            'data_schema': self.data_schema,
+            'answers_tools_count_fig': self.answers_tools_count_fig,
+            'words_count_fig': self.words_count_fig,
+
+            # model training
+            'dataset_repo_id': self.dataset_repo_id,
+            'dataset_version_label': self.dataset_commit_dict["version_label"],
+            'dataset_commit_datetime': self.dataset_commit_dict["commit_datetime"],
+            'dataset_commit_hash': self.dataset_commit_dict["commit_hash"],
+            'dataset_augmentation_rate': self.actual_augmentation_rate,
+            'dataset_enrichment_rate': self.enrichment_rate,
+
+            'model_repo_id': self.model_repo_id,
+            'model_version_label': self.model_commit_dict["version_label"],
+            'model_commit_datetime': self.model_commit_dict["commit_datetime"],
+            'model_commit_hash': self.model_commit_dict["commit_hash"],
+
+            'cpt_log_history_fig': self.cpt_log_history_fig,
+            'sft_log_history_fig': self.sft_log_history_fig,
+
+            'validation_completions_fig': self.validation_completions_fig,
+
+            'pipeline_parameters_dict': {"cpt": self.cpt_training_args,
+                                         "sft": self.sft_training_args},
+
+            'metrics_dict': self.perf_metrics,
+
+            'task_obj_python_cmd': task_obj_python_cmd,
+            'dag_svg': mf_dag_svg(self)
+        }
+        self.html = get_html(params)
+        #############################
+        current
+        #############################
+
+        self.next(self.pipeline_to_hub)
+
+
+    @step
+    def pipeline_to_hub(self):
+        """
+        publish versioned source-code and pipeline-card
+        for ths run on the Hugging Face Hub.
+        """
+
+        model_commit_datetime = \
+            self.model_commit_dict["commit_datetime"]
+        timestamp_str = \
+            "{:%Y%m%d_%H%M%S}".format(model_commit_datetime) + \
+            "{:03d}".format(model_commit_datetime.microsecond//1000) + \
+            "_UTC"
+        subfolder_name = \
+            "v" + self.model_commit_dict["version_label"] + \
+            "_" + timestamp_str
+        commit_datetime = datetime.utcnow()
+
+        ###############################
+        #         source-code         #
+        ###############################
+        # We upload only herein file  #
+        # plus user-provided versions #
+        # of the customizable ones    #
+        # (if any).                   #
+        ###############################
+        custom_source_files = [os.path.abspath(__file__)]
+        if (
+            self.pipeline_card_artifacts_path != \
+            self.default_pipeline_card_module_dir
+        ):
+            candidate_source_files = [
+                "pipeline_card.py",
+                "template.html",
+                "dataset_readme.py",
+                "dataset_readme_template.md",
+                "model_readme.py",
+                "model_readme_template.md"
+            ]
+            for candidate_source_file in candidate_source_files:
+                file_fullpath = os.path.join(
+                    self.pipeline_card_artifacts_path,
+                    candidate_source_file)
+                if os.path.exists(file_fullpath):
+                    custom_source_files.append(file_fullpath)
+
+        source_code_commit_hash = \
+            push_files_to_hub_repo_branch(
+                repo_id=self.model_repo_id,
+                branch_name="retrain-pipelines_source-code",
+                file_fullnames=custom_source_files,
+                include_requirements_txt=True,
+                path_in_repo=subfolder_name,
+                commit_message=\
+                    "source-code for model version " + \
+                    subfolder_name + \
+                    f"- retrain-pipelines {__version__}",
+                repo_type="model",
+                hf_token=os.getenv("HF_TOKEN", None)
+            )
+        print(source_code_commit_hash)
+        self.source_code_commit_dict = {
+            "repo_id": self.model_repo_id,
+            "branch_name": "retrain-pipelines_source-code",
+            "commit_datetime": commit_datetime,
+            "commit_hash": source_code_commit_hash
+        }
+        ###############################
+
+        ###############################
+        #        pipeline-card        #
+        ###############################
+        pipeline_card_fullname = None
+        for run_step in current.run.steps():
+            task = list(run_step.tasks())[0]
+            task_name = task.path_components[2]
+            if "pipeline_card" == task_name:
+                pipeline_card = get_cards(
+                    task, id='custom', type='html')[0]
+                pipeline_card_fullname = os.path.realpath(
+                    os.path.join(
+                        task.metadata_dict.get("ds-root", None),
+                        mf_config.CARD_SUFFIX, pipeline_card.path
+                    ))
+                print(pipeline_card_fullname)
+                break
+        pipeline_card_commit_hash = \
+            push_files_to_hub_repo_branch(
+                repo_id=self.model_repo_id,
+                branch_name="retrain-pipelines_pipeline-card",
+                file_fullnames=[pipeline_card_fullname],
+                path_in_repo=subfolder_name,
+                commit_message=\
+                    "pipeline-card for model version " + \
+                    subfolder_name + \
+                    f"- retrain-pipelines {__version__}",
+                repo_type="model",
+                hf_token=os.getenv("HF_TOKEN", None)
+            )
+        print(pipeline_card_commit_hash)
+        self.pipeline_card_commit_dict = {
+            "repo_id": self.model_repo_id,
+            "branch_name": "retrain-pipelines_pipeline-card",
+            "commit_datetime": commit_datetime,
+            "commit_hash": pipeline_card_commit_hash
+        }
+        ###############################
+
+        self.next(self.deploy)
+
+
+    @step
+    def deploy(self):
+        """
+        placeholder for the serving SDK deploy call
+        (on the target production platform).
+        Include any artifact you want,
+        consider including the portable pipelione-card
+        itself !
+        """
+
+        if (
+            self.model_version_blessed and
+            (self.local_serve_is_ready == LocalServeReadinessEnum.SUCCESS)
+        ):
+            pass # your code here
+
+        self.next(self.load_test)
+
+
+    @step
+    def load_test(self):
+        """
+        placeholder
+        """
+
+        if (
+            self.model_version_blessed and
+            (self.local_serve_is_ready == LocalServeReadinessEnum.SUCCESS)
+        ):
+            pass # your code here
+
+        self.next(self.end)
+
+
+    @step
+    def end(self):
+        pass
+
+
+if __name__ == "__main__":
+    UnslothFuncCallFlow()
+