---
name: routir
description: Use RoutIR — call a running endpoint with the client, stand up a local server that mixes locally-hosted services with services proxied from a remote master server, and extend RoutIR with new bi-encoders, rerankers, and document collections by writing a small config (and optionally one Python file referenced via `file_imports`). No checkout of the routir source is required.
---
# RoutIR skill
RoutIR is an HTTP/gRPC service that hosts retrieval engines (dense, sparse,
rerankers, fusion, query expanders) behind a uniform API and composes them
into multi-stage pipelines with a small DSL. This skill covers the five
things you actually do with it:
1. Call a running RoutIR endpoint with the client.
2. Stand up your own RoutIR server, optionally importing services from a
master server so users can pipeline local + remote engines together.
3. Wrap a bi-encoder retrieval index.
4. Wrap a cross-encoder reranker.
5. Serve document collections (single- or multi-view).
**Run everything through `uvx`.** This project forbids bare `pip`,
`python`, `pytest`, `ruff`. Every command below uses `uvx` so the deps land
in a throwaway venv instead of your conda environment.
The full reference for extending RoutIR with Python lives in
[`examples/CLAUDE.md`](examples/CLAUDE.md); pipeline / DSL / config reference
in the project [`CLAUDE.md`](CLAUDE.md). This skill is the operational
overview — read those when you need depth.
---
## 1. Use the client (and the pipeline DSL)
`routir.client.Client` is the sync facade (auto-uses gRPC when the server
advertises it, falls back to REST). `AsyncClient` is the async version with
the same surface.
```python
from routir.client import Client
with Client(endpoint="http://compute01:5000", api_key="…optional…") as c:
print(c.avail()) # what's available
print(c.transport) # "grpc" or "rest"
# search a single hosted index
r = c.search(service="qwen3-neuclir", query="…", limit=20)
# r["scores"] -> {doc_id: float}
# score a list of passages against a query
r = c.score(service="my-reranker", query="…", passages=["p1", "p2", …])
# r["scores"] -> [float, float, …] (one per passage)
# fetch document content
r = c.content(collection="neuclir", id="doc-id-123", view="asr")
# r["text"] or r["bytes"]
# run a composed pipeline (the main entry point)
r = c.pipeline(
pipeline="{dense%1000, bm25%1000}RRF%100 >> rerank@asr%20",
query="…",
collection="neuclir", # needed when a stage reranks
runtime_kwargs={"rerank": {"some_engine_kwarg": 0.5}}, # optional
)
```
**Endpoint scheme rules:** `http(s)://…` → REST; `grpc(s)://…` → gRPC;
bare `host:port` is treated as `http://` and may auto-upgrade to gRPC if
the server advertises `grpc_port` via `/avail`. Pass
`transport="rest"` to force REST.
### Pipeline DSL — quick reference
| DSL form | Meaning |
| --------------------------------------- | ----------------------------------------------------------------------------- |
| `svc%N` | Call `svc`, keep top *N*. |
| `A%1000 >> B%20` | Sequential: A retrieves 1000, B reranks down to 20. B gets role `rerank`. |
| `{A%K, B%K}Merger%N` | Parallel: A and B run concurrently; `Merger.fuse_batch` fuses to top *N*. |
| `Expander{A%K, B%K}Merger%N` | Expander makes sub-queries; each runs A and B in parallel; all fused. |
| `svc[alias]%N` | Name a stage so `runtime_kwargs` can target it by alias. |
| `svc@view%N` | Pick a named view of the collection at rerank time (multi-modal). |
**Built-in mergers:** `RRF` (reciprocal rank fusion) and `ScoreFusion` are
always available. You only need to write your own fusion engine if you need
something neither covers.
**Aliases:** `pipeline_aliases` in the server config let you give a long
pipeline a short name (`"ragtime2": "{zho%100, rus%100, …}ScoreFusion"`);
the alias is then usable everywhere a service name is. The call-site `%N`
re-caps the alias's outer-most stage.
**Bound the result set.** `/pipeline` ignores a top-level `limit`; the
result-set size is whatever the final `%N` produced. If your pipeline has
no `%N`, you'll get the full inner result.
`scripts/query.py` is the canonical batch-query script that writes the
[JSONL run format documented in `CLAUDE.md`](CLAUDE.md#search-results-output-format).
Copy it when scripting evaluations.
---
## 2. Stand up a local server (and import a master server)
Minimal `my-config.json`:
```json
{
"services": [
{ "name": "my-bm25", "engine": "PyseriniBM25",
"config": {"index_path": "/path/to/bm25-index"},
"cache": 1024, "cache_ttl": 600,
"batch_size": 16, "max_wait_time": 0.05 }
],
"collections": [
{ "name": "my-corpus", "doc_path": "/data/corpus.jsonl" }
],
"server_imports": [
"http://master-host:5000"
],
"file_imports": [
"./my_extension.py"
]
}
```
**`server_imports` is the master-import mechanism.** At startup the local
server queries `/avail` on every entry and registers a `Relay`-backed local
proxy for every remote search / score / content service it doesn't already
host. After that, your users can write pipelines that freely mix local and
remote services as if they were all on one box:
```text
{my-bm25%1000, remote-qwen3%1000}RRF%100 >> remote-cross-encoder%20
```
Notes that bite people:
- **Local services win on name collision.** Remote services with a name
already registered locally are skipped, not overridden.
- **Each entry can be a dict** with `endpoint`, `grpc_endpoint`, `api_key`,
`transport`, etc., not just a bare URL — handy when the master is gRPC.
- **Cache relayed content.** Reranking against a remote collection
re-fetches every doc per query without a cache. Set
`"relay_content_cache": 4096` (and `relay_content_cache_ttl`) at the
top level of the config, or wire a Redis URL.
**Serve it:**
```bash
# REST only, port 5000
uvx --with "routir[dense] @ ." \
routir my-config.json --port 5000
# REST + gRPC; add --with for every runtime dep your engines need
uvx --with transformers --with torch --with "routir[dense,grpc] @ ." \
routir my-config.json --port 5000 --grpc --grpc-port 50051
# With auth (prefer env over --api_key; CLI args show up in ps)
ROUTIR_API_KEY=sekret uvx --with "routir @ ." routir my-config.json --port 5000
```
**Verify it works**:
```bash
curl http://localhost:5000/avail # lists search/score/fuse/collection
curl http://localhost:5000/ping # always unauthenticated; for liveness
```
`/avail` is also what `server_imports` uses for discovery — if your
service doesn't appear there, no one else will see it either.
**Per-service knobs in `services[]`** that you'll routinely tune:
| key | what it does |
| ------------------ | -------------------------------------------------------------------------------------------------- |
| `cache` | LRU capacity for this service's results. `-1` (default) disables. |
| `cache_ttl` | TTL in seconds. Applies to LRU and Redis. |
| `cache_key_fields` | Request fields that go into the cache key. Default `["query", "limit"]`; add `"subset"` etc. when they affect results. |
| `cache_redis_url` | Use Redis instead of in-memory LRU. |
| `batch_size` | Max requests batched into one engine call. Default 32. |
| `max_wait_time` | Max seconds to wait for a batch to fill. Default 0.05; raise for throughput, lower for latency. |
| `scoring_disabled` | Set to `true` to refuse to register `/score` for an engine that *can* score, when you only want search. |
---
## 3. Wrap a bi-encoder with an index
**Default path: write zero Python.** RoutIR ships `Qwen3` and
`SentenceTransformerEngine` and they cover the majority of dense models
purely through config. Reach for a custom engine only when neither fits.
### 3a. Zero-code: external query encoder via OpenAI-compatible API (preferred)
This is the right answer almost always — it keeps the query encoder on its
own GPU/process (vLLM, llama.cpp, sglang, …), so RoutIR doesn't have to
share VRAM with it and you can scale the encoder independently.
Run your query encoder as an OpenAI-compatible `/v1/embeddings` server
(vLLM, llama.cpp, sglang, TEI — any of them) and point RoutIR at it:
```json
{
"services": [
{
"name": "qwen3-neuclir",
"engine": "Qwen3",
"cache": 1024, "cache_ttl": 600,
"batch_size": 32, "max_wait_time": 0.05,
"config": {
"index_path": "/path/to/faiss-index",
"embedding_base_url": "http://gpu-host:8000/v1/",
"embedding_model_name": "Qwen/Qwen3-Embedding-8B",
"api_key": "…or set OPENAI_API_KEY env…",
"k_scale": 5
}
}
]
}
```
Likewise `SentenceTransformerEngine` covers ME5-Instruct, ArcticEmbed, BGE-M3,
Jina-v3, etc. by setting `embedding_model_name`, `instruction`,
`prompt_name_query`, `task_query`, `normalize_embeddings`. The full key list
is on the class in `src/routir/models/st.py`.
**`index_path`** is a directory with `index.faiss` + `index.ids` (one doc id
per line, in the same order as FAISS vectors). The `hfds:<org/repo>` prefix
auto-downloads from Hugging Face Datasets at startup
(e.g. `"index_path": "hfds:routir/neuclir-qwen3-8b-faiss-PQ2048x4fs"`).
### 3b. Run the query encoder in-process (less ideal)
If you must (e.g. small model, no spare encoder host, sharing VRAM is fine),
either drop `embedding_base_url` from the configs above — both engines fall
back to local `transformers`/`sentence-transformers` — or write a tiny
custom engine if neither fits. The cost is real: the query encoder now
contends with everything else this RoutIR process is doing.
If you do need to write your own, use `file_imports` to ship one `.py`
without touching the routir checkout:
```python
# my_extension.py
from typing import Dict, List
import faiss, numpy as np
from routir.models.abstract import Engine
class MyBiencoderEngine(Engine):
def __init__(self, name=None, config=None, **kwargs):
super().__init__(name, config, **kwargs)
self.encoder = load_my_query_encoder(config["model"])
self.index = faiss.read_index(f"{config['index_path']}/index.faiss")
with open(f"{config['index_path']}/index.ids") as f:
self.doc_ids = [ln.strip() for ln in f]
async def search_batch(self, queries: List[str], limit=1000, **kwargs) -> List[Dict[str, float]]:
if isinstance(limit, int):
limit = [limit] * len(queries)
q_emb = self.encoder.encode(queries) # (N, d)
scores, idx = self.index.search(q_emb.astype(np.float32),
k=max(limit) * 2)
return [
dict(list(zip([self.doc_ids[i] for i in idx[qi]], scores[qi].tolist()))[:k])
for qi, k in enumerate(limit)
]
```
```json
{
"file_imports": ["./my_extension.py"],
"services": [{
"name": "my-dense",
"engine": "MyBiencoderEngine",
"config": {"model": "org/name", "index_path": "/path/to/index"},
"batch_size": 16, "max_wait_time": 0.05
}]
}
```
The class **name** in `engine:` must match the Python class name exactly —
that's how lookup works. There is no registration boilerplate.
If your engine can also score query-passage pairs cheaply (true for dense
models — it's just `q @ p.T`), implement `score_batch` on the same class
too and a `/score` endpoint registers automatically. See `Qwen3.score_batch`
for the canonical implementation.
**Building the FAISS index** is out of scope here — produce
`index.faiss` + `index.ids` any way you like and point `index_path` at the
directory.
---
## 4. Wrap a reranker as a scorer
A reranker is an `Engine` that implements `score_batch`. The input shape is
intentionally flat-with-strides and trips everyone up the first time, so
read this section before you write the method.
### The contract
```python
async def score_batch(
self,
queries: List[str], # N queries
passages: List[str], # flattened over ALL queries
candidate_length: List[int] = None, # passages-per-query strides
**kwargs,
) -> List[List[float]]: # one score list per query
```
- `passages` is **not** grouped per query — it's one flat list across the
whole batch. `candidate_length[i]` says how many *consecutive* passages
belong to `queries[i]`. So `sum(candidate_length) == len(passages)` and
`len(candidate_length) == len(queries)`.
- The return value **is** grouped: one list of floats per query, in
passage order, higher = more relevant.
- `candidate_length=None` defaults to `[len(passages)]` (all passages
belong to a single query).
### The canonical implementation skeleton — copy this
```python
# my_reranker.py — ship via file_imports
from typing import List
from routir.models.abstract import Engine
class MyReranker:
"""Sync model wrapper. Knows nothing about RoutIR or async."""
def __init__(self, model_path: str, batch_size: int = 32):
self.model = load_my_model(model_path)
self.batch_size = batch_size
def score(self, pairs):
# pairs: List[Tuple[str, str]] of (query, passage)
# return: List[float] of the same length, higher = more relevant
out = []
for i in range(0, len(pairs), self.batch_size):
out.extend(self.model.predict(pairs[i:i + self.batch_size]))
return out
class MyRerankerEngine(Engine):
"""RoutIR adapter. Implements score_batch ONLY."""
def __init__(self, name=None, config=None, **kwargs):
super().__init__(name, config, **kwargs)
self.model = MyReranker(
model_path=config.get("model", "org/reranker"),
batch_size=config.get("batch_size", 32),
)
async def score_batch(self, queries, passages, candidate_length=None, **kwargs):
if candidate_length is None:
candidate_length = [len(passages)]
assert len(candidate_length) == len(queries)
assert sum(candidate_length) == len(passages)
# 1) expand queries so each passage has its query alongside it
expanded = [queries[i] for i, n in enumerate(candidate_length) for _ in range(n)]
pairs = list(zip(expanded, passages))
# 2) score all pairs at once (sync call inside async is fine for GPU work)
flat = self.model.score(pairs)
# 3) regroup back into per-query lists, in original passage order
out, cursor = [], 0
for n in candidate_length:
out.append(flat[cursor:cursor + n])
cursor += n
return out
```
Config:
```json
{
"file_imports": ["./my_reranker.py"],
"services": [{
"name": "my-reranker",
"engine": "MyRerankerEngine",
"config": {"model": "org/reranker", "batch_size": 32},
"cache": 1024, "cache_ttl": 600,
"batch_size": 16, "max_wait_time": 0.05
}]
}
```
Smoke test:
```bash
curl -X POST http://localhost:5000/score -H 'Content-Type: application/json' \
-d '{"service":"my-reranker","query":"q","passages":["p1","p2","p3"]}'
# -> {"scores": [...], ...}
```
In a pipeline, the reranker is the second stage:
```text
my-bm25%1000 >> my-reranker%20
```
The pipeline runs `my-bm25` to get 1000 candidates, fetches each candidate's
text from the collection's `/content` endpoint, then calls your
`score_batch(queries, flattened_passages, candidate_length)` and keeps the
top 20.
### Gotchas
- **Don't override `score`** (the singular one). The base class implements
it as a one-element call to `score_batch`. Overriding both is a known
legacy footgun.
- **Return floats in the original passage order.** The pipeline pairs your
output back to candidate ids positionally — don't sort inside
`score_batch`.
- **Fallbacks beat crashes.** If a single pair errors, return `0.5` (or any
neutral score) for it rather than blowing up the whole batch.
- **Truncate to your model's context.** Long passages will OOM or get
silently clipped by tokenizers.
- **Bytes/multimodal rerankers** (keyframes, audio): set
`accepts_view_kind = "bytes"` as a class attribute. Then `passages` is
`List[List[bytes]]` (one inner list per doc, possibly empty / multi-blob).
REST `/score` refuses bytes engines — reach them via pipeline DSL or gRPC
`Score`. See `examples/CLAUDE.md` for the full multimodal contract.
- **If candidates aren't supplied by the pipeline** (i.e. you're hitting
`/score` standalone or the engine should run its own first-stage), look at
the `Reranker` base class in `src/routir/models/abstract.py` — it
implements `search_batch` for you via an `upstream_service` +
`text_service`. Most users only need `score_batch` and the pipeline
supplies candidates.
---
## 5. Serve a document collection (single- and multi-view)
Collections power `/content` and feed text/bytes to reranker stages. Every
collection has one or more named **views**; a view picks a backend (JSONL,
tar, local files) and a modality (`text` or `bytes`).
### Single text view (the common case)
Either form works. The "legacy" form is still fully supported and is the
shortest path.
```json
{
"collections": [
{ "name": "my-corpus", "doc_path": "/data/corpus.jsonl",
"id_field": "docid", "content_field": "text" }
]
}
```
Equivalent modern form (use this for anything non-trivial):
```json
{
"collections": [{
"name": "my-corpus",
"views": {
"text": {
"kind": "text",
"source": {
"source": "text_jsonl",
"doc_path": "/data/corpus.jsonl",
"id_field": "docid",
"content_fields": ["title", "body"],
"sep": "\n"
}
}
}
}]
}
```
JSONL access is O(1): RoutIR builds a `.offsetmap` sidecar on first open.
Random access then uses byte offsets.
### Multi-view
```json
{
"collections": [{
"name": "multivent",
"default_view": "asr",
"views": {
"asr": {
"kind": "text",
"source": {"source": "text_jsonl",
"doc_path": "/data/mv.jsonl",
"id_field": "chunk_id",
"content_fields": "asr_transcript"}
},
"keyframe": {
"kind": "bytes",
"source": {
"source": "tar",
"tar_template": "/data/mv/shard_{shard:06d}.tar",
"shard_resolver": {
"kind": "manifest",
"path": "/data/mv/catalog.csv",
"id_column": "chunk_id",
"shard_column": "shard_index"
},
"matcher": {"kind": "glob", "pattern": "{id}.kf_uni5s.t*.jpg"},
"mime": "image/jpeg",
"cache_dir": "/expscratch/me/routir/.cache"
}
}
}
}]
}
```
Pipeline stages pick a view with `@view`:
```text
{asr-search%1000, kf-search%1000}RRF%200 >> kf-reranker@keyframe%20
```
The `kf-reranker` stage gets the `keyframe` view's bytes; `asr-reranker`
(if you added one) would `@asr` and receive strings.
`default_view` is used when a stage omits `@view`. Required to be set
explicitly when there are multiple views; auto-elected when there's only one.
### View backends
| `source:` | Kind | Use for |
| ------------- | ------ | ---------------------------------------------------- |
| `text_jsonl` | text | One JSONL file (optionally gzipped). The default. |
| `local_path` | bytes | One-or-many files per id (`path_template` or `path_glob`). |
| `tar` | bytes | Sharded `.tar` archives with a member-name matcher. |
For tar collections, the **shard resolver** decides which tar holds a given
id: `manifest` (CSV/TSV `id,shard`), `modulo` (hash-based), or `substring`
(slice of the id is the shard token).
### Sidecar caches and warmup — important on shared filesystems
`tar` and `text_jsonl` backends build sidecar indexes
(`.taridx`, `.offsetmap`) on first access. Scanning 50–500 K members in a
tar on a cold request will stall.
- **Always set `cache_dir`** per view when the dataset mount is read-only
(most shared FS). Point at an in-tree `./.cache/` by convention.
- **Pre-build sidecars off the login node** with
`scripts/warmup_slurm.sh <config.json> [view]`. One sbatch job per
*view* — never per shard. See [`CLAUDE.md`](CLAUDE.md#warming-up-sidecar-caches)
for the rule and the exact command.
### Caching `/content`
`CollectionConfig.cache` is 256 entries by default and keys on
`(view, id)` — bytes views especially need this. Crank it up for hot
corpora; set to `0` to disable.
For `server_imports`ed remote collections, enable `relay_content_cache` at
the top level of the config; otherwise every reranked candidate triggers a
remote fetch.
---
## Where to look when this skill isn't enough
- **DSL grammar and execution model**: `src/routir/pipeline/parser.py` and
`src/routir/pipeline/pipeline.py`. The grammar at the top of `parser.py`
is short and authoritative.
- **Engine contract**: `src/routir/models/abstract.py` — the docstrings
on `search_batch`, `score_batch`, `decompose_query_batch`, `fuse_batch`
are the source of truth.
- **Reference engine implementations**: `src/routir/models/` (built-ins)
and `examples/*_extension.py` (file-imports pattern).
- **Config schema**: `src/routir/config/config.py` — every JSON field
the server accepts is a Pydantic field with a docstring.
- **Detailed extension how-to**: [`examples/CLAUDE.md`](examples/CLAUDE.md).
- **Project conventions and warmup**: [`CLAUDE.md`](CLAUDE.md).
