Implementation:Neuml Txtai Cluster

Knowledge Sources	Neuml_Txtai
Domains	Distributed_Search, API
Last Updated	2026-02-09 17:00 GMT

Overview

Cluster aggregates multiple embeddings shards into a single logical embeddings instance, distributing search and indexing operations across remote HTTP endpoints using asynchronous requests.

Description

The Cluster class provides a distributed embeddings search layer that fans out queries to multiple txtai API shards and merges the results. It uses aiohttp for concurrent asynchronous HTTP requests to all configured shard URLs, and leverages the Aggregate SQL processor to combine partial results with proper handling of aggregate functions, GROUP BY, and ORDER BY clauses. Document sharding for writes uses a deterministic hash-based assignment (adler32 for string IDs, modulo for integer IDs) to distribute documents evenly across shards.

Usage

Use Cluster when a single embeddings index is too large for one machine or when you want to horizontally scale search throughput across multiple txtai API instances. It is the primary mechanism for distributed search in txtai deployments.

Code Reference

Source Location

• Repository: Neuml_Txtai
• File: src/python/txtai/api/cluster.py
• Lines: 1-295

Signature

class Cluster:
    def __init__(self, config=None):
        """
        Creates a new Cluster.

        Args:
            config: cluster configuration
        """

Import

from txtai.api import Cluster

Key Methods

Method	Description
search(query, limit=None, weights=None, index=None, parameters=None, graph=False)	Searches all shards for the given query via HTTP GET, aggregates results using SQL-aware merging, and returns the top results up to limit.
batchsearch(queries, limit=None, weights=None, index=None, parameters=None, graph=False)	Runs multiple queries across all shards via HTTP POST, combines results per query, and applies aggregation and limits.
add(documents)	Distributes documents across shards using hash-based assignment and sends each shard's batch via HTTP POST.
index()	Triggers index building on all shards via HTTP GET.
upsert()	Triggers upsert operation on all shards via HTTP GET.
delete(ids)	Sends delete requests to all shards and returns the list of deleted IDs.
reindex(config, function=None)	Recreates the embeddings index on all shards with a new configuration.
count()	Returns the sum of element counts across all shards.
shard(documents)	Splits documents into per-shard lists using hash-based deterministic assignment (adler32 for strings, modulo for integers, random for None).
execute(method, action, data=None)	Core async HTTP execution engine. Dispatches GET or POST requests to all shard URLs concurrently via aiohttp and asyncio.

I/O Contract

Inputs

Name	Type	Required	Description
config	dict	Yes	Cluster configuration. Must contain a shards key with a list of shard URL strings (e.g., ["http://host1:8080", "http://host2:8080"]).
query	str	Yes (for search)	Search query string. Passed as a URL parameter to each shard's search endpoint.
queries	list of str	Yes (for batchsearch)	List of search query strings for batch processing.
documents	list of dict	Yes (for add)	List of documents to add. Each document should have an id key used for shard assignment.
limit	int	No	Maximum results to return. Defaults to 10.
weights	float	No	Hybrid score weights for combining semantic and keyword search results.
index	str	No	Named index to search, if applicable.
parameters	dict	No	Named parameters to bind to SQL query placeholders.

Outputs

Name	Type	Description
search results	list of dict	Aggregated and sorted results from all shards. Format matches txtai API search response: list of {"id": ..., "score": ...} dicts, or richer dicts when database content is enabled.
batchsearch results	list of list of dict	Per-query aggregated results from all shards.
count	int	Sum of element counts across all shards.
deleted ids	list	Flattened list of IDs deleted across all shards.

Usage Examples

Basic Usage

from txtai.api import Cluster

# Configure a cluster with two shards
config = {
    "shards": [
        "http://shard1:8080",
        "http://shard2:8080",
    ]
}

cluster = Cluster(config)

# Add documents - automatically distributed across shards
documents = [
    {"id": "doc1", "text": "Machine learning fundamentals"},
    {"id": "doc2", "text": "Natural language processing techniques"},
    {"id": "doc3", "text": "Computer vision with deep learning"},
    {"id": "doc4", "text": "Reinforcement learning applications"},
]

cluster.add(documents)
cluster.index()

# Search across all shards
results = cluster.search("deep learning", limit=3)
for result in results:
    print(f"ID: {result['id']}, Score: {result['score']:.4f}")

# Get total count across all shards
print(f"Total documents: {cluster.count()}")

Batch Search

from txtai.api import Cluster

config = {"shards": ["http://shard1:8080", "http://shard2:8080"]}
cluster = Cluster(config)

# Run multiple queries at once
queries = ["machine learning", "natural language", "computer vision"]
results = cluster.batchsearch(queries, limit=5)

for query, query_results in zip(queries, results):
    print(f"Query: {query}")
    for result in query_results:
        print(f"  ID: {result['id']}, Score: {result['score']:.4f}")

Related Pages

• Principle:Neuml_Txtai_Distributed_Search