Como construir uma pipeline de agentes para evitar custos descontrolados e loops infinitos no 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
Qualquer engenheiro de IA que já tenha colocado um agente baseado em LangGraph ou AutoGen em produção em um ambiente corporativo provavelmente já passou por uma experiência de gelar o sangue. É aquele cenário em que você deixa o Claude Code CLI rodando como um processo em segundo plano e, por causa de uma única falha no código, ele fica chamando ferramentas repetidamente por conta própria, consumindo centenas de dólares em tokens em apenas 10 minutos.
Este não é um problema que possa ser resolvido simplesmente escrevendo bons prompts. É necessário resolver gargalos em nível de sistema que acontecem no ambiente de produção real, como falhas de controle em segundo plano, esperas infinitas entre nós e travamentos de memória no navegador. Compilamos quatro soluções em código e arquitetura prontos para uso em pipelines de produção.
-pQuando você executa modelos como o Claude 3 Opus da Anthropic no modo assíncrono sem interação (flag -p), se um teste falhar, o modelo chama as ferramentas novamente de imediato, sem intervenção do usuário. Se um loop infinito ocorrer nesse momento, o histórico de inferência multi-step se acumula continuamente, consumindo uma quantidade enorme de tokens a cada turno. Processos zumbis que não são encerrados e permanecem na memória do sistema também são criados nessa situação.
Para evitar esse fenômeno, é necessário um wrapper que imponha limites no nível da CLI e possa forçar o encerramento do processo no nível do script.
--max-budget-usd e limite o número de chamadas de ferramentas com --max-turns.Read, Grep e Glob usando a opção --allowedTools e adicione a flag --bare para remover o overhead de carregamento de plugins.asyncio.subprocess do Python para anexar uma classe wrapper que encerra (kill) o processo após um determinado tempo.`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)}
`
Aplicando este wrapper, o custo por validação não ultrapassará US$ 0,50. Isso evita explosões de orçamento causadas por loops travados.
O RetryPolicy padrão do LangGraph foi projetado para lidar com erros de rede. Essa política padrão não funciona corretamente quando um LLM escreve uma lógica incorreta e falha na validação. No momento em que o limite de tentativas é excedido, um GraphRecursionError é gerado, travando todo o sistema.
Se o mesmo erro se repetir continuamente, você deve criar um disjuntor (Circuit Breaker) dentro do State para interromper a execução e transferir o controle.
verification_attempts e uma variável last_error_signature ao State.Human-in-the-loop para redirecionar a tarefa com segurança para uma thread 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()
`
Como ele quebra o loop infinito após duas falhas consecutivas e muda imediatamente para um estado de espera, evita que o agente entre em um impasse (deadlock) e trave todo o processo.
Nós que tiram capturas de tela ou validam estruturas DOM executam o Chrome Headless Shell. O problema é que, no momento em que esse código entra em um ambiente Docker ou GitHub Actions Runner, o renderizador do Chromium trava imediatamente. A capacidade padrão da memória compartilhada (/dev/shm) do Docker é de apenas 64 MB, o que gera o erro Failed to reserve shared memory ao capturar telas.
Ao executar um navegador em um ambiente de contêiner, você deve ajustar as seguintes configurações:
--shm-size=2g e adicione as flags --disable-dev-shm-usage e --no-sandbox às opções do Chromium.--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/
`
Aumentando o espaço de memória e isolando totalmente as sessões, o nó de validação do navegador executa até o fim sem falhar, mesmo em ambientes em segundo plano.
O State principal do LangGraph é uma memória compartilhada usada por todos os nós. Se você injetar códigos HTML longos ou logs de execução inteiros capturados por um nó de validação diretamente nesse State, a janela de contexto será preenchida de uma só vez na próxima chamada ao LLM. A velocidade de salvamento no banco de dados também diminui visivelmente.
Você deve usar uma abordagem onde dados volumosos são salvos em arquivos externos, deixando apenas o caminho de referência no State.
validation_status e o artifact_ref_path, que armazenará o caminho do arquivo de resultado.@traceable do LangSmith para conectar o fluxo de logs.`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
}
`
Isso previne que logs desnecessários sejam misturados aos prompts do LLM e torna a identificação de problemas muito mais fácil, pois basta abrir o arquivo no caminho especificado quando ocorrer um erro.