Cómo construir una canalización de agentes que evite la explosión de costos y los bucles infinitos en 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
Cualquier ingeniero de IA que haya puesto en producción agentes basados en LangGraph o AutoGen en un entorno empresarial habrá experimentado, al menos una vez, una situación escalofriante: dejar Claude Code CLI ejecutándose como un proceso en segundo plano y que, debido a un solo fallo en el código, el agente comience a llamar a las herramientas una y otra vez de forma autónoma, consumiendo cientos de dólares en tokens en cuestión de 10 minutos.
Este no es un problema que se resuelva simplemente redactando mejores prompts. Es necesario solucionar cuellos de botella a nivel de sistema que explotan en entornos de producción reales, como fallos en el control de procesos en segundo plano, esperas infinitas entre nodos y bloqueos de memoria en el navegador. A continuación, he recopilado cuatro arquitecturas y sus correspondientes códigos de respuesta listos para usar en canalizaciones de producción.
-pAl ejecutar modelos como Claude 3 Opus de Anthropic en modo asíncrono sin interacción (-p), el sistema vuelve a llamar inmediatamente a las herramientas tras un fallo en las pruebas sin la intervención del usuario. Si se produce un bucle infinito en este punto, el historial de razonamiento multipasos se acumula continuamente, consumiendo una cantidad enorme de tokens en cada turno. En este escenario también se generan procesos zombi que permanecen en la memoria del sistema sin finalizar.
Para evitar este fenómeno, se necesita un tope a nivel de CLI y un envoltorio (wrapper) en el script que pueda forzar la finalización del proceso.
--max-budget-usd y limite el número de llamadas a herramientas con --max-turns.Read, Grep y Glob mediante la opción --allowedTools, y añada la marca --bare para eliminar la sobrecarga de carga de plugins.asyncio.subprocess de Python para forzar la finalización (Kill) del proceso tras un tiempo determinado.`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)}
`
Al aplicar este wrapper, el costo por validación no superará los $0.50. De esta forma, se pueden bloquear de raíz las explosiones de gastos causadas por bucles descontrolados.
La directiva RetryPolicy predeterminada de LangGraph está diseñada para errores de red. Cuando un LLM escribe mal la lógica y falla en la validación, esta política predeterminada no funciona adecuadamente. Al superar el límite de reintentos, se lanza un error GraphRecursionError y todo el sistema se detiene.
Si el mismo error se repite continuamente, se debe crear un cortacircuitos (Circuit Breaker) dentro del estado (State) para detener la ejecución y transferir el control.
verification_attempts y una variable last_error_signature al State.Human-in-the-loop para devolver la tarea de forma segura a un hilo superior.`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()
`
Dado que rompe el bucle infinito y pasa inmediatamente al estado de espera tras dos fallos consecutivos, evita que el agente quede bloqueado en un punto muerto y paralice todo el proceso.
Los nodos que toman capturas de pantalla o validan la estructura del DOM ejecutan Chrome Headless Shell. El problema es que tan pronto como este código ingresa a un entorno Docker o GitHub Actions Runner, el renderizador de Chromium falla de inmediato. La memoria compartida predeterminada (/dev/shm) de Docker es de solo 64 MB, lo que provoca el error Failed to reserve shared memory al capturar la pantalla.
Al iniciar el navegador en un entorno de contenedores, se debe ajustar la siguiente configuración:
--shm-size=2g y añada las marcas --disable-dev-shm-usage y --no-sandbox en las opciones de Chromium.--user-data-dir=/tmp/session_$RUN_ID para evitar conflictos de cookies o almacenamiento local entre ejecuciones.`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/
`
Al aumentar el espacio de memoria y aislar completamente las sesiones, el nodo de validación del navegador se ejecuta sin fallos hasta el final, incluso en entornos en segundo plano.
El State principal de LangGraph es una memoria compartida utilizada por todos los nodos. Si se introduce directamente en este State el código HTML extenso o la totalidad de los registros de ejecución capturados por el nodo de validación, la ventana de contexto se llenará de golpe en la siguiente llamada al LLM. Además, la velocidad de guardado en la base de datos disminuirá notablemente.
Se debe utilizar un enfoque en el que los datos de gran volumen se extraigan a un archivo externo, dejando únicamente la ruta de referencia en el State.
State principal solo validation_status y artifact_ref_path, que contendrá la ruta del archivo de resultados.@traceable de LangSmith para vincular el flujo de registros con fines de rastreo.`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
}
`
Esto evita que se mezclen registros innecesarios en el prompt del LLM y, si ocurre un problema, resulta mucho más sencillo identificar la causa, ya que solo es necesario abrir el archivo ubicado en la ruta especificada.