Designing Machine Learning Systems That Capture Real-World Signals
Modern machine learning systems increasingly rely on behavioral signals that reflect real-world actions rather than purely digital interactions. Purchases, checkouts, and user engagement events all encode information about user intent, economic activity, and preference formation. But the gap between raw event logs and a model that can actually rank or predict something useful is enormous — and most of the engineering effort lives in that gap.
This article walks through the architectural decisions behind ML systems that convert raw behavioral data into dense representations consumed by ranking and prediction models. I’ll cover the full pipeline from event ingestion through feature computation, embedding learning, and serving, with concrete code examples at each stage.
System Architecture Overview
Before diving into components, here is the end-to-end architecture of a typical behavioral ML system:
┌──────────────┐ ┌───────────────────┐ ┌─────────────────┐
│ │ │ │ │ │
│ Raw Events │────▶│ Event Processor │────▶│ Feature Store │
│ (Kafka/ │ │ (Normalization, │ │ (Redis/ │
│ Kinesis) │ │ Sessionization, │ │ DynamoDB) │
│ │ │ Deduplication) │ │ │
└──────────────┘ └───────────────────┘ └────────┬────────┘
│
▼
┌───────────────────┐ ┌─────────────────┐
│ │ │ │
│ Ranking Service │◀────│ Embedding Model │
│ (Real-time │ │ (Transformer / │
│ inference) │ │ Two-Tower) │
│ │ │ │
└───────────────────┘ └─────────────────┘
Each box here represents a system boundary with its own scaling characteristics, failure modes, and data contracts. The rest of this post walks through each one.
From Raw Signals to Structured Representations
Behavioral events are noisy and heterogeneous. A typical pipeline ingests page views, clicks, purchases, checkout events, and advertiser interactions. These events arrive asynchronously, at wildly different rates, and with inconsistent schemas across sources.
The first step is event normalization — mapping heterogeneous event types into a common schema that downstream systems can consume without knowing the original source.
Event Processing Pipeline
Here is a simplified event processor that normalizes raw events into feature vectors. In production, this would run as a streaming job (Flink, Spark Structured Streaming, or a custom consumer), but the core logic looks like this:
from dataclasses import dataclass, field
from typing import Optional
import numpy as np
from datetime import datetime, timezone
@dataclass
class RawEvent:
user_id: str
event_type: str # "click", "purchase", "page_view", "checkout"
item_id: str
timestamp: float
metadata: dict = field(default_factory=dict)
@dataclass
class NormalizedEvent:
user_id: str
item_id: str
timestamp: float
event_vector: np.ndarray # fixed-size numerical representation
# Event type weights reflecting downstream signal strength.
# Purchases carry far more intent signal than page views.
EVENT_WEIGHTS = {
"page_view": 0.1,
"click": 0.3,
"add_to_cart": 0.6,
"checkout": 0.8,
"purchase": 1.0,
}
# One-hot dimension per event type
EVENT_TYPE_DIM = len(EVENT_WEIGHTS)
METADATA_DIM = 8
TOTAL_DIM = EVENT_TYPE_DIM + METADATA_DIM + 2 # +2 for weight and time features
def normalize_event(event: RawEvent, reference_time: float) -> NormalizedEvent:
"""Convert a raw event into a fixed-size numerical vector.
The vector encodes:
- One-hot event type
- Scalar weight reflecting signal strength
- Time decay relative to a reference point
- Hashed metadata features (price bucket, category, etc.)
"""
event_types = list(EVENT_WEIGHTS.keys())
one_hot = np.zeros(EVENT_TYPE_DIM, dtype=np.float32)
if event.event_type in EVENT_WEIGHTS:
idx = event_types.index(event.event_type)
one_hot[idx] = 1.0
weight = EVENT_WEIGHTS.get(event.event_type, 0.05)
# Exponential time decay — events older than ~7 days contribute
# very little. The half-life is tunable per use case.
hours_elapsed = (reference_time - event.timestamp) / 3600.0
time_decay = np.exp(-0.01 * max(hours_elapsed, 0))
# Hash metadata into a fixed-size vector. In production you'd use
# a proper feature hashing scheme (e.g., MurmurHash mod N).
meta_vector = np.zeros(METADATA_DIM, dtype=np.float32)
for key, value in event.metadata.items():
bucket = hash((key, str(value))) % METADATA_DIM
meta_vector[bucket] += 1.0
# L2-normalize metadata to bound its contribution
norm = np.linalg.norm(meta_vector)
if norm > 0:
meta_vector /= norm
feature_vector = np.concatenate([
one_hot,
np.array([weight, time_decay], dtype=np.float32),
meta_vector,
])
return NormalizedEvent(
user_id=event.user_id,
item_id=event.item_id,
timestamp=event.timestamp,
event_vector=feature_vector,
)
def build_user_sequence(
events: list[RawEvent],
max_seq_len: int = 128,
) -> np.ndarray:
"""Aggregate a user's raw events into a padded sequence matrix.
Returns shape (max_seq_len, TOTAL_DIM). Most recent events come last.
Older events beyond max_seq_len are dropped.
"""
now = datetime.now(timezone.utc).timestamp()
normalized = [normalize_event(e, now) for e in events]
normalized.sort(key=lambda e: e.timestamp)
# Take the most recent max_seq_len events
normalized = normalized[-max_seq_len:]
seq = np.zeros((max_seq_len, TOTAL_DIM), dtype=np.float32)
for i, evt in enumerate(normalized):
offset = max_seq_len - len(normalized) + i
seq[offset] = evt.event_vector
return seq
A few things worth calling out:
The choice of exponential time decay with a tunable half-life is deliberate. In practice, you want recent events to dominate for short-term intent (what is this user looking for right now?) while still retaining long-term preference signals. We found that a half-life of roughly 7 days works well for e-commerce, but this varies dramatically by domain.
The build_user_sequence function produces a padded matrix of shape (max_seq_len, feature_dim). This is the input format expected by the transformer-based embedding model described below.
Why Sequence Order Matters
A naive approach might aggregate events into a single summary vector (e.g., mean pooling over all events). This throws away temporal structure, which is one of the richest signals available. Consider the difference between:
- User views running shoes → views hiking boots → purchases trail runners
- User purchases trail runners → views hiking boots → views running shoes
Same set of events, completely different intent. The first user is actively shopping; the second already bought and is browsing casually. A sequence model captures this distinction; a bag-of-events model cannot.
Production Feature Computation
Between the event processor and the embedding model sits the feature store — the system responsible for computing, storing, and serving features at inference time. This is where most production complexity lives.
Here is an example of a feature computation module that maintains sliding-window aggregates per user:
import time
from collections import defaultdict
from typing import Any
class SlidingWindowFeatureComputer:
"""Compute real-time aggregate features over sliding time windows.
Maintains per-user counters for configurable windows (e.g., 1h, 24h, 7d).
In production, this would be backed by Redis sorted sets or a dedicated
feature store like Feast/Tecton.
"""
WINDOWS = {
"1h": 3600,
"24h": 86400,
"7d": 604800,
}
def __init__(self):
# user_id -> list of (timestamp, event_type, item_id)
self._events: dict[str, list[tuple[float, str, str]]] = defaultdict(list)
def ingest(self, user_id: str, event_type: str, item_id: str,
ts: Optional[float] = None):
ts = ts or time.time()
self._events[user_id].append((ts, event_type, item_id))
def compute_features(self, user_id: str) -> dict[str, Any]:
"""Return a feature dict suitable for model input.
Features computed:
- event_count_{window}: total events in window
- unique_items_{window}: distinct items interacted with
- purchase_rate_{window}: purchases / total events
- avg_event_gap_{window}: mean seconds between consecutive events
"""
now = time.time()
events = self._events.get(user_id, [])
features = {}
for window_name, window_seconds in self.WINDOWS.items():
cutoff = now - window_seconds
window_events = [(ts, et, iid) for ts, et, iid in events
if ts >= cutoff]
count = len(window_events)
features[f"event_count_{window_name}"] = count
features[f"unique_items_{window_name}"] = len(
set(iid for _, _, iid in window_events)
)
purchases = sum(1 for _, et, _ in window_events if et == "purchase")
features[f"purchase_rate_{window_name}"] = (
purchases / count if count > 0 else 0.0
)
if count >= 2:
timestamps = sorted(ts for ts, _, _ in window_events)
gaps = [timestamps[i+1] - timestamps[i]
for i in range(len(timestamps) - 1)]
features[f"avg_event_gap_{window_name}"] = sum(gaps) / len(gaps)
else:
features[f"avg_event_gap_{window_name}"] = 0.0
return features
These windowed features serve two purposes. First, they provide direct signal to the ranking model (a user with 50 events in the last hour is in a very different browsing mode than one with 2 events in the last week). Second, they act as conditioning context for the embedding model — the same event sequence should produce different embeddings depending on the user’s current activity level.
Representation Learning at Scale
The core of the system is the embedding model — the component that transforms a variable-length sequence of behavioral events into a fixed-size dense vector. This vector needs to capture both long-term preferences and short-term intent.
Transformer-Based Embedding Model
We use a transformer encoder architecture with a few modifications tuned for behavioral sequences rather than natural language:
import torch
import torch.nn as nn
import math
class PositionalEncoding(nn.Module):
"""Sinusoidal positional encoding, but computed over actual timestamps
rather than integer positions. This lets the model reason about real
time gaps between events."""
def __init__(self, d_model: int, max_len: int = 512):
super().__init__()
pe = torch.zeros(max_len, d_model)
position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1)
div_term = torch.exp(
torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model)
)
pe[:, 0::2] = torch.sin(position * div_term)
pe[:, 1::2] = torch.cos(position * div_term)
self.register_buffer("pe", pe.unsqueeze(0)) # (1, max_len, d_model)
def forward(self, x: torch.Tensor) -> torch.Tensor:
return x + self.pe[:, :x.size(1)]
class BehavioralEmbeddingModel(nn.Module):
"""Transformer encoder that maps a sequence of behavioral event vectors
to a single dense user embedding.
Architecture choices:
- Input projection from event feature dim to model dim
- Standard transformer encoder blocks with pre-norm (more stable training)
- Attention pooling over the sequence to produce a single vector
(better than CLS token for variable-length behavioral sequences)
- Final projection to embedding space with L2 normalization
"""
def __init__(
self,
event_dim: int = 15, # dimension of each event vector
d_model: int = 128, # transformer hidden dimension
nhead: int = 4,
num_layers: int = 3,
embedding_dim: int = 64, # final embedding size
dropout: float = 0.1,
max_seq_len: int = 128,
):
super().__init__()
# Project raw event features into model dimension
self.input_proj = nn.Sequential(
nn.Linear(event_dim, d_model),
nn.LayerNorm(d_model),
nn.GELU(),
)
self.pos_encoder = PositionalEncoding(d_model, max_len=max_seq_len)
encoder_layer = nn.TransformerEncoderLayer(
d_model=d_model,
nhead=nhead,
dim_feedforward=d_model * 4,
dropout=dropout,
activation="gelu",
batch_first=True,
norm_first=True, # pre-norm for training stability
)
self.transformer = nn.TransformerEncoder(
encoder_layer, num_layers=num_layers
)
# Attention pooling: learn a query vector that attends over
# the full sequence to produce a single summary vector.
self.attn_pool_query = nn.Parameter(torch.randn(1, 1, d_model))
self.attn_pool = nn.MultiheadAttention(
d_model, num_heads=1, batch_first=True
)
# Project to final embedding space
self.output_proj = nn.Sequential(
nn.Linear(d_model, embedding_dim),
nn.LayerNorm(embedding_dim),
)
def forward(
self,
event_sequence: torch.Tensor, # (batch, seq_len, event_dim)
padding_mask: torch.Tensor | None = None, # (batch, seq_len), True = padded
) -> torch.Tensor:
"""Returns L2-normalized embeddings of shape (batch, embedding_dim)."""
x = self.input_proj(event_sequence) # (batch, seq_len, d_model)
x = self.pos_encoder(x)
x = self.transformer(x, src_key_padding_mask=padding_mask)
# Attention pooling
batch_size = x.size(0)
query = self.attn_pool_query.expand(batch_size, -1, -1)
pooled, _ = self.attn_pool(query, x, x, key_padding_mask=padding_mask)
pooled = pooled.squeeze(1) # (batch, d_model)
embedding = self.output_proj(pooled) # (batch, embedding_dim)
# L2 normalize — this is critical for retrieval via approximate
# nearest neighbor search (cosine similarity = dot product on
# unit vectors, which HNSW/ScaNN can index efficiently).
embedding = nn.functional.normalize(embedding, p=2, dim=-1)
return embedding
A few design decisions worth explaining:
Pre-norm transformer blocks (setting norm_first=True) place layer normalization before the attention and FFN sublayers rather than after. This consistently produces more stable training dynamics for behavioral data, which tends to have higher variance than text.
Attention pooling instead of a CLS token or mean pooling. With behavioral sequences, the informative events are often a small fraction of the total (one purchase among dozens of page views). Attention pooling lets the model learn to focus on high-signal events. In ablation experiments, this typically gives a 2-4% improvement in retrieval recall over mean pooling.
L2 normalization of the output embedding is essential for serving. Production retrieval systems (FAISS, ScaNN, Pinecone) operate on cosine similarity or inner product. With L2-normalized vectors, cosine similarity reduces to a dot product, which these systems index extremely efficiently using approximate nearest neighbor algorithms like HNSW.
A common mistake is training with cosine similarity loss but serving un-normalized embeddings. The ANN index will return different results than what the model was optimized for. Always normalize at the model output layer, not as a post-processing step.
Training Objective
The model is trained with a contrastive loss that pulls together embeddings of users with similar behavioral patterns and pushes apart those with different patterns. In practice, we use in-batch negatives with a temperature-scaled cross-entropy loss:
def contrastive_loss(
user_embeddings: torch.Tensor, # (batch, dim)
item_embeddings: torch.Tensor, # (batch, dim)
temperature: float = 0.07,
) -> torch.Tensor:
"""InfoNCE / NT-Xent loss with in-batch negatives.
Each (user, item) pair at the same index is a positive pair.
All other combinations within the batch are negative pairs.
"""
# Similarity matrix: (batch, batch)
logits = torch.matmul(user_embeddings, item_embeddings.T) / temperature
# Labels: the diagonal entries are the positive pairs
labels = torch.arange(logits.size(0), device=logits.device)
# Symmetric loss: user→item and item→user
loss_u2i = nn.functional.cross_entropy(logits, labels)
loss_i2u = nn.functional.cross_entropy(logits.T, labels)
return (loss_u2i + loss_i2u) / 2
The temperature parameter controls how hard the negatives are. Lower temperatures make the loss focus more on hard negatives (items that are similar but not the target), which generally improves embedding quality but can destabilize training if set too low. We typically start at 0.07 and tune from there.
Practical Challenges in Production
Data Sparsity and Cold Start
For new users with no behavioral history, the embedding model cannot produce a meaningful representation. We handle this with a fallback stack:
- If the user has ≥5 events, use the full transformer embedding
- If the user has 1-4 events, use a simplified mean-pooling embedding (skip the transformer)
- If the user has zero events, fall back to a popularity-based ranking
This tiered approach avoids serving garbage embeddings for cold-start users while still personalizing as soon as we have signal.
Concept Drift
User behavior changes over time — seasonal patterns, trend shifts, and external events all cause concept drift. We address this in two ways:
- Online feature updates: The sliding-window features described above naturally decay old signal. A user who was interested in winter coats in December will have those events drop out of the 7-day window by January.
- Periodic model retraining: The embedding model is retrained weekly on the most recent 90 days of data. We use a warm-start from the previous checkpoint, which reduces training time by roughly 60% while still adapting to distributional shifts.
Latency Constraints
The ranking service has a strict p99 latency budget of 50ms. This constrains several architectural choices:
- Embeddings are precomputed and cached in the feature store, not computed at request time
- The ANN index uses HNSW with a tuned
ef_searchparameter balancing recall vs. latency - Feature lookups use Redis with connection pooling; a cache miss falls back to a precomputed daily snapshot
Conclusion
The ability to convert raw behavioral events into structured representations lies at the heart of modern ML systems. The pipeline — event normalization, feature computation, embedding learning, and retrieval — involves engineering decisions at every stage that directly impact model quality and system reliability. As behavioral data grows richer and more diverse, the systems that extract signal from it will need to be equally sophisticated in how they process, represent, and serve that information.
Mattia Gaggi is an applied machine learning engineer working on production ML systems.
Last updated 2026-04-13