Comment construire un pipeline d'agents pour empêcher l'explosion des coûts du CLI Claude Code et les boucles infinies
makedream29 juillet 2026
0
Computing/Software
Comments (0)
Log in to leave a comment
No posts yet
makedreammakedreamLog in to leave a comment
No posts yet
Tout ingénieur IA ayant déployé des agents basés sur LangGraph ou AutoGen dans un environnement d'entreprise a déjà vécu cette expérience effrayante : vous laissez tourner le CLI Claude Code en arrière-plan, et à cause d'un simple bug dans le code, l'agent se met à rappeler les outils en boucle, consommant pour des centaines de dollars de jetons en seulement 10 minutes.
Il ne s'agit pas d'un problème que l'on peut résoudre simplement en rédigeant de meilleurs prompts. Il faut s'attaquer aux goulots d'étranglement au niveau système qui surviennent dans un environnement de production réel, tels que les échecs de contrôle en arrière-plan, les attentes infinies entre les nœuds ou les crashs mémoire du navigateur. Voici quatre solutions logicielles et architectures prêtes à être utilisées dans votre pipeline de production.
-pLorsqu'un modèle comme Claude 3 Opus d'Anthropic est exécuté en mode asynchrone sans dialogue (avec le fanion -p), il rappelle immédiatement les outils en cas d'échec de test sans intervention humaine. Si une boucle infinie se produit à ce moment-là, l'historique de raisonnement multi-étapes s'accumule continuellement, consommant une quantité énorme de jetons à chaque tour. C'est également là que se forment des processus zombies qui ne se terminent pas et restent dans la mémoire système.
Pour empêcher ce phénomène, vous avez besoin d'un wrapper capable d'imposer des limites au niveau du CLI et de forcer l'arrêt du processus au niveau du script.
--max-budget-usd pour spécifier une limite de dépense par exécution et limitez le nombre d'appels d'outils avec --max-turns.Read, Grep et Glob avec l'option --allowedTools, et ajoutez le fanion --bare pour supprimer la surcharge de chargement des plugins.asyncio.subprocess en Python pour attacher une classe wrapper qui tue le processus après un certain délai.`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)}
`
En appliquant ce wrapper, le coût par vérification ne dépassera pas un maximum de $0.50. Vous pouvez ainsi bloquer les explosions de dépenses causées par des boucles défaillantes.
La politique de réessai par défaut (RetryPolicy) de LangGraph est conçue pour gérer les erreurs réseau. Lorsque le LLM écrit une mauvaise logique et échoue à la vérification, cette politique par défaut ne fonctionne pas correctement. Dès que la limite de réessais est dépassée, elle lève une GraphRecursionError et l'ensemble du système s'arrête.
Si la même erreur se répète continuellement, vous devez créer un coupe-circuit (Circuit Breaker) à l'intérieur de l'état (State) pour interrompre l'exécution et passer le contrôle.
verification_attempts et une variable last_error_signature dans le State.Human-in-the-loop pour renvoyer la tâche en toute sécurité au thread supérieur.`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()
`
En cas de deux échecs consécutifs, le système brise la boucle infinie et passe immédiatement en état d'attente, ce qui empêche l'agent d'entrer dans un blocage qui paralyserait l'ensemble du processus.
Les nœuds qui prennent des captures d'écran ou vérifient la structure du DOM exécutent Chrome Headless Shell. Le problème est que dès que ce code est introduit dans un environnement Docker ou un runner GitHub Actions, le moteur de rendu Chromium plante immédiatement. La capacité par défaut de la mémoire partagée de Docker (/dev/shm) n'étant que de 64 Mo, une erreur Failed to reserve shared memory survient lors de la capture d'écran.
Lorsque vous lancez un navigateur dans un environnement de conteneur, vous devez ajuster les paramètres suivants :
--shm-size=2g, et ajoutez les fanions --disable-dev-shm-usage et --no-sandbox aux options de Chromium.--user-data-dir=/tmp/session_$RUN_ID pour éviter les conflits de cookies ou de stockage local entre les exécutions.`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/
`
En augmentant l'espace mémoire et en isolant complètement la session, le nœud de vérification du navigateur s'exécute jusqu'au bout sans planter, même dans un environnement en arrière-plan.
Le State principal de LangGraph est une mémoire partagée utilisée par tous les nœuds. Si vous injectez directement de longs codes HTML capturés par le nœud de vérification ou l'intégralité des logs d'exécution dans ce State, la fenêtre de contexte sera instantanément remplie lors du prochain appel du LLM. La vitesse de sauvegarde en base de données ralentit également de manière drastique.
Il convient d'exporter les données volumineuses vers des fichiers externes et de ne conserver que les chemins de référence dans le State.
validation_status et artifact_ref_path, qui contiendra le chemin du fichier de résultat.@traceable de LangSmith pour relier le flux de logs à des fins de suivi.`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
}
`
Cela évite d'inclure des logs inutiles dans le prompt du LLM, et en cas de problème, il suffit d'ouvrir le fichier au chemin indiqué, ce qui rend l'identification de la cause racine bien plus aisée.