Wipro: Coding & Production Systems Round

Base Question:

How do you write a Python wrapper/decorator to handle intermittent failures when calling external LLM APIs with exponential backoff?

Candidate Approach & Code Solution:

Implement an exponential backoff loop bounded by a maximum retry count to prevent infinite execution loops in production environments.

import time
import random
import logging
from typing import Callable, Any

logging.basicConfig(level=logging.INFO)

def call_llm_api_stub() -> str:
    """Simulated LLM API call that may fail intermittently."""
    if random.random() < 0.7:
        raise TimeoutError("504 Gateway Timeout: LLM provider unresponsive")
    return "LLM Response: Generation Successful"

def execute_with_exponential_backoff(
    api_func: Callable[[], Any],
    max_retries: int = 3,
    base_delay: float = 1.0,
    backoff_factor: float = 2.0
) -> Any:
    """
    Executes an API call with configurable retry counts and exponential delays.
    """
    delay = base_delay
    for attempt in range(1, max_retries + 1):
        try:
            logging.info(f"Attempt {attempt}/{max_retries}: Calling LLM API...")
            return api_func()
        except Exception as err:
            logging.warning(f"Attempt {attempt} failed: {err}")
            if attempt == max_retries:
                logging.error("Max retries reached. Escalating exception.")
                raise err

            time.sleep(delay)
            delay *= backoff_factor

Follow-Up Questions & Production Hardening

• Candidate Answer: No, non-transient client errors (\(4xx\)) should never be retried. A 400 indicates a malformed payload or invalid prompt parameters, and a 401 indicates authentication failure. Retrying these will yield identical failures, waste compute, and exhaust rate limits. Retries should strictly target transient network faults (\(502, 503, 504\)) and rate limits (\(429\)).

• Candidate Answer: Apply Full Jitter to the exponential delay. If thousands of worker threads encounter a micro-outage simultaneously, synchronized retries cause a "Thundering Herd" problem, repeatedly overloading the downstream API server at exact interval spikes. Jitter spreads out retry traffic across randomized time windows.

Base Question:

Frequent identical user queries significantly inflate LLM API billing and increase latency. How do you implement a caching system to mitigate this?

Candidate Approach & Code Solution:

Implement a query-level key-value lookup. Hashing the normalized prompt string generates a cache key to return pre-computed responses instantly.

import hashlib
from typing import Dict, Optional

class SimpleLLMCache:
    def __init__(self):
        self._cache: Dict[str, str] = {}

    def _generate_key(self, query: str) -> str:
        # Normalize whitespace and lower string before hashing
        normalized_query = " ".join(query.strip().lower().split())
        return hashlib.sha256(normalized_query.encode('utf-8')).hexdigest()

    def get(self, query: str) -> Optional[str]:
        key = self._generate_key(query)
        return self._cache.get(key)

    def set(self, query: str, response: str) -> None:
        key = self._generate_key(query)
        self._cache[key] = response

Follow-Up Questions & Production Hardening

• Candidate Answer:
• Memory Volatility: In-memory caches clear when the service container restarts or crashes.
• Instance Isolation: Modern applications deploy across multiple stateless server instances (e.g., Kubernetes pods). In-memory dicts create isolated, inconsistent local caches, resulting in duplicate LLM calls across pods.
• Fix: Use a centralized distributed cache like Redis, paired with an LRU (Least Recently Used) eviction policy to control memory footprint.

• Candidate Answer: No. Caching non-deterministic or real-time context queries (e.g., "What is the current stock price of Apple?", "Generate a personalized greeting for User X") causes stale, invalid outputs.

Base Question:

Large documents often exceed the context window of an LLM. How do you split raw text into chunks, and why is chunk overlap necessary?

Candidate Approach & Code Solution:

Implement a sliding-window text chunker using configurable character limits and overlap intervals to maintain continuity across chunk boundaries.

from typing import List

def chunk_text_with_overlap(text: str, chunk_size: int, chunk_overlap: int) -> List[str]:
    """
    Splits text into overlapping chunks using character-level sliding windows.
    """
    if chunk_overlap >= chunk_size:
        raise ValueError("chunk_overlap must be strictly less than chunk_size")

    chunks = []
    start = 0
    text_length = len(text)

    while start < text_length:
        end = start + chunk_size
        chunk = text[start:end]
        chunks.append(chunk)

        # Advance the window by stride (size - overlap)
        start += (chunk_size - chunk_overlap)

    return chunks

# Test Execution
sample_doc = "Generative AI systems require scalable data pipelines and effective chunking strategies."
chunks = chunk_text_with_overlap(sample_doc, chunk_size=30, chunk_overlap=10)

Follow-Up Questions & Production Hardening

• Candidate Answer: Text chunking creates artificial boundaries. If a critical semantic fact spans across the boundary of Chunk A and Chunk B (e.g., a entity subject in Chunk A and its primary action verb in Chunk B), splitting without overlap fragments the context. Embedding models will lose the core context of both fragments. Overlap preserves semantic continuity across contiguous chunks.

• Candidate Answer: No. Fixed-character splitting ignores semantic structure, often cutting sentences or code blocks mid-word.
• Recursive Character Splitting: Hierarchy-aware splitting that attempts to preserve structural boundaries by splitting sequentially on \n\n (paragraphs), \n (lines), (words), and finally "" (characters).
• Semantic Chunking: Computes similarity between adjacent sentences using embeddings, placing split points only where semantic distance spikes significantly.

import re
from typing import List

def recursive_character_tokenizer_concept(text: str, max_chunk_size: int) -> List[str]:
    """
    Splits content primarily on paragraph delimiters, falling back to lower-level
    separators only when paragraph blocks exceed max_chunk_size.
    """
    paragraphs = text.split("\n\n")
    final_chunks = []
    current_chunk = []
    current_length = 0

    for para in paragraphs:
        if current_length + len(para) <= max_chunk_size:
            current_chunk.append(para)
            current_length += len(para) + 2  # account for delimiter
        else:
            if current_chunk:
                final_chunks.append("\n\n".join(current_chunk))
            current_chunk = [para]
            current_length = len(para)

    if current_chunk:
        final_chunks.append("\n\n".join(current_chunk))

    return final_chunks

Base Question:

A vector database returns the top 20 candidate documents based on cosine similarity, but you must pass only the 5 most relevant chunks to the LLM context. Write a function to extract top results.

Candidate Code Solution (Initial Sorting):

from typing import List, Dict, Any

def get_top_k_documents(retrieved_docs: List[Dict[str, Any]], top_k: int = 5) -> List[Dict[str, Any]]:
    """
    Sorts retrieved vector search results by similarity score and returns top_k.
    """
    sorted_docs = sorted(retrieved_docs, key=lambda x: x.get("score", 0.0), reverse=True)
    return sorted_docs[:top_k]

Follow-Up Questions & Production Hardening

• Candidate Answer: High vector similarity does not guarantee information diversity. If a database contains repeated policy updates or redundant log entries, standard cosine sorting returns \(5\) near-duplicate chunks. This consumes valuable context window space without adding useful information.

• Candidate Answer: Implement a Two-Stage Retrieval Architecture utilizing Over-Retrieval + Re-Ranking. 1. Stage 1 (High Recall): Retrieve an expanded candidate set (e.g., Top \(K=30\)) from the vector store using fast approximate nearest neighbors (ANN). 2. Stage 2 (High Precision & Diversity): Pass candidates through a Cross-Encoder Re-Ranker (e.g., Cohere Rerank or bge-reranker-large) or run Maximal Marginal Relevance (MMR) to maximize information coverage while penalizing redundant text embeddings.

                    [ User Query ]
                          │
                          ▼
             [ Stage 1: Vector Search ]
              (Fast ANN, High Recall)
                          │
                          ▼
            [ Top 30 Candidate Documents ]
                          │
                          ▼
              [ Stage 2: Cross-Encoder ]
            (Contextual Re-Ranking & MMR)
                          │
                          ▼
         [ Top 5 Diverse & Relevant Chunks ]
                          │
                          ▼
                     [ LLM Context ]

Base Question:

An downstream service expects the LLM to return valid JSON, but the model occasionally returns malformed JSON, markdown formatting (json ...), or conversational prefix/suffix text. How do you handle this?

Candidate Approach & Code Solution:

Extract the raw JSON string using regex patterns, parse it safely, and validate it against an expected model structure.

import re
import json
from typing import Dict, Any

def sanitize_and_parse_json(raw_llm_output: str) -> Dict[str, Any]:
    """
    Strips conversational text and markdown blocks to parse underlying JSON safely.
    """
    # Pattern to extract content enclosed inside markdown json blocks or raw braces
    json_pattern = r"```(?:json)?\s*(\{.*?\})\s*```|(\{.*\})"
    match = re.search(json_pattern, raw_llm_output, re.DOTALL)

    if match:
        clean_json_str = match.group(1) or match.group(2)
    else:
        clean_json_str = raw_llm_output.strip()

    try:
        return json.loads(clean_json_str)
    except json.JSONDecodeError as err:
        raise ValueError(f"Failed to parse extracted text into valid JSON: {err}")

Follow-Up Questions & Production Hardening

• Candidate Answer: 1. Schema Enforcement: Use Pydantic to enforce strict structural and data-type validation beyond basic JSON syntax checks. 2. Self-Correction Retry Loop: Pass the malformed generation and the specific validation error message back to the LLM in a targeted follow-up prompt, requesting a corrected payload. 3. Graceful Fallbacks: If the model fails after \(N\) retry attempts, log the payload for monitoring, fire an alert, and trigger a structured fallback response to maintain upstream service stability.