How to Build an Agent Pipeline to Prevent Claude Code CLI Cost Spikes and Infinite Loops
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
Any AI engineer who has deployed LangGraph or AutoGen-based agents to production in an enterprise environment has likely experienced a cold sweat moment: running Claude Code CLI as a background process, only to watch a single code defect cause it to repeatedly call tools on loop—burning through hundreds of dollars in tokens in just 10 minutes.
This isn't a problem you can solve simply by writing better prompts. You need to fix system-level bottlenecks that explode in actual production environments, such as background control failures, infinite node waits, and browser memory crashes. Here are four battle-tested implementation patterns and architectures ready for production deployment.
-p Non-Interactive ModeWhen running models like Anthropic's Claude 3 Opus in non-interactive asynchronous mode (using the -p flag), test failures trigger immediate tool re-invocations without user intervention. If an infinite loop occurs here, multi-step reasoning history continually accumulates, consuming massive token volumes with every single turn. This is also when non-terminating processes get left behind in system memory as zombie processes.
To prevent this, you need a wrapper that enforces hard limits at the CLI level and forcefully terminates processes at the script level.
--max-budget-usd flag during CLI execution, and restrict tool invocation limits with --max-turns.Read, Grep, and Glob using the --allowedTools option, and append the --bare flag to strip plugin loading overhead.asyncio.subprocess that kills the process after a specified timeout.`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)}
`
By implementing this wrapper, costs will not exceed a maximum of $0.50 per validation run, preventing unexpected cost explosions caused by infinite loops.
LangGraph's default RetryPolicy is designed for network error recovery. When an LLM fails validation due to faulty logic, this default policy fails to handle it properly. Once the retry threshold is crossed, it throws a GraphRecursionError, bringing the entire system to a halt.
You must build a Circuit Breaker inside the State that halts execution and yields control when identical errors repeat continuously.
verification_attempts counter and a last_error_signature variable to the State.Human-in-the-loop node to safely route execution to a parent thread.`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()
`
By breaking the loop after two consecutive failures and immediately switching to a standby state, you prevent agents from deadlocking and freezing the entire pipeline.
Nodes that capture screenshots or validate DOM structures rely on Chrome Headless Shell. The issue arises when running this code in Docker or GitHub Actions Runner environments, where the Chromium renderer crashes almost immediately. Docker's default shared memory (/dev/shm) is capped at a mere 64MB, leading to Failed to reserve shared memory errors during screen captures.
When spawning browsers in containerized environments, you should apply the following configurations:
--shm-size=2g, and append the --disable-dev-shm-usage and --no-sandbox flags to Chromium options.--user-data-dir=/tmp/session_$RUN_ID to avoid cookie or local storage conflicts across runs.`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/
`
Expanding shared memory space and isolating sessions ensures browser validation nodes complete reliably without crashing, even in background worker environments.
LangGraph's main State acts as a shared memory pool accessible across all nodes. Dumping raw, multi-megabyte HTML captures or execution logs directly into this State rapidly saturates the context window on subsequent LLM invocations. Database write performance also degrades noticeably.
You should offload bulk data to external storage and maintain only reference paths within the State.
validation_status and an artifact_ref_path for file references.@traceable decorator to maintain end-to-end observability.`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
}
`
This architecture prevents superfluous log data from polluting LLM prompts, and makes troubleshooting significantly easier since developers only need to inspect designated artifact files when issues arise.