Taipy: Framework Full-Stack untuk Membangun Aplikasi Web AI yang Production-Ready

Membangun aplikasi AI yang siap produksi sering kali membutuhkan kombinasi berbagai tools: framework frontend untuk UI, backend untuk logika bisnis, dan pipeline orchestration untuk proses data. Taipy hadir sebagai solusi all-in-one yang memungkinkan data scientist dan ML engineer membangun aplikasi web AI lengkap hanya dengan Python.

Dalam tutorial ini, kita akan mempelajari Taipy secara mendalam, mulai dari instalasi, GUI building, pipeline orchestration, hingga deployment aplikasi ML dashboard yang siap produksi.

Apa Itu Taipy?

Taipy adalah framework open-source Python yang dirancang khusus untuk membangun aplikasi data dan AI. Berbeda dengan Streamlit atau Gradio yang fokus pada prototyping, Taipy dirancang untuk aplikasi production-grade dengan fitur lengkap:

• Taipy GUI: Membangun antarmuka web interaktif dengan visual elements
• Taipy Core: Mengelola data pipelines, scenarios, dan task scheduling
• Multi-page support: Aplikasi multi-halaman dengan routing
• State management: Pengelolaan state yang robust untuk aplikasi kompleks
• Scalability: Mendukung concurrent users dan heavy workloads

Instalasi dan Setup

Instalasi Taipy

pip install taipy

Untuk instalasi dengan dependencies tambahan:

# Dengan dukungan PostgreSQL
pip install taipy[postgres]

Dengan semua extras
pip install taipy[all]

Verifikasi instalasi
python -c "import taipy; print(taipy.version)"

Struktur Proyek

taipy-ml-dashboard/
├── main.py
├── config.py
├── pages/
│   ├── home.py
│   ├── dashboard.py
│   ├── predictions.py
│   └── settings.py
├── algorithms/
│   ├── preprocessing.py
│   ├── training.py
│   └── inference.py
├── data/
│   └── dataset.csv
└── requirements.txt

Taipy GUI: Membangun Antarmuka Web

Visual Elements Dasar

Taipy menggunakan syntax Markdown-like yang sederhana untuk mendefinisikan UI:

from taipy.gui import Gui

Variabel state
name = "Data Scientist"
message = "Selamat datang di Taipy!"

Definisi halaman dengan Markdown syntax
page = """
Aplikasi AI Dashboard

Halo, <|{name}|>!

<|{message}|text|>

<|Klik Saya|button|onaction=onbuttonclick|>

"""

def onbuttonclick(state):

    state.message = f"Halo {state.name}! Dashboard siap digunakan."

if name == "main":
    Gui(page=page).run(title="AI Dashboard", port=5000)

from taipy.gui import Gui

State variables
username = ""
age = 25
role = "Data Scientist"
roles = ["Data Scientist", "ML Engineer", "Data Analyst", "AI Researcher"]
isactive = True

page = """
Form Registrasi Tim AI

<|{username}|input|label=Nama Lengkap|>

<|{age}|slider|min=18|max=65|label=Usia|>

<|{role}|selector|lov={roles}|dropdown|label=Pilih Role|>

<|{isactive}|toggle|label=Status Aktif|>


<|Submit|button|onaction=onsubmit|>


Data yang Diinput:
Nama: <|{username}|text|>
Usia: <|{age}|text|>
Role: <|{role}|text|>
Aktif: <|{isactive}|text|>
"""


def onsubmit(state):

    print(f"Registered: {state.username}, {state.age}, {state.role}")

Gui(page=page).run()

from taipy.gui import Gui, State

counter = 0
history = []

def onincrement(state: State):
    state.counter += 1
    state.history = state.history + [f"Counter: {state.counter}"]

def onreset(state: State):

    state.counter = 0
    state.history = []

def oninit(state: State):
    """Dipanggil saat user baru membuka aplikasi"""
    state.counter = 0
    state.history = []

page = """
State Management Demo

Counter: <|{counter}|text|>

<|Tambah|button|onaction=onincrement|>
<|Reset|button|onaction=onreset|>

Riwayat:
<|{history}|text|>
"""

Gui(page=page).run()

Callbacks dan Event Handling

from taipy.gui import Gui, notify

datasetpath = ""

dataloaded = False
rowcount = 0

columns = []

def onfileupload(state, varname, payload):
    """Callback saat file diupload"""
    import pandas as pd

    filepath = payload["args"][0]["path"]

    state.datasetpath = filepath


    try:
        df = pd.readcsv(filepath)

        state.rowcount = len(df)
        state.columns = list(df.columns)
        state.dataloaded = True

        notify(state, "success", f"Dataset berhasil dimuat: {state.rowcount} baris")
    except Exception as e:
        notify(state, "error", f"Gagal memuat dataset: {str(e)}")

page = """
Upload Dataset

<|{datasetpath}|fileselector|extensions=.csv,.xlsx|onaction=onfileupload|label=Upload Dataset|>


<|part|render={dataloaded}|
Informasi Dataset
Jumlah baris: <|{rowcount}|text|>

Kolom: <|{columns}|text|>
|>

"""

Gui(page=page).run()

from taipy.gui import Gui
import pandas as pd
import numpy as np

Generate sample data
np.random.seed(42)
dates = pd.daterange("2025-01-01", periods=100, freq="D")
data = pd.DataFrame({
    "Tanggal": dates,
    "Akurasi": np.cumsum(np.random.randn(100)  0.5) + 85,

    "Loss": np.abs(np.cumsum(np.random.randn(100)  0.3)) + 0.1,
    "Prediksi": np.random.randint(100, 1000, 100),
})

Confusion matrix data
confusiondata = pd.DataFrame({

    "x": ["Positif", "Positif", "Negatif", "Negatif"],
    "y": ["Positif", "Negatif", "Positif", "Negatif"],
    "z": [850, 50, 30, 870],
})

page = """
ML Metrics Dashboard

Tren Akurasi Model
<|{data}|chart|type=scatter|x=Tanggal|y=Akurasi|mode=lines+markers|title=Akurasi Model Over Time|>

Training Loss
<|{data}|chart|type=scatter|x=Tanggal|y=Loss|mode=lines|color=#ff6347|title=Training Loss|>

Distribusi Prediksi
<|{data}|chart|type=histogram|x=Prediksi|title=Distribusi Jumlah Prediksi|>

Confusion Matrix
<|{confusiondata}|chart|type=heatmap|x=x|y=y|z=z|title=Confusion Matrix|>
"""

Gui(page=page).run()

Taipy Core: Pipeline dan Scenario Management

Taipy Core adalah engine untuk mengelola data pipelines dan business scenarios:

from taipy.core import Config, Scope
import pandas as pd
from sklearn.ensemble import RandomForestClassifier
from sklearn.modelselection import traintestsplit

from sklearn.metrics import accuracyscore
import pickle

Fungsi-fungsi pipeline
def loaddata(path: str) -> pd.DataFrame:

    """Memuat dataset dari file CSV"""
    df = pd.readcsv(path)
    print(f"Dataset dimuat: {len(df)} baris, {len(df.columns)} kolom")
    return df

def preprocess(df: pd.DataFrame) -> tuple:
    """Preprocessing data: split features dan target"""
    X = df.drop("target", axis=1)
    y = df["target"]
    Xtrain, Xtest, ytrain, ytest = traintestsplit(
        X, y, testsize=0.2, randomstate=42
    )
    return Xtrain, Xtest, ytrain, ytest

def trainmodel(Xtrain, ytrain, nestimators: int = 100) -> object:
    """Training model Random Forest"""
    model = RandomForestClassifier(nestimators=nestimators, randomstate=42)

    model.fit(Xtrain, ytrain)

    return model

def evaluate(model, Xtest, ytest) -> dict:

    """Evaluasi model dan return metrics"""
    predictions = model.predict(Xtest)
    accuracy = accuracyscore(ytest, predictions)
    return {"accuracy": accuracy, "npredictions": len(predictions)}


Konfigurasi Data Nodes
datapathcfg = Config.configuredatanode(

    id="datapath",
    defaultdata="data/dataset.csv",

    scope=Scope.SCENARIO,
)

rawdatacfg = Config.configuredatanode(id="rawdata", scope=Scope.SCENARIO)
traindatacfg = Config.configuredatanode(id="traindata", scope=Scope.SCENARIO)

testdatacfg = Config.configuredatanode(id="testdata", scope=Scope.SCENARIO)
modelcfg = Config.configuredatanode(id="trainedmodel", scope=Scope.SCENARIO)
metricscfg = Config.configuredatanode(id="metrics", scope=Scope.SCENARIO)


Konfigurasi Tasks
loadtaskcfg = Config.configuretask(
    id="loaddata",

    function=loaddata,
    input=[datapathcfg],
    output=[rawdatacfg],
)

preprocesstaskcfg = Config.configuretask(

    id="preprocess",
    function=preprocess,
    input=[rawdatacfg],

    output=[traindatacfg, testdatacfg],

)

traintaskcfg = Config.configuretask(
    id="trainmodel",

    function=trainmodel,
    input=[traindatacfg],
    output=[modelcfg],

)

evaluatetaskcfg = Config.configuretask(
    id="evaluate",
    function=evaluate,
    input=[modelcfg, testdatacfg],

    output=[metricscfg],
)

Konfigurasi Scenario
scenariocfg = Config.configurescenario(
    id="mlpipeline",

    taskconfigs=[
        loadtaskcfg,
        preprocesstaskcfg,
        traintaskcfg,
        evaluatetaskcfg,
    ],
)

Menjalankan Scenario

import taipy.core as tp

Inisialisasi Taipy Core
tp.Core().run()

Membuat dan menjalankan scenario
scenario = tp.createscenario(scenariocfg)
tp.submit(scenario)

Membaca hasil
metrics = scenario.metrics.read()
print(f"Model Accuracy: {metrics['accuracy']:.4f}")

Membuat scenario baru dengan parameter berbeda
scenario2 = tp.createscenario(scenariocfg)

scenario2.datapath.write("data/datasetv2.csv")
tp.submit(scenario2)

from taipy.gui import Gui
import taipy.gui.builder as tgb

Halaman Home
homepage = """
Selamat Datang di ML Dashboard

Aplikasi ini menyediakan:
Dashboard: Monitoring metrik model
Prediksi: Jalankan prediksi baru
Settings: Konfigurasi model

<|Dashboard|button|onaction=navigatedashboard|>
<|Prediksi|button|onaction=navigatepredictions|>
"""

Halaman Dashboard
dashboardpage = """

Dashboard Monitoring

Performa Model
<|{metricsdata}|chart|type=bar|x=Model|y=Accuracy|title=Perbandingan Akurasi Model|>

Tren Harian
<|{dailydata}|chart|type=scatter|x=Date|y=Requests|mode=lines|title=Request Harian|>


<|{dailydata}|table|>
"""

Halaman Prediksi
predictionpage = """

Prediksi Baru

Input Features
<|{feature1}|input|label=Feature 1|>
<|{feature2}|input|label=Feature 2|>

<|{feature3}|input|label=Feature 3|>

<|Jalankan Prediksi|button|onaction=runprediction|>

<|part|render={predictionresult is not None}|

Hasil Prediksi
Kelas: <|{predictionresult}|text|>
Confidence: <|{confidence}|text|>%
|>

"""

Root page dengan navbar
rootpage = """

<|navbar|>
"""

def navigatedashboard(state):
    state.navigate("/dashboard")

def navigatepredictions(state):

    state.navigate("/predictions")

Setup multi-page
pages = {
    "/": rootpage,
    "home": homepage,

    "dashboard": dashboardpage,
    "predictions": predictionpage,

}

Gui(pages=pages).run(title="ML Dashboard", port=5000)

from taipy.gui import Gui, notify
import pandas as pd
import numpy as np
from sklearn.datasets import loadiris
from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.modelselection import traintestsplit, crossvalscore

from sklearn.metrics import classificationreport, confusionmatrix


Load dan prepare data
iris = loadiris()
df = pd.DataFrame(iris.data, columns=iris.featurenames)

df["target"] = iris.target
df["species"] = df["target"].map({0: "Setosa", 1: "Versicolor", 2: "Virginica"})

X = df[iris.featurenames]
y = df["target"]
Xtrain, Xtest, ytrain, ytest = traintestsplit(X, y, testsize=0.2, randomstate=42)

Train models
models = {
    "Random Forest": RandomForestClassifier(nestimators=100, randomstate=42),
    "Gradient Boosting": GradientBoostingClassifier(nestimators=100, randomstate=42),
}

results = []
trainedmodels = {}

for name, model in models.items():
    model.fit(Xtrain, ytrain)

    scores = crossvalscore(model, X, y, cv=5)

    results.append({
        "Model": name,
        "Accuracy": round(scores.mean()  100, 2),

        "Std": round(scores.std()  100, 2),
    })
    trainedmodels[name] = model

metricsdf = pd.DataFrame(results)


Feature importance
rfmodel = trainedmodels["Random Forest"]

importancedf = pd.DataFrame({
    "Feature": iris.featurenames,

    "Importance": rfmodel.featureimportances,
}).sortvalues("Importance", ascending=False)


State variables
selectedmodel = "Random Forest"
modelnames = list(models.keys())

sepallength = 5.1
sepalwidth = 3.5

petallength = 1.4
petalwidth = 0.2

predictionresult = None
confidence = 0
predictionhistory = pd.DataFrame(columns=["Waktu", "Model", "Prediksi", "Confidence"])


def runprediction(state):
    """Jalankan prediksi dengan model terpilih"""
    model = trainedmodels[state.selectedmodel]
    features = np.array([[
        state.sepallength,

        state.sepalwidth,
        state.petallength,

        state.petalwidth
    ]])

    pred = model.predict(features)[0]
    proba = model.predictproba(features)[0]


    speciesmap = {0: "Setosa", 1: "Versicolor", 2: "Virginica"}
    state.predictionresult = speciesmap[pred]
    state.confidence = round(max(proba) * 100, 2)

    # Update history
    newrow = pd.DataFrame([{

        "Waktu": pd.Timestamp.now().strftime("%Y-%m-%d %H:%M:%S"),
        "Model": state.selectedmodel,
        "Prediksi": state.predictionresult,

        "Confidence": f"{state.confidence}%",
    }])
    state.predictionhistory = pd.concat(
        [newrow, state.predictionhistory], ignoreindex=True

    )
    notify(state, "success", f"Prediksi: {state.predictionresult} ({state.confidence}%)")

Dashboard page
dashboard = """
ML Model Dashboard

Perbandingan Model
<|{metricsdf}|chart|type=bar|x=Model|y=Accuracy|title=Akurasi Model (%)|>


<|{metricsdf}|table|>

Feature Importance (Random Forest)
<|{importancedf}|chart|type=bar|x=Feature|y=Importance|title=Feature Importance|>


Distribusi Dataset
<|{df}|chart|type=scatter|x=sepal length (cm)|y=petal length (cm)|color=species|title=Scatter Plot: Sepal vs Petal Length|>
"""

Prediction page
prediction = """
Prediksi Spesies Iris

Pilih Model
<|{selectedmodel}|selector|lov={modelnames}|dropdown|label=Model|>


Input Features
<|{sepallength}|slider|min=4.0|max=8.0|step=0.1|label=Sepal Length (cm)|>
<|{sepalwidth}|slider|min=2.0|max=5.0|step=0.1|label=Sepal Width (cm)|>

<|{petallength}|slider|min=1.0|max=7.0|step=0.1|label=Petal Length (cm)|>
<|{petalwidth}|slider|min=0.1|max=3.0|step=0.1|label=Petal Width (cm)|>


<|Prediksi|button|onaction=runprediction|>


<|part|render={predictionresult is not None}|
Hasil Prediksi
Spesies: <|{predictionresult}|text|> (Confidence: <|{confidence}|text|>%)

|>

Riwayat Prediksi
<|{predictionhistory}|table|>
"""

pages = {
    "/": "<|navbar|>",
    "Dashboard": dashboard,
    "Prediction": prediction,
}

if name == "main":
    Gui(pages=pages).run(
        title="Iris ML Dashboard",
        port=5000,
        darkmode=True,

    )

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 5000

CMD ["python", "main.py"]

# docker-compose.yml
version: "3.8"
services:
  taipy-app:
    build: .
    ports:
"5000:5000"
    volumes:

./data:/app/data
    environment:

TAIPYPORT=5000
TAIPYHOST=0.0.0.0

    restart: unless-stopped

from taipy.gui import Gui

if name == "main":
    gui = Gui(pages=pages)
    gui.run(
        title="Production ML Dashboard",
        host="0.0.0.0",
        port=5000,
        debug=False,
        usereloader=False,
        darkmode=True,

        watermark="PT RUBYTHALIB DATA KONSULTA",
    )

server {
    listen 80;
    servername ml-dashboard.example.com;

    location / {
        proxypass http://localhost:5000;

        proxysetheader Host $host;

        proxysetheader X-Real-IP $remoteaddr;
        proxysetheader X-Forwarded-For $proxyaddxforwardedfor;
        proxyhttpversion 1.1;
        proxysetheader Upgrade $httpupgrade;

        proxysetheader Connection "upgrade";

    }
}

import pytest
from algorithms.preprocessing import preprocess

def testpreprocess():
    import pandas as pd
    df = pd.DataFrame({"feature1": [1, 2, 3], "target": [0, 1, 0]})
    Xtrain, Xtest, ytrain, ytest = preprocess(df)
    assert len(Xtrain) > 0

    assert len(Xtest) > 0

Kesimpulan

Taipy adalah framework yang sangat powerful untuk membangun aplikasi AI/ML production-grade. Dengan kombinasi Taipy GUI untuk antarmuka web dan Taipy Core untuk pipeline orchestration, developer dapat membangun aplikasi end-to-end tanpa perlu menggunakan banyak tools terpisah.

Keunggulan utama Taipy dibandingkan alternatif seperti Streamlit dan Gradio adalah kemampuannya menangani aplikasi kompleks dengan state management yang robust, multi-page support native, dan built-in pipeline orchestration. Ini menjadikannya pilihan ideal untuk proyek AI yang serius dan membutuhkan skalabilitas tinggi.

Mulailah dengan contoh sederhana, lalu secara bertahap tambahkan kompleksitas sesuai kebutuhan proyek Anda. Dokumentasi resmi Taipy di docs.taipy.io menyediakan banyak contoh dan referensi tambahan.