Оптимизация расходов на ИИ-кодинг: руководство по рабочему процессу разработчика (2026)

from openai import OpenAI

client = OpenAI(
    api_key="tr-your-key-here",
    base_url="https://api.teamorouter.com/v1",
)

def classify_task(prompt: str) -> str:
    """Classify a task by its complexity to route to the right model tier."""
    prompt_lower = prompt.lower()

    # Simple: pattern-matching recognizable low-complexity tasks
    simple_patterns = [
        "format", "lint", "indent", "prettify", "add comment",
        "add docstring", "add type hints", "rename variable",
        "simple test", "unit test for", "boilerplate",
    ]
    if any(p in prompt_lower for p in simple_patterns):
        return "simple"

    # Complex: architecture, multi-file, hard debugging
    complex_patterns = [
        "architecture", "design system", "refactor across",
        "multi-module", "hard bug", "race condition",
        "deadlock", "memory leak", "distributed",
        "design pattern for", "system design",
    ]
    if any(p in prompt_lower for p in complex_patterns):
        return "complex"

    # Medium: everything else — feature work, single-file refactors
    return "medium"

def route_model(prompt: str, use_free_tier: bool = True) -> str:
    """Route a prompt to the most cost-effective model for its complexity."""
    tier = classify_task(prompt)

    routing_table = {
        "simple":  "deepseek-v4-flash-free" if use_free_tier else "deepseek-v4-flash",
        "medium":  "deepseek-v4-flash",
        "complex": "deepseek-v4-pro",
    }
    return routing_table[tier]

def ai_complete(prompt: str) -> str:
    """Send a prompt to the appropriate model based on task complexity."""
    model = route_model(prompt)
    resp = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
    )
    return resp.choices[0].message.content

# Usage
result = ai_complete("Add type hints to all functions in this module.")
# -> routed to deepseek-v4-flash-free (simple task, zero cost)

from openai import OpenAI
import time
from collections import deque

client = OpenAI(
    api_key="tr-your-key-here",
    base_url="https://api.teamorouter.com/v1",
)

class FreeTierManager:
    """Track free tier usage and fall back to paid models when quota is exhausted."""

    def __init__(self, daily_limit: int = 200):
        self.daily_limit = daily_limit
        self.usage_timestamps = deque()
        self.free_model = "deepseek-v4-flash-free"
        self.paid_fallback = "deepseek-v4-flash"

    def _prune_old_entries(self):
        """Remove timestamps older than 24 hours."""
        cutoff = time.time() - 86400
        while self.usage_timestamps and self.usage_timestamps[0] < cutoff:
            self.usage_timestamps.popleft()

    def get_model(self, task_tier: str) -> str:
        """Return the best model for a task tier, respecting free tier limits."""
        self._prune_old_entries()

        if task_tier == "simple" and len(self.usage_timestamps) < self.daily_limit:
            self.usage_timestamps.append(time.time())
            return self.free_model

        # Free tier exhausted or task needs paid quality
        if task_tier == "simple":
            return self.paid_fallback
        elif task_tier == "medium":
            return "deepseek-v4-flash"
        else:
            return "deepseek-v4-pro"


free_tier = FreeTierManager(daily_limit=200)

def smart_complete(prompt: str) -> str:
    tier = classify_task(prompt)
    model = free_tier.get_model(tier)
    resp = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
    )
    return resp.choices[0].message.content

# Good: cache-friendly structure
messages = [
    {
        "role": "system",
        "content": (
            "You are a senior Python engineer. Follow these conventions:\n"
            "- Use type hints on all function signatures\n"
            "- Prefer dataclasses over dicts for structured data\n"
            "- Use pathlib over os.path\n"
            "- Max line length: 100 characters\n"
            "- Docstrings: Google style"
        ),
    },
    # Static reference context — also cached
    {
        "role": "user",
        "content": (
            "Project structure:\n"
            "src/models/ - data models\n"
            "src/services/ - business logic\n"
            "src/api/ - FastAPI routes\n"
            "tests/ - pytest test suite\n"
            "Requirements: pytest, fastapi, sqlalchemy, pydantic"
        ),
    },
    # Dynamic request — varies per call, but prefix is cached
    {
        "role": "user",
        "content": "Add a new endpoint to create a user with email validation.",
    },
]

resp = client.chat.completions.create(
    model="deepseek-v4-flash",
    messages=messages,
)

import json
from openai import OpenAI

client = OpenAI(
    api_key="tr-your-key-here",
    base_url="https://api.teamorouter.com/v1",
)

# Routing tiers: cheapest to most expensive
TIERS = [
    {"model": "deepseek-v4-flash-free", "max_retries": 0, "quality": "low"},
    {"model": "deepseek-v4-flash", "max_retries": 1, "quality": "medium"},
    {"model": "deepseek-v4-pro", "max_retries": 1, "quality": "high"},
    {"model": "claude-sonnet-4-6", "max_retries": 0, "quality": "frontier"},
]

def validate_output(task_type: str, output: str) -> bool:
    """Check if the output is acceptable for the given task type."""
    if not output or len(output.strip()) < 10:
        return False

    # For code tasks, check for obvious failures
    if task_type in ("code", "refactor", "test"):
        failure_markers = [
            "I cannot", "I'm unable", "I apologize",
            "as an AI", "I don't have", "unfortunately",
            "```\n```",  # empty code block
        ]
        if any(m in output.lower() for m in failure_markers):
            return False

    return True

def hybrid_complete(prompt: str, task_type: str = "general") -> str:
    """Try tiers sequentially until output passes validation."""
    last_error = None

    for tier in TIERS:
        for attempt in range(tier["max_retries"] + 1):
            try:
                resp = client.chat.completions.create(
                    model=tier["model"],
                    messages=[{"role": "user", "content": prompt}],
                    temperature=0.3 if attempt > 0 else 0.0,
                )
                output = resp.choices[0].message.content

                if validate_output(task_type, output):
                    return output

            except Exception as e:
                last_error = e
                continue

        # This tier failed all attempts; escalate to next tier
        continue

    raise RuntimeError(f"All tiers exhausted. Last error: {last_error}")

# Usage
code = hybrid_complete(
    "Write a Python decorator that retries a function with exponential backoff.",
    task_type="code",
)

import datetime
import pytz

BEIJING_TZ = pytz.timezone("Asia/Shanghai")

PEAK_WINDOWS = [
    (datetime.time(9, 0), datetime.time(12, 0)),   # Morning peak
    (datetime.time(14, 0), datetime.time(18, 0)),  # Afternoon peak
]

def is_peak_time() -> bool:
    """Check if current Beijing time is within a peak pricing window."""
    now = datetime.datetime.now(BEIJING_TZ).time()
    for start, end in PEAK_WINDOWS:
        if start <= now < end:
            return True
    return False

def get_deepseek_model(base_model: str) -> str:
    """Return the base model — your gateway handles peak pricing transparently."""
    # TeamoRouter's Agentic Routing can be configured to prefer cheaper
    # models during peak hours automatically. For direct DeepSeek API,
    # you would check is_peak_time() and potentially switch models.
    return base_model

# For batch workloads, use a scheduler that defers to off-peak:
def schedule_batch_job(job_fn, prefer_off_peak: bool = True):
    """Run a job now, or defer to off-peak if configured."""
    if prefer_off_peak and is_peak_time():
        # Calculate seconds until next off-peak window
        now = datetime.datetime.now(BEIJING_TZ)
        next_off_peak = now.replace(hour=18, minute=0, second=0, microsecond=0)
        if now.hour >= 18:
            next_off_peak = next_off_peak + datetime.timedelta(days=1)
            next_off_peak = next_off_peak.replace(hour=0, minute=0)
        wait_seconds = (next_off_peak - now).total_seconds()
        print(f"Peak time. Deferring job for {wait_seconds/3600:.1f} hours.")
        # In production: enqueue to a job queue with delay
        # For now: just warn
        return None

    return job_fn()

import tiktoken
from dataclasses import dataclass, field
from typing import Dict

@dataclass
class TokenBudget:
    """Token budget configuration per task type."""
    daily_limit: int
    per_request_limit: int
    used_today: int = 0

class TokenTracker:
    """Track token usage and enforce budgets."""

    def __init__(self):
        self.budgets: Dict[str, TokenBudget] = {
            "simple_edit":     TokenBudget(daily_limit=100_000, per_request_limit=8_000),
            "test_write":      TokenBudget(daily_limit=200_000, per_request_limit=16_000),
            "feature_work":    TokenBudget(daily_limit=500_000, per_request_limit=32_000),
            "architecture":    TokenBudget(daily_limit=1_000_000, per_request_limit=64_000),
        }
        self.enc = tiktoken.get_encoding("cl100k_base")

    def count_tokens(self, text: str) -> int:
        return len(self.enc.encode(text))

    def check_budget(self, task_type: str, prompt: str) -> bool:
        budget = self.budgets.get(task_type)
        if not budget:
            return True  # No budget configured

        prompt_tokens = self.count_tokens(prompt)
        if prompt_tokens > budget.per_request_limit:
            print(f"WARNING: Prompt ({prompt_tokens} tokens) exceeds "
                  f"per-request limit ({budget.per_request_limit}) for {task_type}")
            return False

        if budget.used_today + prompt_tokens > budget.daily_limit:
            print(f"WARNING: Daily budget for {task_type} exceeded "
                  f"({budget.used_today}/{budget.daily_limit} tokens used)")
            return False

        return True

    def record_usage(self, task_type: str, tokens: int):
        if task_type in self.budgets:
            self.budgets[task_type].used_today += tokens


tracker = TokenTracker()

def budgeted_complete(prompt: str, task_type: str = "feature_work") -> str | None:
    if not tracker.check_budget(task_type, prompt):
        return None  # Budget exceeded; caller should handle

    model = route_model(prompt)
    resp = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
    )
    output = resp.choices[0].message.content

    # Record usage
    prompt_tokens = tracker.count_tokens(prompt)
    output_tokens = tracker.count_tokens(output)
    tracker.record_usage(task_type, prompt_tokens + output_tokens)

    return output

"""
Cost-optimized LLM routing for AI coding workflows.
Drop-in replacement for any openai.Completions-based client.

Strategies applied:
  1. Model tiering by task complexity
  2. Free tier utilization (200 req/day)
  3. Prompt caching (stable system prompts)
  4. Hybrid routing with escalation
  5. Off-peak awareness (warns on peak)
  6. Token budgeting
"""

from openai import OpenAI
import time
from collections import deque
from functools import lru_cache


class OptimizedRouter:
    def __init__(self, api_key: str, base_url: str = "https://api.teamorouter.com/v1"):
        self.client = OpenAI(api_key=api_key, base_url=base_url)
        self.free_usage = deque()
        self.free_daily_limit = 200

    # --- Free tier tracking ---
    def _free_remaining(self) -> int:
        cutoff = time.time() - 86400
        while self.free_usage and self.free_usage[0] < cutoff:
            self.free_usage.popleft()
        return max(0, self.free_daily_limit - len(self.free_usage))

    def _consume_free(self) -> bool:
        if self._free_remaining() > 0:
            self.free_usage.append(time.time())
            return True
        return False

    # --- Task classification ---
    @staticmethod
    @lru_cache(maxsize=1024)
    def classify(prompt: str) -> str:
        p = prompt.lower()
        if any(k in p for k in ["format", "lint", "add type hint", "add docstring",
                                 "add comment", "rename", "simple test"]):
            return "simple"
        if any(k in p for k in ["architecture", "system design", "distributed",
                                 "race condition", "deadlock", "memory leak"]):
            return "complex"
        return "medium"

    # --- Model routing ---
    def route(self, prompt: str) -> str:
        tier = self.classify(prompt)

        if tier == "simple":
            return "deepseek-v4-flash-free" if self._consume_free() else "deepseek-v4-flash"
        elif tier == "medium":
            return "deepseek-v4-flash"
        else:
            return "deepseek-v4-pro"

    # --- The main call ---
    def complete(self, prompt: str, system: str | None = None) -> str:
        model = self.route(prompt)
        messages = []
        if system:
            messages.append({"role": "system", "content": system})
        messages.append({"role": "user", "content": prompt})

        resp = self.client.chat.completions.create(
            model=model,
            messages=messages,
        )
        return resp.choices[0].message.content


# Usage
router = OptimizedRouter(api_key="tr-your-key-here")

result = router.complete(
    prompt="Add type hints to all functions in this module.",
    system="You are a Python engineer. Write clean, typed, production-quality code.",
)

{
  "cline.provider": "openai-compatible",
  "cline.baseUrl": "https://api.teamorouter.com/v1",
  "cline.apiKey": "tr-your-key-here",
  "cline.model": "deepseek-v4-flash"
}

{
  "provider": {
    "openai": {
      "baseUrl": "https://api.teamorouter.com/v1",
      "apiKey": "tr-your-key-here"
    }
  },
  "model": "deepseek-v4-flash"
}

# Simple task -> V4 Flash (cheapest paid)
curl -s https://api.teamorouter.com/v1/chat/completions \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer tr-your-key-here" \
  -d '{
    "model": "deepseek-v4-flash",
    "messages": [
      {"role": "user", "content": "Generate pytest fixtures for a FastAPI test suite."}
    ]
  }' | jq -r '.choices[0].message.content'

# Complex task -> V4 Pro (more capable)
curl -s https://api.teamorouter.com/v1/chat/completions \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer tr-your-key-here" \
  -d '{
    "model": "deepseek-v4-pro",
    "messages": [
      {"role": "user", "content": "Design the database schema for a multi-tenant SaaS platform."}
    ]
  }' | jq -r '.choices[0].message.content'

# .ai-router.toml — project-level routing configuration
[router]
base_url = "https://api.teamorouter.com/v1"
api_key = "tr-your-key-here"

[router.tiers.simple]
model = "deepseek-v4-flash-free"
fallback = "deepseek-v4-flash"
daily_free_limit = 200

[router.tiers.medium]
model = "deepseek-v4-flash"

[router.tiers.complex]
model = "deepseek-v4-pro"
fallback = "claude-sonnet-4-6"

[router.budgets]
daily_total_tokens = 2_000_000
per_request_tokens = 64_000

[router.scheduling]
avoid_peak_hours = true
peak_timezone = "Asia/Shanghai"
peak_windows = [
    "09:00-12:00",
    "14:00-18:00",
]