Как построить пайплайн агентов, предотвращающий утечку бюджета и бесконечные циклы в Claude Code CLI

Любой AI-инженер, запускавший агентов на базе LangGraph или AutoGen в продакшене корпоративной среды, хотя бы раз сталкивался с ситуацией, от которой стынет кровь. Вы запускаете Claude Code CLI в качестве фонового процесса, а из-за единственного бага в коде агент начинает бесконечно вызывать инструменты по кругу, сжигая токены на сотни долларов за каких-то 10 минут.

Эту проблему не решить простым составлением правильных промптов. Необходимо устранять узкие места системного уровня, возникающие в реальной эксплуатации: сбои фонового контроля, бесконечные ожидания между узлами и падения браузера по памяти. Мы собрали четыре архитектурных подхода с готовым кодом, которые можно сразу внедрить в ваш продакшен-пайплайн.

1. Предотвращение утечки токенов и появления процессов-зомби при выполнении с опцией `-p`

При запуске моделей уровня Anthropic Claude 3 Opus в асинхронном режиме без диалога (флаг -p), в случае ошибки тестирования агент сразу же снова вызывает инструменты без участия пользователя. Если при этом возникает бесконечный цикл, история многошаговых рассуждений начинает стремительно расти, поедая огромные объемы токенов на каждом шаге. В этот же момент появляются процессы-зомби, которые не завершаются и остаются в системной памяти.

Чтобы предотвратить это, необходима обертка (wrapper), устанавливающая лимиты на уровне CLI и принудительно завершающая процесс на уровне скрипта.

При вызове CLI задайте лимит расходов на один запуск с помощью флага --max-budget-usd и ограничьте количество вызовов инструментов через --max-turns.
Используйте параметр --allowedTools, чтобы разрешить только инструменты только для чтения, такие как Read, Grep, Glob, и добавьте флаг --bare, чтобы исключить накладные расходы на загрузку плагинов.
Добавьте класс-обертку на Python с использованием asyncio.subprocess, который принудительно завершает (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)}

С этой оберткой стоимость одной проверки не превысит $0.50. Вы гарантированно защитите себя от незапланированных трат, вызванных зацикливанием.

2. Реализация Circuit Breaker для прерывания бесконечных повторов в узлах графа

Стандартная RetryPolicy в LangGraph предназначена для обработки сетевых ошибок. Когда LLM делает логическую ошибку в коде и проверка проваливается, эта стандартная политика не помогает. Как только количество повторов превышает лимит, система выбрасывает GraphRecursionError и полностью останавливается.

Если одна и та же ошибка повторяется регулярно, необходимо создать внутри State автомат защиты (Circuit Breaker), который прервет выполнение и передаст управление дальше.

Добавьте в State счетчик verification_attempts и переменную last_error_signature.
При переходе ко второй попытке после первого сбоя подмените модель на более легкую (например, Claude 3.5 Haiku), чтобы сократить время маршрутизации.
Если количество одинаковых ошибок превышает 2, передайте управление узлу Human-in-the-loop, чтобы безопасно перенаправить задачу в родительский поток.

`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()

После двух неудач подряд схема разрывает бесконечный цикл и переводит процесс в режим ожидания, предотвращая тупиковые ситуации и зависание всего пайплайна.

3. Изоляция памяти Chrome Headless Shell в CI/CD и Docker

Узлы, выполняющие снятие скриншотов или валидацию DOM-структуры, запускают Chrome Headless Shell. Проблема заключается в том, что как только этот код попадает в окружение Docker или GitHub Actions Runner, рендерер Chromium моментально падает. Размер разделяемой памяти по умолчанию в Docker (/dev/shm) составляет всего 64 МБ, из-за чего при захвате экрана возникает ошибка Failed to reserve shared memory.

Для стабильного запуска браузера в контейнерах необходимо применить следующие настройки:

При запуске Docker увеличьте объем разделяемой памяти с помощью --shm-size=2g, а в параметры Chromium добавьте флаги --disable-dev-shm-usage и --no-sandbox.
Чтобы избежать конфликтов куки и локального хранилища между прогонами, создавайте для каждой сессии изолированную директорию вида --user-data-dir=/tmp/session_$RUN_ID.
Добавьте шаг сохранения артефактов в CI/CD workflow, чтобы вы могли проанализировать причины упавших проверок.

`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: ${{ secrets.ANTHROPIC_API_KEY }} SESSION_STORAGE_DIR: "$ {{ github.workspace }}/artifacts/sessions/${{ github.run_id }}"
run: |
mkdir -p $SESSION_STORAGE_DIR npx ts-node src/cli_validation_entry.ts --user-data-dir="$ SESSION_STORAGE_DIR"
- uses: actions/upload-artifact@v4
if: always()
with:
name: validation-artifacts-${{ github.run_id }}
path: ${{ github.workspace }}/artifacts/

Выделение достаточного объема памяти и полная изоляция сессий позволят узлам браузерной валидации успешно работать даже в фоновом режиме без сбоев.

4. Вынос мониторинга логов без засорения основного State

Основное состояние (Main State) в LangGraph — это общая память, доступная всем узлам. Если узел валидации будет записывать туда длинный HTML-код или полные логи выполнения, контекстное окно LLM переполнится на следующем же шаге. Скорость записи в БД также ощутимо снизится.

Объемные данные следует сохранять во внешние файлы, оставляя в State только пути-ссылки на них.

Оставьте в Main State только поля validation_status и artifact_ref_path для пути к файлу результатов.
Логи объемом в несколько мегабайт, генерируемые узлом проверки, сохраняйте в виде JSON-файлов в локальный каталог или на S3.
Для трассировки используйте декоратор @traceable из LangSmith, чтобы связать поток логов в единую цепочку.

`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
}

Это предотвращает попадание лишних логов в промпт LLM, а при возникновении проблем вы сможете легко локализовать причину, просто открыв нужный файл по указанному пути.

Как построить пайплайн агентов, предотвращающий утечку бюджета и бесконечные циклы в Claude Code CLI

1. Предотвращение утечки токенов и появления процессов-зомби при выполнении с опцией `-p`

При вызове CLI задайте лимит расходов на один запуск с помощью флага --max-budget-usd и ограничьте количество вызовов инструментов через --max-turns.
Используйте параметр --allowedTools, чтобы разрешить только инструменты только для чтения, такие как Read, Grep, Glob, и добавьте флаг --bare, чтобы исключить накладные расходы на загрузку плагинов.
Добавьте класс-обертку на Python с использованием asyncio.subprocess, который принудительно завершает (kill) процесс по истечении заданного времени.

`python
import asyncio
import os
from typing import Any, Dict

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)}

2. Реализация Circuit Breaker для прерывания бесконечных повторов в узлах графа

Добавьте в State счетчик verification_attempts и переменную last_error_signature.
При переходе ко второй попытке после первого сбоя подмените модель на более легкую (например, Claude 3.5 Haiku), чтобы сократить время маршрутизации.
Если количество одинаковых ошибок превышает 2, передайте управление узлу Human-in-the-loop, чтобы безопасно перенаправить задачу в родительский поток.

`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"]

return {"verification_attempts": attempts, "last_error_signature": "SyntaxError", "verification_status": "FAILED"}

def human_in_the_loop_node(state: AgentGraphState) -> dict:
return {"verification_status": "CIRCUIT_BROKEN"}

3. Изоляция памяти Chrome Headless Shell в CI/CD и Docker

Для стабильного запуска браузера в контейнерах необходимо применить следующие настройки:

При запуске Docker увеличьте объем разделяемой памяти с помощью --shm-size=2g, а в параметры Chromium добавьте флаги --disable-dev-shm-usage и --no-sandbox.
Чтобы избежать конфликтов куки и локального хранилища между прогонами, создавайте для каждой сессии изолированную директорию вида --user-data-dir=/tmp/session_$RUN_ID.
Добавьте шаг сохранения артефактов в CI/CD workflow, чтобы вы могли проанализировать причины упавших проверок.

`yaml
name: Agent Background Verification Pipeline
on: [push]

4. Вынос мониторинга логов без засорения основного State

Объемные данные следует сохранять во внешние файлы, оставляя в State только пути-ссылки на них.

Оставьте в Main State только поля validation_status и artifact_ref_path для пути к файлу результатов.
Логи объемом в несколько мегабайт, генерируемые узлом проверки, сохраняйте в виде JSON-файлов в локальный каталог или на S3.
Для трассировки используйте декоратор @traceable из LangSmith, чтобы связать поток логов в единую цепочку.

`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

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
}

Как построить пайплайн агентов, предотвращающий утечку бюджета и бесконечные циклы в Claude Code CLI

Related Video

Anthropic только что устранила главный недостаток графовой инженерии

Как построить пайплайн агентов, предотвращающий утечку бюджета и бесконечные циклы в Claude Code CLI

1. Предотвращение утечки токенов и появления процессов-зомби при выполнении с опцией `-p`

2. Реализация Circuit Breaker для прерывания бесконечных повторов в узлах графа

3. Изоляция памяти Chrome Headless Shell в CI/CD и Docker

4. Вынос мониторинга логов без засорения основного State

Comments (0)

Как построить пайплайн агентов, предотвращающий утечку бюджета и бесконечные циклы в Claude Code CLI

1. Предотвращение утечки токенов и появления процессов-зомби при выполнении с опцией `-p`

2. Реализация Circuit Breaker для прерывания бесконечных повторов в узлах графа

3. Изоляция памяти Chrome Headless Shell в CI/CD и Docker

4. Вынос мониторинга логов без засорения основного State

Как построить пайплайн агентов, предотвращающий утечку бюджета и бесконечные циклы в Claude Code CLI

1. Предотвращение утечки токенов и появления процессов-зомби при выполнении с опцией -p

2. Реализация Circuit Breaker для прерывания бесконечных повторов в узлах графа

3. Изоляция памяти Chrome Headless Shell в CI/CD и Docker

4. Вынос мониторинга логов без засорения основного State

Comments (0)

Как построить пайплайн агентов, предотвращающий утечку бюджета и бесконечные циклы в Claude Code CLI

1. Предотвращение утечки токенов и появления процессов-зомби при выполнении с опцией -p

2. Реализация Circuit Breaker для прерывания бесконечных повторов в узлах графа

3. Изоляция памяти Chrome Headless Shell в CI/CD и Docker

4. Вынос мониторинга логов без засорения основного State

1. Предотвращение утечки токенов и появления процессов-зомби при выполнении с опцией `-p`

1. Предотвращение утечки токенов и появления процессов-зомби при выполнении с опцией `-p`