klingai-cost-controls

Kling AI Cost Controls

Overview

Prevent unexpected spending with per-request cost estimation, daily budget enforcement, threshold alerts, and usage dashboards. Credits are consumed per task based on duration, mode, and audio.

Credit Cost Reference

Budget Guard

import time
from dataclasses import dataclass, field

@dataclass
class BudgetGuard:
    """Enforce daily credit budget with alerting."""
    daily_limit: int = 1000
    alert_threshold: float = 0.8  # alert at 80%
    _used: int = 0
    _reset_time: float = field(default_factory=time.time)
    _alerts_sent: set = field(default_factory=set)

    def _check_reset(self):
        if time.time() - self._reset_time > 86400:
            self._used = 0
            self._reset_time = time.time()
            self._alerts_sent.clear()

    def estimate_credits(self, duration: int = 5, mode: str = "standard",
                         audio: bool = False) -> int:
        base = {(5, "standard"): 10, (5, "professional"): 35,
                (10, "standard"): 20, (10, "professional"): 70}
        credits = base.get((duration, mode), 10)
        if audio:
            credits *= 5
        return credits

    def check(self, credits_needed: int) -> bool:
        self._check_reset()

        # Check alert threshold
        usage_pct = (self._used + credits_needed) / self.daily_limit
        if usage_pct >= self.alert_threshold and "80pct" not in self._alerts_sent:
            self._alerts_sent.add("80pct")
            self._on_alert(f"Budget at {usage_pct:.0%} ({self._used + credits_needed}/{self.daily_limit})")

        if self._used + credits_needed > self.daily_limit:
            raise RuntimeError(
                f"Daily budget exceeded: {self._used} + {credits_needed} > {self.daily_limit} credits"
            )
        return True

    def record(self, credits: int):
        self._used += credits

    def _on_alert(self, message: str):
        """Override for custom alerting (Slack, email, PagerDuty)."""
        print(f"ALERT: {message}")

    @property
    def remaining(self) -> int:
        self._check_reset()
        return max(0, self.daily_limit - self._used)

    @property
    def usage_report(self) -> dict:
        self._check_reset()
        return {
            "used": self._used,
            "limit": self.daily_limit,
            "remaining": self.remaining,
            "usage_pct": f"{(self._used / self.daily_limit) * 100:.1f}%",
        }

Pre-Batch Cost Check

def pre_batch_check(prompts: list, budget: BudgetGuard,
                    duration: int = 5, mode: str = "standard"):
    """Estimate and validate batch cost before submission."""
    per_video = budget.estimate_credits(duration, mode)
    total = len(prompts) * per_video

    print(f"Batch estimate: {len(prompts)} videos x {per_video} credits = {total} credits")
    print(f"Budget remaining: {budget.remaining}")

    if total > budget.remaining:
        raise RuntimeError(
            f"Batch needs {total} credits but only {budget.remaining} remaining. "
            f"Reduce to {budget.remaining // per_video} videos or lower mode."
        )
    return total

Cost-Aware Client

class CostAwareKlingClient:
    """Kling client that enforces budget on every request."""

    def __init__(self, base_client, budget: BudgetGuard):
        self.client = base_client
        self.budget = budget

    def text_to_video(self, prompt: str, **kwargs):
        credits = self.budget.estimate_credits(
            kwargs.get("duration", 5),
            kwargs.get("mode", "standard"),
            kwargs.get("audio", False),
        )
        self.budget.check(credits)

        result = self.client.text_to_video(prompt, **kwargs)
        self.budget.record(credits)
        return result

Optimization Strategies

Usage Tracking

import json
from datetime import datetime

class UsageTracker:
    """Log every generation for cost analysis."""

    def __init__(self, log_file: str = "kling_usage.jsonl"):
        self.log_file = log_file

    def log(self, task_id: str, credits: int, model: str,
            duration: int, mode: str, prompt: str):
        entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "task_id": task_id,
            "credits": credits,
            "model": model,
            "duration": duration,
            "mode": mode,
            "prompt_preview": prompt[:100],
        }
        with open(self.log_file, "a") as f:
            f.write(json.dumps(entry) + "\n")

Prerequisites

• An approved credit budget, synthetic or rights-cleared test brief, authorized workspace, policy review, draft-only destination, redaction policy, and rollback owner.

Instructions

1. Measure credit controls with a bounded sandbox canary and log aggregate consumption only; never store prompt previews, private data, asset URLs, or credentials in cost records.
2. Verify the approved budget, policy/rights outcome, destination, retention, and cancellation path before submitting work.
3. Halt and cancel queued drafts on a credit anomaly, policy concern, or retention drift; restore the prior budget configuration before retrying.
4. Retain a redacted cost receipt only for the approved window, then remove temporary assets and test records.

Output

Produce a cost-control receipt with environment, approved/observed aggregate credits, model/duration category, policy/rights and draft-only checks, cancellation or rollback action, owner approval, and cleanup proof. Exclude prompts, asset URLs, identities, and secrets.

Error Handling

Examples

env=ci-sandbox; category=standard-5s; budget=40-credits; observed=20-credits; policy=pass; destination=draft-only; cleanup=verified supports review.

Resources

• Pricing
• Developer Portal