So verhindern Sie Kostenexplosionen und Endlosschleifen der Claude Code CLI beim Aufbau von Agenten-Pipelines
makedream٢٩ يوليو ٢٠٢٦
0
Computing/Software
Comments (0)
Log in to leave a comment
No posts yet
makedreammakedreamLog in to leave a comment
No posts yet
Jeder KI-Ingenieur, der im Unternehmensumfeld bereits Agenten auf Basis von LangGraph oder AutoGen in der Praxis eingesetzt hat, dürfte schon einmal eine unangenehme Überraschung erlebt haben: Man lässt die Claude Code CLI als Hintergrundprozess laufen, und wegen eines einzigen Fehlers im Code ruft sie autonom immer wieder Werkzeuge auf — wodurch innerhalb von 10 Minuten Token im Wert von hunderten Dollar verbrannt werden.
Dies ist kein Problem, das sich einfach durch besseres Prompt-Engineering lösen lässt. Es gilt, Engpässe auf Systemebene zu beseitigen, die in realen Produktionsumgebungen auftreten — etwa fehlende Hintergrundsteuerung, endloses Warten zwischen Knoten oder Browser-Speicherabstürze. Im Folgenden finden Sie vier einsatzbereite Codebeispiele und Architekturmuster für Ihre Produktionspipeline.
-pWenn Sie ein Modell wie Claudes 3 Opus von Anthropic im asynchronen Modus ohne Dialog (-p-Flag) ausführen, ruft es bei einem Testfehlschlag sofort wieder Werkzeuge auf, ohne dass ein Benutzer eingreifen kann. Gerät das System hierbei in eine Endlosschleife, sammelt sich der Verlauf der mehrstufigen Schlussfolgerung (Multi-Step Reasoning) kontinuierlich an, was bei jedem Durchlauf enorme Mengen an Token verbraucht. Auch Zombie-Prozesse, die nicht beendet werden und im Systemarbeitsspeicher verbleiben, entstehen in diesem Szenario.
Um dieses Phänomen zu verhindern, wird ein Wrapper benötigt, der Obergrenzen auf CLI-Ebene festlegt und den Prozess auf Skriptebene erzwingen kann, sich zu beenden.
--max-budget-usd ein Ausgabenlimit pro Ausführung an und begrenzen Sie die Anzahl der Werkzeugaufrufe mit --max-turns.--allowedTools nur schreibgeschützte Werkzeuge wie Read, Grep und Glob, und fügen Sie das Flag --bare hinzu, um den Overhead durch das Laden von Plugins zu vermeiden.asyncio.subprocess von Python, um einen Wrapper zu implementieren, der den Prozess nach Ablauf einer bestimmten Zeitspanne abbricht (Kill).`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)}
`
Durch den Einsatz dieses Wrappers übersteigen die Kosten pro Validierungslauf maximal $0.50 nicht. Damit lassen sich Kostenexplosionen durch festgefahrene Schleifen effektiv verhindern.
Die Standard-RetryPolicy von LangGraph ist für die Bewältigung von Netzwerkfehlern ausgelegt. Wenn das LLM fehlerhaften Code generiert und die Validierung fehlschlägt, greift diese Standardrichtlinie nicht richtig. Sobald das Limit für Wiederholungsversuche überschritten wird, wird ein GraphRecursionError ausgelöst und das gesamte System stoppt.
Wenn sich derselbe Fehler wiederholt, muss im State ein Schutzschalter (Circuit Breaker) eingebaut werden, der die Ausführung abbricht und die Kontrolle wiedergibt.
verification_attempts und eine Variable last_error_signature hinzu.Human-in-the-loop-Knoten, um die Aufgabe sicher an einen übergeordneten Thread zu delegieren.`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()
`
Da die Endlosschleife nach zwei aufeinanderfolgenden Fehlschlägen unterbrochen und sofort in einen Wartezustand gewechselt wird, wird verhindert, dass der Agent in einen Deadlock gerät und der gesamte Prozess blockiert wird.
Knoten, die Screenshots erstellen oder die DOM-Struktur validieren, führen eine Chrome Headless Shell aus. Das Problem entsteht, sobald dieser Code in einer Docker- oder GitHub Actions Runner-Umgebung ausgeführt wird: Der Chromium-Renderer stürzt sofort ab. Da die Standardgröße des gemeinsamen Speichers (/dev/shm) in Docker nur 64 MB beträgt, tritt beim Erfassen des Bildschirms der Fehler Failed to reserve shared memory auf.
Beim Starten des Browsers in einer Container-Umgebung müssen folgende Einstellungen vorgenommen werden:
--shm-size=2g und fügen Sie den Chromium-Optionen die Flags --disable-dev-shm-usage sowie --no-sandbox hinzu.--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/
`
Durch die Vergrößerung des Speicherbereichs und die vollständige Trennung der Sitzungen läuft der Browser-Validierungsknoten auch in Hintergrundumgebungen ohne Abstürze bis zum Ende durch.
Der Haupt-State von LangGraph ist ein gemeinsamer Speicher, auf den alle Knoten zugreifen. Wenn ein Validierungsknoten lange HTML-Codes oder vollständige Ausführungsprotokolle direkt in diesen State schreibt, ist das Kontextfenster beim nächsten LLM-Aufruf schlagartig überfüllt. Auch die Speichergeschwindigkeit in der Datenbank verlangsamt sich spürbar.
Große Datenmengen sollten in externe Dateien ausgelagert werden, sodass im State nur noch der Referenzpfad verbleibt.
validation_status und den Pfad zur Ergebnisdatei in artifact_ref_path.@traceable-Dekorator von LangSmith zur Nachverfolgung, um den Protokollfluss zu verknüpfen.`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
}
`
Dadurch wird verhindert, dass unnötige Protokolle in die LLM-Prompts gelangen. Tritt ein Problem auf, muss lediglich die Datei am angegebenen Pfad geöffnet werden, was die Ursachenanalyse erheblich erleichtert.