Как построить пайплайн агентов, предотвращающий утечку бюджета и бесконечные циклы в 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
Любой AI-инженер, запускавший агентов на базе LangGraph или AutoGen в продакшене корпоративной среды, хотя бы раз сталкивался с ситуацией, от которой стынет кровь. Вы запускаете Claude Code CLI в качестве фонового процесса, а из-за единственного бага в коде агент начинает бесконечно вызывать инструменты по кругу, сжигая токены на сотни долларов за каких-то 10 минут.
Эту проблему не решить простым составлением правильных промптов. Необходимо устранять узкие места системного уровня, возникающие в реальной эксплуатации: сбои фонового контроля, бесконечные ожидания между узлами и падения браузера по памяти. Мы собрали четыре архитектурных подхода с готовым кодом, которые можно сразу внедрить в ваш продакшен-пайплайн.
-pПри запуске моделей уровня Anthropic Claude 3 Opus в асинхронном режиме без диалога (флаг -p), в случае ошибки тестирования агент сразу же снова вызывает инструменты без участия пользователя. Если при этом возникает бесконечный цикл, история многошаговых рассуждений начинает стремительно расти, поедая огромные объемы токенов на каждом шаге. В этот же момент появляются процессы-зомби, которые не завершаются и остаются в системной памяти.
Чтобы предотвратить это, необходима обертка (wrapper), устанавливающая лимиты на уровне CLI и принудительно завершающая процесс на уровне скрипта.
--max-budget-usd и ограничьте количество вызовов инструментов через --max-turns.--allowedTools, чтобы разрешить только инструменты только для чтения, такие как Read, Grep, Glob, и добавьте флаг --bare, чтобы исключить накладные расходы на загрузку плагинов.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. Вы гарантированно защитите себя от незапланированных трат, вызванных зацикливанием.
Стандартная RetryPolicy в LangGraph предназначена для обработки сетевых ошибок. Когда LLM делает логическую ошибку в коде и проверка проваливается, эта стандартная политика не помогает. Как только количество повторов превышает лимит, система выбрасывает GraphRecursionError и полностью останавливается.
Если одна и та же ошибка повторяется регулярно, необходимо создать внутри State автомат защиты (Circuit Breaker), который прервет выполнение и передаст управление дальше.
verification_attempts и переменную last_error_signature.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()
`
После двух неудач подряд схема разрывает бесконечный цикл и переводит процесс в режим ожидания, предотвращая тупиковые ситуации и зависание всего пайплайна.
Узлы, выполняющие снятие скриншотов или валидацию DOM-структуры, запускают Chrome Headless Shell. Проблема заключается в том, что как только этот код попадает в окружение Docker или GitHub Actions Runner, рендерер Chromium моментально падает. Размер разделяемой памяти по умолчанию в Docker (/dev/shm) составляет всего 64 МБ, из-за чего при захвате экрана возникает ошибка Failed to reserve shared memory.
Для стабильного запуска браузера в контейнерах необходимо применить следующие настройки:
--shm-size=2g, а в параметры Chromium добавьте флаги --disable-dev-shm-usage и --no-sandbox.--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/
`
Выделение достаточного объема памяти и полная изоляция сессий позволят узлам браузерной валидации успешно работать даже в фоновом режиме без сбоев.
Основное состояние (Main State) в LangGraph — это общая память, доступная всем узлам. Если узел валидации будет записывать туда длинный HTML-код или полные логи выполнения, контекстное окно LLM переполнится на следующем же шаге. Скорость записи в БД также ощутимо снизится.
Объемные данные следует сохранять во внешние файлы, оставляя в State только пути-ссылки на них.
validation_status и artifact_ref_path для пути к файлу результатов.@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, а при возникновении проблем вы сможете легко локализовать причину, просто открыв нужный файл по указанному пути.