Cara Membangun Pipeline Agen untuk Mencegah Pembengkakan Biaya dan Infinite Loop pada Claude Code CLI
makedream2026年7月29日
0
Computing/Software
Comments (0)
Log in to leave a comment
No posts yet
makedreammakedreamLog in to leave a comment
No posts yet
Setiap AI engineer yang pernah mengimplementasikan agen berbasis LangGraph atau AutoGen ke lingkungan produksi di tingkat enterprise pasti pernah mengalami situasi yang mendebarkan. Situasi di mana Anda menjalankan Claude Code CLI sebagai proses latar belakang (background process), lalu karena satu cacat kode, agen tersebut terus-menerus memanggil ulang tool secara mandiri hingga menghabiskan token senilai ratusan dolar hanya dalam waktu 10 menit.
Masalah ini tidak bisa diselesaikan hanya dengan menulis prompt yang baik. Anda harus mengatasi kemacetan (bottleneck) tingkat sistem yang benar-benar terjadi di lingkungan operasional nyata, seperti kegagalan kontrol background process, kondisi menunggu tanpa akhir (infinite waiting) antar-node, hingga crash memori pada browser. Berikut adalah ringkasan empat arsitektur dan kode penanganan yang siap digunakan langsung di pipeline produksi Anda.
-pKetika Anda menjalankan model seperti Claude 3 Opus dari Anthropic dalam mode asinkron tanpa interaksi percakapan (flag -p), model akan langsung memanggil ulang tool tanpa campur tangan pengguna saat terjadi kegagalan pengujian. Jika terjadi infinite loop pada kondisi ini, riwayat inferensi multi-step akan terus menumpuk dan menguras token dalam jumlah yang sangat besar di setiap turn. Proses zombie yang tidak berhenti dan tetap mengendap di memori sistem juga terbentuk pada saat ini.
Untuk mencegah fenomena ini, Anda memerlukan wrapper yang dapat menetapkan batas atas di tingkat CLI serta memaksa proses berhenti secara paksa di tingkat skrip.
--max-budget-usd, dan batasi jumlah pemanggilan tool dengan --max-turns.--allowedTools untuk hanya mengizinkan tool read-only seperti Read, Grep, dan Glob, serta tambahkan flag --bare untuk menghilangkan overhead pemuatan plugin.asyncio.subprocess di Python untuk membunuh (Kill) proses jika telah melewati batas waktu tertentu.`python
import asyncio
import os
from typing import Any, Dict
class ClaudeCodeWrapper:
def init(self, max_budget_usd: float = 0.50, max_turns: int = 5, timeout_seconds: float = 120.0):
self.max_budget_usd = max_budget_usd
self.max_turns = max_turns
self.timeout_seconds = timeout_seconds
async def execute_validation(self, prompt: str, target_dir: str) -> Dict[str, Any]:
cmd = [
"claude", "-p", prompt,
"--max-budget-usd", str(self.max_budget_usd),
"--max-turns", str(self.max_turns),
"--allowedTools", "Read", "Grep", "Glob", "Bash(pytest *)",
"--add-dir", target_dir,
"--bare"
]
env = os.environ.copy()
env["CLAUDE_CODE_SIMPLE"] = "1"
try:
process = await asyncio.create_subprocess_exec(
*cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, cwd=target_dir, env=env
)
try:
stdout_data, stderr_data = await asyncio.wait_for(process.communicate(), timeout=self.timeout_seconds)
except asyncio.TimeoutError:
process.kill()
await process.wait()
return {"status": "TIMEOUT_EXCEEDED", "exit_code": -1, "error": f"Timeout after {self.timeout_seconds}s"}
if process.returncode != 0:
return {"status": "CLI_ERROR", "exit_code": process.returncode, "error": stderr_data.decode("utf-8")}
return {"status": "SUCCESS", "exit_code": 0, "raw_output": stdout_data.decode("utf-8")}
except Exception as e:
return {"status": "WRAPPER_EXCEPTION", "exit_code": -2, "error": str(e)}
`
Dengan menerapkan wrapper ini, biaya per satu kali validasi tidak akan melebihi maksimal $0.50. Anda dapat mencegah ledakan biaya yang disebabkan oleh loop yang tersangkut.
RetryPolicy bawaan pada LangGraph dirancang untuk menangani kesalahan jaringan. Kebijakan bawaan ini tidak akan bekerja dengan baik ketika LLM salah menulis logika sehingga validasi gagal. Begitu melewati batas percobaan ulang, sistem akan melempar GraphRecursionError dan menghentikan seluruh sistem.
Jika kesalahan yang sama terus berulang, Anda harus membuat Pemutus Arus (Circuit Breaker) di dalam State untuk menghentikan eksekusi dan menyerahkan kendali.
verification_attempts dan variabel last_error_signature ke dalam State.Human-in-the-loop untuk mengembalikan tugas secara aman ke thread tingkat atas.`python
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from langchain_anthropic import ChatAnthropic
class AgentGraphState(TypedDict):
task_prompt: str
verification_attempts: int
last_error_signature: str
verification_status: Literal["PENDING", "PASSED", "FAILED", "CIRCUIT_BROKEN"]
def verification_node(state: AgentGraphState) -> dict:
attempts = state.get("verification_attempts", 0) + 1
model_name = "claude-3-5-haiku-20241022" if attempts >= 2 else "claude-3-opus-20240229"
llm = ChatAnthropic(model=model_name, temperature=0.0)
return {"verification_attempts": attempts, "last_error_signature": "SyntaxError", "verification_status": "FAILED"}
def circuit_breaker_router(state: AgentGraphState) -> str:
if state.get("verification_status") == "PASSED":
return "proceed"
if state.get("verification_attempts", 0) >= 2:
return "trigger_fallback"
return "retry"
def human_in_the_loop_node(state: AgentGraphState) -> dict:
return {"verification_status": "CIRCUIT_BROKEN"}
builder = StateGraph(AgentGraphState)
builder.add_node("verify", verification_node)
builder.add_node("hitl_fallback", human_in_the_loop_node)
builder.add_conditional_edges("verify", circuit_breaker_router, {
"proceed": END, "retry": "verify", "trigger_fallback": "hitl_fallback"
})
builder.add_edge("hitl_fallback", END)
graph_app = builder.compile()
`
Karena infinite loop akan terputus dan langsung beralih ke status siaga setelah 2 kali kegagalan berturut-turut, fenomena di mana agen mengalami kebuntuan (deadlock) dan menghentikan seluruh proses dapat dicegah.
Node yang mengambil tangkapan layar (screenshot) atau memvalidasi struktur DOM akan menjalankan Chrome Headless Shell. Masalahnya adalah kode ini akan langsung crash pada renderer Chromium begitu masuk ke lingkungan Docker atau GitHub Actions Runner. Kapasitas memori bersama (shared memory) bawaan Docker (/dev/shm) hanya sebesar 64MB, sehingga terjadi error Failed to reserve shared memory saat mengambil tangkapan layar.
Saat menjalankan browser di lingkungan kontainer, Anda harus menyesuaikan pengaturan berikut:
--shm-size=2g, lalu tambahkan flag --disable-dev-shm-usage dan --no-sandbox pada opsi Chromium.--user-data-dir=/tmp/session_$RUN_ID.`yaml
name: Agent Background Verification Pipeline
on: [push]
jobs:
headless-browser-validation:
runs-on: ubuntu-latest
container:
image: node:20-buster
options: --shm-size=2g --user root
steps:
- uses: actions/checkout@v4
- name: Install Chrome
run: |
apt-get update && apt-get install -y wget gnupg
wget -q -O - https://dl-ssl.google.com/linux/linux_signing_key.pub | apt-key add -
sh -c 'echo "deb [arch=amd64] http://dl.google.com/linux/chrome/deb/ stable main" >> /etc/apt/sources.list.d/google.list'
apt-get update && apt-get install -y google-chrome-stable --no-install-recommends
- name: Run Agent Verification
env:
ANTHROPIC_API_KEY: {{ github.workspace }}/artifacts/sessions/${{ github.run_id }}"
run: |
mkdir -p SESSION_STORAGE_DIR"
- uses: actions/upload-artifact@v4
if: always()
with:
name: validation-artifacts-${{ github.run_id }}
path: ${{ github.workspace }}/artifacts/
`
Dengan memperluas ruang memori dan memisahkan sesi secara total, node validasi browser akan berjalan hingga selesai tanpa terhenti bahkan di lingkungan background process.
State utama pada LangGraph adalah memori bersama yang digunakan oleh semua node. Jika Anda memasukkan seluruh log eksekusi atau kode HTML panjang yang ditangkap oleh node validasi ke dalam State ini, context window akan langsung penuh saat pemanggilan LLM pada turn berikutnya. Kecepatan penyimpanan database juga akan melambat secara signifikan.
Gunakan metode memindahkan data berukuran besar ke file eksternal dan hanya menyisakan jalur referensi (reference path) di dalam State.
validation_status dan artifact_ref_path yang akan menampung jalur file hasil.@traceable dari LangSmith untuk pelacakan guna menghubungkan alur log.`python
import json, os, time
from typing import TypedDict, Dict, Any
from langsmith import traceable
class LightMainState(TypedDict):
session_id: str
target_component: str
validation_status: str
artifact_ref_path: str
@traceable(run_type="llm", name="ClaudeCodeValidationSkill", tags=["claude-code-cli", "isolated-node"])
def run_context_free_validation(state: LightMainState) -> Dict[str, Any]:
session_id = state["session_id"]
raw_execution_stdout = "DUMP LOG DATA " * 10000
artifact_dir = f"./artifacts/{session_id}"
os.makedirs(artifact_dir, exist_ok=True)
artifact_full_path = os.path.join(artifact_dir, f"artifact_{int(time.time())}.json")
with open(artifact_full_path, "w", encoding="utf-8") as f:
json.dump({"session_id": session_id, "stdout": raw_execution_stdout}, f, indent=2)
return {
"validation_status": "PASSED",
"artifact_ref_path": artifact_full_path
}
`
Hal ini dapat mencegah masuknya log yang tidak perlu ke dalam prompt LLM. Jika terjadi masalah, Anda cukup membuka file di jalur yang ditentukan sehingga identifikasi penyebab masalah menjadi jauh lebih mudah.