📌 Baca Juga (Scripting & Automation): Membuat Bot Telegram Pintar dengan Python, LangChain, dan Retrieval‑Augmented Generation
Orkestrasi Skrip Python dengan Temporal.io: Menyederhanakan Workflow Asinkron Berbasis Event
Dalam era otomatisasi AI dan workflow asinkron yang semakin kompleks, mengelola proses yang melibatkan banyak tugas dan dependensi sering menjadi tantangan. Temporal.io hadir sebagai platform orkestrasi yang memudahkan pengembang Python untuk menulis, menjalankan, dan memantau alur kerja (workflow) dengan cara yang lebih terstruktur dan fault‑tolerant. Artikel ini membahas konsep dasar Temporal, cara mengintegrasikannya ke dalam skrip Python, serta contoh konkret penerapan dalam proyek otomasi AI.
Pengenalan Temporal.io
Temporal adalah sistem open‑source yang memfokuskan pada workflow orchestration dan stateful execution. Dengan arsitektur yang terinspirasi dari Cadence (sebelumnya), Temporal memungkinkan:
- Menangani retry dan fault‑tolerance secara otomatis
- Memperlihatkan histori eksekusi secara transparan
- Menjaga idempotensi aktivitas (activity) sehingga tidak terjadi duplikasi
- Menyediakan API gRPC yang dapat diakses dari berbagai bahasa, termasuk Python
Berbeda dengan library async sederhana seperti asyncio, Temporal menambahkan lapisan logika bisnis yang memastikan setiap langkah workflow dapat dipulihkan setelah kegagalan jaringan atau crash server.
Kenapa Python? Mengapa Temporal?
Python tetap menjadi bahasa paling populer untuk pemrograman dan otomatisasi AI berkat ekosistem library yang kaya (NumPy, Pandas, TensorFlow, PyTorch). Namun, ketika workflow Anda melibatkan:
- Pengambilan data dari API eksternal
- Pengolahan batch dengan GPU
- Pengiriman notifikasi ke sistem lain (email, Slack, Kafka)
- Retry otomatis setelah kegagalan
Temporal menawarkan solusi yang tidak hanya mengurangi boilerplate, tapi juga meningkatkan keterbacaan kode dan kehandalan sistem.
Instalasi & Setup
1. Menyiapkan Temporal Server
Anda dapat menjalankan Temporal secara lokal menggunakan Docker Compose. Salin konfigurasi berikut ke file docker-compose.yml:
version: '3'
services:
temporal:
image: temporalio/server:latest
ports:
- "7233:7233"
environment:
- TEMPORAL_DB=postgres
- POSTGRES_HOST=postgres
- POSTGRES_PORT=5432
- POSTGRES_USER=temporal
- POSTGRES_PASSWORD=temporal
- POSTGRES_DB=temporal
postgres:
image: postgres:13
environment:
POSTGRES_USER: temporal
POSTGRES_PASSWORD: temporal
POSTGRES_DB: temporal
Jalankan dengan:
docker-compose up -d
2. Instalasi SDK Python
Temporal menyediakan SDK Python yang dapat diinstal lewat pip:
pip install temporalio
Struktur Dasar Workflow dan Activity
Temporal memisahkan workflow (logika alur) dari activity (unit pekerjaan). Berikut contoh sederhana:
Workflow
from temporalio import workflow
from temporalio.client import Client
@workflow.defn
class DataPipelineWorkflow:
@workflow.run
async def run(self, dataset_id: str) -> str:
# Langkah 1: Ambil data
raw_data = await workflow.execute_activity(
fetch_data, dataset_id, schedule_to_start_timeout=timedelta(seconds=30)
)
# Langkah 2: Proses data
processed = await workflow.execute_activity(
process_data, raw_data, schedule_to_start_timeout=timedelta(seconds=60)
)
# Langkah 3: Simpan hasil
await workflow.execute_activity(
store_result, processed, schedule_to_start_timeout=timedelta(seconds=30)
)
return "Selesai"
Activity
from temporalio import activity
@activity.defn
async def fetch_data(dataset_id: str) -> dict:
# Simulasi panggilan API
return {"id": dataset_id, "data": [1, 2, 3]}
@activity.defn
async def process_data(raw: dict) -> dict:
# Contoh pemrosesan sederhana
raw["data"] = [x * 2 for x in raw["data"]]
return raw
@activity.defn
async def store_result(processed: dict):
# Simulasi penyimpanan ke database
print(f"Data tersimpan: {processed}")
Menjalankan Workflow
Berikut contoh cara memanggil workflow dari skrip Python:
import asyncio
from datetime import timedelta
async def main():
client = await Client.connect("localhost:7233")
result = await client.execute_workflow(
DataPipelineWorkflow.run,
"dataset-123",
id="data-pipeline-123",
task_queue="data-queue",
workflow_execution_timeout=timedelta(minutes=10)
)
print(result)
asyncio.run(main())
Integrasi dengan Otomatisasi AI
Temporal sangat cocok untuk pipeline AI yang memerlukan langkah-langkah berikut:
- Pre‑processing data (data cleaning, augmentasi)
- Training model (menggunakan GPU, batch size besar)
- Evaluasi model (cross‑validation, metrics)
- Deployment (pushing ke model registry, API endpoint)
Setiap langkah dapat diimplementasikan sebagai activity. Misalnya, aktivitas train_model dapat memanggil skrip train.py yang menjalankan TensorFlow atau PyTorch. Dengan Temporal, Anda dapat menambahkan retry policy yang cerdas: jika GPU tidak tersedia, sistem akan menunda eksekusi hingga resource tersedia kembali.
Contoh: Training Model dengan Retry
from temporalio import activity, workflow
@activity.defn
async def train_model(config: dict) -> str:
# Simulasi panggilan ke skrip training
# Jika gagal, raise exception
if config.get("gpu_available") is False:
raise RuntimeError("GPU tidak tersedia")
# Proses training...
return "model-id-456"
@workflow.defn
class ModelTrainingWorkflow:
@workflow.run
async def run(self, config: dict) -> str:
return await workflow.execute_activity(
train_model,
config,
schedule_to_start_timeout=timedelta(minutes=5),
retry_policy=activity.RetryPolicy(
maximum_attempts=5,
backoff_coefficient=2.0,
initial_interval=timedelta(seconds=30),
maximum_interval=timedelta(minutes=5)
)
)
Monitoring & Debugging
Temporal menyediakan UI Temporal Web yang menampilkan semua workflow dan activity. Di sana Anda dapat:
- Melihat status (running, succeeded, failed)
- Menelusuri log aktivitas
- Mengatur retry policy secara manual
- Meninjau histori kejadian untuk audit trail
Untuk log lebih detail, Anda dapat mengintegrasikan 💡 Baca Juga & Rekomendasi Jaringan AI Network:structlog
Posting Komentar