Scaling, Performance & Production Guardrails

1. Latency Optimization Strategies

LLM latency has two components: time-to-first-token (TTFT) and inter-token latency (ITL). TTFT dominates user-perceived responsiveness, while ITL affects the smoothness of streaming output. Different optimization strategies target different components.

Rate Limiting with Token Buckets

import time
from dataclasses import dataclass, field

@dataclass
class TokenBucket:
    """Token bucket rate limiter for LLM API requests."""
    capacity: float          # max tokens in bucket
    refill_rate: float       # tokens added per second
    tokens: float = field(init=False)
    last_refill: float = field(init=False)

    def __post_init__(self):
        self.tokens = self.capacity
        self.last_refill = time.monotonic()

    def consume(self, cost: float = 1.0) -> bool:
        now = time.monotonic()
        elapsed = now - self.last_refill
        self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate)
        self.last_refill = now

        if self.tokens >= cost:
            self.tokens -= cost
            return True
        return False

# Allow 10 requests/sec, burst up to 50
limiter = TokenBucket(capacity=50, refill_rate=10)
for i in range(5):
    allowed = limiter.consume()
    print(f"Request {i}: {'allowed' if allowed else 'rate-limited'}")

2. Backpressure and Queue Management

import asyncio
from collections import deque

class BackpressureQueue:
    """Bounded async queue with backpressure signaling."""

    def __init__(self, max_size: int = 100, warn_threshold: float = 0.8):
        self.queue = asyncio.Queue(maxsize=max_size)
        self.max_size = max_size
        self.warn_threshold = warn_threshold
        self.rejected = 0

    @property
    def utilization(self) -> float:
        return self.queue.qsize() / self.max_size

    async def enqueue(self, item, timeout: float = 5.0):
        try:
            await asyncio.wait_for(
                self.queue.put(item), timeout=timeout
            )
            return {"status": "queued", "position": self.queue.qsize()}
        except asyncio.TimeoutError:
            self.rejected += 1
            return {"status": "rejected", "reason": "queue_full"}

    def health_status(self):
        util = self.utilization
        if util > self.warn_threshold:
            return "degraded"
        return "healthy"

3. Production Guardrails

NeMo Guardrails Configuration

# config.yml for NeMo Guardrails
models:
  - type: main
    engine: openai
    model: gpt-4o-mini

rails:
  input:
    flows:
      - self check input      # Block harmful prompts
      - check jailbreak       # Detect jailbreak attempts
  output:
    flows:
      - self check output     # Filter harmful outputs
      - check hallucination   # Verify factual claims

# Colang rails definition
define user ask about competitors
  "What do you think about [competitor]?"
  "Compare yourself to [competitor]"

define flow
  user ask about competitors
  bot refuse to discuss competitors
  "I'm not able to provide comparisons with other products."

Llama Guard for Content Safety

from transformers import AutoModelForCausalLM, AutoTokenizer
import torch

def check_safety(conversation: list[dict], model_name="meta-llama/Llama-Guard-3-8B"):
    """Classify conversation safety using Llama Guard 3."""
    tokenizer = AutoTokenizer.from_pretrained(model_name)
    model = AutoModelForCausalLM.from_pretrained(
        model_name, torch_dtype=torch.bfloat16, device_map="auto"
    )

    chat = tokenizer.apply_chat_template(conversation, tokenize=False)
    inputs = tokenizer(chat, return_tensors="pt").to(model.device)

    with torch.no_grad():
        output = model.generate(**inputs, max_new_tokens=100, pad_token_id=0)

    result = tokenizer.decode(output[0][inputs["input_ids"].shape[1]:])
    is_safe = "safe" in result.lower()
    return {"safe": is_safe, "raw_output": result.strip()}

# Example usage
convo = [
    {"role": "user", "content": "How do I make a simple pasta recipe?"}
]
print(check_safety(convo))