---

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; pipeline / DSL / config reference in the project 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.

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. Copy it when scripting evaluations.

---

2. Stand up a local server (and import a master server)

Minimal my-config.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:

{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:

# 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:

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:

{
    "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:

# 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)
        ]

{
    "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 na