Claude Code CLI लागत विस्फोट और अनंत लूप को रोकने के लिए एजेंट पाइपलाइन निर्माण विधि
makedreamJuly 29, 2026
0
Computing/Software
Comments (0)
Log in to leave a comment
No posts yet
makedreammakedreamLog in to leave a comment
No posts yet
एंटरप्राइज वातावरण में LangGraph या AutoGen आधारित एजेंट को उत्पादन (production) में लागू करने वाले किसी भी AI इंजीनियर ने कभी न कभी एक भयावह अनुभव जरूर किया होगा। Claude Code CLI को बैकग्राउंड प्रोसेस के रूप में चलाकर छोड़ देने पर, कोड में सिर्फ एक खराबी के कारण वह खुद ही टूल को बार-बार कॉल करता रहता है और 10 मिनट के भीतर सैकड़ों डॉलर के टोकन खपत कर जाता है।
यह केवल प्रॉम्प्ट को अच्छे से लिखने से हल होने वाली समस्या नहीं है। बैकग्राउंड कंट्रोल में विफलता, नोड्स के बीच अनंत प्रतीक्षा, ब्राउज़र मेमोरी क्रैश जैसी वास्तविक ऑपरेटिंग वातावरण में उत्पन्न होने वाली सिस्टम-लेवल की बाधाओं को हल करना होगा। प्रोडक्शन पाइपलाइन में तुरंत उपयोग किए जा सकने वाले चार रेस्पॉन्स कोड और आर्किटेक्चर यहाँ प्रस्तुत हैं।
-p विकल्प निष्पादित करते समय टोकन विस्फोट और प्रोसेस ज़ोंबीकरण को रोकनाAnthropic के Claude 3 Opus जैसे मॉडल को बिना बातचीत वाले एसिंक्रोनस मोड (-p फ्लैग) में चलाने पर, परीक्षण विफल होने पर उपयोगकर्ता के हस्तक्षेप के बिना तुरंत टूल को फिर से कॉल किया जाता है। इस समय यदि अनंत लूप चलता है, तो मल्टी-स्टेप तर्क इतिहास लगातार जमा होता रहता है और हर टर्न में भारी मात्रा में टोकन की खपत होती है। प्रोसेस समाप्त न होकर सिस्टम मेमोरी में बना रहने वाला ज़ोंबी प्रोसेस भी इसी समय उत्पन्न होता है।
इस स्थिति को रोकने के लिए 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 से अधिक नहीं होगी। लूप के उलझने से होने वाले खर्च के विस्फोट को रोका जा सकता है।
LangGraph की डिफ़ॉल्ट RetryPolicy नेटवर्क त्रुटियों से निपटने के लिए है। जब LLM तर्क गलत लिखता है और सत्यापन विफल हो जाता है, तो यह डिफ़ॉल्ट नीति ठीक से काम नहीं करती है। पुनः प्रयास की सीमा पार होते ही GraphRecursionError थ्रो होता है और पूरा सिस्टम रुक जाता है।
यदि वही त्रुटि बार-बार दोहराई जाती है, तो निष्पादन को रोककर नियंत्रण हस्तांतरित करने वाला सर्किट ब्रेकर (Circuit Breaker) State के भीतर बनाया जाना चाहिए।
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()
`
लगातार 2 बार विफल होने पर यह अनंत लूप को तोड़कर तुरंत स्टैंडबाय स्थिति में बदल जाता है, जिससे एजेंट के गतिरोध (deadlock) में फंसने और पूरी प्रक्रिया के रुकने की समस्या से बचाव होता है।
स्क्रीनशॉट लेने या DOM संरचना को सत्यापित करने वाले नोड Chrome Headless Shell चलाते हैं। समस्या यह है कि जैसे ही यह कोड Docker या GitHub Actions Runner वातावरण में जाता है, Chromium रेंडरर तुरंत क्रैश हो जाता है। Docker की डिफ़ॉल्ट शेयर की गई मेमोरी (/dev/shm) की क्षमता केवल 64MB होती है, जिससे स्क्रीन कैप्चर करते समय 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/
`
मेमोरी स्पेस बढ़ाने और सेशन को पूरी तरह से अलग करने पर, बैकग्राउंड वातावरण में भी ब्राउज़र सत्यापन नोड क्रैश हुए बिना अंत तक चलता रहता है।
LangGraph का मुख्य State एक साझा मेमोरी है जिसका उपयोग सभी नोड मिलकर करते हैं। यदि सत्यापन नोड द्वारा कैप्चर किया गया लंबा HTML कोड या संपूर्ण निष्पादन लॉग सीधे इस State में डाल दिया जाता है, तो अगली बार LLM को कॉल करते समय कॉन्टेक्स्ट विंडो तुरंत भर जाती है। DB सेव करने की गति भी काफी धीमी हो जाती है।
बड़े डेटा को बाहरी फ़ाइलों में निकालना चाहिए और State में केवल संदर्भ पथ (reference path) छोड़ना चाहिए।
validation_status और परिणाम फ़ाइल पथ रखने के लिए artifact_ref_path छोड़ें।@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 प्रॉम्प्ट में अनावश्यक लॉग के शामिल होने को रोकता है, और कोई समस्या आने पर केवल निर्दिष्ट पथ की फ़ाइल खोलकर देखना होता है, जिससे कारण की पहचान करना काफी आसान हो जाता है।