Pergeseran Paradigma: Pipeline Linier vs Closed-Loop Feedback

Pipeline pemrosesan data linier beroperasi berdasarkan asumsi input deterministik yang diproses melalui serangkaian transformasi statis: Input → Step A → Step B → Step C → Output. Pendekatan DAG (Directed Acyclic Graph) ini optimal untuk ETL konvensional, agregasi log, dan batch processing karena alur kegagalan mudah diprediksi dan di-retry per langkah.

Namun, integrasi komponen non-deterministik seperti Large Language Models (LLM), optimasi heuristik, atau sistem kontrol adaptif memerlukan validasi berulang terhadap output sementara. Menggunakan pipeline linier untuk kasus ini memaksa penulisan rantai kondisional statis yang rapuh. Arsitektur closed-loop feedback mengatasi ini dengan memperlakukan evaluasi sebagai mekanisme kontrol: output dari proses dievaluasi terhadap kriteria eksplisit, lalu dikembalikan sebagai input terarah untuk iterasi berikutnya hingga kondisi konvergen terpenuhi.

Matriks Evaluasi Trade-off Arsitektur

Penerapan pola loop engineering menghadirkan kompromi teknis yang signifikan dibanding alur sekuensial:

  • State Management: Pipeline linier bersifat stateless antar prosesor atau mengandalkan checkpoint sederhana. Closed-loop engine membutuhkan state machine persisten untuk melacak riwayat perbaikan, context drift, dan diferensial metrik per iterasi guna mencegah modifikasi siklis berulang.
  • Latensi Runtime: Pipeline linier memiliki latensi $O(N)$ proporsional terhadap jumlah langkah. Feedback loop memiliki latensi dinamis $O(k \cdot M)$ di mana $k$ adalah jumlah iterasi evaluasi yang diperlukan untuk konvergensi. P99 latency menjadi jauh lebih sulit diprediksi.
  • Biaya Komputasi: Tanpa batas deterministik, closed-loop engine rentan memicu konsumsi resource eksponensial per request. Hal ini krusial saat setiap iterasi memanggil API pihak ketiga atau inference model lokal.
  • Beban Maintainability: Pipeline linier mudah diuji secara modular dengan unit test per step. Pengujian closed-loop engine membutuhkan validasi boundary conditions, deteksi state flapping, dan pembuktian matematis bahwa termination condition selalu tercapai.

Implementasi Minimal Closed-Loop Feedback Engine

Berikut implementasi Python standar dari loop engine yang mengisolasi generator, evaluator deterministik, serta kontrol auto-termination untuk mencegah runaway execution.

from dataclasses import dataclass, field
from typing import Callable, Generic, List, Optional, Tuple, TypeVar

T = TypeVar("T")
Feedback = str

@dataclass(frozen=True)
class LoopBudget:
    max_iterations: int = 5
    max_eval_score: float = 1.0
    min_score_threshold: float = 0.85

@dataclass
class LoopState(Generic[T]):
    current_value: T
    iteration: int = 0
    score: float = 0.0
    history: List[Tuple[float, Feedback]] = field(default_factory=list)
    terminated_early: bool = False

def execute_closed_loop(
    initial_input: T,
    budget: LoopBudget,
    step_fn: Callable[[T, Optional[Feedback]], T],
    eval_fn: Callable[[T], Tuple[float, Feedback]],
) -> LoopState[T]:
    state = LoopState(current_value=initial_input)

    for i in range(1, budget.max_iterations + 1):
        state.iteration = i
        feedback = state.history[-1][1] if state.history else None
        
        # Eksekusi regenerasi/transformasi berdasarkan evaluasi sebelumnya
        candidate = step_fn(state.current_value, feedback)
        score, eval_feedback = eval_fn(candidate)
        
        state.current_value = candidate
        state.score = score
        state.history.append((score, eval_feedback))

        # Kontrol terminasi: ambang batas terpenuhi (konvergen)
        if score >= budget.min_score_threshold:
            state.terminated_early = True
            return state

    # Terminasi batas budget tercapai tanpa konvergen sempurna
    return state

# --- Runnable Assertion Check ---
def _mock_step(val: int, feedback: Optional[str]) -> int:
    # ponytail: optimasi linier statis, upgrade ke model adaptif saat input non-linear
    delta = 2 if feedback == "increase" else 1
    return val + delta

def _mock_eval(val: int) -> Tuple[float, str]:
    target = 10
    score = min(val / target, 1.0)
    return score, "increase" if val < target else "ok"

result = execute_closed_loop(
    initial_input=0,
    budget=LoopBudget(max_iterations=8, min_score_threshold=0.9),
    step_fn=_mock_step,
    eval_fn=_mock_eval
)

assert result.terminated_early is True
assert result.score >= 0.9
assert result.iteration <= 8

Dua baris desain di balik implementasi ini: validasi termination check dilakukan setelah state mutation tercatat lengkap, dan state history disimpan imutabel guna kebutuhan observabilitas pasca-eksekusi.

Mekanisme Pencegahan Runaway Execution

Sistem closed-loop yang gagal mencapai convergence threshold akan terus berjalan hingga timeout infrastruktur jika tidak dipagari secara deterministik. Terapkan tiga guardrail berikut:

  1. Hard Iteration Ceiling: Batas absolut jumlah loop (misal $k = 5$). Jangan izinkan feedback loop dinamis menaikkan batas ini secara internal.
  2. Stagnation Detection (Epsilon Delta): Bandingkan delta skor antar loop berturut-turut. Jika $|score_t - score_{t-1}| < \epsilon$ selama 2 iterasi, eksekusi harus dihentikan paksa (early exit) karena sistem telah terjebak di local minima atau loop osilasi.
  3. Degradation Fallback: Sediakan fallback data structure yang dapat dikembalikan jika iterasi habis tanpa mencapai status valid. Jangan lempar unhandled exception; kembalikan status degradation eksplisit bersama trace riwayat evaluasi.

Panduan Keputusan: Kapan Memilih Closed-Loop

Gunakan pipeline pemrosesan data linier jika transformasi bersifat stateless, waktu per tahapan dapat diprediksi secara deterministik, dan keberhasilan langkah $N$ menjamin integritas langkah $N+1$.

Pilihlah closed-loop feedback jika output dari sebuah proses membutuhkan evaluasi verifikatif yang tidak dapat dipetakan melalui pemetaan if/else sederhana (misal sintesis kode, perbaikan parsing data terkorupsi, atau rekayasa prompt multi-turn). Pastikan biaya komputasi per iterasi telah diperhitungkan dalam SLA latensi sistem Anda.