embeddings(bge-m3): vectorize sparse/colbert aggregation on GPU

The per-token Python loops that built the sparse {token_id: weight} dict and
the colbert vector lists held the GIL between forward passes, starving the GPU
(steady ~10-36% util with brief bursts). Replace them with GPU tensor ops:

- sparse: relu(sparse_linear·h) with invalid positions (specials/padding)
  zeroed, then max-pooled per token id via a single scatter_reduce(amax) into a
  (B, vocab) matrix; per-row nonzero extraction is O(nnz), not O(L).
- colbert: one C-level tolist per row over its valid tokens (vs per-token).
- valid mask computed on-device via torch.isin.

Output byte-matches the previous per-token logic (dense/sparse/colbert within
1e-4). Benefit grows with batch size, so larger client batches translate to GPU
work instead of Python-loop time.
Co-Authored-By: 's avatarClaude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Mw2KQiswmD69T45fTfjKwW
parent 8fb4f85a
...@@ -1257,51 +1257,56 @@ def _bge_m3_encode(model_obj, texts: List[str], types, max_length: int = 8192): ...@@ -1257,51 +1257,56 @@ def _bge_m3_encode(model_obj, texts: List[str], types, max_length: int = 8192):
relu(sparse_linear·h), max-pooled per token id, special tokens dropped; colbert = relu(sparse_linear·h), max-pooled per token id, special tokens dropped; colbert =
L2-normalized colbert_linear·h per content token.""" L2-normalized colbert_linear·h per content token."""
import torch import torch
import torch.nn.functional as F
tokenizer, model, heads, device = model_obj.model tokenizer, model, heads, device = model_obj.model
enc = tokenizer(texts, padding=True, truncation=True, enc = tokenizer(texts, padding=True, truncation=True,
max_length=int(max_length), return_tensors='pt') max_length=int(max_length), return_tensors='pt')
enc = {k: v.to(device) for k, v in enc.items()} enc = {k: v.to(device) for k, v in enc.items()}
with torch.no_grad(): input_ids = enc['input_ids'] # (B, L) on device
out = model(**enc) attn = enc['attention_mask'] # (B, L) on device
last = out.last_hidden_state # (B, L, H) n = input_ids.shape[0]
input_ids = enc['input_ids'].cpu()
attn = enc['attention_mask'].cpu()
specials = set(getattr(tokenizer, 'all_special_ids', []) or [])
n = len(texts)
results = [dict() for _ in range(n)] results = [dict() for _ in range(n)]
if 'dense' in types: with torch.no_grad():
dense = torch.nn.functional.normalize(last[:, 0], p=2, dim=-1).cpu() out = model(**enc)
for i in range(n): last = out.last_hidden_state # (B, L, H)
results[i]['dense'] = dense[i].tolist()
# Valid = attended AND not a special token (CLS/SEP/PAD/UNK/MASK). Computed on
if 'sparse' in types: # the GPU so aggregation never falls back to a Python per-token loop (the GIL
w = torch.relu(heads['sparse'](last)).squeeze(-1).cpu() # (B, L) # bottleneck that starved the card between forward passes).
for i in range(n): specials = sorted(set(getattr(tokenizer, 'all_special_ids', []) or []))
d = {} if specials:
ids_i, w_i, m_i = input_ids[i], w[i], attn[i] valid = attn.bool() & ~torch.isin(
for j in range(ids_i.shape[0]): input_ids, torch.tensor(specials, device=device))
if int(m_i[j]) == 0: else:
continue valid = attn.bool()
tid = int(ids_i[j])
if tid in specials: if 'dense' in types:
continue dense = F.normalize(last[:, 0], p=2, dim=-1).cpu().tolist() # (B, H)
val = float(w_i[j]) for i in range(n):
if val > d.get(tid, 0.0): results[i]['dense'] = dense[i]
d[tid] = val
results[i]['sparse'] = d if 'sparse' in types:
# relu(sparse_linear·h), invalid positions zeroed, then max-pooled per
if 'colbert' in types: # token id via a single GPU scatter_reduce (amax) into a (B, vocab) matrix.
col = torch.nn.functional.normalize(heads['colbert'](last), p=2, dim=-1).cpu() w = torch.relu(heads['sparse'](last)).squeeze(-1).float() # (B, L)
for i in range(n): w = w.masked_fill(~valid, 0.0)
vecs = [] vocab = int(getattr(model.config, 'vocab_size', 0)) or int(input_ids.max()) + 1
ids_i, m_i = input_ids[i], attn[i] sparse_mat = torch.zeros(n, vocab, device=device, dtype=w.dtype)
for j in range(ids_i.shape[0]): sparse_mat.scatter_reduce_(1, input_ids, w, reduce='amax', include_self=True)
if int(m_i[j]) == 0 or int(ids_i[j]) in specials: sparse_mat = sparse_mat.cpu()
continue for i in range(n):
vecs.append(col[i][j].tolist()) row = sparse_mat[i]
results[i]['colbert'] = vecs nz = torch.nonzero(row, as_tuple=True)[0] # token ids > 0
results[i]['sparse'] = dict(zip(nz.tolist(), row[nz].tolist()))
if 'colbert' in types:
col = F.normalize(heads['colbert'](last), p=2, dim=-1).cpu() # (B, L, H)
valid_cpu = valid.cpu()
for i in range(n):
# one C-level tolist per row over its valid tokens (vs per-token)
results[i]['colbert'] = col[i][valid_cpu[i]].tolist()
return results return results
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment