LLM Failover Machine Coding Question | Python

LLM Failover Resilience.py

# Online Python compiler (interpreter) to run Python online.
# Write Python 3 code in this online editor and run it.


# LLM Failover router 
# Prvovider - OpenAI, Azure, Google, Anthropic, Cohere, HuggingFace
# Configurable - 
# Ideal scenaria - 100% requests would goes to primary provider
# when primary fails / degreaded state then 95% goes to secondary provider and 5% goes to primary provider for testing
# when primary recovers then 100% goes to primary provider
# when both are down keep 5% to primary provider for testing and 5% goes to secondary 
# Reject 90% of the requests

# degradation state - x consecutive failures
# uptime state - y consecutive successes 

# Stragety() - plug and play models
# Circuit Breaker - maintain state + count 

# Tasks - optional

# Task - "string"

# Track the failures 
# Track the state 
# Proces request 

# Classes to be created 

# Traffic Stretegy - keep tracks of succes and failures taking place
# Provider - contains information of prpovider
# Router - process - requests
# Circuit Breaker - changes the accordingly
import random

class TrafficStrategy:
    def __init__(self):
        pass
    
    def chooseProvider(self, p_provider, s_provider):
        p_state = p_provider.state
        s_state = s_provider.state
        
        r = random.randint(1, 100)

        if p_state == "CLOSED":
            return p_provider
            
        if p_state == "OPEN" and s_state == "CLOSED":
            return s_provider if r <= 5 else p_provider
            
        if p_state == "OPEN" and s_state == "OPEN":
            if r<=5:
                return p_provider
            elif r<=10: 
                return s_provider
            else:
                return None
                
        return s_provider
        
class Router:
    def __init__(self, p_provider, s_provider, toBeUsedProvider):
        self.p_provider = p_provider
        self.s_provider = s_provider
        self.toBeUsedProvider = toBeUsedProvider
        
    def route(self, request): 
        provider = self.toBeUsedProvider.chooseProvider(self.p_provider, self.s_provider)
        
        if provider is None:
            return "request rejected"

        
        try:
            res = provider.call(request)
            print(f"{request.id} procssed")
        except Exception as e:
            print(f"{request.id} failed ")

class Provider:
    def __init__(self, name):
        self.name = name
        
    
    def call(self, request):
        if request.id in [1,2,4,5]:
            raise Exception(f"{self.name} failed")
        return f"{self.name} success"
        
        

class CircuitBreaker: 
    def __init__(self, provider, fail_threshold, success_threshold):
        self.provider = provider
        self.fail_threshold = fail_threshold
        self.success_threshold = success_threshold
        
        self.fail_count = 0
        self.success_count = 0
        
        self.state = "CLOSED"
        
    def call(self, request):
        if self.state == "OPEN":
            self.state = "HALF_OPEN"
            
        try:
            res = self.provider.call(request)
            self.success_count+=1
            self.fail_count = 0
        
            if self.state == "HALF_OPEN" and self.success_count >= self.success_threshold: 
                self.state = "CLOSED"
            
            return res
            
        except Exception as e: 
            self.fail_count+=1
            self.success_count = 0
            
            if self.fail_count>=self.fail_threshold:
                self.state = "OPEN"
                
            raise e
            
                
class Request:
    def __init__(self, id):
        self.id = id
            
            
primary = CircuitBreaker(Provider("OpenAI"), fail_threshold=3, success_threshold=2)
secondary = CircuitBreaker(Provider("Observe.ai"), fail_threshold=2, success_threshold=2)


router = Router(primary, secondary, TrafficStrategy())

def run_test(test_name, requests):
    print("\n" + "=" * 50)
    print(f"TEST: {test_name}")
    print("=" * 50)

    for r in requests:
        router.route(r)


run_test(
    "Normal traffic, mostly success",
    [Request(i) for i in range(1, 6)]
)

run_test(
    "Primary failure triggers OPEN",
    [Request(i) for i in range(1, 12)]
)


run_test(
    "Traffic after circuit is OPEN",
    [Request(i) for i in range(12, 20)]
)