防止 Claude Code CLI 费用暴涨与无限循环的 Agent Pipeline 构建指南

在企业级环境中，凡是将基于 LangGraph 或 AutoGen 的 Agent 推向生产实战的 AI 工程师，恐怕都经历过让人脊背发凉的一幕：将 Claude Code CLI 作为后台进程运行后，仅因一个代码缺陷，Agent 就开始独自不断重新调用工具，短短 10 分钟内便消耗了价值数百美元的 Token。

这绝非仅靠精心编写 Prompt 就能解决的问题。必须解决后台控制失效、节点间无限等待、浏览器内存崩溃等在真实生产环境中爆发的系统级瓶颈。本文整理了四种可直接应用于生产 Pipeline 的应对代码与架构设计。

1. 防止携带 `-p` 选项运行时 Token 暴涨与进程僵尸化

当以无对话的异步模式（-p 标志）运行 Anthropic 的 Claude 3 Opus 等模型时，一旦测试失败，系统会在没有用户干预的情况下立即重新调用工具。此时若陷入无限循环，多步推理历史将不断累积，每一轮都会消耗海量的 Token。同时，进程无法正常终止并残留在系统内存中的僵尸进程也由此产生。

要阻止这一现象，需要从 CLI 层级设置上限，并在脚本层挂载能够强制终止进程的 Wrapper。

在 CLI 调用时，通过 --max-budget-usd 标志指定单次运行的支出上限，并通过 --max-turns 限制工具调用次数。
使用 --allowedTools 选项仅允许 Read、Grep、Glob 等只读工具，并加上 --bare 标志以移除插件加载的额外开销。
使用 Python 的 asyncio.subprocess 挂载一个 Wrapper 类，用于在超过设定时间后强制 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)}

应用此 Wrapper 后，单次验证的费用最高不会超过 $0.50，能够有效拦截因循环逻辑异常而导致的费用爆炸。

2. 实现截断图节点无限重试的熔断器（Circuit Breaker）

LangGraph 默认的 RetryPolicy 是为了应对网络错误而设计的。当 LLM 编写的代码逻辑存在错误导致验证失败时，该默认策略无法正常起效。一旦超过重试上限，就会抛出 GraphRecursionError 并导致整个系统中断。

如果重复发生相同的错误，需要在 State 内部构建一个熔断器（Circuit Breaker），用于挂起执行并将控制权移交出去。

在 State 中增加 verification_attempts 计数器和 last_error_signature 变量。
1 次失败后进入第 2 次重试时，将模型替换为相对轻量化的模型（如 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()

由于连续 2 次失败后会截断无限循环并立即转换为等待状态，因此可以防止 Agent 陷入死锁导致整个流程停滞的情况。

3. CI/CD 与 Docker 内部 Chrome Headless Shell 的内存隔离

用于截屏或验证 DOM 结构的节点需要运行 Chrome Headless Shell。问题在于，这段代码一旦进入 Docker 或 GitHub Actions Runner 环境，Chromium 渲染器就会立刻崩溃。这是因为 Docker 默认的共享内存（/dev/shm）容量只有 64MB，在捕获画面时会触发 Failed to reserve shared memory 错误。

在容器环境中启动浏览器时，必须配置以下设置：

运行 Docker 时使用 --shm-size=2g 选项增大共享内存，并在 Chromium 选项中加入 --disable-dev-shm-usage 及 --no-sandbox 标志。
为避免多次运行间发生 Cookie 或 Local Storage 冲突，按 Session 生成格式为 --user-data-dir=/tmp/session_$RUN_ID 的独立目录。
在 CI/CD 工作流中配置 Artifact 存储步骤，以便后续排查失败原因。

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

通过扩充内存空间并实现 Session 的完全隔离，即使在后台环境中，浏览器验证节点也不会崩溃，能够顺畅运行至结束。

4. 拆分日志监控，保持主 State 清洁

LangGraph 的主 State 是所有节点共同使用的共享内存。如果直接将验证节点捕获的冗长 HTML 代码或完整执行日志原封不动地塞进该 State 中，在下一次调用 LLM 时，上下文窗口（Context Window）会被瞬间填满，数据库的存储速度也会明显变慢。

应该采用将大容量数据提取为外部文件、而在 State 中仅保留引用路径的做法。

主 State 中仅保留 validation_status 以及用于存放结果文件路径的 artifact_ref_path。
验证节点输出的数 MB 级别的文本日志，以 JSON 文件形式导出至本地目录或 S3。
为便于追溯，挂载 LangSmith 的 @traceable 装饰器以串联日志流。

`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 Prompt 中，并且在出现问题时只需打开指定路径的文件即可，使得原因排查变得更加轻松。

防止 Claude Code CLI 费用暴涨与无限循环的 Agent Pipeline 构建指南

1. 防止携带 `-p` 选项运行时 Token 暴涨与进程僵尸化

要阻止这一现象，需要从 CLI 层级设置上限，并在脚本层挂载能够强制终止进程的 Wrapper。

在 CLI 调用时，通过 --max-budget-usd 标志指定单次运行的支出上限，并通过 --max-turns 限制工具调用次数。
使用 --allowedTools 选项仅允许 Read、Grep、Glob 等只读工具，并加上 --bare 标志以移除插件加载的额外开销。
使用 Python 的 asyncio.subprocess 挂载一个 Wrapper 类，用于在超过设定时间后强制 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)}

应用此 Wrapper 后，单次验证的费用最高不会超过 $0.50，能够有效拦截因循环逻辑异常而导致的费用爆炸。

2. 实现截断图节点无限重试的熔断器（Circuit Breaker）

如果重复发生相同的错误，需要在 State 内部构建一个熔断器（Circuit Breaker），用于挂起执行并将控制权移交出去。

在 State 中增加 verification_attempts 计数器和 last_error_signature 变量。
1 次失败后进入第 2 次重试时，将模型替换为相对轻量化的模型（如 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"}

由于连续 2 次失败后会截断无限循环并立即转换为等待状态，因此可以防止 Agent 陷入死锁导致整个流程停滞的情况。

3. CI/CD 与 Docker 内部 Chrome Headless Shell 的内存隔离

在容器环境中启动浏览器时，必须配置以下设置：

运行 Docker 时使用 --shm-size=2g 选项增大共享内存，并在 Chromium 选项中加入 --disable-dev-shm-usage 及 --no-sandbox 标志。
为避免多次运行间发生 Cookie 或 Local Storage 冲突，按 Session 生成格式为 --user-data-dir=/tmp/session_$RUN_ID 的独立目录。
在 CI/CD 工作流中配置 Artifact 存储步骤，以便后续排查失败原因。

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

通过扩充内存空间并实现 Session 的完全隔离，即使在后台环境中，浏览器验证节点也不会崩溃，能够顺畅运行至结束。

4. 拆分日志监控，保持主 State 清洁

应该采用将大容量数据提取为外部文件、而在 State 中仅保留引用路径的做法。

主 State 中仅保留 validation_status 以及用于存放结果文件路径的 artifact_ref_path。
验证节点输出的数 MB 级别的文本日志，以 JSON 文件形式导出至本地目录或 S3。
为便于追溯，挂载 LangSmith 的 @traceable 装饰器以串联日志流。

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

这样可以防止无关日志混入 LLM Prompt 中，并且在出现问题时只需打开指定路径的文件即可，使得原因排查变得更加轻松。

防止 Claude Code CLI 费用暴涨与无限循环的 Agent Pipeline 构建指南

Related Video

Anthropic 刚刚修复了图工程的最大缺陷

防止 Claude Code CLI 费用暴涨与无限循环的 Agent Pipeline 构建指南

1. 防止携带 `-p` 选项运行时 Token 暴涨与进程僵尸化

2. 实现截断图节点无限重试的熔断器（Circuit Breaker）

3. CI/CD 与 Docker 内部 Chrome Headless Shell 的内存隔离

4. 拆分日志监控，保持主 State 清洁

Comments (0)

防止 Claude Code CLI 费用暴涨与无限循环的 Agent Pipeline 构建指南

1. 防止携带 `-p` 选项运行时 Token 暴涨与进程僵尸化

2. 实现截断图节点无限重试的熔断器（Circuit Breaker）

3. CI/CD 与 Docker 内部 Chrome Headless Shell 的内存隔离

4. 拆分日志监控，保持主 State 清洁

防止 Claude Code CLI 费用暴涨与无限循环的 Agent Pipeline 构建指南

1. 防止携带 -p 选项运行时 Token 暴涨与进程僵尸化

2. 实现截断图节点无限重试的熔断器（Circuit Breaker）

3. CI/CD 与 Docker 内部 Chrome Headless Shell 的内存隔离

4. 拆分日志监控，保持主 State 清洁

Comments (0)

防止 Claude Code CLI 费用暴涨与无限循环的 Agent Pipeline 构建指南

1. 防止携带 -p 选项运行时 Token 暴涨与进程僵尸化

2. 实现截断图节点无限重试的熔断器（Circuit Breaker）

3. CI/CD 与 Docker 内部 Chrome Headless Shell 的内存隔离

4. 拆分日志监控，保持主 State 清洁

1. 防止携带 `-p` 选项运行时 Token 暴涨与进程僵尸化

1. 防止携带 `-p` 选项运行时 Token 暴涨与进程僵尸化