Implementation:Neuml Txtai Sparse Scoring

Knowledge Sources	txtai txtai Documentation
Domains	Sparse_Retrieval, Learned_Embeddings
Last Updated	2026-02-09 00:00 GMT

Overview

Sparse vector scoring backend using learned sparse embeddings with threaded async encoding and ANN-based retrieval.

Description

The Sparse class implements a scoring backend that transforms text into learned sparse vectors using a sparse embedding model, then indexes and retrieves them via a sparse approximate nearest neighbor (ANN) index. Unlike traditional keyword-based scoring (e.g., BM25, TF-IDF), this approach uses a neural model to produce sparse vectors where dimensions correspond to vocabulary terms weighted by learned importance scores.

The class uses a threaded encoding pipeline for efficient batch processing during indexing. A background thread reads batches from a Queue, encodes them via the sparse vector model, and accumulates the results. The main thread feeds documents to the queue and blocks on completion when the index is built. This producer-consumer pattern is controlled by the start(), stop(), encode(), and stream() methods.

The indexing flow proceeds as:

1. insert() starts the encoding thread (if not running) and pushes document batches onto the queue.
2. index() or upsert() sends the end-of-stream message, waits for the encoding thread to complete, then builds or appends to the sparse ANN index.
3. search() encodes query text, runs ANN search, and optionally normalizes scores.

Score normalization is applied when self.isnormalize is enabled and the vector model does not already produce normalized vectors. The normalization divides each score by the maximum possible score (derived from the query's self-dot-product divided by a scale factor, defaulting to 30.0).

Usage

Use this scoring backend for learned sparse retrieval that captures semantic term importance beyond simple term frequency. It is typically created through the ScoringFactory as part of an embeddings configuration. It can serve as the sparse component in a hybrid dense+sparse retrieval setup.

Code Reference

Source Location

• Repository: txtai
• File: src/python/txtai/scoring/sparse.py
• Lines: L1-218

Class Definition

class Sparse(Scoring):
    """
    Sparse vector scoring.
    """

Constructor Signature

def __init__(self, config=None, models=None):

The constructor maps configuration keys (vectormethod -> method, vectornormalize -> normalize), loads the sparse vector model via SparseVectorsFactory.create, and initializes threading and ANN state to None.

Import

from txtai.scoring import Sparse  # via ScoringFactory

I/O Contract

insert(documents, index=None, checkpoint=None)

Name	Type	Required	Description
documents	iterable of (uid, document, tags)	Yes	Documents to insert. Each document can be a string, a list of strings (joined with spaces), or a dict with "text" or "object" keys.
index	int or None	No	Not used by Sparse; present for interface compatibility.
checkpoint	any	No	Checkpoint directory passed to the encoding thread for recovery support.

search(query, limit=3)

Name	Type	Required	Description
query	str	Yes	Query text to search for.
limit	int	No	Maximum number of results. Defaults to 3.

Returns: A list of (uid, score) tuples. Scores are optionally normalized.

batchsearch(queries, limit=3, threads=True)

Name	Type	Required	Description
queries	list of str	Yes	List of query texts.
limit	int	No	Maximum number of results per query. Defaults to 3.
threads	bool	No	Accepted for interface compatibility. Defaults to True.

Returns: A list of result lists, one per query.

Key Methods

insert(documents, index=None, checkpoint=None)

Starts the background encoding thread (if not already running), extracts text from documents, and pushes the batch onto the processing queue.

delete(ids)

Delegates to self.ann.delete(ids) to remove entries from the sparse ANN index.

index(documents=None)

Optionally inserts documents, then stops the encoding thread and builds a new sparse ANN index from the accumulated embeddings:

embeddings = self.stop()
if embeddings is not None:
    self.ann = SparseANNFactory.create(self.config)
    self.ann.index(embeddings)

upsert(documents=None)

Similar to index() but appends to an existing ANN index rather than creating a new one. Falls back to index() if no ANN exists yet.

search(query, limit=3)

Delegates to batchsearch with a single query.

batchsearch(queries, limit=3, threads=True)

Encodes queries using the sparse vector model, runs ANN search, and optionally normalizes scores:

embeddings = self.model.batchtransform((None, query, None) for query in queries)
scores = self.ann.search(embeddings, limit)
return self.normalize(embeddings, scores) if self.isnormalize and not self.model.isnormalize else scores

load(path) / save(path) / close()

Standard persistence methods that delegate to the sparse ANN index. close() also clears the model, ANN, thread, and queue references.

Threading Architecture

The encoding pipeline uses a producer-consumer pattern:

start(checkpoint)

Creates a Queue(5) with a capacity of 5 batches and starts a background thread running self.encode(checkpoint).

stop()

Sends the end-of-stream sentinel (Sparse.COMPLETE = 1) to the queue, joins the thread, and returns the accumulated embeddings stored in self.data.

encode(checkpoint)

The thread target function. Streams data from the queue via self.stream(), feeds it through self.model.vectors(), and stores the results in self.data. Also saves the embedding dimensions to config.

stream()

Generator that reads batches from the queue and yields individual documents until the end-of-stream sentinel is received:

def stream(self):
    batch = self.queue.get()
    while batch != Sparse.COMPLETE:
        yield from batch
        batch = self.queue.get()

Score Normalization

normalize(queries, scores)

Normalizes scores when the model does not produce pre-normalized vectors:

scale = 30.0 if isinstance(self.isnormalize, bool) else self.isnormalize
maxscores = self.model.dot(queries, queries)

for x, result in enumerate(scores):
    maxscore = max(maxscores[x][x] / scale, scale)
    maxscore = max(maxscore, result[0][1]) if result else maxscore
    results.append([(uid, score / maxscore) for uid, score in result])

The normalization scale factor defaults to 30.0 but can be set to a custom float via the normalize config key. Each score is divided by the maximum of the scaled self-similarity or the top result score.

Configuration

Key	Type	Default	Description
method or vectormethod	str	None	Sparse vector model method.
normalize or vectornormalize	bool or float	True	Enable score normalization. A float sets the scale factor (default 30.0).
batch	int	1024	Batch size for the encoding pipeline.
dimensions	int	(auto-detected)	Number of sparse dimensions, set automatically after encoding.
columns.text	str	"text"	Key to extract text content from dict documents.
columns.object	str	"object"	Fallback key when "text" is not found.

Inheritance Chain

Sparse -> Scoring

The Scoring base class provides the interface contract and column configuration parsing.

Usage Examples

Configuring Sparse Scoring in Embeddings

from txtai.embeddings import Embeddings

# Hybrid search with dense + sparse scoring
embeddings = Embeddings({
    "path": "sentence-transformers/all-MiniLM-L6-v2",
    "content": True,
    "hybrid": True,
    "scoring": {
        "method": "sparse"
    }
})

embeddings.index([
    (0, "Neural network architectures for NLP", None),
    (1, "Database indexing strategies", None),
    (2, "Transformer attention mechanisms", None)
])

# Search uses both dense and sparse scoring
results = embeddings.search("attention in neural networks", limit=3)

Direct Sparse Scoring Usage

from txtai.scoring import Sparse

config = {
    "path": "neuml/txtai-intro",
    "batch": 512,
    "normalize": True
}

scoring = Sparse(config)

# Index documents
documents = [
    (0, "Machine learning classification methods", None),
    (1, "Deep learning with PyTorch", None),
    (2, "Statistical analysis techniques", None)
]
scoring.index(documents)

# Search
results = scoring.search("deep learning", limit=2)
for uid, score in results:
    print(f"ID: {uid}, Score: {score:.4f}")

scoring.close()

Related Pages

Implements Principle

• Principle:Neuml_Txtai_Sparse_Retrieval

Requires Environment

• Environment:Neuml_Txtai_Python_Scoring_Environment