Circuit Breaker untuk AI Agent — Mencegah Infinite Loop

AI trading bot saya pernah menghabiskan $47 token dalam 20 menit karena agent masuk ke loop yang tidak bisa keluar. Tidak ada alert, tidak ada hard stop — bot terus memanggil LLM sampai saya notice di dashboard billing.

Itu momen saya belajar bahwa circuit breaker bukan optional — ini adalah komponen wajib sebelum agent apapun masuk ke production.

Mengapa Agent Bisa Infinite Loop

Tool selalu gagal — agent terus memanggil tool yang error tapi tidak tahu cara lain
Contradictory instructions — dua instruksi saling bertentangan, agent tidak bisa mencapai END
Parsing error — LLM menghasilkan format yang salah, agent retry terus-menerus
Overambitious goal — tujuan yang tidak bisa diselesaikan dengan tools yang ada

Circuit Breaker State Machine — CLOSED → OPEN → HALF_OPEN

Prinsip Circuit Breaker

Lebih baik agent berhenti dan melaporkan "tidak bisa menyelesaikan" daripada terus berjalan sampai menghabiskan semua budget. Failure yang explicit jauh lebih baik dari silent runaway.

Layer 1: Max Iterations + Timeout

Paling dasar, paling cepat diimplementasikan. LangChain AgentExecutor sudah menyediakan ini:

Python · AgentExecutor Limits

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    max_iterations=10,           # hard stop setelah 10 langkah
    max_execution_time=60,       # timeout 60 detik
    early_stopping_method="generate",  # LLM buat final answer saat stop
    handle_parsing_errors=True,  # jangan crash saat format output salah
    verbose=True                  # log setiap step untuk debugging
)

Layer 2: Token Budget Tracking

Max iterations tidak cukup kalau setiap iteration memanggil tool yang mahal. Token budget memberikan hard cap pada biaya per execution:

Python · Token Budget

from langchain.callbacks import get_openai_callback

def run_with_budget(agent_executor, input_data, max_tokens=10000):
    with get_openai_callback() as cb:
        try:
            result = agent_executor.invoke(input_data)
            if cb.total_tokens > max_tokens:
                log_warning(f"Token budget exceeded: {cb.total_tokens}")
            return {
                "result": result,
                "tokens": cb.total_tokens,
                "cost_usd": cb.total_cost
            }
        except Exception as e:
            return {"error": str(e), "tokens": cb.total_tokens}

Layer 3: Consecutive Failure Detection

Dari AI trading bot — ini yang paling berguna di production. Kalau agent gagal 3 kali berturut-turut, kemungkinan besar ada masalah sistemik yang tidak bisa diselesaikan dengan retry:

Python · CircuitBreaker Class

from dataclasses import dataclass
from datetime import datetime, timedelta

@dataclass
class CircuitBreaker:
    max_failures: int = 3
    reset_timeout: int = 300  # detik
    failure_count: int = 0
    last_failure_time: datetime = None
    state: str = "CLOSED"  # CLOSED/OPEN/HALF_OPEN

    def call(self, func, *args, **kwargs):
        if self.state == "OPEN":
            elapsed = datetime.now() - self.last_failure_time
            if elapsed > timedelta(seconds=self.reset_timeout):
                self.state = "HALF_OPEN"
            else:
                raise Exception("Circuit breaker OPEN — sistem sedang recovery")
        try:
            result = func(*args, **kwargs)
            self.failure_count = 0
            self.state = "CLOSED"
            return result
        except Exception:
            self.failure_count += 1
            self.last_failure_time = datetime.now()
            if self.failure_count >= self.max_failures:
                self.state = "OPEN"
                alert_team(f"Circuit breaker tripped: {self.failure_count} failures")
            raise

# Pemakaian
breaker = CircuitBreaker(max_failures=3, reset_timeout=300)
result = breaker.call(agent_executor.invoke, input_data)

⚠️ Selalu Ada Alert saat Circuit Breaker Trip

Circuit breaker yang trip tanpa alert sama saja tidak ada. Minimal kirim Telegram message atau email ke diri sendiri. Di production saya pakai Telegram bot yang notif langsung ke HP saat breaker trip.

Layer	Implementasi	Melindungi dari	Kapan Tidak Cukup
Max iterations	`max_iterations=10`	Loop tanpa akhir dalam satu run	Tidak melindungi dari tool yang mahal per call
Timeout	`max_execution_time=60`	Agent yang hang menunggu response	Tidak melindungi dari banyak run yang gagal
Token budget	`get_openai_callback()`	Tagihan yang membengkak per run	Tidak mencegah failure berulang lintas run
Circuit breaker	`CircuitBreaker class`	Consecutive failure sistemik	✓ Melengkapi semua layer di atas

Urutan Implementasi yang Disarankan

Mulai dari max_iterations + timeout

Paling mudah, paling cepat — 2 baris kode di AgentExecutor. Pasang ini sebelum deploy apapun ke production, bahkan untuk testing internal.

Tambahkan token budget tracking

Wrap agent call dengan get_openai_callback(). Log total_tokens dan total_cost setiap run ke database. Ini juga memberikan data yang berguna untuk optimasi cost ke depannya.

Implementasikan CircuitBreaker untuk production

Setelah ada data failure pattern dari logging (step 02), set threshold yang realistis. Selalu test circuit breaker secara eksplisit — simulate failure dan verifikasi breaker trip dan alert terkirim.

Kesimpulan

Tiga layer yang wajib sebelum agent production: max iterations + timeout, token budget tracking, dan consecutive failure detection. Implementasikan secara berurutan — layer 1 paling cepat, tapi layer 3 yang paling berguna jangka panjang.

Agent yang gagal dengan pesan yang jelas jauh lebih baik dari agent yang terus jalan sampai tagihan bulan ini mengejutkan Anda.

Episode terakhir: Evaluasi Output LLM — framework sistematis untuk mengukur kualitas jawaban agent, bukan hanya "kelihatannya bagus."

#AIAgent #Python #LangChain #ProductionAI #CircuitBreaker #BuildInPublic