TL;DR — This project demonstrates a deep understanding of probabilistic decoding, entropy management, and production‑grade Python engineering. By implementing temperature scaling, top‑p nucleus sampling, and dynamic entropy throttling from scratch, you showcase the ability to reason about LLM inference quality and efficiency trade‑offs. The code is fully runnable, testable, and extensible, making it a concrete talking point in interviews.
In modern LLM serving, the decoder is the bridge between raw model logits and the text users read. A naive greedy or beam‑search decoder produces repetitive, low‑diversity output, while a poorly tuned sampling strategy can generate incoherent or unsafe content. An adaptive temperature‑scaled top‑p (nucleus) sampler with dynamic entropy throttling gives you a single, tunable knob that balances creativity, coherence, and latency. This post walks you through building a pure‑Python implementation that you can drop into any inference pipeline, demonstrate on a résumé, and extend toward production.
Why This Project Stands Out on a CV
- Probabilistic reasoning – You will manipulate logits, compute Shannon entropy, and apply temperature scaling, showing you understand the mathematics behind generative models.
- Algorithm design – Implementing nucleus sampling and an entropy‑based throttle requires careful state management and real‑time decision logic, a skill that maps directly to algorithm‑heavy roles.
- Production‑grade Python – The code uses type hints, dataclasses, and a clean module structure, signaling you can write maintainable, testable software.
- Systems thinking – By exposing knobs for dynamic throttling, you demonstrate awareness of latency, throughput, and resource constraints in LLM serving.
- Interview talking point – Hiring managers love concrete projects that combine theory (entropy, sampling) with practice (runnable code, benchmarks). This project gives you a story for both ML engineering and full‑stack inference positions.
Architecture Overview
The sampler is composed of five loosely‑coupled components:
- LogitProcessor – Receives raw logits from the model and applies temperature scaling.
- NucleusFilter – Implements top‑p (nucleus) filtering by cumulatively masking low‑probability tokens until the cumulative mass exceeds
p. - EntropyThrottler – Computes the Shannon entropy of the filtered distribution and decides whether to tighten or relax the filter based on a target entropy window.
- Sampler – Ties the above together, performing the final categorical draw using
torch.multinomialornumpy.random.choice. - CLI / Driver – A small
__main__block that reads synthetic logits (or a real model’s output) and prints sampled tokens, useful for quick demos and testing.
A textual diagram:
Model Logits ──► LogitProcessor (temperature) ──► NucleusFilter (top‑p)
│
▼
EntropyThrottler (adjust p)
│
▼
Sampler (multinomial)
│
▼
Output Token(s)
Each component is a pure function or a small class, making unit testing straightforward and allowing you to swap implementations later (e.g., switch to a different entropy estimator).
Building It Step by Step
Below are the core steps with runnable Python snippets. The full module can be saved as adaptive_sampler.py.
Step 1 – Project Skeleton
mkdir adaptive_sampler
cd adaptive_sampler
touch adaptive_sampler.py
Step 2 – Data Types and Imports
from __future__ import annotations
import math
from dataclasses import dataclass
from typing import List, Tuple
import numpy as np
Step 3 – Temperature Scaling
@dataclass
class TemperatureConfig:
temperature: float = 1.0 # 0 < temperature <= 1.0 for sharper, >1.0 for softer
def apply_temperature(logits: np.ndarray, cfg: TemperatureConfig) -> np.ndarray:
"""Divide logits by temperature before softmax."""
if cfg.temperature <= 0:
raise ValueError("Temperature must be positive.")
return logits / cfg.temperature
Step 4 – Softmax and Entropy
def softmax(x: np.ndarray) -> np.ndarray:
e = np.exp(x - np.max(x))
return e / e.sum()
def shannon_entropy(probs: np.ndarray) -> float:
"""Compute entropy in bits."""
probs = probs[probs > 0] # avoid log(0)
return -np.sum(probs * np.log2(probs))
Step 5 – Nucleus (Top‑p) Filtering
def nucleus_filter(probs: np.ndarray, p: float) -> np.ndarray:
"""Zero out tokens outside the top‑p mass."""
sorted_idx = np.argsort(probs)[::-1]
sorted_probs = probs[sorted_idx]
cumsum = np.cumsum(sorted_probs)
cutoff = np.searchsorted(cumsum, p, side="right")
keep_idx = sorted_idx[: cutoff + 1]
mask = np.zeros_like(probs, dtype=bool)
mask[keep_idx] = True
filtered = probs * mask
filtered /= filtered.sum() # renormalize
return filtered
Step 6 – Dynamic Entropy Throttler
@dataclass
class ThrottleConfig:
target_entropy: float = 2.0 # desired entropy in bits
tolerance: float = 0.5 # allowed deviation
class EntropyThrottler:
def __init__(self, cfg: ThrottleConfig):
self.cfg = cfg
def adjust_p(self, current_entropy: float, current_p: float) -> float:
"""Increase p if entropy too low, decrease if too high."""
deviation = current_entropy - self.cfg.target_entropy
if abs(deviation) <= self.cfg.tolerance:
return current_p
# Simple linear adjustment; you can replace with a PID controller.
step = 0.05 * deviation
new_p = float(np.clip(current_p - step, 0.01, 0.99))
return new_p
Step 7 – The Sampler Class
@dataclass
class AdaptiveSamplerConfig:
temperature: float = 1.0
initial_p: float = 0.9
target_entropy: float = 2.0
tolerance: float = 0.5
class AdaptiveSampler:
def __init__(self, cfg: AdaptiveSamplerConfig):
self.temp_cfg = TemperatureConfig(temperature=cfg.temperature)
self.throttle_cfg = ThrottleConfig(
target_entropy=cfg.target_entropy, tolerance=cfg.tolerance
)
self.p = cfg.initial_p
self.throttler = EntropyThrottler(self.throttle_cfg)
def sample(self, logits: np.ndarray) -> int:
# 1. Temperature scaling
scaled = apply_temperature(logits, self.temp_cfg)
probs = softmax(scaled)
# 2. Nucleus filter
filtered = nucleus_filter(probs, self.p)
# 3. Entropy measurement
entropy = shannon_entropy(filtered)
# 4. Dynamic throttling
self.p = self.throttler.adjust_p(entropy, self.p)
# 5. Draw token
return int(np.random.choice(len(filtered), p=filtered))
Step 8 – Quick Demo
if __name__ == "__main__":
# Simulate logits for 10 tokens
rng = np.random.default_rng(42)
synthetic_logits = rng.normal(size=10)
sampler = AdaptiveSampler(AdaptiveSamplerConfig())
token = sampler.sample(synthetic_logits)
print(f"Sampled token index: {token}")
Run with:
python adaptive_sampler.py
You should see a token index printed each time you execute the script.
Running and Testing It
Local Execution – Save the code above as
adaptive_sampler.pyand runpython adaptive_sampler.py. The script prints a sampled token; repeat to observe diversity.Unit Tests – Create
test_sampler.pyusingpytest:
import numpy as np
from adaptive_sampler import AdaptiveSampler, AdaptiveSamplerConfig
def test_temperature_scaling():
cfg = AdaptiveSamplerConfig(temperature=0.5)
sampler = AdaptiveSampler(cfg)
logits = np.array([1.0, 2.0, 3.0])
# Ensure no exception
sampler.sample(logits)
def test_entropy_adjustment():
sampler = AdaptiveSampler(AdaptiveSamplerConfig())
# Low‑entropy distribution should increase p
low_entropy_logits = np.array([10.0, 0.0, 0.0])
sampler.sample(low_entropy_logits)
assert sampler.p > 0.9 # p should rise
Run with pytest -v.
- Integration with a Real Model – If you have a Hugging Face model, replace the synthetic logits with
model(input_ids).logits[:, -1, :]. The sampler will work unchanged.
Extending It: Your Roadmap to Senior‑Level
Persistence Layer – Store throttling decisions and sampled sequences in SQLite or Redis. Why it matters: enables offline analysis of decoding behavior and supports replay for debugging.
Horizontal Scaling – Wrap the sampler in a FastAPI service and deploy multiple instances behind a load balancer; use a message queue (e.g., RabbitMQ) for request dispatch. Why it matters: meets throughput demands of production traffic.
Observability – Expose metrics (entropy, p‑value, latency) to Prometheus via
prometheus_client. Why it matters: operators can detect drift in decoding quality and trigger alerts.Fault Tolerance – Implement retry logic with exponential backoff for model inference calls; use circuit‑breaker patterns to avoid cascading failures. Why it matters: ensures robustness when the underlying model service degrades.
Benchmarking Suite – Add
pytest‑benchmarkto compare greedy, fixed‑temperature, and adaptive sampling on standard metrics (perplexity, diversity, latency). Why it matters: provides evidence for adopting the adaptive approach in production.Integration with Hugging Face Transformers – Subclass
LogitsProcessorfromtransformersto plug your sampler directly into the generation pipeline, enabling one‑line usage:model.generate(..., logits_processor=MyLogitsProcessor()). Why it matters: aligns your code with the dominant LLM ecosystem, making it instantly usable by the community.
Key Takeaways
- The project demonstrates a blend of probabilistic theory, algorithmic design, and production‑ready Python.
- It signals readiness for ML engineering, LLM inference, and systems‑focused roles.
- The code is modular, testable, and can be extended with persistence, scaling, observability, fault tolerance, benchmarking, and framework integration.
- By showcasing dynamic entropy throttling, you highlight an ability to balance quality and efficiency— a core concern in real‑world LLM serving.
Further Reading
- Holtzman, A., et al. The Curious Case of Neural Text Degeneration. arXiv:1904.09740 – Introduces nucleus sampling and discusses entropy.
- Holtzman, A., et al. Nucleus Sampling for Language Models. arXiv:2004.05778 – Detailed analysis of top‑p filtering.
- Hugging Face, LogitsProcessor documentation. https://huggingface.co/docs/transformers/main_classes/model#transformers.LogitsProcessor – How to integrate custom decoding logic.
- PyTorch, torch.multinomial reference. https://pytorch.org/docs/stable/generated/torch.multinomial.html – Underlying sampling primitive.
Building something like this? I’m an AI/systems contractor open to new projects. Book a 30-min call.