Skip to content

Model clients

varagity.models.registry

Model client factory (spec §12).

The reference implementation's util_model.py had a real bug here: it branched on the builtin type instead of its model_type parameter, so the reasoning/tool branches were dead code. This factory dispatches on the parameter (regression-tested) and validates it.

v1 serves a single LLM, so reasoning and tool are aliases of default — one llama.cpp server hosts one model (spec §21 #9); they become separate servers/models post-v1 without touching callers. v2 adds rerank (spec_v2 §5.4): the infinity-served cross-encoder.

get_model(model_type) 

get_model(
    model_type: Literal["embedding"],
) -> EmbeddingsClient

get_model(model_type: Literal['rerank']) -> RerankClient

get_model(
    model_type: Literal["default", "reasoning", "tool"],
) -> LLMClient

get_model(
    model_type: str,
) -> EmbeddingsClient | LLMClient | RerankClient

Build the client for a model type.

Parameters:

Name Type Description Default
model_type str

One of :data:MODEL_TYPES. embedding resolves to the infinity embeddings client; rerank to the infinity cross-encoder client (spec_v2 §5.4); default — and its v1 aliases reasoning and tool — to the llama.cpp chat client.

 required

Returns:

Type Description
EmbeddingsClient | LLMClient | RerankClient

A configured client for that backend.

Raises:

Type Description
ValueError

If model_type is not a known type (message lists the available ones).
Source code in varagity/models/registry.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def get_model(model_type: str) -> EmbeddingsClient | LLMClient | RerankClient:
    """Build the client for a model type.

    Args:
        model_type: One of :data:`MODEL_TYPES`. ``embedding`` resolves to the
            infinity embeddings client; ``rerank`` to the infinity
            cross-encoder client (spec_v2 §5.4); ``default`` — and its v1
            aliases ``reasoning`` and ``tool`` — to the llama.cpp chat
            client.

    Returns:
        A configured client for that backend.

    Raises:
        ValueError: If ``model_type`` is not a known type (message lists the
            available ones).
    """
    if model_type == "embedding":
        return EmbeddingsClient()
    if model_type == "rerank":
        return RerankClient()
    if model_type in _LLM_TYPES:
        return LLMClient()
    raise ValueError(f"Unknown model_type {model_type!r}. Available: {list(MODEL_TYPES)}")

varagity.models.embeddings

Infinity embeddings client with the two e5-instruct modes.

multilingual-e5-large-instruct formats asymmetrically (spec §9.5, research §2.3), and getting it wrong degrades recall silently:

  • Passages (ingestion) are embedded as raw text — no prefix. The model card is explicit: "No need to add instruction for retrieval documents."
  • Queries are wrapped as Instruct: {task}\nQuery: {query}.

Both modes live here so no caller can format inconsistently.

EmbeddingsClient

Client for the infinity /v1/embeddings endpoint.

Batches requests by EMBEDDING_BATCH_SIZE and retries transient failures with exponential backoff (tenacity owns retries — the SDK's built-in retry is disabled so behavior is single-layered and testable).

Source code in varagity/models/embeddings.py 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
class EmbeddingsClient:
    """Client for the infinity ``/v1/embeddings`` endpoint.

    Batches requests by ``EMBEDDING_BATCH_SIZE`` and retries transient
    failures with exponential backoff (``tenacity`` owns retries — the SDK's
    built-in retry is disabled so behavior is single-layered and testable).
    """

    def __init__(
        self,
        *,
        base_url: str | None = None,
        api_key: str | None = None,
        model: str | None = None,
        batch_size: int | None = None,
    ) -> None:
        """Create the client.

        Args:
            base_url: OpenAI-compatible base URL; defaults to
                ``settings.EMBEDDING_API_URL``.
            api_key: Bearer token; defaults to ``settings.EMBEDDING_API_KEY``.
            model: Served model name; defaults to ``settings.EMBEDDING_MODEL``.
            batch_size: Passages per request; defaults to
                ``settings.EMBEDDING_BATCH_SIZE``.
        """
        settings = get_settings()
        self.model = model or settings.EMBEDDING_MODEL
        self.batch_size = batch_size or settings.EMBEDDING_BATCH_SIZE
        self._client = openai.OpenAI(
            base_url=base_url or settings.EMBEDDING_API_URL,
            api_key=api_key or settings.EMBEDDING_API_KEY,
            max_retries=0,  # tenacity owns retries (see class docstring)
        )

    def embed_passages(self, texts: list[str], verbose: int | None = None) -> list[list[float]]:
        """Embed passages for indexing (e5 **passage mode** — no prefix).

        Args:
            texts: Passage texts, embedded exactly as given.
            verbose: Console verbosity (0–2); defaults to
                ``settings.DEFAULT_VERBOSE``.

        Returns:
            One embedding vector per input text, in input order.

        Raises:
            ValueError: If ``verbose`` is invalid.
            openai.APIError: If a request still fails after retries.
        """
        check_verbose(get_settings().DEFAULT_VERBOSE if verbose is None else verbose)
        for i, text in enumerate(texts):
            n_tokens = count_tokens(text)
            if n_tokens >= TOKEN_WARN_THRESHOLD:
                logger.warning(
                    "passage %d is ~%d tokens (≥%d): e5 truncates at 512 — "
                    "check CHUNK_SIZE / context blurb length",
                    i,
                    n_tokens,
                    TOKEN_WARN_THRESHOLD,
                )
        embeddings: list[list[float]] = []
        for start in range(0, len(texts), self.batch_size):
            embeddings.extend(self._embed_batch(texts[start : start + self.batch_size]))
        return embeddings

    def embed_query(self, query: str, verbose: int | None = None) -> list[float]:
        """Embed a search query (e5 **query mode** — instruction-wrapped).

        Args:
            query: The user's query; wrapped via :func:`format_query` before
                embedding.
            verbose: Console verbosity (0–2); defaults to
                ``settings.DEFAULT_VERBOSE``.

        Returns:
            The query's embedding vector.

        Raises:
            ValueError: If ``verbose`` is invalid.
            openai.APIError: If the request still fails after retries.
        """
        check_verbose(get_settings().DEFAULT_VERBOSE if verbose is None else verbose)
        return self._embed_batch([format_query(query)])[0]

    @retry(
        retry=retry_if_exception_type(_RETRYABLE_ERRORS),
        wait=wait_exponential(multiplier=0.5, max=10),
        stop=stop_after_attempt(4),
        before_sleep=before_sleep_log(logger, logging.WARNING),
        reraise=True,
    )
    def _embed_batch(self, texts: list[str]) -> list[list[float]]:
        """Send one embeddings request, retrying transient failures.

        Args:
            texts: The batch of already-formatted texts.

        Returns:
            One embedding per text, re-ordered to input order via the
            response's ``index`` field.
        """
        response = self._client.embeddings.create(model=self.model, input=texts)
        return [item.embedding for item in sorted(response.data, key=lambda item: item.index)]

__init__(*, base_url=None, api_key=None, model=None, batch_size=None)

Create the client.

Parameters:

Name Type Description Default
base_url str | None

OpenAI-compatible base URL; defaults to settings.EMBEDDING_API_URL.

 None
api_key str | None

Bearer token; defaults to settings.EMBEDDING_API_KEY.

 None
model str | None

Served model name; defaults to settings.EMBEDDING_MODEL.

 None
batch_size int | None

Passages per request; defaults to settings.EMBEDDING_BATCH_SIZE.

 None
Source code in varagity/models/embeddings.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def __init__(
    self,
    *,
    base_url: str | None = None,
    api_key: str | None = None,
    model: str | None = None,
    batch_size: int | None = None,
) -> None:
    """Create the client.

    Args:
        base_url: OpenAI-compatible base URL; defaults to
            ``settings.EMBEDDING_API_URL``.
        api_key: Bearer token; defaults to ``settings.EMBEDDING_API_KEY``.
        model: Served model name; defaults to ``settings.EMBEDDING_MODEL``.
        batch_size: Passages per request; defaults to
            ``settings.EMBEDDING_BATCH_SIZE``.
    """
    settings = get_settings()
    self.model = model or settings.EMBEDDING_MODEL
    self.batch_size = batch_size or settings.EMBEDDING_BATCH_SIZE
    self._client = openai.OpenAI(
        base_url=base_url or settings.EMBEDDING_API_URL,
        api_key=api_key or settings.EMBEDDING_API_KEY,
        max_retries=0,  # tenacity owns retries (see class docstring)
    )

embed_passages(texts, verbose=None)

Embed passages for indexing (e5 passage mode — no prefix).

Parameters:

Name Type Description Default
texts list[str]

Passage texts, embedded exactly as given.

 required
verbose int | None

Console verbosity (0–2); defaults to settings.DEFAULT_VERBOSE.

 None

Returns:

Type Description
list[list[float]]

One embedding vector per input text, in input order.

Raises:

Type Description
ValueError

If verbose is invalid.
APIError

If a request still fails after retries.
Source code in varagity/models/embeddings.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def embed_passages(self, texts: list[str], verbose: int | None = None) -> list[list[float]]:
    """Embed passages for indexing (e5 **passage mode** — no prefix).

    Args:
        texts: Passage texts, embedded exactly as given.
        verbose: Console verbosity (0–2); defaults to
            ``settings.DEFAULT_VERBOSE``.

    Returns:
        One embedding vector per input text, in input order.

    Raises:
        ValueError: If ``verbose`` is invalid.
        openai.APIError: If a request still fails after retries.
    """
    check_verbose(get_settings().DEFAULT_VERBOSE if verbose is None else verbose)
    for i, text in enumerate(texts):
        n_tokens = count_tokens(text)
        if n_tokens >= TOKEN_WARN_THRESHOLD:
            logger.warning(
                "passage %d is ~%d tokens (≥%d): e5 truncates at 512 — "
                "check CHUNK_SIZE / context blurb length",
                i,
                n_tokens,
                TOKEN_WARN_THRESHOLD,
            )
    embeddings: list[list[float]] = []
    for start in range(0, len(texts), self.batch_size):
        embeddings.extend(self._embed_batch(texts[start : start + self.batch_size]))
    return embeddings

embed_query(query, verbose=None)

Embed a search query (e5 query mode — instruction-wrapped).

Parameters:

Name Type Description Default
query str

The user's query; wrapped via :func:format_query before embedding.

 required
verbose int | None

Console verbosity (0–2); defaults to settings.DEFAULT_VERBOSE.

 None

Returns:

Type Description
list[float]

The query's embedding vector.

Raises:

Type Description
ValueError

If verbose is invalid.
APIError

If the request still fails after retries.
Source code in varagity/models/embeddings.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def embed_query(self, query: str, verbose: int | None = None) -> list[float]:
    """Embed a search query (e5 **query mode** — instruction-wrapped).

    Args:
        query: The user's query; wrapped via :func:`format_query` before
            embedding.
        verbose: Console verbosity (0–2); defaults to
            ``settings.DEFAULT_VERBOSE``.

    Returns:
        The query's embedding vector.

    Raises:
        ValueError: If ``verbose`` is invalid.
        openai.APIError: If the request still fails after retries.
    """
    check_verbose(get_settings().DEFAULT_VERBOSE if verbose is None else verbose)
    return self._embed_batch([format_query(query)])[0]

format_query(query)

Wrap a query in the e5-instruct format (query mode).

Parameters:

Name Type Description Default
query str

The user's search query, unformatted.

 required

Returns:

Type Description
str

Instruct: {task}\nQuery: {query} with the model card's default
str

retrieval task.
Source code in varagity/models/embeddings.py 
46
47
48
49
50
51
52
53
54
55
56
def format_query(query: str) -> str:
    r"""Wrap a query in the e5-instruct format (query mode).

    Args:
        query: The user's search query, unformatted.

    Returns:
        ``Instruct: {task}\nQuery: {query}`` with the model card's default
        retrieval task.
    """
    return f"Instruct: {E5_QUERY_TASK}\nQuery: {query}"

varagity.models.rerank

infinity cross-encoder rerank client (spec_v2 §5.4).

infinity's /rerank route is not an OpenAI-SDK method, so this client posts with httpx directly (same tenacity retry posture as the other model clients). The wire contract is infinity's Cohere-protocol schema, verified from its pymodels.py: request {model, query, documents, return_documents, top_n}; response {"results": [{"index": i, "relevance_score": s}, …]} where index is the 0-based position in the input documents list and results arrive sorted by relevance.

Only a served cross-encoder (bge-reranker-v2-m3) can fulfill /rerank — infinity structurally rejects bi-encoders (e5, jina) with a 4xx, which this client surfaces as a clear :class:ValueError instead of a bare HTTP error.

RerankClient

Client for the infinity /rerank endpoint.

Retries transient failures with exponential backoff (tenacity owns retries, mirroring the embeddings/LLM clients). The whole candidate pool goes in one request — batching across the pool is the server's concern (infinity's '32;4' batch cap on the reranker).

Source code in varagity/models/rerank.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
class RerankClient:
    """Client for the infinity ``/rerank`` endpoint.

    Retries transient failures with exponential backoff (``tenacity`` owns
    retries, mirroring the embeddings/LLM clients). The whole candidate
    pool goes in one request — batching across the pool is the server's
    concern (infinity's ``'32;4'`` batch cap on the reranker).
    """

    def __init__(
        self,
        *,
        base_url: str | None = None,
        api_key: str | None = None,
        model: str | None = None,
    ) -> None:
        """Create the client.

        Args:
            base_url: OpenAI-style base URL (the ``/v1`` prefix infinity
                serves under); defaults to ``settings.RERANK_API_URL``.
            api_key: Bearer token; defaults to ``settings.RERANK_API_KEY``.
            model: Served reranker name; defaults to
                ``settings.RERANK_MODEL``. Must be a cross-encoder — the
                server rejects bi-encoders at ``/rerank``.
        """
        settings = get_settings()
        self.model = model or settings.RERANK_MODEL
        self._base_url = (base_url or settings.RERANK_API_URL).rstrip("/")
        self._api_key = api_key or settings.RERANK_API_KEY

    def rerank(
        self,
        query: str,
        documents: list[str],
        *,
        top_n: int | None = None,
        verbose: int | None = None,
    ) -> list[RerankResult]:
        """Score documents against a query with the cross-encoder.

        Args:
            query: The user's query, unformatted (cross-encoders attend
                over the raw pair — no e5-style instruction wrapping).
            documents: Candidate texts to score, best-first order not
                required.
            top_n: Keep only the ``top_n`` most relevant results
                (server-side cut); ``None`` scores and returns every
                document.
            verbose: Console verbosity (0–2); defaults to
                ``settings.DEFAULT_VERBOSE``.

        Returns:
            Relevance results, sorted most-relevant first; each ``index``
            points back into ``documents``. Empty input returns ``[]``
            without a request.

        Raises:
            ValueError: If ``verbose`` is invalid, or the server rejected
                the request permanently (e.g. ``RERANK_MODEL`` is not a
                served cross-encoder).
            httpx.HTTPError: If the request still fails after retries.
        """
        check_verbose(get_settings().DEFAULT_VERBOSE if verbose is None else verbose)
        if not documents:
            return []
        payload: dict[str, Any] = {
            "model": self.model,
            "query": query,
            "documents": documents,
            "return_documents": False,
        }
        if top_n is not None:
            payload["top_n"] = top_n
        try:
            data = self._post(payload)
        except httpx.HTTPStatusError as error:
            if _is_retryable(error):  # transient, retries exhausted — not a misconfig
                raise
            raise ValueError(
                f"rerank request rejected (HTTP {error.response.status_code}) for model "
                f"{self.model!r}: {error.response.text}. Only a served cross-encoder "
                "(e.g. BAAI/bge-reranker-v2-m3) can fulfill /rerank — bi-encoders "
                "(e5, jina) are structurally rejected."
            ) from error
        return [RerankResult.model_validate(item) for item in data["results"]]

    @retry(
        retry=retry_if_exception(_is_retryable),
        wait=wait_exponential(multiplier=0.5, max=10),
        stop=stop_after_attempt(4),
        before_sleep=before_sleep_log(logger, logging.WARNING),
        reraise=True,
    )
    def _post(self, payload: dict[str, Any]) -> dict[str, Any]:
        """Send one rerank request, retrying transient failures.

        Args:
            payload: The full request body.

        Returns:
            The parsed response body.

        Raises:
            httpx.HTTPStatusError: On a non-2xx response (retried for
                5xx/429, then reraised).
            httpx.TransportError: On connection/timeout trouble after
                retries.
        """
        response = httpx.post(
            f"{self._base_url}/rerank",
            json=payload,
            headers={"Authorization": f"Bearer {self._api_key}"},
            timeout=TIMEOUT_SECONDS,
        )
        response.raise_for_status()
        result: dict[str, Any] = response.json()
        return result

__init__(*, base_url=None, api_key=None, model=None)

Create the client.

Parameters:

Name Type Description Default
base_url str | None

OpenAI-style base URL (the /v1 prefix infinity serves under); defaults to settings.RERANK_API_URL.

 None
api_key str | None

Bearer token; defaults to settings.RERANK_API_KEY.

 None
model str | None

Served reranker name; defaults to settings.RERANK_MODEL. Must be a cross-encoder — the server rejects bi-encoders at /rerank.

 None
Source code in varagity/models/rerank.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def __init__(
    self,
    *,
    base_url: str | None = None,
    api_key: str | None = None,
    model: str | None = None,
) -> None:
    """Create the client.

    Args:
        base_url: OpenAI-style base URL (the ``/v1`` prefix infinity
            serves under); defaults to ``settings.RERANK_API_URL``.
        api_key: Bearer token; defaults to ``settings.RERANK_API_KEY``.
        model: Served reranker name; defaults to
            ``settings.RERANK_MODEL``. Must be a cross-encoder — the
            server rejects bi-encoders at ``/rerank``.
    """
    settings = get_settings()
    self.model = model or settings.RERANK_MODEL
    self._base_url = (base_url or settings.RERANK_API_URL).rstrip("/")
    self._api_key = api_key or settings.RERANK_API_KEY

rerank(query, documents, *, top_n=None, verbose=None)

Score documents against a query with the cross-encoder.

Parameters:

Name Type Description Default
query str

The user's query, unformatted (cross-encoders attend over the raw pair — no e5-style instruction wrapping).

 required
documents list[str]

Candidate texts to score, best-first order not required.

 required
top_n int | None

Keep only the top_n most relevant results (server-side cut); None scores and returns every document.

 None
verbose int | None

Console verbosity (0–2); defaults to settings.DEFAULT_VERBOSE.

 None

Returns:

Type Description
list[RerankResult]

Relevance results, sorted most-relevant first; each index
list[RerankResult]

points back into documents. Empty input returns []
list[RerankResult]

without a request.

Raises:

Type Description
ValueError

If verbose is invalid, or the server rejected the request permanently (e.g. RERANK_MODEL is not a served cross-encoder).
HTTPError

If the request still fails after retries.
Source code in varagity/models/rerank.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
def rerank(
    self,
    query: str,
    documents: list[str],
    *,
    top_n: int | None = None,
    verbose: int | None = None,
) -> list[RerankResult]:
    """Score documents against a query with the cross-encoder.

    Args:
        query: The user's query, unformatted (cross-encoders attend
            over the raw pair — no e5-style instruction wrapping).
        documents: Candidate texts to score, best-first order not
            required.
        top_n: Keep only the ``top_n`` most relevant results
            (server-side cut); ``None`` scores and returns every
            document.
        verbose: Console verbosity (0–2); defaults to
            ``settings.DEFAULT_VERBOSE``.

    Returns:
        Relevance results, sorted most-relevant first; each ``index``
        points back into ``documents``. Empty input returns ``[]``
        without a request.

    Raises:
        ValueError: If ``verbose`` is invalid, or the server rejected
            the request permanently (e.g. ``RERANK_MODEL`` is not a
            served cross-encoder).
        httpx.HTTPError: If the request still fails after retries.
    """
    check_verbose(get_settings().DEFAULT_VERBOSE if verbose is None else verbose)
    if not documents:
        return []
    payload: dict[str, Any] = {
        "model": self.model,
        "query": query,
        "documents": documents,
        "return_documents": False,
    }
    if top_n is not None:
        payload["top_n"] = top_n
    try:
        data = self._post(payload)
    except httpx.HTTPStatusError as error:
        if _is_retryable(error):  # transient, retries exhausted — not a misconfig
            raise
        raise ValueError(
            f"rerank request rejected (HTTP {error.response.status_code}) for model "
            f"{self.model!r}: {error.response.text}. Only a served cross-encoder "
            "(e.g. BAAI/bge-reranker-v2-m3) can fulfill /rerank — bi-encoders "
            "(e5, jina) are structurally rejected."
        ) from error
    return [RerankResult.model_validate(item) for item in data["results"]]

RerankResult

Bases: BaseModel

One document's cross-encoder relevance judgment.

Attributes:

Name Type Description
index int

0-based position of the document in the request's documents list (the Cohere protocol — results come back sorted by relevance, so index is the join key back to the input order).
relevance_score float

The cross-encoder's query–document relevance.
Source code in varagity/models/rerank.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
class RerankResult(BaseModel):
    """One document's cross-encoder relevance judgment.

    Attributes:
        index: 0-based position of the document in the request's
            ``documents`` list (the Cohere protocol — results come back
            sorted by relevance, so ``index`` is the join key back to the
            input order).
        relevance_score: The cross-encoder's query–document relevance.
    """

    index: int
    relevance_score: float

varagity.models.llm

llama.cpp chat client and reasoning-model response cleanup (spec §12).

The llama.cpp server speaks the OpenAI /v1 surface, so the openai SDK pointed at BASE_MODEL_API_URL is the client. Responses from reasoning models carry <think>…</think> blocks; callers strip them with :func:clean_response (answers and context blurbs) or, on the streaming path (spec_v2 §4.3), classify them delta-by-delta with :class:varagity.models.stream.ThinkStreamSplitter.

The served model's own tokenizer is reachable too, through llama.cpp's non-OpenAI /tokenize route (:meth:LLMClient.count_served_tokens) — for the callers whose content has to fit the window exactly.

GenerationTimings dataclass

One reading of llama.cpp's timings counters (spec_v3 §7).

llama.cpp extends the OpenAI chunk with a top-level timings object — a sibling of choices, not nested inside usage. It is the server's own cumulative decode accounting, which is why the throughput figure here is the model's and not a guess from chunk arrival times (browser scheduling and network jitter would both taint that).

The field is the capability probe: only llama.cpp emits it, so an absent timings means "this server can't report throughput" and the readout simply never appears. No version sniffing, no config flag.

Attributes:

Name Type Description
predicted_n int

Tokens decoded so far this generation (cumulative).
predicted_ms float

Milliseconds spent decoding them (cumulative).
Source code in varagity/models/llm.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
@dataclass(frozen=True)
class GenerationTimings:
    """One reading of llama.cpp's ``timings`` counters (spec_v3 §7).

    llama.cpp extends the OpenAI chunk with a top-level ``timings`` object —
    a sibling of ``choices``, *not* nested inside ``usage``. It is the
    server's own cumulative decode accounting, which is why the throughput
    figure here is the model's and not a guess from chunk arrival times
    (browser scheduling and network jitter would both taint that).

    The field is the **capability probe**: only llama.cpp emits it, so an
    absent ``timings`` means "this server can't report throughput" and the
    readout simply never appears. No version sniffing, no config flag.

    Attributes:
        predicted_n: Tokens decoded so far this generation (cumulative).
        predicted_ms: Milliseconds spent decoding them (cumulative).
    """

    predicted_n: int
    predicted_ms: float

    @property
    def tokens_per_second(self) -> float | None:
        """Decode throughput, or ``None`` when the counters can't support one.

        Computed from the raw counters rather than read from the server's
        own ``predicted_per_second`` — the same arithmetic llama.cpp's web
        UI does, and it keeps the parse tolerant of builds that omit the
        derived field.

        Returns:
            Tokens per second, or ``None`` if nothing has been decoded yet
            (``predicted_ms`` is 0 on the first chunk, which would divide
            by zero).
        """
        if self.predicted_n < 1 or self.predicted_ms <= 0:
            return None
        return self.predicted_n / self.predicted_ms * 1000.0

tokens_per_second property

Decode throughput, or None when the counters can't support one.

Computed from the raw counters rather than read from the server's own predicted_per_second — the same arithmetic llama.cpp's web UI does, and it keeps the parse tolerant of builds that omit the derived field.

Returns:

Type Description
float | None

Tokens per second, or None if nothing has been decoded yet
float | None

(predicted_ms is 0 on the first chunk, which would divide
float | None

by zero).

LLMClient

Chat-completions client for the llama.cpp /v1 endpoint.

Retries transient failures with exponential backoff (tenacity owns retries — the SDK's built-in retry is disabled so behavior is single-layered and testable). Generation defaults (model, max_tokens, temperature) resolve from settings at construction and can be overridden per call.

Source code in varagity/models/llm.py 
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
class LLMClient:
    """Chat-completions client for the llama.cpp ``/v1`` endpoint.

    Retries transient failures with exponential backoff (``tenacity`` owns
    retries — the SDK's built-in retry is disabled so behavior is
    single-layered and testable). Generation defaults (``model``,
    ``max_tokens``, ``temperature``) resolve from settings at construction
    and can be overridden per call.
    """

    def __init__(
        self,
        *,
        base_url: str | None = None,
        api_key: str | None = None,
        model: str | None = None,
        max_tokens: int | None = None,
        temperature: float | None = None,
    ) -> None:
        """Create the client.

        Args:
            base_url: OpenAI-compatible base URL; defaults to
                ``settings.BASE_MODEL_API_URL``.
            api_key: Bearer token; defaults to ``settings.BASE_MODEL_API_KEY``
                (llama.cpp ignores it, but the SDK requires one).
            model: Model name sent in requests; defaults to
                ``settings.BASE_MODEL`` (llama.cpp serves a single model and
                echoes this field — provenance, not routing).
            max_tokens: Generation cap; defaults to ``settings.MAX_TOKENS``.
            temperature: Sampling temperature; defaults to
                ``settings.LLM_TEMPERATURE``.
        """
        settings = get_settings()
        self.model = model or settings.BASE_MODEL
        self.max_tokens = max_tokens if max_tokens is not None else settings.MAX_TOKENS
        self.temperature = temperature if temperature is not None else settings.LLM_TEMPERATURE
        resolved_base_url = base_url or settings.BASE_MODEL_API_URL
        self._api_key = api_key or settings.BASE_MODEL_API_KEY
        self._client = openai.OpenAI(
            base_url=resolved_base_url,
            api_key=self._api_key,
            max_retries=0,  # tenacity owns retries (see class docstring)
        )
        self._tokenize_url = f"{_server_root(resolved_base_url)}{_TOKENIZE_PATH}"
        # Latched when the server permanently rejects /tokenize (a backend
        # other than llama.cpp): later counts answer None without a request,
        # so one warning stands for the client's lifetime.
        self._tokenize_unsupported = False

    def generate(
        self,
        messages: Sequence[ChatCompletionMessageParam],
        *,
        max_tokens: int | None = None,
        temperature: float | None = None,
        verbose: int | None = None,
    ) -> str:
        """Generate one chat completion.

        The raw response text is returned — reasoning models may include a
        ``<think>…</think>`` stage; callers post-process with
        :func:`clean_response` (spec §10.2, §11.1).

        Args:
            messages: OpenAI-format chat messages.
            max_tokens: Generation cap for this call; defaults to the
                client's ``max_tokens``.
            temperature: Sampling temperature for this call; defaults to the
                client's ``temperature``.
            verbose: Console verbosity (0–2); defaults to
                ``settings.DEFAULT_VERBOSE``.

        Returns:
            The assistant message's text content (empty string if the model
            returned no content).

        Raises:
            ValueError: If ``verbose`` is invalid, or the prompt alone
                overflows ``LLM_CONTEXT_TOKENS``. (The cap is silently
                clamped when *prompt + cap* would overflow — llama.cpp hard-
                fails such requests instead of stopping at the boundary.)
            openai.APIError: If the request still fails after retries.
        """
        check_verbose(get_settings().DEFAULT_VERBOSE if verbose is None else verbose)
        return self._create(
            messages,
            max_tokens=_fit_max_tokens(
                messages, self.max_tokens if max_tokens is None else max_tokens
            ),
            temperature=self.temperature if temperature is None else temperature,
        )

    def generate_stream(
        self,
        messages: Sequence[ChatCompletionMessageParam],
        *,
        max_tokens: int | None = None,
        temperature: float | None = None,
        verbose: int | None = None,
        on_usage: Callable[[CompletionUsage], None] | None = None,
        on_timings: Callable[[GenerationTimings], None] | None = None,
    ) -> Generator[str, None, None]:
        """Generate one chat completion, streamed as raw text deltas.

        The stream's *establishment* (request sent, response headers read) is
        retried like :meth:`generate`; once tokens flow, a mid-stream failure
        surfaces immediately — replaying a half-consumed stream would emit
        duplicate text. Reasoning stages are **not** stripped: ``<think>``
        tags pass through in the deltas for
        :class:`~varagity.models.stream.ThinkStreamSplitter` to classify.
        Servers that extract reasoning into the non-standard
        ``reasoning_content`` delta field instead (llama.cpp under some
        ``--reasoning-format`` settings) are normalized to the same shape:
        those deltas are re-wrapped in synthesized ``<think>…</think>`` tags,
        so downstream sees one contract either way.

        Closing the returned iterator (``.close()``, or abandoning a ``for``
        loop) closes the underlying HTTP stream, which is what aborts a
        llama.cpp generation early — the client-disconnect path (spec_v2
        §4.3 cancellation) relies on it.

        Args:
            messages: OpenAI-format chat messages.
            max_tokens: Generation cap for this call; defaults to the
                client's ``max_tokens``.
            temperature: Sampling temperature for this call; defaults to the
                client's ``temperature``.
            verbose: Console verbosity (0–2); defaults to
                ``settings.DEFAULT_VERBOSE``.
            on_usage: Called with the server-reported token usage when the
                final stream chunk carries it (requested via
                ``stream_options.include_usage``; servers that don't report
                usage simply never trigger it).
            on_timings: Called with llama.cpp's cumulative decode counters
                as each chunk carrying them arrives — many times per
                generation, so callers that surface this are expected to
                throttle. Never fires on a server that doesn't emit
                ``timings``, which is what keeps the throughput readout
                llama.cpp-only (see :class:`GenerationTimings`).

        Returns:
            A generator of raw text deltas, in generation order —
            specifically a *generator*: ``close()`` is part of the contract
            (fakes standing in for this method must return one too).

        Raises:
            ValueError: If ``verbose`` is invalid, or the prompt alone
                overflows ``LLM_CONTEXT_TOKENS`` (see :meth:`generate` — the
                same clamp applies here).
            openai.APIError: If establishing the stream still fails after
                retries, or the stream breaks mid-generation.
        """
        check_verbose(get_settings().DEFAULT_VERBOSE if verbose is None else verbose)
        stream = self._create_stream(
            messages,
            max_tokens=_fit_max_tokens(
                messages, self.max_tokens if max_tokens is None else max_tokens
            ),
            temperature=self.temperature if temperature is None else temperature,
        )
        return _iter_stream_deltas(stream, on_usage, on_timings)

    def count_served_tokens(self, text: str) -> int | None:
        """Count ``text`` with the served model's own tokenizer.

        The cl100k approximation (:mod:`varagity.tokens`) undercounts what
        the served model sees whenever text is digit-dense: this model gives
        every digit its own token where cl100k packs up to three, so a
        timestamp-and-coordinate export measured 1.45× its cl100k estimate.
        llama.cpp exposes its tokenizer at ``/tokenize`` (server root, not
        under ``/v1``); the request is ``{"content": text}`` and the
        response ``{"tokens": [...]}``, special tokens not added. Transient
        failures retry like the chat calls.

        Args:
            text: The text to count.

        Returns:
            The served token count, or ``None`` when the server cannot give
            one: the route is not served (a non-llama.cpp backend — latched,
            so later calls answer ``None`` without a request), the server
            stayed unreachable through the retries, or the body was not the
            expected shape. Each case logs a warning; callers fall back to a
            bound of their own. Never raises for a server-side reason.
        """
        if self._tokenize_unsupported:
            return None
        try:
            body = self._post_tokenize(text)
        except httpx.HTTPStatusError as error:
            status = error.response.status_code
            if _is_retryable_http(error):
                logger.warning(
                    "served tokenizer at %s failed (HTTP %d) after retries",
                    self._tokenize_url,
                    status,
                )
            else:
                self._tokenize_unsupported = True
                logger.warning(
                    "served tokenizer at %s rejected the request (HTTP %d) — not a llama.cpp "
                    "server? Served token counts are off for this client",
                    self._tokenize_url,
                    status,
                )
            return None
        except (httpx.HTTPError, ValueError) as error:  # ValueError: a body that is not JSON
            logger.warning("served tokenizer at %s unreachable: %s", self._tokenize_url, error)
            return None
        tokens = body.get("tokens") if isinstance(body, dict) else None
        if not isinstance(tokens, list):
            logger.warning(
                "served tokenizer at %s answered without a token list", self._tokenize_url
            )
            return None
        return len(tokens)

    @retry(
        retry=retry_if_exception(_is_retryable_http),
        wait=wait_exponential(multiplier=0.5, max=10),
        stop=stop_after_attempt(4),
        before_sleep=before_sleep_log(logger, logging.WARNING),
        reraise=True,
    )
    def _post_tokenize(self, text: str) -> object:
        """Send one ``/tokenize`` request, retrying transient failures.

        Args:
            text: The text to tokenize.

        Returns:
            The parsed response body (shape-checked by the caller).

        Raises:
            httpx.HTTPStatusError: On a non-2xx response (retried for
                5xx/429, then reraised).
            httpx.TransportError: On connection/timeout trouble after
                retries.
            ValueError: If the body is not JSON.
        """
        response = httpx.post(
            self._tokenize_url,
            json={"content": text},
            headers={"Authorization": f"Bearer {self._api_key}"},
            timeout=_TOKENIZE_TIMEOUT_SECONDS,
        )
        response.raise_for_status()
        body: object = response.json()
        return body

    @retry(
        retry=retry_if_exception_type(_RETRYABLE_ERRORS),
        wait=wait_exponential(multiplier=0.5, max=10),
        stop=stop_after_attempt(4),
        before_sleep=before_sleep_log(logger, logging.WARNING),
        reraise=True,
    )
    def _create_stream(
        self,
        messages: Sequence[ChatCompletionMessageParam],
        *,
        max_tokens: int,
        temperature: float,
    ) -> Stream[ChatCompletionChunk]:
        """Open one streaming chat-completions request, retrying establishment.

        Args:
            messages: OpenAI-format chat messages.
            max_tokens: Generation cap for this request.
            temperature: Sampling temperature for this request.

        Returns:
            The established SDK stream (not yet consumed).
        """
        return self._client.chat.completions.create(
            model=self.model,
            messages=messages,
            max_tokens=max_tokens,
            temperature=temperature,
            stream=True,
            stream_options={"include_usage": True},
            # llama.cpp extension: put its `timings` counters on *every*
            # chunk, not just the last one — the live throughput readout.
            # Servers that don't know the field ignore it; the final chunk
            # carries `timings` regardless, so a backend that rejected it
            # outright would still lose only the live half.
            extra_body={"timings_per_token": True},
        )

    @retry(
        retry=retry_if_exception_type(_RETRYABLE_ERRORS),
        wait=wait_exponential(multiplier=0.5, max=10),
        stop=stop_after_attempt(4),
        before_sleep=before_sleep_log(logger, logging.WARNING),
        reraise=True,
    )
    def _create(
        self,
        messages: Sequence[ChatCompletionMessageParam],
        *,
        max_tokens: int,
        temperature: float,
    ) -> str:
        """Send one chat-completions request, retrying transient failures.

        Args:
            messages: OpenAI-format chat messages.
            max_tokens: Generation cap for this request.
            temperature: Sampling temperature for this request.

        Returns:
            The assistant message's text content (``""`` when absent).
        """
        response = self._client.chat.completions.create(
            model=self.model,
            messages=messages,
            max_tokens=max_tokens,
            temperature=temperature,
        )
        return response.choices[0].message.content or ""

__init__(*, base_url=None, api_key=None, model=None, max_tokens=None, temperature=None)

Create the client.

Parameters:

Name Type Description Default
base_url str | None

OpenAI-compatible base URL; defaults to settings.BASE_MODEL_API_URL.

 None
api_key str | None

Bearer token; defaults to settings.BASE_MODEL_API_KEY (llama.cpp ignores it, but the SDK requires one).

 None
model str | None

Model name sent in requests; defaults to settings.BASE_MODEL (llama.cpp serves a single model and echoes this field — provenance, not routing).

 None
max_tokens int | None

Generation cap; defaults to settings.MAX_TOKENS.

 None
temperature float | None

Sampling temperature; defaults to settings.LLM_TEMPERATURE.

 None
Source code in varagity/models/llm.py 
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
def __init__(
    self,
    *,
    base_url: str | None = None,
    api_key: str | None = None,
    model: str | None = None,
    max_tokens: int | None = None,
    temperature: float | None = None,
) -> None:
    """Create the client.

    Args:
        base_url: OpenAI-compatible base URL; defaults to
            ``settings.BASE_MODEL_API_URL``.
        api_key: Bearer token; defaults to ``settings.BASE_MODEL_API_KEY``
            (llama.cpp ignores it, but the SDK requires one).
        model: Model name sent in requests; defaults to
            ``settings.BASE_MODEL`` (llama.cpp serves a single model and
            echoes this field — provenance, not routing).
        max_tokens: Generation cap; defaults to ``settings.MAX_TOKENS``.
        temperature: Sampling temperature; defaults to
            ``settings.LLM_TEMPERATURE``.
    """
    settings = get_settings()
    self.model = model or settings.BASE_MODEL
    self.max_tokens = max_tokens if max_tokens is not None else settings.MAX_TOKENS
    self.temperature = temperature if temperature is not None else settings.LLM_TEMPERATURE
    resolved_base_url = base_url or settings.BASE_MODEL_API_URL
    self._api_key = api_key or settings.BASE_MODEL_API_KEY
    self._client = openai.OpenAI(
        base_url=resolved_base_url,
        api_key=self._api_key,
        max_retries=0,  # tenacity owns retries (see class docstring)
    )
    self._tokenize_url = f"{_server_root(resolved_base_url)}{_TOKENIZE_PATH}"
    # Latched when the server permanently rejects /tokenize (a backend
    # other than llama.cpp): later counts answer None without a request,
    # so one warning stands for the client's lifetime.
    self._tokenize_unsupported = False

count_served_tokens(text)

Count text with the served model's own tokenizer.

The cl100k approximation (:mod:varagity.tokens) undercounts what the served model sees whenever text is digit-dense: this model gives every digit its own token where cl100k packs up to three, so a timestamp-and-coordinate export measured 1.45× its cl100k estimate. llama.cpp exposes its tokenizer at /tokenize (server root, not under /v1); the request is {"content": text} and the response {"tokens": [...]}, special tokens not added. Transient failures retry like the chat calls.

Parameters:

Name Type Description Default
text str

The text to count.

 required

Returns:

Name Type Description
int | None

The served token count, or None when the server cannot give
one int | None

the route is not served (a non-llama.cpp backend — latched,
int | None

so later calls answer None without a request), the server
int | None

stayed unreachable through the retries, or the body was not the
int | None

expected shape. Each case logs a warning; callers fall back to a
int | None

bound of their own. Never raises for a server-side reason.
Source code in varagity/models/llm.py 
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
def count_served_tokens(self, text: str) -> int | None:
    """Count ``text`` with the served model's own tokenizer.

    The cl100k approximation (:mod:`varagity.tokens`) undercounts what
    the served model sees whenever text is digit-dense: this model gives
    every digit its own token where cl100k packs up to three, so a
    timestamp-and-coordinate export measured 1.45× its cl100k estimate.
    llama.cpp exposes its tokenizer at ``/tokenize`` (server root, not
    under ``/v1``); the request is ``{"content": text}`` and the
    response ``{"tokens": [...]}``, special tokens not added. Transient
    failures retry like the chat calls.

    Args:
        text: The text to count.

    Returns:
        The served token count, or ``None`` when the server cannot give
        one: the route is not served (a non-llama.cpp backend — latched,
        so later calls answer ``None`` without a request), the server
        stayed unreachable through the retries, or the body was not the
        expected shape. Each case logs a warning; callers fall back to a
        bound of their own. Never raises for a server-side reason.
    """
    if self._tokenize_unsupported:
        return None
    try:
        body = self._post_tokenize(text)
    except httpx.HTTPStatusError as error:
        status = error.response.status_code
        if _is_retryable_http(error):
            logger.warning(
                "served tokenizer at %s failed (HTTP %d) after retries",
                self._tokenize_url,
                status,
            )
        else:
            self._tokenize_unsupported = True
            logger.warning(
                "served tokenizer at %s rejected the request (HTTP %d) — not a llama.cpp "
                "server? Served token counts are off for this client",
                self._tokenize_url,
                status,
            )
        return None
    except (httpx.HTTPError, ValueError) as error:  # ValueError: a body that is not JSON
        logger.warning("served tokenizer at %s unreachable: %s", self._tokenize_url, error)
        return None
    tokens = body.get("tokens") if isinstance(body, dict) else None
    if not isinstance(tokens, list):
        logger.warning(
            "served tokenizer at %s answered without a token list", self._tokenize_url
        )
        return None
    return len(tokens)

generate(messages, *, max_tokens=None, temperature=None, verbose=None)

Generate one chat completion.

The raw response text is returned — reasoning models may include a <think>…</think> stage; callers post-process with :func:clean_response (spec §10.2, §11.1).

Parameters:

Name Type Description Default
messages Sequence[ChatCompletionMessageParam]

OpenAI-format chat messages.

 required
max_tokens int | None

Generation cap for this call; defaults to the client's max_tokens.

 None
temperature float | None

Sampling temperature for this call; defaults to the client's temperature.

 None
verbose int | None

Console verbosity (0–2); defaults to settings.DEFAULT_VERBOSE.

 None

Returns:

Type Description
str

The assistant message's text content (empty string if the model
str

returned no content).

Raises:

Type Description
ValueError

If verbose is invalid, or the prompt alone overflows LLM_CONTEXT_TOKENS. (The cap is silently clamped when prompt + cap would overflow — llama.cpp hard- fails such requests instead of stopping at the boundary.)
APIError

If the request still fails after retries.
Source code in varagity/models/llm.py 
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
def generate(
    self,
    messages: Sequence[ChatCompletionMessageParam],
    *,
    max_tokens: int | None = None,
    temperature: float | None = None,
    verbose: int | None = None,
) -> str:
    """Generate one chat completion.

    The raw response text is returned — reasoning models may include a
    ``<think>…</think>`` stage; callers post-process with
    :func:`clean_response` (spec §10.2, §11.1).

    Args:
        messages: OpenAI-format chat messages.
        max_tokens: Generation cap for this call; defaults to the
            client's ``max_tokens``.
        temperature: Sampling temperature for this call; defaults to the
            client's ``temperature``.
        verbose: Console verbosity (0–2); defaults to
            ``settings.DEFAULT_VERBOSE``.

    Returns:
        The assistant message's text content (empty string if the model
        returned no content).

    Raises:
        ValueError: If ``verbose`` is invalid, or the prompt alone
            overflows ``LLM_CONTEXT_TOKENS``. (The cap is silently
            clamped when *prompt + cap* would overflow — llama.cpp hard-
            fails such requests instead of stopping at the boundary.)
        openai.APIError: If the request still fails after retries.
    """
    check_verbose(get_settings().DEFAULT_VERBOSE if verbose is None else verbose)
    return self._create(
        messages,
        max_tokens=_fit_max_tokens(
            messages, self.max_tokens if max_tokens is None else max_tokens
        ),
        temperature=self.temperature if temperature is None else temperature,
    )

generate_stream(messages, *, max_tokens=None, temperature=None, verbose=None, on_usage=None, on_timings=None)

Generate one chat completion, streamed as raw text deltas.

The stream's establishment (request sent, response headers read) is retried like :meth:generate; once tokens flow, a mid-stream failure surfaces immediately — replaying a half-consumed stream would emit duplicate text. Reasoning stages are not stripped: <think> tags pass through in the deltas for :class:~varagity.models.stream.ThinkStreamSplitter to classify. Servers that extract reasoning into the non-standard reasoning_content delta field instead (llama.cpp under some --reasoning-format settings) are normalized to the same shape: those deltas are re-wrapped in synthesized <think>…</think> tags, so downstream sees one contract either way.

Closing the returned iterator (.close(), or abandoning a for loop) closes the underlying HTTP stream, which is what aborts a llama.cpp generation early — the client-disconnect path (spec_v2 §4.3 cancellation) relies on it.

Parameters:

Name Type Description Default
messages Sequence[ChatCompletionMessageParam]

OpenAI-format chat messages.

 required
max_tokens int | None

Generation cap for this call; defaults to the client's max_tokens.

 None
temperature float | None

Sampling temperature for this call; defaults to the client's temperature.

 None
verbose int | None

Console verbosity (0–2); defaults to settings.DEFAULT_VERBOSE.

 None
on_usage Callable[[CompletionUsage], None] | None

Called with the server-reported token usage when the final stream chunk carries it (requested via stream_options.include_usage; servers that don't report usage simply never trigger it).

 None
on_timings Callable[[GenerationTimings], None] | None

Called with llama.cpp's cumulative decode counters as each chunk carrying them arrives — many times per generation, so callers that surface this are expected to throttle. Never fires on a server that doesn't emit timings, which is what keeps the throughput readout llama.cpp-only (see :class:GenerationTimings).

 None

Returns:

Type Description
None

A generator of raw text deltas, in generation order —
None

specifically a generator: close() is part of the contract
None

(fakes standing in for this method must return one too).

Raises:

Type Description
ValueError

If verbose is invalid, or the prompt alone overflows LLM_CONTEXT_TOKENS (see :meth:generate — the same clamp applies here).
APIError

If establishing the stream still fails after retries, or the stream breaks mid-generation.
Source code in varagity/models/llm.py 
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
def generate_stream(
    self,
    messages: Sequence[ChatCompletionMessageParam],
    *,
    max_tokens: int | None = None,
    temperature: float | None = None,
    verbose: int | None = None,
    on_usage: Callable[[CompletionUsage], None] | None = None,
    on_timings: Callable[[GenerationTimings], None] | None = None,
) -> Generator[str, None, None]:
    """Generate one chat completion, streamed as raw text deltas.

    The stream's *establishment* (request sent, response headers read) is
    retried like :meth:`generate`; once tokens flow, a mid-stream failure
    surfaces immediately — replaying a half-consumed stream would emit
    duplicate text. Reasoning stages are **not** stripped: ``<think>``
    tags pass through in the deltas for
    :class:`~varagity.models.stream.ThinkStreamSplitter` to classify.
    Servers that extract reasoning into the non-standard
    ``reasoning_content`` delta field instead (llama.cpp under some
    ``--reasoning-format`` settings) are normalized to the same shape:
    those deltas are re-wrapped in synthesized ``<think>…</think>`` tags,
    so downstream sees one contract either way.

    Closing the returned iterator (``.close()``, or abandoning a ``for``
    loop) closes the underlying HTTP stream, which is what aborts a
    llama.cpp generation early — the client-disconnect path (spec_v2
    §4.3 cancellation) relies on it.

    Args:
        messages: OpenAI-format chat messages.
        max_tokens: Generation cap for this call; defaults to the
            client's ``max_tokens``.
        temperature: Sampling temperature for this call; defaults to the
            client's ``temperature``.
        verbose: Console verbosity (0–2); defaults to
            ``settings.DEFAULT_VERBOSE``.
        on_usage: Called with the server-reported token usage when the
            final stream chunk carries it (requested via
            ``stream_options.include_usage``; servers that don't report
            usage simply never trigger it).
        on_timings: Called with llama.cpp's cumulative decode counters
            as each chunk carrying them arrives — many times per
            generation, so callers that surface this are expected to
            throttle. Never fires on a server that doesn't emit
            ``timings``, which is what keeps the throughput readout
            llama.cpp-only (see :class:`GenerationTimings`).

    Returns:
        A generator of raw text deltas, in generation order —
        specifically a *generator*: ``close()`` is part of the contract
        (fakes standing in for this method must return one too).

    Raises:
        ValueError: If ``verbose`` is invalid, or the prompt alone
            overflows ``LLM_CONTEXT_TOKENS`` (see :meth:`generate` — the
            same clamp applies here).
        openai.APIError: If establishing the stream still fails after
            retries, or the stream breaks mid-generation.
    """
    check_verbose(get_settings().DEFAULT_VERBOSE if verbose is None else verbose)
    stream = self._create_stream(
        messages,
        max_tokens=_fit_max_tokens(
            messages, self.max_tokens if max_tokens is None else max_tokens
        ),
        temperature=self.temperature if temperature is None else temperature,
    )
    return _iter_stream_deltas(stream, on_usage, on_timings)

clean_response(text)

Strip reasoning-model <think>…</think> stages from a response.

Carried from the reference implementation and hardened for the shapes a llama.cpp-served reasoning model actually emits:

  • balanced <think>…</think> blocks (however many) are removed;
  • an orphaned </think> (some chat templates consume the opening tag) drops everything up to and including it;
  • an unclosed <think> (hit the token cap mid-reasoning) drops everything from it onward.

Parameters:

Name Type Description Default
text str

The raw LLM response.

 required

Returns:

Type Description
str

The response without reasoning stages, whitespace-stripped.
Source code in varagity/models/llm.py 
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
def clean_response(text: str) -> str:
    """Strip reasoning-model ``<think>…</think>`` stages from a response.

    Carried from the reference implementation and hardened for the shapes a
    llama.cpp-served reasoning model actually emits:

    * balanced ``<think>…</think>`` blocks (however many) are removed;
    * an orphaned ``</think>`` (some chat templates consume the opening tag)
      drops everything up to and including it;
    * an unclosed ``<think>`` (hit the token cap mid-reasoning) drops
      everything from it onward.

    Args:
        text: The raw LLM response.

    Returns:
        The response without reasoning stages, whitespace-stripped.
    """
    cleaned = _THINK_BLOCK_RE.sub("", text)
    cleaned = _UNCLOSED_THINK_RE.sub("", cleaned)
    # Any closer left now is orphaned: the text before it is reasoning.
    _, closer, after = cleaned.partition("</think>")
    if closer:
        cleaned = after
    return cleaned.strip()

varagity.models.stream

Streaming <think> splitter — the token-stream twin of clean_response.

:func:varagity.models.llm.clean_response strips reasoning stages from a complete response; the SSE chat path (spec_v2 §4.3) needs the same classification while tokens are still arriving, so reasoning deltas stream as reasoning events and everything else as token events. :class:ThinkStreamSplitter is that state machine. It handles the shapes clean_response hardens against:

  • balanced <think>…</think> blocks (however many), with tags split across token boundaries;
  • an unclosed <think> (generation hit the token cap mid-reasoning) — everything after the opener classifies as reasoning;
  • an orphaned </think> (some chat templates consume the opening tag).

The orphan case is the one genuinely streaming-hostile shape: the text before the closer was reasoning, but a splitter cannot reclassify deltas it already emitted as answer. It flags the stream instead (:attr:ThinkStreamSplitter.saw_orphan_closer), and callers that persist the answer reconcile by running clean_response over the accumulated raw text — the streaming display is best-effort, the stored answer is exact.

Kind = Literal['reasoning', 'answer'] module-attribute

Classification of one streamed text fragment.

ThinkStreamSplitter

Classify streamed LLM deltas as reasoning or answer text.

Feed raw deltas in arrival order with :meth:feed; each call returns the fragments that became classifiable (tag characters themselves are never emitted). Call :meth:finalize once at end-of-stream to flush any held- back partial tag.

Attributes:

Name Type Description
saw_orphan_closer

True once a </think> arrived with no open <think> block — the already-emitted answer fragments were actually reasoning. Callers needing exact clean_response semantics re-clean the accumulated raw text when set.
Source code in varagity/models/stream.py 
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
class ThinkStreamSplitter:
    """Classify streamed LLM deltas as ``reasoning`` or ``answer`` text.

    Feed raw deltas in arrival order with :meth:`feed`; each call returns the
    fragments that became classifiable (tag characters themselves are never
    emitted). Call :meth:`finalize` once at end-of-stream to flush any held-
    back partial tag.

    Attributes:
        saw_orphan_closer: ``True`` once a ``</think>`` arrived with no open
            ``<think>`` block — the already-emitted ``answer`` fragments were
            actually reasoning. Callers needing exact ``clean_response``
            semantics re-clean the accumulated raw text when set.
    """

    def __init__(self) -> None:
        """Start in the answer state with nothing pending."""
        self._pending = ""
        self._in_reasoning = False
        self.saw_orphan_closer = False

    def feed(self, delta: str) -> list[tuple[Kind, str]]:
        """Consume one raw delta and return the newly classifiable fragments.

        Args:
            delta: The next raw text fragment from the LLM stream (may be
                empty; may split a tag anywhere).

        Returns:
            ``(kind, text)`` tuples in stream order; possibly empty while the
            splitter holds back a potential partial tag.
        """
        self._pending += delta
        events: list[tuple[Kind, str]] = []
        while self._pending:
            if self._in_reasoning:
                if not self._scan_reasoning(events):
                    break
            elif not self._scan_answer(events):
                break
        return events

    def finalize(self) -> list[tuple[Kind, str]]:
        """Flush the hold-back at end-of-stream.

        A pending fragment can no longer become a tag once the stream ends:
        it classifies under the current state (reasoning inside an unclosed
        ``<think>`` block, answer otherwise).

        Returns:
            At most one ``(kind, text)`` tuple.
        """
        if not self._pending:
            return []
        kind: Kind = "reasoning" if self._in_reasoning else "answer"
        events: list[tuple[Kind, str]] = [(kind, self._pending)]
        self._pending = ""
        return events

    def _scan_reasoning(self, events: list[tuple[Kind, str]]) -> bool:
        """Consume pending text inside a ``<think>`` block.

        Only ``</think>`` is meaningful here — a nested ``<think>`` is inert
        reasoning text, matching ``clean_response``'s non-greedy block regex.

        Args:
            events: Output accumulator to append classified fragments to.

        Returns:
            ``True`` if a state transition consumed a tag (scan again);
            ``False`` when the pending text is exhausted for this call.
        """
        index = self._pending.find(_CLOSE_TAG)
        if index != -1:
            if index:
                events.append(("reasoning", self._pending[:index]))
            self._pending = self._pending[index + len(_CLOSE_TAG) :]
            self._in_reasoning = False
            return True
        emit_up_to = len(self._pending) - _partial_tag_suffix(self._pending, (_CLOSE_TAG,))
        if emit_up_to:
            events.append(("reasoning", self._pending[:emit_up_to]))
            self._pending = self._pending[emit_up_to:]
        return False

    def _scan_answer(self, events: list[tuple[Kind, str]]) -> bool:
        """Consume pending text outside any ``<think>`` block.

        Both tags are meaningful here: ``<think>`` opens a reasoning block;
        an orphaned ``</think>`` is dropped and flagged (see the module
        docstring on why it cannot be reclassified retroactively).

        Args:
            events: Output accumulator to append classified fragments to.

        Returns:
            ``True`` if a state transition consumed a tag (scan again);
            ``False`` when the pending text is exhausted for this call.
        """
        matches = [
            (index, tag)
            for tag in (_OPEN_TAG, _CLOSE_TAG)
            if (index := self._pending.find(tag)) != -1
        ]
        if matches:
            index, tag = min(matches)
            if index:
                events.append(("answer", self._pending[:index]))
            self._pending = self._pending[index + len(tag) :]
            if tag == _OPEN_TAG:
                self._in_reasoning = True
            else:
                self.saw_orphan_closer = True
            return True
        emit_up_to = len(self._pending) - _partial_tag_suffix(
            self._pending, (_OPEN_TAG, _CLOSE_TAG)
        )
        if emit_up_to:
            events.append(("answer", self._pending[:emit_up_to]))
            self._pending = self._pending[emit_up_to:]
        return False

__init__()

Start in the answer state with nothing pending.

Source code in varagity/models/stream.py 
71
72
73
74
75
def __init__(self) -> None:
    """Start in the answer state with nothing pending."""
    self._pending = ""
    self._in_reasoning = False
    self.saw_orphan_closer = False

feed(delta)

Consume one raw delta and return the newly classifiable fragments.

Parameters:

Name Type Description Default
delta str

The next raw text fragment from the LLM stream (may be empty; may split a tag anywhere).

 required

Returns:

Type Description
list[tuple[Kind, str]]

(kind, text) tuples in stream order; possibly empty while the
list[tuple[Kind, str]]

splitter holds back a potential partial tag.
Source code in varagity/models/stream.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def feed(self, delta: str) -> list[tuple[Kind, str]]:
    """Consume one raw delta and return the newly classifiable fragments.

    Args:
        delta: The next raw text fragment from the LLM stream (may be
            empty; may split a tag anywhere).

    Returns:
        ``(kind, text)`` tuples in stream order; possibly empty while the
        splitter holds back a potential partial tag.
    """
    self._pending += delta
    events: list[tuple[Kind, str]] = []
    while self._pending:
        if self._in_reasoning:
            if not self._scan_reasoning(events):
                break
        elif not self._scan_answer(events):
            break
    return events

finalize()

Flush the hold-back at end-of-stream.

A pending fragment can no longer become a tag once the stream ends: it classifies under the current state (reasoning inside an unclosed <think> block, answer otherwise).

Returns:

Type Description
list[tuple[Kind, str]]

At most one (kind, text) tuple.
Source code in varagity/models/stream.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def finalize(self) -> list[tuple[Kind, str]]:
    """Flush the hold-back at end-of-stream.

    A pending fragment can no longer become a tag once the stream ends:
    it classifies under the current state (reasoning inside an unclosed
    ``<think>`` block, answer otherwise).

    Returns:
        At most one ``(kind, text)`` tuple.
    """
    if not self._pending:
        return []
    kind: Kind = "reasoning" if self._in_reasoning else "answer"
    events: list[tuple[Kind, str]] = [(kind, self._pending)]
    self._pending = ""
    return events