إدارة التكلفة

الوكلاء يمكنهم استنزاف ميزانيات API بسرعة. حلقة هاربة أو برومبتات مطولة يمكنها أن تكلف مئات الدولارات في دقائق. إدارة التكلفة ليست اختيارية—إنها بقاء.

فهم تكاليف الوكيل

محرك التكلفة	التأثير	التخفيف
رموز الإدخال	برومبتات النظام، السياق، نتائج الأدوات	الضغط، التخزين المؤقت
رموز الإخراج	ردود مطولة، تفسيرات متكررة	برومبتات موجزة، max_tokens
حلقات الأدوات	الوكيل يستدعي نفس الأداة مراراً	كشف الحلقات، حدود
اختيار النموذج	Opus مقابل Sonnet مقابل Haiku	الحجم المناسب لكل مهمة
إعادات المحاولة الفاشلة	رموز مهدرة على الأخطاء	معالجة أخطاء أفضل

⚠ Prices change frequently. The values above are for illustration only and may be out of date. Always verify current pricing directly with the provider before making cost decisions: Anthropic · OpenAI · Google Gemini · Google Vertex AI · AWS Bedrock · Azure OpenAI · Mistral · Cohere · Together AI · DeepSeek · Groq · Fireworks AI · Perplexity · xAI · Cursor · GitHub Copilot · Windsurf.

إنفاذ الميزانية

حدود صارمة توقف الإنفاق فعلاً:

from dataclasses import dataclass, field
from datetime import datetime, timedelta

@dataclass
class Budget:
    max_tokens: int
    max_requests: int
    max_cost_usd: float
    period: timedelta
    tokens_used: int = 0
    requests_made: int = 0
    cost_incurred: float = 0.0
    period_start: datetime = field(default_factory=datetime.now)

class BudgetEnforcer:
    def __init__(self):
        self.budgets = {}  # user_id -> Budget

    def set_budget(self, user_id: str, budget: Budget):
        self.budgets[user_id] = budget

    def check_budget(self, user_id: str, estimated_tokens: int) -> tuple[bool, str]:
        """تحقق إذا الطلب يناسب الميزانية."""
        budget = self.budgets.get(user_id)
        if not budget:
            return True, ""  # لا ميزانية محددة

        # إعادة التعيين إذا انتهت الفترة
        if datetime.now() - budget.period_start > budget.period:
            budget.tokens_used = 0
            budget.requests_made = 0
            budget.cost_incurred = 0.0
            budget.period_start = datetime.now()

        # تحقق من جميع الحدود
        if budget.tokens_used + estimated_tokens > budget.max_tokens:
            return False, f"تم تجاوز ميزانية الرموز: {budget.tokens_used}/{budget.max_tokens}"

        if budget.requests_made >= budget.max_requests:
            return False, f"تم الوصول لحد الطلبات: {budget.requests_made}/{budget.max_requests}"

        estimated_cost = self.estimate_cost(estimated_tokens)
        if budget.cost_incurred + estimated_cost > budget.max_cost_usd:
            return False, f"تم تجاوز ميزانية التكلفة: ${budget.cost_incurred:.2f}/${budget.max_cost_usd:.2f}"

        return True, ""

    def record_usage(self, user_id: str, tokens: int, cost: float):
        """تسجيل الاستخدام الفعلي بعد الطلب."""
        budget = self.budgets.get(user_id)
        if budget:
            budget.tokens_used += tokens
            budget.requests_made += 1
            budget.cost_incurred += cost

    def estimate_cost(self, tokens: int, model: str = "claude-sonnet-4-6") -> float:
        """تقدير التكلفة بناءً على عدد الرموز."""
        # الأسعار حتى أبريل 2026
        prices = {
            "claude-opus-4-7": {"input": 5.0, "output": 25.0},
            "claude-sonnet-4-6": {"input": 3.0, "output": 15.0},
            "claude-haiku-4-5-20251001": {"input": 1.0, "output": 5.0}
        }
        rate = prices.get(model, prices["claude-sonnet-4-6"])
        # افترض تقسيم 50/50 إدخال/إخراج للتقدير
        return (tokens / 2 * rate["input"] + tokens / 2 * rate["output"]) / 1_000_000

اختيار النموذج حسب المهمة

استخدم أرخص نموذج يعمل:

class ModelSelector:
    def __init__(self):
        self.model_tiers = {
            "haiku": {
                "model": "claude-haiku-4-5-20251001",
                "tasks": ["classification", "extraction", "simple_qa", "routing"],
                "max_complexity": 2
            },
            "sonnet": {
                "model": "claude-sonnet-4-6",
                "tasks": ["coding", "analysis", "summarization", "tool_use"],
                "max_complexity": 7
            },
            "opus": {
                "model": "claude-opus-4-7",
                "tasks": ["research", "complex_reasoning", "creative", "multi_step"],
                "max_complexity": 10
            }
        }

    def select_model(self, task_type: str, complexity: int = 5) -> str:
        """اختيار النموذج الأكثر فعالية من حيث التكلفة للمهمة."""
        for tier_name in ["haiku", "sonnet", "opus"]:
            tier = self.model_tiers[tier_name]
            if task_type in tier["tasks"] and complexity <= tier["max_complexity"]:
                return tier["model"]

        return self.model_tiers["sonnet"]["model"]  # الافتراضي

# الاستخدام
selector = ModelSelector()
model = selector.select_model("classification", complexity=1)  # يرجع haiku
model = selector.select_model("coding", complexity=6)  # يرجع sonnet
model = selector.select_model("research", complexity=9)  # يرجع opus

كشف الحلقات والحدود

منع الوكلاء الهاربين:

class LoopDetector:
    def __init__(self, max_iterations: int = 20, similarity_threshold: float = 0.9):
        self.max_iterations = max_iterations
        self.similarity_threshold = similarity_threshold
        self.history = []

    def check_iteration(self, tool_call: dict) -> tuple[bool, str]:
        """تحقق إذا الوكيل عالق في حلقة."""
        # تحقق من إجمالي التكرارات
        if len(self.history) >= self.max_iterations:
            return False, f"تم الوصول لأقصى تكرارات ({self.max_iterations})"

        # تحقق من الاستدعاءات المتطابقة المتكررة
        call_signature = f"{tool_call['name']}:{json.dumps(tool_call['args'], sort_keys=True)}"

        identical_count = sum(1 for h in self.history[-5:] if h == call_signature)
        if identical_count >= 3:
            return False, "تم اكتشاف استدعاءات أدوات متطابقة متكررة"

        self.history.append(call_signature)
        return True, ""

    def reset(self):
        self.history = []

تخزين البرومبت مؤقتاً

إعادة استخدام السياق المكلف:

class PromptCache:
    def __init__(self, ttl_seconds: int = 300):
        self.cache = {}
        self.ttl = ttl_seconds

    def get_cached_context(self, context_key: str) -> str | None:
        """استرجاع السياق المخزن إذا لا يزال صالحاً."""
        if context_key in self.cache:
            entry = self.cache[context_key]
            if time.time() - entry["timestamp"] < self.ttl:
                return entry["content"]
            del self.cache[context_key]
        return None

    def cache_context(self, context_key: str, content: str):
        """تخزين السياق المكلف لإعادة الاستخدام."""
        self.cache[context_key] = {
            "content": content,
            "timestamp": time.time()
        }

# مثال: تخزين برومبتات النظام ومحتويات الملفات الكبيرة
async def get_agent_context(project_id: str) -> str:
    cache_key = f"project:{project_id}"

    cached = prompt_cache.get_cached_context(cache_key)
    if cached:
        return cached  # وفر الرموز!

    # بناء السياق المكلف
    context = await build_project_context(project_id)
    prompt_cache.cache_context(cache_key, context)
    return context

لوحة تحكم التكلفة

تتبع وتصور الإنفاق:

class CostDashboard:
    def __init__(self):
        self.usage_log = []

    def log_request(self, user_id: str, model: str, input_tokens: int, output_tokens: int):
        """تسجيل كل طلب مع تفصيل التكلفة الكامل."""
        cost = self.calculate_cost(model, input_tokens, output_tokens)
        self.usage_log.append({
            "timestamp": datetime.now(),
            "user_id": user_id,
            "model": model,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "cost_usd": cost
        })

    def get_daily_summary(self, date: datetime = None) -> dict:
        """الحصول على ملخص التكلفة ليوم محدد."""
        date = date or datetime.now()
        day_start = date.replace(hour=0, minute=0, second=0)
        day_end = day_start + timedelta(days=1)

        day_logs = [l for l in self.usage_log
                    if day_start <= l["timestamp"] < day_end]

        return {
            "date": date.date().isoformat(),
            "total_requests": len(day_logs),
            "total_tokens": sum(l["input_tokens"] + l["output_tokens"] for l in day_logs),
            "total_cost_usd": sum(l["cost_usd"] for l in day_logs),
            "by_model": self.group_by_model(day_logs),
            "by_user": self.group_by_user(day_logs)
        }

ملاحظة نيردية: أفضل تحسين للتكلفة هو عدم إجراء الاستدعاء أصلاً. هل يمكنك تخزينه مؤقتاً؟ هل يمكن لنموذج أصغر فعله؟ هل المستخدم يحتاج هذه الميزة فعلاً؟

الوحدة التالية: إلى أين تتجه الوكلاء. :::