Streaming & Timeouts

import openai
import time

def stream_response(messages, timeout=30):
    """Stream LLM response with timeout and error handling."""
    start = time.time()
    buffer = ""

    try:
        response = openai.ChatCompletion.create(
            model="gpt-4-turbo",
            messages=messages,
            stream=True,
            timeout=timeout
        )

        for chunk in response:
            # Check timeout
            if time.time() - start > timeout:
                yield {"type": "timeout", "partial": buffer}
                return

            delta = chunk["choices"][0].get("delta", {})
            if "content" in delta:
                buffer += delta["content"]
                yield {"type": "token", "content": delta["content"]}

        yield {"type": "done", "full_response": buffer, "elapsed": time.time() - start}

    except openai.error.Timeout:
        yield {"type": "error", "error": "Request timed out", "partial": buffer}
    except openai.error.RateLimitError:
        yield {"type": "error", "error": "Rate limited, retrying...", "partial": buffer}
    except Exception as e:
        yield {"type": "error", "error": str(e), "partial": buffer}

import threading

class CancellableRequest:
    def __init__(self):
        self.cancelled = threading.Event()
        self.result = None

    def start(self, messages, timeout=30):
        """Start a streaming request that can be cancelled."""
        def _stream():
            for event in stream_response(messages, timeout):
                if self.cancelled.is_set():
                    return
                if event["type"] == "done":
                    self.result = event
                elif event["type"] == "token":
                    yield event

        self.generator = _stream()

    def cancel(self):
        """Cancel the request."""
        self.cancelled.set()

# Usage
request = CancellableRequest()
request.start(messages)

# User clicks cancel
request.cancel()

import time

class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=60):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failures = 0
        self.state = "closed"  # closed = normal, open = blocked, half-open = testing
        self.last_failure = 0

    def record_success(self):
        """Record a successful call."""
        self.failures = 0
        self.state = "closed"

    def record_failure(self):
        """Record a failed call."""
        self.failures += 1
        self.last_failure = time.time()

        if self.failures >= self.failure_threshold:
            self.state = "open"

    def allow_request(self):
        """Check if a request is allowed."""
        if self.state == "closed":
            return True

        if self.state == "open":
            if time.time() - self.last_failure > self.recovery_timeout:
                self.state = "half-open"
                return True
            return False

        if self.state == "half-open":
            return True  # Allow one test request

        return False

# Usage
breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=60)

def call_with_breaker(messages):
    if not breaker.allow_request():
        return fallback_response(messages)

    try:
        result = call_llm(messages)
        breaker.record_success()
        return result
    except Exception as e:
        breaker.record_failure()
        return fallback_response(messages)

def graceful_llm_call(messages, feature):
    """Call with graceful degradation."""
    # Try primary model
    try:
        return call_llm("gpt-4-turbo", messages, timeout=15)
    except TimeoutError:
        pass

    # Try faster model
    try:
        return call_llm("gpt-3.5-turbo", messages, timeout=5)
    except TimeoutError:
        pass

    # Try local model
    try:
        return call_local_llama(messages)
    except Exception:
        pass

    # Return cached or template response
    cached = semantic_cache.get(" ".join(m["content"] for m in messages))
    if cached:
        return cached["response"]

    # Final fallback: template response
    return get_template_response(feature)