موازنة الحمل بين النماذج

لماذا توجيه الطلبات عبر نماذج متعددة؟

التوجيه بناءً على التعقيد

COMPLEXITY_CLASSIFIER_PROMPT = '''Classify the complexity of this user request.
Return ONLY one word: SIMPLE, MODERATE, or COMPLEX.

SIMPLE: greeting, factual lookup, single-step question, direct answer needed
MODERATE: multi-step explanation, comparison, short analysis, code snippet
COMPLEX: deep analysis, long code generation, reasoning chain, specialized domain

Request: {request}'''

import openai

client_mini = openai.OpenAI(api_key='YOUR_API_KEY')

def classify_complexity(request):
    response = client_mini.chat.completions.create(
        model='gpt-4o-mini',  # always use cheap model for classifier
        messages=[{'role': 'user', 'content':
            COMPLEXITY_CLASSIFIER_PROMPT.format(request=request)}],
        max_tokens=5,
        temperature=0
    )
    label = response.choices[0].message.content.strip().upper()
    if label not in ('SIMPLE', 'MODERATE', 'COMPLEX'):
        label = 'MODERATE'  # safe default
    return label

for req in ['Hi', 'Explain quicksort', 'Design a distributed systems architecture']:
    print(f'{req[:40]}: {classify_complexity(req)}')

فئة موجّه النموذج

MODEL_ROUTES = {
    'SIMPLE': {
        'model': 'gpt-4o-mini',
        'max_tokens': 300,
        'cost_per_1k_input': 0.00015,
        'use_case': 'greetings, FAQ, simple factual questions'
    },
    'MODERATE': {
        'model': 'gpt-4o',
        'max_tokens': 1500,
        'cost_per_1k_input': 0.0025,
        'use_case': 'explanations, analysis, code snippets'
    },
    'COMPLEX': {
        'model': 'claude-opus-4-5',
        'max_tokens': 4096,
        'cost_per_1k_input': 0.015,
        'use_case': 'deep reasoning, long code, specialized domains'
    }
}

class ModelRouter:
    def __init__(self):
        self.routes = MODEL_ROUTES
        self.call_counts = {k: 0 for k in MODEL_ROUTES}

    def route(self, request, messages):
        complexity = classify_complexity(request)
        route = self.routes[complexity]
        self.call_counts[complexity] += 1
        print(f'Routing "{request[:40]}" -> {route["model"]} ({complexity})')
        return route['model'], route['max_tokens']

    def cost_report(self):
        total = sum(self.call_counts.values())
        for complexity, count in self.call_counts.items():
            pct = count / total * 100 if total else 0
            print(f'{complexity}: {count} calls ({pct:.0f}%)')

التوجيه مع مراعاة التكلفة

class CostAwareRouter(ModelRouter):
    def __init__(self, daily_budget_usd=100.0):
        super().__init__()
        self.daily_budget = daily_budget_usd
        self.daily_spent = 0.0

    def estimate_cost(self, model, input_tokens, max_output_tokens):
        route = next((r for r in self.routes.values() if r['model'] == model), None)
        if not route:
            return 0.0
        return (
            (input_tokens / 1000) * route['cost_per_1k_input'] +
            (max_output_tokens / 1000) * route['cost_per_1k_input'] * 3
        )

    def route_with_budget(self, request, messages, user_tier='free'):
        complexity = classify_complexity(request)

        # Downgrade if budget is exhausted or user is on free tier
        budget_remaining = self.daily_budget - self.daily_spent
        if budget_remaining < 0.01 or user_tier == 'free':
            complexity = 'SIMPLE'  # downgrade to cheapest model
            print('Budget constraint: routing to SIMPLE model')

        route = self.routes[complexity]
        input_tokens = sum(len(m['content'].split()) for m in messages) * 1.3
        cost = self.estimate_cost(route['model'], input_tokens, route['max_tokens'])
        self.daily_spent += cost
        return route['model'], route['max_tokens']

التوجيه مع مراعاة زمن الاستجابة

import time

# Model latency profiles (approximate P95 values)
MODEL_LATENCY_P95 = {
    'gpt-4o-mini': 1.5,       # seconds
    'gpt-4o': 4.0,
    'claude-haiku-4-5': 1.2,
    'claude-sonnet-4-5': 3.0,
    'claude-opus-4-5': 6.0
}

SLA_LATENCY_BUDGET = 3.0  # seconds

def route_with_latency_constraint(complexity, sla_seconds=SLA_LATENCY_BUDGET):
    route = MODEL_ROUTES[complexity]
    p95_latency = MODEL_LATENCY_P95.get(route['model'], 5.0)

    if p95_latency > sla_seconds:
        # Find fastest model under SLA
        affordable_models = [
            (lat, m) for m, lat in MODEL_LATENCY_P95.items()
            if lat <= sla_seconds
        ]
        if affordable_models:
            fastest = min(affordable_models)[1]
            print(f'Latency constraint: downgrading from {route["model"]} to {fastest}')
            return fastest
    return route['model']

print('COMPLEX request under 3s SLA:', route_with_latency_constraint('COMPLEX'))

التوجيه مع مراعاة القدرات

MODEL_CAPABILITIES = {
    'gpt-4o-mini': {
        'vision': True,
        'function_calling': True,
        'context_window': 128000,
        'code_interpreter': False
    },
    'gpt-4o': {
        'vision': True,
        'function_calling': True,
        'context_window': 128000,
        'code_interpreter': True
    },
    'claude-opus-4-5': {
        'vision': True,
        'function_calling': True,
        'context_window': 200000,
        'code_interpreter': False
    }
}

def capability_aware_route(required_capabilities, context_length=0):
    candidates = []
    for model, caps in MODEL_CAPABILITIES.items():
        if context_length > caps['context_window']:
            continue
        if all(caps.get(cap, False) for cap in required_capabilities):
            candidates.append(model)

    if not candidates:
        raise ValueError(f'No model supports: {required_capabilities}')

    # Among capable models, pick cheapest
    cost_rank = ['gpt-4o-mini', 'claude-opus-4-5', 'gpt-4o']
    for model in cost_rank:
        if model in candidates:
            return model
    return candidates[0]

print(capability_aware_route(['vision', 'function_calling'], context_length=5000))

سلاسل التراجع

FALLBACK_CHAINS = {
    'primary': 'claude-opus-4-5',
    'fallback': 'gpt-4o',
    'emergency': 'gpt-4o-mini'
}

def call_with_fallback(messages, chain=FALLBACK_CHAINS):
    providers = [
        ('anthropic', chain['primary']),
        ('openai', chain['fallback']),
        ('openai', chain['emergency'])
    ]

    for provider, model in providers:
        try:
            print(f'Trying {model}...')
            if provider == 'anthropic':
                import anthropic
                ac = anthropic.Anthropic(api_key='YOUR_KEY')
                resp = ac.messages.create(
                    model=model, max_tokens=500, messages=messages
                )
                return resp.content[0].text
            else:
                import openai
                oc = openai.OpenAI(api_key='YOUR_KEY')
                resp = oc.chat.completions.create(
                    model=model, messages=messages, max_tokens=500
                )
                return resp.choices[0].message.content
        except Exception as e:
            print(f'{model} failed: {e}. Trying next...')

    raise RuntimeError('All models in fallback chain failed')

تسجيل قرارات التوجيه

import json
from datetime import datetime

ROUTING_LOG_FILE = 'routing_decisions.jsonl'

def log_routing_decision(request_id, request_text, complexity,
                          model_selected, cost_estimate, latency_ms,
                          user_tier='free'):
    entry = {
        'timestamp': datetime.utcnow().isoformat(),
        'request_id': request_id,
        'request_preview': request_text[:50],
        'complexity': complexity,
        'model': model_selected,
        'cost_estimate_usd': round(cost_estimate, 6),
        'latency_ms': round(latency_ms),
        'user_tier': user_tier
    }
    with open(ROUTING_LOG_FILE, 'a') as f:
        f.write(json.dumps(entry) + '\n')

# Analyze routing log to tune thresholds
def analyze_routing_log():
    from collections import Counter
    model_counts = Counter()
    total_cost = 0.0
    with open(ROUTING_LOG_FILE) as f:
        for line in f:
            e = json.loads(line)
            model_counts[e['model']] += 1
            total_cost += e['cost_estimate_usd']
    print('Model distribution:', dict(model_counts))
    print(f'Total estimated cost: ${total_cost:.4f}')

اختبار النماذج بتقنية A/B في الإنتاج

import random

class ABModelRouter:
    def __init__(self, control_model, treatment_model, treatment_pct=10):
        self.control = control_model
        self.treatment = treatment_model
        self.treatment_pct = treatment_pct
        self.assignment_log = {}  # request_id: 'control' | 'treatment'

    def route(self, request_id):
        if request_id in self.assignment_log:
            # Sticky assignment: same user always gets same model
            return self.assignment_log[request_id]

        if random.random() * 100 < self.treatment_pct:
            assignment = 'treatment'
            model = self.treatment
        else:
            assignment = 'control'
            model = self.control

        self.assignment_log[request_id] = assignment
        return model, assignment

# Usage
ab_router = ABModelRouter(
    control_model='gpt-4o',
    treatment_model='claude-opus-4-5',
    treatment_pct=10  # 10% get new model
)

for user_id in range(5):
    result = ab_router.route(f'user_{user_id}')
    print(f'user_{user_id}: {result}')

فحوصات سلامة النماذج

import time

def health_check(model, provider='openai', timeout=5):
    '''
    Returns True if model is healthy, False if timed out or errored.
    '''
    test_prompt = 'Reply with exactly: OK'
    try:
        start = time.time()
        if provider == 'openai':
            import openai
            client = openai.OpenAI(api_key='YOUR_API_KEY')
            resp = client.chat.completions.create(
                model=model,
                messages=[{'role': 'user', 'content': test_prompt}],
                max_tokens=5,
                timeout=timeout
            )
            text = resp.choices[0].message.content.strip()
        elif provider == 'anthropic':
            import anthropic
            client = anthropic.Anthropic(api_key='YOUR_API_KEY')
            resp = client.messages.create(
                model=model, max_tokens=5,
                messages=[{'role': 'user', 'content': test_prompt}],
            )
            text = resp.content[0].text.strip()
        latency = (time.time() - start) * 1000
        healthy = 'ok' in text.lower()
        print(f'{model}: {"HEALTHY" if healthy else "DEGRADED"} ({latency:.0f}ms)')
        return healthy
    except Exception as e:
        print(f'{model}: UNHEALTHY ({e})')
        return False

# Run health checks before routing critical traffic
# health_check('gpt-4o-mini', provider='openai')
# health_check('claude-haiku-4-5', provider='anthropic')

تحليل أثر التكلفة

def cost_impact_analysis(daily_requests=10000):
    # Traffic distribution
    traffic = {'SIMPLE': 0.60, 'MODERATE': 0.30, 'COMPLEX': 0.10}

    # Avg tokens per request (input + output)
    avg_tokens = {'SIMPLE': 500, 'MODERATE': 2000, 'COMPLEX': 5000}

    # Pricing per 1K tokens (blended input+output)
    pricing = {'SIMPLE': 0.00030, 'MODERATE': 0.01000, 'COMPLEX': 0.04500}
    premium_price = 0.04500  # if we used COMPLEX model for everything

    routed_cost = 0.0
    premium_cost = 0.0

    for complexity, pct in traffic.items():
        requests = daily_requests * pct
        tokens = avg_tokens[complexity]
        routed_cost += requests * (tokens / 1000) * pricing[complexity]
        premium_cost += requests * (tokens / 1000) * premium_price

    savings_pct = (1 - routed_cost / premium_cost) * 100
    print(f'Daily requests: {daily_requests:,}')
    print(f'With routing:  ${routed_cost:,.2f}/day')
    print(f'Without routing: ${premium_cost:,.2f}/day')
    print(f'Savings: {savings_pct:.0f}% (${premium_cost - routed_cost:,.2f}/day)')

cost_impact_analysis()

تحقق سريع

ملخص توجيه النماذج