LLM ops engineer coding interview questions

11 questions on coding for llm ops engineer candidates. Expect prompts such as “Parse and recover malformed JSON from LLM output robustly” and “Implement a token-bucket rate limiter for an LLM API in Python”, each with a worked answer outline and the follow-ups interviewers push on.

Showing 1 to 11 of 11 coding questions.

As asked

LLMs often produce near-valid JSON with trailing commas, unquoted keys, or truncated output. Write a Python function that attempts to parse LLM JSON output, applies common recovery heuristics, and raises a typed error with a diagnostic message if all recovery attempts fail.

Sample answer outline

A strong answer tries json.loads first, then applies ordered heuristics: strip markdown fences, truncation recovery by finding the last complete object bracket, trailing comma removal via regex, and finally a permissive parser like json5 or demjson3. Each step is logged so the caller can diagnose which recovery was applied. A custom ParseError carries the original text, the attempted repairs, and the position of the first parse error.

Reference implementation (python)

Python

import json, re
from dataclasses import dataclass

@dataclass
class ParseError(Exception):
    original: str
    repairs_tried: list[str]
    detail: str

def parse_llm_json(text: str) -> dict | list:
    attempts = []
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        attempts.append('direct')
    stripped = re.sub(r'^```[a-z]*\n?|```$', '', text.strip(), flags=re.MULTILINE).strip()
    try:
        return json.loads(stripped)
    except json.JSONDecodeError:
        attempts.append('strip-fences')
    no_trailing = re.sub(r',\s*(\}|\])', r'\1', stripped)
    try:
        return json.loads(no_trailing)
    except json.JSONDecodeError as e:
        attempts.append('trailing-commas')
        for end in range(len(no_trailing), 0, -1):
            try:
                return json.loads(no_trailing[:end])
            except json.JSONDecodeError:
                continue
        raise ParseError(original=text, repairs_tried=attempts, detail=str(e))

Expect these follow-ups

When would you use constrained decoding instead of post-hoc repair?
How does your function behave if the model emits a valid JSON array nested inside prose text?

json-parsingllm-outputerror-handlingpythonstructured-output

As asked

Write a thread-safe token bucket rate limiter class in Python that limits requests by output tokens per minute per API key. The class must support async usage and handle distributed state via Redis.

Sample answer outline

A strong answer implements a Redis-backed Lua script for atomic check-and-consume to avoid race conditions, uses an async Redis client (aioredis or redis.asyncio), stores (current_tokens, last_refill_timestamp) per key, and refills on each check rather than on a background timer. The Lua script does the math in one round trip. The class exposes an async context manager or a consume(n_tokens) coroutine that returns a bool and a retry-after seconds value.

Reference implementation (python)

Python

import time
import redis.asyncio as redis

LUA_SCRIPT = """
local key = KEYS[1]
local capacity = tonumber(ARGV[1])
local refill_rate = tonumber(ARGV[2])
local cost = tonumber(ARGV[3])
local now = tonumber(ARGV[4])
local data = redis.call('HMGET', key, 'tokens', 'ts')
local tokens = tonumber(data[1]) or capacity
local ts = tonumber(data[2]) or now
local elapsed = now - ts
tokens = math.min(capacity, tokens + elapsed * refill_rate)
if tokens < cost then
  return {0, math.ceil((cost - tokens) / refill_rate)}
end
tokens = tokens - cost
redis.call('HMSET', key, 'tokens', tokens, 'ts', now)
redis.call('EXPIRE', key, 3600)
return {1, 0}
"""

class TokenBucketLimiter:
    def __init__(self, redis_url: str, capacity: int, refill_rate: float):
        self.r = redis.from_url(redis_url)
        self.capacity = capacity
        self.refill_rate = refill_rate
        self._script = None

    async def consume(self, api_key: str, cost: int) -> tuple[bool, int]:
        if not self._script:
            self._script = self.r.register_script(LUA_SCRIPT)
        result = await self._script(
            keys=[f'rl:{api_key}'],
            args=[self.capacity, self.refill_rate, cost, time.time()]
        )
        return bool(result[0]), int(result[1])

Expect these follow-ups

How would you handle the thundering herd problem if 10,000 keys all refill at the same second?
What changes if you need to rate-limit on input tokens plus output tokens combined?

rate-limitingredisasync-pythonapi-designtoken-budget

As asked

Write a Python async script that sends 10,000 prompts to an OpenAI-compatible LLM API, handles rate limit and server errors with exponential backoff, and writes results to a JSONL file without losing progress if the script is interrupted.

Sample answer outline

A strong answer uses asyncio with a semaphore to cap concurrency, wraps each request in a retry loop with exponential backoff plus jitter on 429 and 5xx responses, writes each completed result immediately to JSONL (so progress survives a crash), and tracks completed IDs in a set loaded from the output file at startup so restarted runs skip already-done items.

Reference implementation (python)

Python

import asyncio, json, random, httpx
from pathlib import Path

DONE_FILE = Path('results.jsonl')

async def call_with_retry(client, prompt, idx, sem, max_retries=5):
    delay = 1.0
    for attempt in range(max_retries):
        async with sem:
            try:
                r = await client.post('/v1/chat/completions', json={
                    'model': 'meta-llama/Llama-3-70b',
                    'messages': [{'role': 'user', 'content': prompt}]
                }, timeout=60)
                if r.status_code == 429 or r.status_code >= 500:
                    raise httpx.HTTPStatusError('retryable', request=r.request, response=r)
                r.raise_for_status()
                return idx, r.json()['choices'][0]['message']['content']
            except httpx.HTTPStatusError:
                await asyncio.sleep(delay + random.uniform(0, 0.5))
                delay = min(delay * 2, 60)
    return idx, None

async def run_batch(prompts):
    done = set()
    if DONE_FILE.exists():
        for line in DONE_FILE.read_text().splitlines():
            done.add(json.loads(line)['idx'])
    sem = asyncio.Semaphore(50)
    async with httpx.AsyncClient(base_url='http://localhost:8000') as client:
        with DONE_FILE.open('a') as f:
            tasks = [call_with_retry(client, p, i, sem)
                     for i, p in enumerate(prompts) if i not in done]
            for coro in asyncio.as_completed(tasks):
                idx, result = await coro
                if result:
                    f.write(json.dumps({'idx': idx, 'output': result}) + '\n')
                    f.flush()

Expect these follow-ups

How would you tune the semaphore size given a rate limit of 5000 tokens/minute?
What changes if the API returns streaming responses and you need to reconstruct the full text?

async-pythonbatch-inferenceretryrate-limitingopenai-api

As asked

Write a Python function that evaluates a model on multiple-choice questions by comparing the log probability of each answer choice, selects the highest-probability choice, and reports accuracy. This is the core of how lm-eval-harness works.

Sample answer outline

For each question, construct one completion request per answer choice where the prompt ends just before the choice text, request logprobs for the choice tokens, sum the log probabilities, and select the choice with the highest sum. This is the log-likelihood scoring approach used in lm-eval-harness for tasks like HellaSwag and MMLU. Accuracy is the fraction of questions where the selected choice matches the gold label. The key subtlety is normalizing by token length to avoid bias toward shorter choices.

Reference implementation (python)

Python

import httpx
from typing import TypedDict

class MCQuestion(TypedDict):
    context: str
    choices: list[str]
    gold: int

def score_choice(client: httpx.Client, model: str, context: str, choice: str) -> float:
    prompt = context + choice
    r = client.post('/v1/completions', json={
        'model': model, 'prompt': prompt, 'max_tokens': 0,
        'echo': True, 'logprobs': 1
    }).json()
    lps = r['choices'][0]['logprobs']['token_logprobs']
    n_ctx = len(r['choices'][0]['logprobs']['tokens']) - len(choice.split())
    choice_lps = [lp for lp in lps[n_ctx:] if lp is not None]
    return sum(choice_lps) / max(len(choice_lps), 1)

def evaluate(questions: list[MCQuestion], base_url: str, model: str) -> float:
    client = httpx.Client(base_url=base_url, timeout=60)
    correct = 0
    for q in questions:
        scores = [score_choice(client, model, q['context'], c) for c in q['choices']]
        if scores.index(max(scores)) == q['gold']:
            correct += 1
    return correct / len(questions)

Expect these follow-ups

Why does length normalization of logprobs matter for multiple-choice accuracy?
How would you parallelize this across 10,000 questions efficiently?

eval-harnesslogprobsbenchmarksmmlupython

As asked

Write a FastAPI endpoint that proxies a streaming chat completion request to a vLLM backend using Server-Sent Events, accumulates the full response for logging, and handles client disconnect without leaving orphaned requests on the backend.

Sample answer outline

A strong answer uses an async generator that reads chunks from the vLLM HTTP response, yields them to the FastAPI StreamingResponse, accumulates content chunks in a buffer, and logs the full request/response on generator exhaustion or on GeneratorExit (which fires on client disconnect). It also cancels the in-flight httpx request on disconnect using request.is_disconnected() checked in the loop or by wrapping the httpx stream in a task that is cancelled on GeneratorExit.

Reference implementation (python)

Python

import asyncio, httpx, json, logging
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse

app = FastAPI()
log = logging.getLogger('proxy')

@app.post('/v1/chat/completions')
async def chat(request: Request):
    body = await request.json()
    body['stream'] = True
    collected = []

    async def generate():
        async with httpx.AsyncClient(base_url='http://vllm:8000') as client:
            async with client.stream('POST', '/v1/chat/completions', json=body) as resp:
                try:
                    async for line in resp.aiter_lines():
                        if await request.is_disconnected():
                            log.info('client disconnected')
                            return
                        if line.startswith('data: '):
                            payload = line[6:]
                            if payload == '[DONE]':
                                yield 'data: [DONE]\n\n'
                                break
                            chunk = json.loads(payload)
                            delta = chunk['choices'][0]['delta'].get('content', '')
                            collected.append(delta)
                            yield f'data: {payload}\n\n'
                finally:
                    log.info('full response: %s', ''.join(collected))

    return StreamingResponse(generate(), media_type='text/event-stream')

Expect these follow-ups

How would you add per-request latency logging that captures time-to-first-chunk separately from total duration?
What happens to the KV cache slot on the vLLM side when the client disconnects mid-generation?

streamingssefastapiproxyasync-python

As asked

You receive a stream of request completion events with a timestamp and duration in milliseconds. Write a Python class that maintains a rolling 60-second p99 latency using an efficient data structure, updating on every event and answering queries in O(log n) time.

Sample answer outline

A strong answer uses a sorted container (SortedList from the sortedcontainers library) or a binary-indexed tree paired with a deque of (timestamp, duration) events. On each add, append to the deque and insert into the sorted structure. On query or each add, evict expired events from the front of the deque and remove them from the sorted structure. p99 is then the element at the 99th percentile index of the sorted structure.

Reference implementation (python)

Python

from collections import deque
from sortedcontainers import SortedList
import time

class RollingP99:
    def __init__(self, window_seconds: int = 60):
        self.window = window_seconds
        self._events: deque[tuple[float, float]] = deque()
        self._sorted = SortedList()

    def add(self, duration_ms: float, ts: float | None = None):
        ts = ts or time.monotonic()
        self._events.append((ts, duration_ms))
        self._sorted.add(duration_ms)
        self._evict(ts)

    def p99(self) -> float | None:
        self._evict(time.monotonic())
        n = len(self._sorted)
        if n == 0:
            return None
        idx = max(0, int(n * 0.99) - 1)
        return self._sorted[idx]

    def _evict(self, now: float):
        cutoff = now - self.window
        while self._events and self._events[0][0] < cutoff:
            _, d = self._events.popleft()
            self._sorted.remove(d)

Expect these follow-ups

What are the tradeoffs between a sorted container and a histogram-based approach (like HdrHistogram) for this use case?
How would you modify this to report multiple percentiles simultaneously without extra passes?

algorithmsdata-structuresmonitoringlatencysliding-window

As asked

Embedding computation is expensive. Write a thread-safe LRU cache in Python that stores float32 numpy arrays keyed by the SHA-256 hash of the input text, has a configurable max size by number of entries, and exposes cache hit rate as a metric.

Sample answer outline

A strong answer uses an OrderedDict with a threading.Lock for thread safety, implements get and put that move accessed keys to the end, evicts the oldest on overflow, and tracks hits and misses as counters for the hit_rate property. It also handles the case where the numpy array must be copied on retrieval to prevent callers from mutating the cached value.

Reference implementation (python)

Python

import hashlib, threading
from collections import OrderedDict
import numpy as np

class EmbeddingLRU:
    def __init__(self, max_size: int = 10_000):
        self._cache: OrderedDict[str, np.ndarray] = OrderedDict()
        self._lock = threading.Lock()
        self.max_size = max_size
        self._hits = 0
        self._misses = 0

    @staticmethod
    def _key(text: str) -> str:
        return hashlib.sha256(text.encode()).hexdigest()

    def get(self, text: str) -> np.ndarray | None:
        k = self._key(text)
        with self._lock:
            if k not in self._cache:
                self._misses += 1
                return None
            self._cache.move_to_end(k)
            self._hits += 1
            return self._cache[k].copy()

    def put(self, text: str, vec: np.ndarray):
        k = self._key(text)
        with self._lock:
            if k in self._cache:
                self._cache.move_to_end(k)
            else:
                if len(self._cache) >= self.max_size:
                    self._cache.popitem(last=False)
                self._cache[k] = vec.copy()

    @property
    def hit_rate(self) -> float:
        total = self._hits + self._misses
        return self._hits / total if total else 0.0

Expect these follow-ups

How would you extend this to a distributed cache shared across multiple embedding server replicas?
What is the memory overhead of storing 10,000 1536-dimensional float32 vectors in this cache?

lru-cacheembeddingdata-structurespythonthread-safety

As asked

Write a Python function that performs a readiness check on a vLLM endpoint. It must verify that the model is loaded (not just the HTTP server), that a test generation completes under a timeout, and that the output token count is within expected bounds. Return a structured result with pass/fail and diagnostics.

Sample answer outline

A strong answer hits /v1/models to confirm the model name is present, then sends a short generation request with a 10-second timeout and checks that the response contains at least 1 token and at most max_expected_tokens. The function returns a dataclass or dict with ok=bool, latency_ms, token_count, and error_detail. This is the logic that should back the Kubernetes readiness probe, so it must be fast (under 5 seconds) and never raise an uncaught exception.

Reference implementation (python)

Python

import time, httpx
from dataclasses import dataclass

@dataclass
class HealthResult:
    ok: bool
    latency_ms: float
    token_count: int
    error_detail: str | None

def check_vllm_ready(
    base_url: str,
    model: str,
    probe_prompt: str = 'Hello',
    max_tokens: int = 5,
    timeout: float = 10.0,
) -> HealthResult:
    start = time.monotonic()
    try:
        client = httpx.Client(base_url=base_url, timeout=timeout)
        models = client.get('/v1/models').json()
        names = [m['id'] for m in models.get('data', [])]
        if model not in names:
            return HealthResult(False, 0, 0, f'model {model!r} not in {names}')
        r = client.post('/v1/completions', json={
            'model': model, 'prompt': probe_prompt, 'max_tokens': max_tokens
        }).json()
        tokens = len(r['choices'][0]['text'].split())
        latency = (time.monotonic() - start) * 1000
        if tokens == 0:
            return HealthResult(False, latency, tokens, 'zero tokens returned')
        return HealthResult(True, latency, tokens, None)
    except Exception as exc:
        latency = (time.monotonic() - start) * 1000
        return HealthResult(False, latency, 0, str(exc))

Expect these follow-ups

How would you adapt this to check multiple models behind a single vLLM multi-model endpoint?
What change would you make if this check is running 100 times per minute and you want to avoid polluting serving metrics?

health-checkvllmkubernetespythondeployment

As asked

vLLM's completions API can return logprobs for each generated token. Write a Python function that computes perplexity over a list of reference texts by prompting the model to score each text and accumulating the log probabilities.

Sample answer outline

A strong answer constructs a prompt with an empty or prefix-only first turn, requests logprobs=True with top_logprobs=1, extracts the log probability of each generated token that matches the reference continuation, sums the log probabilities, divides by token count, and exponentiates. It must handle tokenization boundaries carefully, and average correctly over documents of different lengths using per-document perplexity then macro-averaging.

Reference implementation (python)

Python

import math, httpx

def compute_perplexity(texts: list[str], base_url: str, model: str) -> float:
    total_nll = 0.0
    total_tokens = 0
    client = httpx.Client(base_url=base_url, timeout=120)
    for text in texts:
        response = client.post('/v1/completions', json={
            'model': model,
            'prompt': text,
            'max_tokens': 0,
            'echo': True,
            'logprobs': 1,
        }).json()
        logprobs = response['choices'][0]['logprobs']['token_logprobs']
        valid = [lp for lp in logprobs if lp is not None]
        total_nll += -sum(valid)
        total_tokens += len(valid)
    return math.exp(total_nll / total_tokens) if total_tokens else float('inf')

Expect these follow-ups

Why does perplexity computed this way differ from the model's own training loss perplexity?
How would you make this computation efficient for 100,000 reference sentences?

perplexitylogprobsevalvllmpython

As asked

Write a Python async priority queue that schedules LLM inference requests such that enterprise-tier requests are processed before free-tier requests, with fair FIFO ordering within each tier. The queue must be safe for concurrent producers and a single consumer.

Sample answer outline

A strong answer uses asyncio.PriorityQueue with a (priority, sequence_number, request) tuple so that within the same priority, FIFO order is preserved by the monotonically increasing sequence number. The priority is an integer where lower is higher priority (enterprise=0, pro=1, free=2). The consumer awaits queue.get(), processes the request, and calls queue.task_done(). The sequence counter is an integer incremented per enqueue inside the event loop, so no lock is needed.

Reference implementation (python)

Python

import asyncio
from dataclasses import dataclass, field
from itertools import count
from typing import Any

TIER_PRIORITY = {'enterprise': 0, 'pro': 1, 'free': 2}
_seq = count()

@dataclass(order=True)
class PrioritizedRequest:
    priority: int
    seq: int = field(default_factory=lambda: next(_seq))
    request: Any = field(compare=False, default=None)

class TieredScheduler:
    def __init__(self, maxsize: int = 0):
        self._q: asyncio.PriorityQueue[PrioritizedRequest] = asyncio.PriorityQueue(maxsize)

    async def enqueue(self, request: Any, tier: str = 'free'):
        priority = TIER_PRIORITY.get(tier, 9)
        await self._q.put(PrioritizedRequest(priority=priority, request=request))

    async def dequeue(self) -> Any:
        item = await self._q.get()
        self._q.task_done()
        return item.request

    def qsize(self) -> int:
        return self._q.qsize()

Expect these follow-ups

How would you add preemption so that an arriving enterprise request can interrupt an in-progress free-tier generation?
What changes to the scheduler are needed if you want to implement weighted-fair queueing across tiers?

schedulingpriority-queueasync-pythonfairnessmulti-tenant

As asked

LLMs occasionally get stuck in repetition loops, generating the same phrase forever. Write a Python class that monitors a stream of output tokens and raises an alert (or signals termination) when a repetition pattern is detected.

Sample answer outline

A strong answer maintains a rolling buffer of recent tokens (e.g., last 200 tokens) and checks for n-gram repetition: if any n-gram of length 4 to 8 appears more than K times in the buffer, the repetition is flagged. An alternative approach uses the ratio of unique tokens in the buffer to total tokens as a simple entropy signal. The class exposes an add(token) method and a is_looping property. The caller (streaming proxy) checks is_looping after each token and sends a stop signal to vLLM if true.

Reference implementation (python)

Python

from collections import deque, Counter

class RepetitionDetector:
    def __init__(self, window: int = 200, ngram_n: int = 5, max_repeats: int = 6):
        self.window = window
        self.ngram_n = ngram_n
        self.max_repeats = max_repeats
        self._tokens: deque[str] = deque(maxlen=window)

    def add(self, token: str) -> None:
        self._tokens.append(token)

    @property
    def is_looping(self) -> bool:
        buf = list(self._tokens)
        if len(buf) < self.ngram_n * self.max_repeats:
            return False
        ngrams = [
            tuple(buf[i:i + self.ngram_n])
            for i in range(len(buf) - self.ngram_n + 1)
        ]
        counts = Counter(ngrams)
        if counts.most_common(1)[0][1] >= self.max_repeats:
            return True
        unique_ratio = len(set(buf)) / len(buf)
        return unique_ratio < 0.05

Expect these follow-ups

How do you tune the thresholds to avoid false positives for legitimate repetitive content like poems or lists?
How would you integrate this into the streaming SSE proxy from the prior coding question?

streamingoutput-qualitypythonmonitoringrepetition-detection

Tools to sharpen your prep

All tools

import json, re from dataclasses import dataclass @dataclass class ParseError(Exception): original: str repairs_tried: list[str] detail: str def parse_llm_json(text: str) -> dict | list: attempts = [] try: return json.loads(text) except json.JSONDecodeError: attempts.append('direct') stripped = re.sub(r'^```[a-z]*\n?|```$', '', text.strip(), flags=re.MULTILINE).strip() try: return json.loads(stripped) except json.JSONDecodeError: attempts.append('strip-fences') no_trailing = re.sub(r',\s*(\}|\])', r'\1', stripped) try: return json.loads(no_trailing) except json.JSONDecodeError as e: attempts.append('trailing-commas') for end in range(len(no_trailing), 0, -1): try: return json.loads(no_trailing[:end]) except json.JSONDecodeError: continue raise ParseError(original=text, repairs_tried=attempts, detail=str(e))

import time import redis.asyncio as redis LUA_SCRIPT = """ local key = KEYS[1] local capacity = tonumber(ARGV[1]) local refill_rate = tonumber(ARGV[2]) local cost = tonumber(ARGV[3]) local now = tonumber(ARGV[4]) local data = redis.call('HMGET', key, 'tokens', 'ts') local tokens = tonumber(data[1]) or capacity local ts = tonumber(data[2]) or now local elapsed = now - ts tokens = math.min(capacity, tokens + elapsed * refill_rate) if tokens < cost then return {0, math.ceil((cost - tokens) / refill_rate)} end tokens = tokens - cost redis.call('HMSET', key, 'tokens', tokens, 'ts', now) redis.call('EXPIRE', key, 3600) return {1, 0} """ class TokenBucketLimiter: def __init__(self, redis_url: str, capacity: int, refill_rate: float): self.r = redis.from_url(redis_url) self.capacity = capacity self.refill_rate = refill_rate self._script = None async def consume(self, api_key: str, cost: int) -> tuple[bool, int]: if not self._script: self._script = self.r.register_script(LUA_SCRIPT) result = await self._script( keys=[f'rl:{api_key}'], args=[self.capacity, self.refill_rate, cost, time.time()] ) return bool(result[0]), int(result[1])

import asyncio, json, random, httpx from pathlib import Path DONE_FILE = Path('results.jsonl') async def call_with_retry(client, prompt, idx, sem, max_retries=5): delay = 1.0 for attempt in range(max_retries): async with sem: try: r = await client.post('/v1/chat/completions', json={ 'model': 'meta-llama/Llama-3-70b', 'messages': [{'role': 'user', 'content': prompt}] }, timeout=60) if r.status_code == 429 or r.status_code >= 500: raise httpx.HTTPStatusError('retryable', request=r.request, response=r) r.raise_for_status() return idx, r.json()['choices'][0]['message']['content'] except httpx.HTTPStatusError: await asyncio.sleep(delay + random.uniform(0, 0.5)) delay = min(delay * 2, 60) return idx, None async def run_batch(prompts): done = set() if DONE_FILE.exists(): for line in DONE_FILE.read_text().splitlines(): done.add(json.loads(line)['idx']) sem = asyncio.Semaphore(50) async with httpx.AsyncClient(base_url='http://localhost:8000') as client: with DONE_FILE.open('a') as f: tasks = [call_with_retry(client, p, i, sem) for i, p in enumerate(prompts) if i not in done] for coro in asyncio.as_completed(tasks): idx, result = await coro if result: f.write(json.dumps({'idx': idx, 'output': result}) + '\n') f.flush()

import httpx from typing import TypedDict class MCQuestion(TypedDict): context: str choices: list[str] gold: int def score_choice(client: httpx.Client, model: str, context: str, choice: str) -> float: prompt = context + choice r = client.post('/v1/completions', json={ 'model': model, 'prompt': prompt, 'max_tokens': 0, 'echo': True, 'logprobs': 1 }).json() lps = r['choices'][0]['logprobs']['token_logprobs'] n_ctx = len(r['choices'][0]['logprobs']['tokens']) - len(choice.split()) choice_lps = [lp for lp in lps[n_ctx:] if lp is not None] return sum(choice_lps) / max(len(choice_lps), 1) def evaluate(questions: list[MCQuestion], base_url: str, model: str) -> float: client = httpx.Client(base_url=base_url, timeout=60) correct = 0 for q in questions: scores = [score_choice(client, model, q['context'], c) for c in q['choices']] if scores.index(max(scores)) == q['gold']: correct += 1 return correct / len(questions)

import asyncio, httpx, json, logging from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse app = FastAPI() log = logging.getLogger('proxy') @app.post('/v1/chat/completions') async def chat(request: Request): body = await request.json() body['stream'] = True collected = [] async def generate(): async with httpx.AsyncClient(base_url='http://vllm:8000') as client: async with client.stream('POST', '/v1/chat/completions', json=body) as resp: try: async for line in resp.aiter_lines(): if await request.is_disconnected(): log.info('client disconnected') return if line.startswith('data: '): payload = line[6:] if payload == '[DONE]': yield 'data: [DONE]\n\n' break chunk = json.loads(payload) delta = chunk['choices'][0]['delta'].get('content', '') collected.append(delta) yield f'data: {payload}\n\n' finally: log.info('full response: %s', ''.join(collected)) return StreamingResponse(generate(), media_type='text/event-stream')

from collections import deque from sortedcontainers import SortedList import time class RollingP99: def __init__(self, window_seconds: int = 60): self.window = window_seconds self._events: deque[tuple[float, float]] = deque() self._sorted = SortedList() def add(self, duration_ms: float, ts: float | None = None): ts = ts or time.monotonic() self._events.append((ts, duration_ms)) self._sorted.add(duration_ms) self._evict(ts) def p99(self) -> float | None: self._evict(time.monotonic()) n = len(self._sorted) if n == 0: return None idx = max(0, int(n * 0.99) - 1) return self._sorted[idx] def _evict(self, now: float): cutoff = now - self.window while self._events and self._events[0][0] < cutoff: _, d = self._events.popleft() self._sorted.remove(d)

import hashlib, threading from collections import OrderedDict import numpy as np class EmbeddingLRU: def __init__(self, max_size: int = 10_000): self._cache: OrderedDict[str, np.ndarray] = OrderedDict() self._lock = threading.Lock() self.max_size = max_size self._hits = 0 self._misses = 0 @staticmethod def _key(text: str) -> str: return hashlib.sha256(text.encode()).hexdigest() def get(self, text: str) -> np.ndarray | None: k = self._key(text) with self._lock: if k not in self._cache: self._misses += 1 return None self._cache.move_to_end(k) self._hits += 1 return self._cache[k].copy() def put(self, text: str, vec: np.ndarray): k = self._key(text) with self._lock: if k in self._cache: self._cache.move_to_end(k) else: if len(self._cache) >= self.max_size: self._cache.popitem(last=False) self._cache[k] = vec.copy() @property def hit_rate(self) -> float: total = self._hits + self._misses return self._hits / total if total else 0.0

import time, httpx from dataclasses import dataclass @dataclass class HealthResult: ok: bool latency_ms: float token_count: int error_detail: str | None def check_vllm_ready( base_url: str, model: str, probe_prompt: str = 'Hello', max_tokens: int = 5, timeout: float = 10.0, ) -> HealthResult: start = time.monotonic() try: client = httpx.Client(base_url=base_url, timeout=timeout) models = client.get('/v1/models').json() names = [m['id'] for m in models.get('data', [])] if model not in names: return HealthResult(False, 0, 0, f'model {model!r} not in {names}') r = client.post('/v1/completions', json={ 'model': model, 'prompt': probe_prompt, 'max_tokens': max_tokens }).json() tokens = len(r['choices'][0]['text'].split()) latency = (time.monotonic() - start) * 1000 if tokens == 0: return HealthResult(False, latency, tokens, 'zero tokens returned') return HealthResult(True, latency, tokens, None) except Exception as exc: latency = (time.monotonic() - start) * 1000 return HealthResult(False, latency, 0, str(exc))

import math, httpx def compute_perplexity(texts: list[str], base_url: str, model: str) -> float: total_nll = 0.0 total_tokens = 0 client = httpx.Client(base_url=base_url, timeout=120) for text in texts: response = client.post('/v1/completions', json={ 'model': model, 'prompt': text, 'max_tokens': 0, 'echo': True, 'logprobs': 1, }).json() logprobs = response['choices'][0]['logprobs']['token_logprobs'] valid = [lp for lp in logprobs if lp is not None] total_nll += -sum(valid) total_tokens += len(valid) return math.exp(total_nll / total_tokens) if total_tokens else float('inf')

import asyncio from dataclasses import dataclass, field from itertools import count from typing import Any TIER_PRIORITY = {'enterprise': 0, 'pro': 1, 'free': 2} _seq = count() @dataclass(order=True) class PrioritizedRequest: priority: int seq: int = field(default_factory=lambda: next(_seq)) request: Any = field(compare=False, default=None) class TieredScheduler: def __init__(self, maxsize: int = 0): self._q: asyncio.PriorityQueue[PrioritizedRequest] = asyncio.PriorityQueue(maxsize) async def enqueue(self, request: Any, tier: str = 'free'): priority = TIER_PRIORITY.get(tier, 9) await self._q.put(PrioritizedRequest(priority=priority, request=request)) async def dequeue(self) -> Any: item = await self._q.get() self._q.task_done() return item.request def qsize(self) -> int: return self._q.qsize()

from collections import deque, Counter class RepetitionDetector: def __init__(self, window: int = 200, ngram_n: int = 5, max_repeats: int = 6): self.window = window self.ngram_n = ngram_n self.max_repeats = max_repeats self._tokens: deque[str] = deque(maxlen=window) def add(self, token: str) -> None: self._tokens.append(token) @property def is_looping(self) -> bool: buf = list(self._tokens) if len(buf) < self.ngram_n * self.max_repeats: return False ngrams = [ tuple(buf[i:i + self.ngram_n]) for i in range(len(buf) - self.ngram_n + 1) ] counts = Counter(ngrams) if counts.most_common(1)[0][1] >= self.max_repeats: return True unique_ratio = len(set(buf)) / len(buf) return unique_ratio < 0.05

Questions

Parse and recover malformed JSON from LLM output robustlyCodingeasyVery common

As asked

Sample answer outline

Reference implementation (python)

Expect these follow-ups

Implement a token-bucket rate limiter for an LLM API in PythonCodingmediumCommon

As asked

Sample answer outline

Reference implementation (python)

Expect these follow-ups

Write a batch inference script that handles partial failures and retriesCodingmediumCommon

As asked

Sample answer outline

Reference implementation (python)

Expect these follow-ups

Implement a multiple-choice eval harness against a vLLM endpointCodingmediumCommon

As asked

Sample answer outline

Reference implementation (python)

Expect these follow-ups

Write a streaming SSE proxy that forwards vLLM responses to clientsCodingmediumCommon

As asked

Sample answer outline

Reference implementation (python)

Expect these follow-ups

Compute rolling p99 latency over a stream of request durationsCodingmediumCommon

As asked

Sample answer outline

Reference implementation (python)

Expect these follow-ups

Implement an LRU cache for embedding vectors keyed by text hashCodingeasyCommon

As asked

Sample answer outline

Reference implementation (python)

Expect these follow-ups

Write a health check script that validates a vLLM endpoint before routing trafficCodingeasyCommon

As asked

Sample answer outline

Reference implementation (python)

Expect these follow-ups

Compute perplexity from vLLM logprobs output for a test corpusCodingmediumOccasional

As asked

Sample answer outline

Reference implementation (python)

Expect these follow-ups

Implement a priority scheduler for LLM requests by user tierCodingmediumOccasional

As asked

Sample answer outline

Reference implementation (python)

Expect these follow-ups

Detect and terminate repetition loops in streaming LLM outputCodingmediumOccasional

As asked

Sample answer outline

Reference implementation (python)

Expect these follow-ups

Related questions

Implement a token-bucket rate limiter for an LLM API in Python

Write a batch inference script that handles partial failures and retries

Implement a multiple-choice eval harness against a vLLM endpoint

Write a streaming SSE proxy that forwards vLLM responses to clients

More llm ops engineer topics

Tools to sharpen your prep

Questions

Parse and recover malformed JSON from LLM output robustlyCodingeasyVery common

As asked

Sample answer outline

Reference implementation (python)

Expect these follow-ups

Implement a token-bucket rate limiter for an LLM API in PythonCodingmediumCommon

As asked

Sample answer outline

Reference implementation (python)

Expect these follow-ups

Write a batch inference script that handles partial failures and retriesCodingmediumCommon

As asked

Sample answer outline

Reference implementation (python)

Expect these follow-ups

Implement a multiple-choice eval harness against a vLLM endpointCodingmediumCommon