كيفية بناء خط أنابيب وكلاء يمنع التكاليف الباهظة والدورات اللانهائية في Claude Code CLI
makedream29 Juli 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 في بيئة المؤسسات قد مر بالتأكيد بتجربة مرعبة واحدة على الأقل. تلك اللحظة التي تترك فيها Claude Code CLI يعمل كعملية في الخلفية، وفقط بسبب خلل بسيط في الكود، يبدأ في إعادة استدعاء الأدوات تلقائياً بمفرده، ليستنزف مئات الدولارات من الرموز (Tokens) في غضون 10 دقائق فقط.
هذه المشكلة لا يمكن حلها لمجرد كتابة المطالبات (Prompts) بشكل جيد. بل يجب معالجة اختناقات مستوى النظام التي تنفجر في بيئات التشغيل الفعلية، مثل فشل التحكم في الخلفية، والانتظار اللانهائي بين العقد، وانهيار ذاكرة المتصفح. لقد قمنا بتلخيص أربعة أكواد واستراتيجيات معمارية قابلة للاستخدام المباشر في خطوط أنابيب الإنتاج.
-pعند تشغيل نماذج مثل Claude 3 Opus من Anthropic في وضع غير متزامن بدون حوار (باستخدام العلامة -p)، فإنه في حالة فشل الاختبار، يعيد استدعاء الأدوات فوراً دون تدخل من المستخدم. في هذه اللحظة، إذا دارت العملية في حلقة لانهائية، تستمر سجلات الاستدلال متعدد الخطوات في التراكم، مما يستهلك كميات هائلة من الرموز في كل دورة. كما تظهر في هذه الأثناء عمليات معلقة (Zombie Processes) تبقى في ذاكرة النظام دون أن تنتهي.
ولمنع هذه الظاهرة، تحتاج إلى غلاف (Wrapper) يفرض حداً أقصى على مستوى CLI ويقوم بإنهاء العملية قسراً على مستوى البرنامج النصي (Script).
--max-budget-usd، وحدد عدد استدعاءات الأدوات باستخدام --max-turns.--allowedTools للسماح بالأدوات المخصصة للقراءة فقط مثل Read و Grep و Glob، وأضف العلامة --bare لإزالة العبء الإضافي لتحميل الملحقات.asyncio.subprocess في Python لإرفاق فئة غلاف (Wrapper Class) تقوم بقتل العملية (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 ويتوقف النظام بأكمله.
إذا تكرر نفس الخطأ باستمرار، يجب إنشاء قاطع دائرة (Circuit Breaker) داخل الحالة (State) يوقف التنفيذ وينقل سلطة التحكم.
verification_attempts ومتغير last_error_signature إلى الحالة (State).Human-in-the-loop لإعادة المهمة بأمان إلى الخيط الأعلى (Upper Thread).`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()
`
نظرًا لأنه يقطع الحلقة اللانهائية عند الفشل لمرتين متتاليتين ويتحول فوراً إلى حالة الانتظار، فإنه يمنع ظاهرة توقف العملية بأكملها نتيجة وقوع الوكيل في حالة جمود (Deadlock).
العقد التي تلتقط لقطات شاشة أو تتحقق من بنية DOM تقوم بتشغيل Chrome Headless Shell. المشكلة هي أنه بمجرد دخول هذا الكود إلى بيئة Docker أو GitHub Actions Runner، يتوقف محرك عرض Chromium عن العمل فوراً. سعة الذاكرة المشتركة الافتراضية في Docker (/dev/shm) هي 64 ميجابايت فقط، مما يسبب خطأ Failed to reserve shared memory عند التقاط الشاشة.
عند تشغيل المتصفح في بيئة الحاويات (Containers)، يجب ضبط الإعدادات التالية:
--shm-size=2g، وأضف العلامتين --disable-dev-shm-usage و --no-sandbox إلى خيارات 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/
`
عن طريق زيادة مساحة الذاكرة وعزل الجلسات تماماً، ستعمل عقدة التحقق من المتصفح حتى النهاية دون أن تتوقف، حتى في بيئات الخلفية.
الحالة الرئيسية (Main State) في LangGraph هي ذاكرة مشتركة تستخدمها جميع العقد معاً. إذا قمت بدفع كود HTML الطويل أو سجلات التنفيذ الكاملة التي التقطتها عقدة التحقق كما هي في هذه الحالة، فستمتلئ نافذة السياق (Context Window) دفعة واحدة في استدعاء 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
}
`
يمكنك منع اختلاط السجلات غير الضرورية في موجه (Prompt) الـ LLM، وعند حدوث مشكلة، يصبح تحديد السبب أسهل بكثير حيث لا يلزم سوى فتح الملف الموجود في المسار المحدد.