Automating the End-to-End Workflow for AI Training Data and Model Files

Getting Started

Core Concept

Automatically send data and model files to the systems that need them

In AI workflows, a wide range of files—including raw data, preprocessing results, training data, model files, and inference results—are used across multiple systems and environments.

By connecting where each file is created with the system that uses it, you can send data to the next processing environment as soon as it is ready, then route generated models and result files to the locations where they are needed.

Data Source
     │
     ▼
Data Processing
     │
     ▼
AI Training
     │
     ▼
Model Output
     │
     ▼
Result Storage

With this setup, AI data and model files can be managed according to the processing environment at each stage, with each file automatically flowing into the next task as soon as it is ready.

Workflow Flow

Connecting the workflow from data collection through processing and result utilization

Collect data generated across multiple systems and send it to the AI processing environment where it is needed.

Once data processing and training are complete, send the generated models and result files to the next system or storage environment for downstream use.

Data Generation
     │
     ▼
Data Collection
     │
     ▼
Preprocessing · AI Processing
     │
     ▼
Model Generation
     │
     ├──────────────┐
     ▼              ▼
Model Storage        Result Delivery
     │              │
     └──────┬───────┘
            ▼
        Next Task

Each stage can use the files and processing results generated by the previous stage as inputs for the next stage.

Automation Benefits

Reduce repetitive large-file movement and management at every stage

As AI workloads grow in data volume and processing stages, the work required to prepare files and send them to the systems that need them grows as well.

By configuring the next task to run automatically based on file creation and processing status, you can connect each stage's file flow in a defined sequence.

This lets you define the flow of data, models, and result files stage by stage and manage the entire AI workflow as a single file workflow.

IT Engineer

Configure an automated workflow for AI data and model files

Data Collection

Bring data from multiple systems into a single workflow

First, connect the systems where data for AI processing is generated or stored to the workflow.

Set each system's file paths and collection targets to bring data from multiple locations into a single processing flow for use in the next stage.

Source A ──┐
           │
Source B ──┼──→ Data Collection
           │
Source C ──┘
                  │
                  ▼
             AI Processing

[Product UI: Flow Canvas screen connecting multiple Source systems and data folders to a single data collection task]

If needed, set conditions such as file paths, names, and extensions to collect only the data required for AI processing.

Processing Connection

Send collected data to the AI processing systems that need it

Once the collected data is ready, connect it to the AI processing systems that perform the next tasks, such as preprocessing, training, or inference.

Specify the systems and file paths used at each processing stage, and configure the output from one stage to become the input for the next system.

Collected Data
       │
       ▼
Preprocessing
       │
       ▼
Training Server
       │
       ▼
Inference / Analysis

With this setup, the files required at each AI processing stage can be automatically sent to the prepared working environment.

Job Conditions

Start the next task based on data and processing status

Each workflow stage can be configured to start the next task when a defined condition is met, such as file creation, completion of data collection, or completion of processing.

For example, you can start preprocessing once data has been collected from multiple systems, then continue to the training task as soon as the preprocessing results are generated.

Data Ready
    │
    ▼
Collection Complete
    │
    ▼
Start Processing
    │
    ▼
Processing Complete
    │
    ▼
Start Training

By linking job conditions, you can run the workflow in sequence from data collection through AI processing and model generation based on the processing status at each stage.

Result Transfer

Send generated models and processing results to the systems that need them

Once training and processing are complete, send the generated model files and result data to the next environment.

Connect the locations where files will be used—such as model storage, inference systems, validation environments, and business systems—and send each output to the appropriate destination.

AI Training
     │
     ▼
Model Generated
     │
 ┌───┴───────────┐
 ▼               ▼
Model Storage  Inference Server
                     │
                     ▼
                Result Output
                     │
                     ▼
               Target System

When a model or result file is used across multiple environments, you can connect destination-specific file flows and automatically extend the workflow to every required working location.

Run Management

Track processing status stage by stage, from data to models and results

For each workflow run, you can view the overall flow together with the status of each stage.

You can track processed files and progress step by step across data collection, preprocessing, AI processing, model generation, and result transfer.

AI Workflow
     │
     ├── Data Collection   ✓
     │
     ├── Processing        ✓
     │
     ├── Model Training    ●
     │
     └── Result Transfer   ○

Key items to check include:

This lets you manage the status of a specific stage alongside the overall AI file workflow.

Exception Handling

Check transfer and processing status, then rerun the stages that need attention

When a stage requires additional investigation during execution, review the data status, system connections, and file transfer results using the task details and execution history.

You can review each stage separately from data collection through AI processing and result transfer, then rerun the required task.

Workflow Run
      │
      ▼
Stage Status
      │
 ┌────┴─────┐
 ▼          ▼
Completed   Check Required
                │
                ▼
           Run Details
                │
       ┌────────┼────────┐
       ▼        ▼        ▼
     Data     Device    Transfer
                │
                ▼
           Retry Stage
                │
                ▼
          Result Check

Developer

Deploy datasets to multiple training nodes and collect node-specific results on a storage system

Integration Setup

Prepare shared API call code and path conventions

import os
import requests

BASE_URL = os.getenv("INNORIX_BASE_URL", "https://app.innorix.com").rstrip("/")
TOKEN = os.environ["INNORIX_ACCESS_TOKEN"]
WORKSPACE_ID = os.getenv("INNORIX_WORKSPACE_ID")   # optional; falls back to the current workspace

STATUS_COMPLETE = 2
TERMINAL = {2, 4, 5, 9, 99}          # complete / error / cancelled / partial / failed
NOT_SUCCEEDED = {4, 5, 9, 99}


def api(method, path, body=None, params=None):
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {TOKEN}",
    }

    if WORKSPACE_ID:
        headers["x-workspace-id"] = WORKSPACE_ID

    response = requests.request(
        method, BASE_URL + path,
        headers=headers, json=body, params=params, timeout=30,
    )

    payload = response.json() if response.content else {}

    if not response.ok:
        raise RuntimeError(payload.get("message") or f"HTTP {response.status_code}")

    return payload.get("data")


def is_terminal(detail):
    return detail.get("isTerminal", detail.get("status") in TERMINAL)

import base64
import time


def encode_path(device_id, raw_path):
    normalized = str(raw_path or "").replace("\\", "/")
    token = base64.b64encode(normalized.encode("utf-8")).decode("ascii")
    return f"{device_id}_ino_{token}"


def now_iso():
    return time.strftime("%Y-%m-%dT%H:%M:%S.000Z", time.gmtime())

Transfer status is determined using the values below. There are five terminal states, and the value representing success is Complete (2).

Path Rules

Include the dataset version and run identifier in the path

As runs accumulate, it becomes difficult to tell which result came from which dataset. By standardizing path rules in functions, you can avoid assembling path strings throughout the code.

DATA_ROOT = "/data"
WORK_ROOT = "/work"
ARCHIVE_ROOT = "/archive"


# source dataset on the storage device
def dataset_path(dataset, version):
    return f"{DATA_ROOT}/datasets/{dataset}/{version}"


# path the training job reads on the node
def node_input_path(dataset, version, run_id):
    return f"{WORK_ROOT}/{run_id}/input/{dataset}/{version}"


# path the node writes its results to
def node_output_path(run_id):
    return f"{WORK_ROOT}/{run_id}/output"


# archive path where results are collected
def archive_path(run_id):
    return f"{ARCHIVE_ROOT}/runs/{run_id}"


run_id = f"r-{time.strftime('%Y%m%d-%H%M%S')}"

Putting the run identifier at the beginning of the path makes it easy to delete or move an entire run at once. Using only the date causes two runs on the same day to get mixed together, while using only the dataset name makes it difficult to distinguish results after a version change.

Data Deployment

Send a dataset to multiple nodes and confirm that every node has received it

One transfer targets one destination. If there are multiple nodes, there are multiple transfers.

def deploy(storage, nodes, dataset, version, run_id):
    source_path = dataset_path(dataset, version)
    target_path = node_input_path(dataset, version, run_id)

    return {
        node: api("POST", "/api/transfers/manual", {
            "sourceDevice": storage,
            "targetDevice": node,
            "targetPath": target_path,
            "sourcePaths": [source_path],
            "sendAllFolder": True,
            "checkIntegrity": True,
            "transferOptions": {"target-action": "overwrite"},
        })["monitorId"]
        for node in nodes
    }

def wait(monitor_id, timeout=3600, interval=3):
    deadline = time.time() + timeout

    while time.time() < deadline:
        detail = api("GET", f"/api/transfers/{monitor_id}")

        if is_terminal(detail):
            return detail

        time.sleep(interval)

    raise TimeoutError(monitor_id)


def failed_files(monitor_id):
    result = api("GET", f"/api/transfers/{monitor_id}/files", params={
        "state": "any", "size": 500,
    }) or {}

    return [r for r in (result.get("children") or [])
            if r.get("status") in NOT_SUCCEEDED]


def retry_failed(monitor_id):
    rows = failed_files(monitor_id)

    if not rows:
        return 0

    api("POST", f"/api/transfers/{monitor_id}/retry", {
        "filesRetry": [
            {"filePath": r["sourceFilePath"], "isDir": bool(r.get("isFolder"))}
            for r in rows
        ]
    })

    return len(rows)

Do not start training if even one node has not received the data. Check every node instead of stopping at the first failure so you can determine the correct retry scope.

transfers = deploy(STORAGE, NODES, "imagenet", "v3", run_id)
results = {node: wait(mid) for node, mid in transfers.items()}

incomplete = [n for n, d in results.items() if d["status"] != STATUS_COMPLETE]

if incomplete:
    raise RuntimeError(f"nodes not fully delivered: {', '.join(incomplete)}")

Use overwrite as the arrival policy. With numbering, copies accumulate on the nodes, making it unclear which copy the training code should read.

Delta Transfer

Send only the data that changed since the last deployment

monitor_id = api("POST", "/api/transfers/manual", {
    "sourceDevice": STORAGE,
    "targetDevice": node,
    "targetPath": node_input_path("imagenet", "v3", run_id),
    "sourcePaths": [dataset_path("imagenet", "v3")],
    "sendAllFolder": True,
    "incremental": True,
    "transferOptions": {"target-action": "overwrite"},
})["monitorId"]

detail = wait(monitor_id)
print(f"transferred {detail['fileCount']} files, {detail['totalSize']} bytes")

It is disabled by default. The node's agent calculates the changes and transfers only files that were added or modified since the last deployment.

Always use overwrite for incremental transfers. With numbering, changed files accumulate with numbered copies, and the training code may continue reading the old files.

Integrity Verification

Verify that the transferred data matches the original

Integrity verification has two steps. Start verification with POST /api/transfers/{monitorId}/verification, then poll the result with GET while the server processes it.

def verify(monitor_id, timeout=1800, interval=10):
    api("POST", f"/api/transfers/{monitor_id}/verification", {})

    deadline = time.time() + timeout

    while time.time() < deadline:
        result = api("GET", f"/api/transfers/{monitor_id}/verification") or {}

        if result.get("verified"):
            return result

        time.sleep(interval)

    raise TimeoutError(f"verification: {monitor_id}")


result = verify(monitor_id)

if result["sourceFileCount"] != result["targetFileCount"]:
    raise RuntimeError("file counts differ - part of the transfer is missing")

if not result["checksumMatched"]:
    for row in result.get("mismatchedFiles") or []:
        print("mismatch:", row)

If the file counts differ, the transfer is incomplete. If the counts match but there are mismatches, the contents were corrupted. The former can be resolved by retransmission; the latter requires finding the cause.

Result Collection

Collect node-specific outputs on a storage system without collisions

Every node uses the same filename, such as model.pt. If everything is collected into one path, the files will overwrite one another.

def collect(nodes, archive, run_id):
    transfers = {}

    for node in nodes:
        target_path = f"{archive_path(run_id)}/{node}"

        transfers[node] = api("POST", "/api/transfers/manual", {
            "sourceDevice": node,
            "targetDevice": archive,
            "targetPath": target_path,
            "sourcePaths": [node_output_path(run_id)],
            "sendAllFolder": True,
            "checkIntegrity": True,
            "transferOptions": {"target-action": "overwrite"},
        })["monitorId"]

    return transfers

/archive/runs/r-20260901-0200/
    dev-gpu-01/
        model.pt
        metrics.json
    dev-gpu-02/
        model.pt
        metrics.json

Including the node identifier in the destination path is the simplest approach. If you avoid collisions with numbering, you cannot tell which node model_1.pt came from.

Collect metrics and logs along with the model files. The cause of a failed training run is recorded in the logs, but those logs will be deleted when the node is reclaimed.

Connect Deployment and Collection

Automatically start collection when deployment finishes

The code above waits for the client to finish before calling the next step. If training takes several hours, the process must stay alive the entire time, and collection will not happen if it dies.

import uuid

flow_id = str(uuid.uuid4())


def build_step(name, source, source_path, target, target_path,
               step, trigger_id=None, webhook=None):
    schedule = {
        "type": "none",
        "startDateType": "now",
        "startDate": now_iso(),
        "timezone": "Asia/Seoul",
    }

    if trigger_id:
        schedule["triggerAutomation"] = {"value": trigger_id}

    body = {
        "name": name,
        "flowName": name,
        "flowId": flow_id,
        "transferType": "normal",
        "timezone": "Asia/Seoul",
        "step": step,
        "isUpcoming": False,
        "details": [
            {
                "senderId": source,
                "receiverId": target,
                "sourceItem": [
                    {
                        "hash": encode_path(source, source_path),
                        "filePath": source_path,
                        "isDir": True,
                    }
                ],
                "targetPath": encode_path(target, target_path),
                "step": step,
                "transferOptions": {
                    "noSchedule": False,
                    "target-action": "overwrite",
                    "send-fileoption": {},
                },
            }
        ],
        "schedules": [schedule],
    }

    if webhook:
        body["processors"] = [{
            "category": "run",
            "type": "http",
            "config": {"url": webhook, "method": "POST"},
        }]

    return body


deploy_id = api("POST", "/api/automations", build_step(
    f"deploy imagenet:v3", STORAGE, dataset_path("imagenet", "v3"),
    node, node_input_path("imagenet", "v3", run_id),
    step=1, webhook=TRAIN_HOOK))["automationId"]

collect_id = api("POST", "/api/automations", build_step(
    f"collect {run_id}", node, node_output_path(run_id),
    ARCHIVE, f"{archive_path(run_id)}/{node}",
    step=2, trigger_id=deploy_id))["automationId"]

There are four items that must be handled correctly in the automation request.

All four items can be omitted and registration will still succeed, but runtime behavior will differ. If a recurring schedule runs only once and then stops, check isUpcoming first.

Once the two requests have been sent, the application's work is done. Save each step's identifier with the execution record so you can check its status later.

Training Invocation and Retry

Start training when the data arrives and retransmit only failed files

The invocation occurs after the transfer completes, and the endpoint that receives the request handles it as follows.

def on_train_hook(payload):
    monitor_id = payload.get("monitorId")

    if monitor_id:
        detail = wait(monitor_id)

        if detail["status"] != STATUS_COMPLETE:
            return abort_run(payload)

    start_training(payload)

Check results for each run in the execution history.

runs = api("GET", f"/api/automations/{deploy_id}/executions") or []
latest = runs[0] if runs else None

if latest and latest["status"] != STATUS_COMPLETE:
    print("deploy failed:", latest["monitorId"],
          "retried", retry_failed(latest["monitorId"]), "files")

When the dataset is large, retransmitting everything because of a few failed files is a major waste.