Claude Code CLI 비용 폭주와 무한 루프를 막는 에이전트 파이프라인 구축법
29 जुलाई 2026
0
컴퓨터/소프트웨어Related Video
14:07앤트로픽, 그래프 엔지니어링의 최대 단점을 마침내 해결했다
AI LABS
Comments (0)
Log in to leave a comment
No posts yet
14:07AI LABS
Log in to leave a comment
No posts yet
엔터프라이즈 환경에서 LangGraph나 AutoGen 기반 에이전트를 실무에 올려본 AI 엔지니어라면 누구나 한번쯤 서늘한 경험을 해봤을 겁니다. Claude Code CLI를 백그라운드 프로세스로 돌려두었더니, 코드 결함 하나 때문에 혼자 도구를 계속 다시 호출하며 10분 만에 수백 달러치 토큰을 깎아 먹는 상황 말입니다.
단순히 프롬프트를 잘 작성한다고 해결될 문제가 아닙니다. 백그라운드 제어 실패, 노드 간 무한 대기, 브라우저 메모리 크래시처럼 실제 운영 환경에서 터지는 시스템 레벨의 병목을 잡아야 합니다. 프로덕션 파이프라인에서 바로 사용할 수 있는 네 가지 대응 코드와 아키텍처를 정리했습니다.
-p 옵션 실행 시 토큰 폭주와 프로세스 좀비화 방지하기Anthropic의 Claude 3 Opus 같은 모델을 대화 없이 작동하는 비동기 모드(-p 플래그)로 실행하면, 테스트 실패 시 유저 개입 없이 곧바로 도구를 다시 호출합니다. 이 때 무한 루프가 돌면 멀티스텝 추론 이력이 계속 쌓이면서 매 턴마다 엄청난 토큰을 잡아먹습니다. 프로세스가 종료되지 않고 시스템 메모리에 남아있는 좀비 프로세스도 이때 생깁니다.
이 현상을 막으려면 CLI 수준에서 상한을 걸고, 스크립트 단에서 프로세스를 강제로 끝낼 수 있는 래퍼가 필요합니다.
--max-budget-usd 플래그로 1회 실행당 지출 한도를 지정하고, --max-turns로 도구 호출 횟수를 제한합니다.--allowedTools 옵션으로 Read, Grep, Glob 같은 읽기 전용 도구만 허용하고, --bare 플래그를 붙여 플러그인 로드 오버헤드를 뺍니다.asyncio.subprocess를 사용해 일정 시간이 지나면 프로세스를 Kill하는 래퍼 클래스를 붙입니다.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)}
이 래퍼를 적용하면 검증 1회당 비용이 최대 $0.50를 넘지 않습니다. 루프가 꼬여 발생하는 지출 폭탄을 차단할 수 있습니다.
LangGraph의 기본 RetryPolicy는 네트워크 오류 대응용입니다. LLM이 로직을 잘못 작성해 검증에 실패할 때는 이 기본 정책이 제대로 동작하지 않습니다. 재시도 한도를 넘기는 순간 GraphRecursionError를 던지며 전체 시스템이 멈춥니다.
동일한 에러가 계속 반복되면 실행을 중단하고 제어권을 넘기는 회로 차단기(Circuit Breaker)를 State 내부에 만들어야 합니다.
verification_attempts 카운터와 last_error_signature 변수를 추가합니다.Human-in-the-loop 노드로 제어권을 넘겨 작업을 안전하게 상위 스레드로 돌립니다.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()
연속 2회 실패 시 무한 루프를 끊고 곧바로 대기 상태로 전환하기 때문에, 에이전트가 교착 상태에 빠져 전체 프로세스가 멈추는 현상을 방지합니다.
스크린샷을 찍거나 DOM 구조를 검증하는 노드는 Chrome Headless Shell을 돌립니다. 문제는 이 코드가 Docker나 GitHub Actions Runner 환경으로 들어가는 순간 Chromium 렌더러가 바로 터진다는 점입니다. Docker의 기본 공유 메모리(/dev/shm) 용량이 64MB밖에 되지 않아 화면을 캡처할 때 Failed to reserve shared memory 오류가 발생합니다.
컨테이너 환경에서 브라우저를 띄울 때는 아래 설정을 맞춰야 합니다.
--shm-size=2g 옵션으로 공유 메모리를 늘리고, Chromium 옵션에 --disable-dev-shm-usage 및 --no-sandbox 플래그를 넣습니다.--user-data-dir=/tmp/session_$RUN_ID 형태의 독립 디렉토리를 세션별로 생성합니다.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: ${{ secrets.ANTHROPIC_API_KEY }}
SESSION_STORAGE_DIR: "${{ github.workspace }}/artifacts/sessions/${{ github.run_id }}"
run: |
mkdir -p $SESSION_STORAGE_DIR
npx ts-node src/cli_validation_entry.ts --user-data-dir="$SESSION_STORAGE_DIR"
- uses: actions/upload-artifact@v4
if: always()
with:
name: validation-artifacts-${{ github.run_id }}
path: ${{ github.workspace }}/artifacts/
메모리 공간을 늘리고 세션을 완전히 분리하면 백그라운드 환경에서도 브라우저 검증 노드가 튕기지 않고 끝까지 돌아갑니다.
LangGraph의 메인 State는 모든 노드가 같이 쓰는 공유 메모리입니다. 검증 노드가 캡처한 긴 HTML 코드나 실행 로그 전체를 이 State에 그대로 밀어 넣으면 다음 턴 LLM 호출 시 컨텍스트 윈도우가 단번에 꽉 찹니다. DB 저장 속도도 눈에 띄게 느려집니다.
대용량 데이터는 외부 파일로 빼고, State에는 참조 경로만 남겨놓는 방식을 써야 합니다.
validation_status와 결과 파일 경로를 담을 artifact_ref_path만 남깁니다.@traceable 데코레이터를 달아 로그 흐름을 연결합니다.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
}
LLM 프롬프트에 불필요한 로그가 섞 들어가는 일을 막을 수 있고, 문제가 생겼을 때 지정된 경로의 파일만 열어보면 되므로 원인 파악이 훨씬 수월해집니다.