Run nightly-retrain completed. Duration 6m 2s. That line sat green in the dashboard and it was a lie. A retrain that ingests docs, generates synthetic data and tunes an 8B model does not finish in six minutes, and mine had not. A caching hit on every step meant the pipeline re-used last week’s artifacts, skipped the tuning entirely, and registered a model no better than the one it replaced. Automating the assistant’s retrain is the right move, and this part builds it as a data science pipeline on OpenShift AI, but a pipeline hides a skipped step as neatly as it hides toil, and knowing where it lies to you is half the job.

Where the assistant stands now

Last part moved the tuned Granite 3.1 8B onto a shared OpenShift AI cluster and turned on the pipelines component, but nothing runs there yet. Up to now every retrain was hand driven: someone SSHed into the RHEL AI box, ran ilab data generate from Part 9, kicked off the multi phase tune from Part 10, scored it as in Part 11, and copied the checkpoint across by hand. That worked for one person and broke the week two teams wanted it. This part turns those four manual steps into one graph that runs on a schedule, reads the company docs from object storage, and stops itself if the tuned model does not beat a bar. Same InstructLab work, no human in the loop.

One naming point saves confusion later. OpenShift AI 3.x renamed this feature to AI pipelines in the dashboard, but the component key in your DataScienceCluster is still datasciencepipelines and the custom resource is still a DataSciencePipelinesApplication. Red Hat did not write the engine either. Underneath sits Kubeflow Pipelines 2.0, an upstream project, running on Argo Workflows 3.4 as its executor; Red Hat’s contribution is the operator that installs it, the security backports, and the support line when a run wedges. If a tutorial talks about kfp-tekton, it predates the Argo switch and its code will not run here.

Pipelines, components and the Argo backend

A pipeline is a directed acyclic graph, a DAG, where each node is a component and each edge is a piece of data. A component, in kfp terms, is a function that runs inside its own container image with its own dependencies. That container boundary is the whole point: your prepare step can run pandas on four CPUs while your tune step runs on a GPU, and neither has to share a Python environment. Argo Workflows is what actually schedules those containers as Kubernetes pods and passes artifacts between them. You will never write Argo YAML by hand, and Red Hat does not support you poking at the embedded Argo directly, but it helps to know that is what a running pipeline is underneath.

Before any code, look at where a real nightly run spends its time, because it reframes what is worth optimising. Ingest, prepare and register are seconds to minutes. Synthetic data generation is tens of minutes. Tuning dominates everything. The chart below is one run of the assistant’s retrain on a single L40S, and the shape holds across runs.

Object storage the pipeline server needs

A pipeline server has nowhere to put artifacts until you give it an S3 bucket. This is the CrashLoopBackOff from Part 13’s failure table, and the fix is a DataSciencePipelinesApplication pointed at real storage. Create the credentials as a secret first, reading the keys from your vault or environment, and put nothing sensitive in the manifest.

# Tested on Red Hat OpenShift AI 3.4 self managed (2.25.7 is the last 2.x)
# Backend: Kubeflow Pipelines 2.0 on Argo Workflows 3.4, kfp Python SDK 2.x
# S3 keys come from the environment, never typed into a file that lands in git
$ oc create secret generic pipeline-s3-creds 
    --from-literal=AWS_ACCESS_KEY_ID=$S3_KEY 
    --from-literal=AWS_SECRET_ACCESS_KEY=$S3_SECRET 
    -n support-assistant
secret/pipeline-s3-creds created

# dspa.yaml, wire the pipeline server to real S3, apiVersion is v1 not v1alpha1
apiVersion: datasciencepipelinesapplications.opendatahub.io/v1
kind: DataSciencePipelinesApplication
metadata:
  name: dspa
  namespace: support-assistant
spec:
  dspVersion: v2
  objectStorage:
    externalStorage:
      bucket: assistant-pipelines
      host: s3.us-east-1.amazonaws.com
      scheme: https
      region: us-east-1
      s3CredentialsSecret:
        secretName: pipeline-s3-creds
        accessKey: AWS_ACCESS_KEY_ID
        secretKey: AWS_SECRET_ACCESS_KEY

$ oc apply -f dspa.yaml
datasciencepipelinesapplication.opendatahub.io/dspa created
$ oc get pods -n support-assistant | grep ds-pipeline
ds-pipeline-dspa-6c9f7d8b9-4wq2t          1/1   Running
ds-pipeline-persistenceagent-dspa-...     1/1   Running
ds-pipeline-scheduledworkflow-dspa-...    1/1   Running
ds-pipeline-metadata-envoy-dspa-...       1/1   Running

Building the retrain pipeline in Python

Components come first. Each is a plain Python function wearing a @dsl.component decorator that names its base image and any extra packages. Data enters and leaves through typed artifact parameters, Input and Output of Dataset, Model or Metrics. You write to artifact.path, a local file path the runtime backs with object storage, and the next component reads from the same path. That indirection is the rule that trips up everyone new, and the failure section shows exactly how.

# retrain_pipeline.py, kfp 2.x, one container per step, data moves as artifacts
from kfp import dsl
from kfp.dsl import Input, Output, Dataset, Model, Metrics

UBI = 'registry.redhat.io/ubi9/python-311'

@dsl.component(base_image=UBI, packages_to_install=['pandas==2.2.2'])
def prepare_corpus(raw: Input[Dataset], clean: Output[Dataset]):
    import pandas as pd
    df = pd.read_parquet(raw.path)
    df = df.dropna(subset=['question', 'answer']).drop_duplicates('question')
    df.to_parquet(clean.path)          # write to the artifact path, not a return
    print(f'prepared {len(df)} qa pairs')

@dsl.component(base_image=UBI, packages_to_install=['instructlab-sdg'])  # [VERIFY] pin
def generate_sdg(clean: Input[Dataset], synthetic: Output[Dataset]):
    # wraps the same SDG flow as ilab data generate from Part 9
    ...

@dsl.component(base_image=UBI)
def evaluate(model: Input[Model], scores: Output[Metrics]) -> float:
    mt_bench = run_mt_bench(model.path)   # scoring harness from Part 11
    scores.log_metric('mt_bench', mt_bench)
    return mt_bench

The pipeline function wires those components into the DAG. Outputs of one become inputs of the next, resource requests go on the step that needs them, and a conditional gates the register step on the evaluation score so a weak model never reaches the registry. Caching is switched off on the tune step on purpose, for reasons the opener already made painful.

@dsl.pipeline(name='nightly-retrain',
              description='ingest, sdg, tune and evaluate the support assistant')
def retrain(bucket_prefix: str, min_mt_bench: float = 6.5):
    ing  = ingest_docs(prefix=bucket_prefix)
    prep = prepare_corpus(raw=ing.outputs['out'])
    sdg  = generate_sdg(clean=prep.outputs['clean'])
    tune = train_granite(data=sdg.outputs['synthetic'])
    tune.set_accelerator_type('nvidia.com/gpu').set_accelerator_limit(1)
    tune.set_caching_options(False)          # never re-use a stale tune
    ev = evaluate(model=tune.outputs['model'])
    with dsl.If(ev.output >= min_mt_bench):
        register_model(model=tune.outputs['model'])

from kfp import compiler
compiler.Compiler().compile(retrain, 'nightly-retrain.yaml')

flowchart LR
  I[ingest docs from S3] --> P[prepare corpus]
  P --> S[generate synthetic data]
  S --> T[tune Granite 3.1 8B]
  T --> E[evaluate MT Bench]
  E -->|score at or above 6.5| R[register model]
  E -->|score below 6.5| X[stop and keep old model]

Compiling and submitting the run

Compiling turns the Python into an intermediate YAML the pipeline server understands. Submitting sends that YAML to the server over its route, authenticated with the workbench’s own service account token, which is mounted into every notebook pod. No static key, no password in the code.

# submit from a workbench, token read from the mounted service account
import kfp
token = open('/var/run/secrets/kubernetes.io/serviceaccount/token').read()
client = kfp.Client(
    host='https://ds-pipeline-dspa-support-assistant.apps.example.com',
    existing_token=token,
)
run = client.create_run_from_pipeline_package(
    'nightly-retrain.yaml',
    arguments={'bucket_prefix': 's3://assistant-pipelines/2026-07/'},
)
print(run.run_id)
# 3f2a9c14-8b0e-4d77-9a51-1c6e0b2f4d88

My first attempt never reached the server. I had written prepare_corpus to return the cleaned DataFrame straight out of the function, the way you would in a notebook, and compilation refused it.

$ python retrain_pipeline.py
Traceback (most recent call last):
  File 'retrain_pipeline.py', line 12, in <module>
  File '.../kfp/dsl/component_factory.py', line 402, in ...
TypeError: Unsupported return type annotation: <class
'pandas.core.frame.DataFrame'>. A component return must be a parameter
type (int, str, bool, float, list, dict) or an artifact.

Here is the mechanism, and it is the single most useful thing to internalise about kfp. Components run in separate pods that share no memory and no disk. kfp moves data between them by writing an artifact to object storage and passing a path, so a return value has to be either a tiny parameter it can serialise into metadata or a file written to Output[...].path. A 1.2 GB corpus is neither a tiny parameter nor a serialisable return; it is a Dataset artifact. Switching the signature to clean: Output[Dataset] and writing the parquet to clean.path fixed it, and the same run that had crashed at compile went through to the server in one call.

Where pipeline runs break

Keep this lookup next to the notebook. Six symptoms cover almost every failed run I have hit on this platform, and each maps to a one line cause and fix.

Resource requests belong on the step, not the pipeline, and this is where a run silently underperforms. If you forget the accelerator limit on the tune step it schedules on a CPU node and runs for a day instead of failing loudly. The table below is the resource shape for the assistant’s retrain, and it is worth copying into any pipeline that mixes cheap and expensive steps.

Make it reproducible before you schedule it

Scheduling is the easy part. A DataSciencePipelinesApplication supports a recurring run, and the dashboard will let you set the nightly cadence in two clicks, so I will not spend a code block on it. What earns the schedule is everything before it: pinned base images, a bucket prefix that points at the current corpus and not a stale one, caching off on every step that must run fresh, and an evaluation gate that stops a bad model reaching the registry. Get those right and a nightly run is safe to ignore. Get the corpus prefix wrong and you retrain on last month’s data for a week before anyone notices, which is a mistake I watched cost real trust in the assistant’s answers. The pipeline you just built is also the same object a CI system triggers, which is where the wider practice of CI and CD for ML pipelines picks up, and the evaluation gate here is the InstructLab specific version of building an eval set before you build the feature.

References

• Red Hat OpenShift AI, working with data science pipelines, KFP 2.0 and DSPA
• Red Hat blog, what is new with data science pipelines, Argo Workflows 3.4
• Kubeflow Pipelines, upstream SDK and component documentation

About The Author

Dr. Pranay Jha

Dr. Pranay Jha is a Cloud and AI Consultant with 18+ years of experience in hybrid cloud, virtualization, and enterprise infrastructure transformation. He specializes in VMware technologies, multi-cloud strategy, and Generative AI solutions. He holds a PhD in Computer Applications with research focused on Cloud and AI, has published multiple research papers, and has been a VMware vExpert since 2016 and a VMUG Community Leader.

See author's posts