llm_router_lib#

Overview#

llm_router_lib bundles Pydantic data‑model definitions and a thin, opinionated client wrapper for the LLM‑Router service.

• The data models live in llm_router_lib/data_models and describe every request payload the router accepts.
• The client (LLMRouterClient) offers a high‑level, Pythonic API that hides HTTP details, retries, and error handling.
• The async client (AsyncLLMRouterClient, httpx‑based) offers the same API for asyncio applications (e.g. FastAPI), plus streaming conversation methods.
• Low‑level service classes (ConversationWithModelService, ExtendedConversationWithModelService, TranslateService, GenerativeAnswerService, health services) perform the actual HTTP calls and can be used directly when finer‑grained control is required.
• HttpRequester (in utils/http.py) is a small wrapper around requests that adds logging, configurable retries, and unified error translation.
• AsyncHttpRequester (in utils/http_async.py) is its httpx‑based asynchronous counterpart with the same retry and error‑translation contract (plus a stream() helper for SSE responses).
• A dedicated exception hierarchy (exceptions.py) maps HTTP errors to meaningful Python exceptions.

In short, llm_router_lib provides both the contract (the “schema”) and a convenient client to consume the router service.

Installation#

The library targets Python 3.10.6 and uses a virtualenv. Install it in editable mode for development:

# Clone the repository (if you haven't already)

git clone https://github.com/radlab-dev-group/llm-router.git
cd llm-router/llm_router_lib

# Create and activate a virtual environment

python3 -m venv .venv
source .venv/bin/activate

# Install the package and its dependencies

pip install -e .

All runtime dependencies (requests, pydantic, plus the packages listed in requirements.txt) are declared in the project’s requirements.txt.

Quick start#

from llm_router_lib import LLMRouterClient

# Initialise the client – point it at the router’s host (do **not** include the `/api` prefix)

client = LLMRouterClient(
    api="http://localhost:8080",  # router host URL
    token="YOUR_ROUTER_TOKEN",  # optional, if router requires auth
)

# Call the standard conversation endpoint with named keyword arguments –
# the client builds the Pydantic request model for you.

response = client.conversation_with_model(
    user_last_statement="Hello, how are you?",
    model="google/gemma-3-12b-it",
    temperature=0.7,
    max_new_tokens=128,
)

# response is a typed `ConversationResponse` model:
print(response.response)  # → the assistant's reply text
print(response.generation_time)  # → seconds taken by the server

You can also pass a ready‑made pydantic request model via the payload keyword (all endpoint parameters are keyword‑only – there are no positional arguments):

from llm_router_lib.data_models.builtin_chat import ConversationWithModelRequest

payload = ConversationWithModelRequest(
    model_name="google/gemma-3-12b-it",
    user_last_statement="Hello, how are you?",
    temperature=0.7,
    max_new_tokens=128,
)

response = client.conversation_with_model(payload=payload)

Note: raw dict payloads are no longer accepted – passing a dict as payload raises TypeError. Build the matching Pydantic model explicitly (e.g. ConversationWithModelRequest(**dict_payload)) or use the named keyword arguments. Calling a generation method with neither payload nor enough named arguments raises NoArgsAndNoPayloadError.

Data models#

All request payloads are defined in llm_router_lib/data_models.
A common base class supplies shared options:

class BaseModelOptions(BaseModel):
    """Options shared across many endpoint models."""
    mask_payload: bool = False
    masker_pipeline: Optional[List[str]] = None

Conversation models#

Utility models (selected examples)#

(All utility models inherit from BaseModelOptions and therefore share the mask_payload and masker_pipeline flags.)

Services (low‑level wrappers)#

If you need direct access to the HTTP layer, the library exposes a set of service classes in llm_router_lib/services:

These services inherit from BaseConversationServiceInterface, which provides call_post and call_get helpers that perform JSON parsing and raise the library‑specific exceptions on failure.

Example: using a service directly#

from llm_router_lib.services.conversation import ConversationWithModelService
from llm_router_lib.utils.http import HttpRequester
import logging

http = HttpRequester(base_url="http://localhost:8080", token="...", timeout=10)
logger = logging.getLogger("demo")

service = ConversationWithModelService(http, logger)
payload = {
    "model_name": "google/gemma-3-12b-it",
    "user_last_statement": "Hi!",
}
response = service.call_post(payload)
print(response)

Thin client wrapper (LLMRouterClient)#

LLMRouterClient aggregates the low‑level services and exposes a concise, high‑level API:

Every endpoint method shares one keyword‑only calling contract: pass a pre‑built Pydantic request model via payload=…, or pass the named domain arguments (plus model and optional temperature / max_new_tokens) and the client builds the request model for you. Raw dict payloads are rejected (TypeError); generation defaults come from the Pydantic models. Calling a generation method with neither payload nor enough named arguments raises NoArgsAndNoPayloadError.

Every method above also accepts payload=<RequestModel> as the alternative to the named arguments (e.g. translate(payload=TranslateModel(...))).

All methods return a typed response model (a Pydantic BaseModel) validated from the JSON body — see RESPONSE_MODELS.md for the full reference and the method→model mapping. Call .model_dump() on the result to get the plain dict back. Errors from the underlying HTTP layer are translated into the following exceptions (defined in exceptions.py):

• LLMRouterError – base class for all library‑specific errors.
• AuthenticationError – HTTP 401/403 (invalid or missing token).
• RateLimitError – HTTP 429 (too many requests).
• ValidationError – HTTP 400 (malformed payload).
• NoArgsAndNoPayloadError – client‑side validation: raised when a generation method is called without payload and without enough named arguments.

:::tip Field naming across models

Models for the built‑in generative endpoints use model_name as the field key (e.g. ConversationWithModelRequest). The OpenAI‑compatible endpoint model uses model instead, to match the official OpenAI API schema:

:::

:::tip Context manager support

All public types (LLMRouterClient, HttpRequester) implement __enter__ / __exit__, so they can be used with the with statement to guarantee resource cleanup:

from llm_router_lib import LLMRouterClient

with LLMRouterClient(api="http://localhost:8080", token="...") as client:
    result = client.conversation_with_model(  # session closed automatically
        user_last_statement="Hi!",
        model="google/gemma-3-12b-it",
    )

:::

Async client (AsyncLLMRouterClient)#

AsyncLLMRouterClient (in async_client.py, backed by httpx) is a drop‑in async alternative to LLMRouterClient for asyncio applications (e.g. FastAPI): the same 14 methods, the same keyword‑only calling contract and the same typed response models, plus streaming support. It shares the retry policy (status codes and exponential back‑off), the bearer‑token handling and the error translation of the synchronous client, and the payload‑building contract (utils/payload.py) is the single source of truth for both clients.

import asyncio
from llm_router_lib import AsyncLLMRouterClient

async def main():
    async with AsyncLLMRouterClient(
        api="http://localhost:8080",  # router host URL
        token="YOUR_ROUTER_TOKEN",
    ) as client:  # client closed automatically
        # same unified contract as the sync client:
        response = await client.conversation_with_model(
            user_last_statement="Hello, how are you?",
            model="google/gemma-3-12b-it",
        )
        print(response.response)

asyncio.run(main())

Streaming#

The two conversation endpoints have streaming variants that send the request with stream: true and yield normalised StreamEvent objects (text – the text delta, raw – the original parsed chunk, done – end of the generation). The router forwards provider streams as‑is, so the client transparently handles both OpenAI‑compatible SSE chunks and Ollama NDJSON chunks:

async def main():
    async with AsyncLLMRouterClient(api="http://localhost:8080", token="...") as client:
        async for event in client.stream_conversation_with_model(
            user_last_statement="Hello!",
            model="google/gemma-3-12b-it",
        ):
            if event.text:
                print(event.text, end="", flush=True)
            if event.done:
                break
        print()

        # or aggregate the whole stream into the final text:
        text = await client.collect_stream_text(
            client.stream_conversation_with_model(
                user_last_statement="Hello!",
                model="google/gemma-3-12b-it",
            )
        )

asyncio.run(main())

Notes:

• stream_timeout (constructor argument, default None) is the read timeout for streamed bodies; None disables it so long generations are not cut off by the short default request timeout.
• Streaming requests are never retried (a partial body would make a blind retry unsafe); unary requests follow the same retry policy as the synchronous client.
• A streamed {"error": ...} chunk raises LLMRouterError.
• httpx is a runtime dependency of the async client (installed with the API requirements – see requirements.txt).

All 14 unary methods of the async client are async equivalents of the sync ones listed above (conversation_with_model, extended_conversation_with_model, the nine utility methods and ping / version / models).

Utilities#

• utils/http.py – HttpRequester
  Handles URL construction, bearer‑token injection, configurable retries (via urllib3.Retry), and unified error mapping. It returns the raw requests.Response after validation.

• utils/http_async.py – AsyncHttpRequester The httpx‑based async counterpart: same URL construction, bearer‑token injection, retry policy and error mapping, plus stream() (an async context manager for SSE responses).

• utils/payload.py – build_payload The shared payload‑building contract (Pydantic model / named arguments / dict rejection) used by both clients.

• utils/stream.py – parse_stream_line / iter_events Normalise raw stream lines (OpenAI SSE, Ollama NDJSON, error chunks, data: [DONE]) into typed StreamEvents.

• exceptions.py – centralised exception definitions (see above).

Development & testing#

The repository includes a small test harness under llm_router_lib/tests. Example usage:

python -m llm_router_lib.tests.llm_router_client

This script spins up a LLMRouterClient instance and runs a suite of end‑to‑end tests covering conversation, extended conversation, translation, generative answering, and health checks.

The asynchronous equivalent (python -m llm_router_lib.tests.async_llm_router_client) exercises AsyncLLMRouterClient against a live router, including a token‑by‑token streaming conversation.

The unit tests (including the async ones – pytest-asyncio) are run with the regular test suite:

pytest llm_router_lib/tests