Skip to content

Index

engine

Inference Engine primitive — LLM runtime management.

Classes

EngineConnectionError

Bases: Exception

Raised when an engine is unreachable.

EngineContextLengthError

Bases: EngineConnectionError

The prompt exceeds the served model's maximum context window.

Subclasses EngineConnectionError so existing except EngineConnectionError handlers keep catching it, while callers that want a distinct, user-facing "conversation too long" message can branch on this type (or the is_context_length_error marker) instead of surfacing a generic engine failure.

InferenceEngine

Bases: ABC

Base class for all inference engine backends.

Subclasses must be registered via @EngineRegistry.register("name") to become discoverable.

Methods:
generate abstractmethod 
generate(
    messages: Sequence[Message],
    *,
    model: str,
    temperature: float = 0.7,
    max_tokens: int = 1024,
    **kwargs: Any,
) -> Dict[str, Any]

Synchronous completion — returns a dict with content and usage.

Source code in src/diapason/engine/_stubs.py 
65
66
67
68
69
70
71
72
73
74
75
@abstractmethod
def generate(
    self,
    messages: Sequence[Message],
    *,
    model: str,
    temperature: float = 0.7,
    max_tokens: int = 1024,
    **kwargs: Any,
) -> Dict[str, Any]:
    """Synchronous completion — returns a dict with ``content`` and ``usage``."""
stream abstractmethod async 
stream(
    messages: Sequence[Message],
    *,
    model: str,
    temperature: float = 0.7,
    max_tokens: int = 1024,
    **kwargs: Any,
) -> AsyncIterator[str]

Yield token strings as they are generated.

Source code in src/diapason/engine/_stubs.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
@abstractmethod
async def stream(
    self,
    messages: Sequence[Message],
    *,
    model: str,
    temperature: float = 0.7,
    max_tokens: int = 1024,
    **kwargs: Any,
) -> AsyncIterator[str]:
    """Yield token strings as they are generated."""
    # NOTE: must contain a yield to satisfy the type checker
    yield ""  # pragma: no cover
stream_full async 
stream_full(
    messages: Sequence[Message],
    *,
    model: str,
    temperature: float = 0.7,
    max_tokens: int = 1024,
    **kwargs: Any,
) -> AsyncIterator["StreamChunk"]

Yield full StreamChunks including tool_calls and finish_reason.

Default implementation wraps stream() for backward compatibility. Engines with native tool-call streaming should override this.

Source code in src/diapason/engine/_stubs.py 
 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
async def stream_full(
    self,
    messages: Sequence[Message],
    *,
    model: str,
    temperature: float = 0.7,
    max_tokens: int = 1024,
    **kwargs: Any,
) -> AsyncIterator["StreamChunk"]:
    """Yield full StreamChunks including tool_calls and finish_reason.

    Default implementation wraps ``stream()`` for backward compatibility.
    Engines with native tool-call streaming should override this.
    """
    async with aclosing(
        self.stream(
            messages,
            model=model,
            temperature=temperature,
            max_tokens=max_tokens,
            **kwargs,
        )
    ) as source:
        async for token in source:
            yield StreamChunk(content=token)
    yield StreamChunk(finish_reason="stop")
list_models abstractmethod 
list_models() -> List[str]

Return identifiers of models available on this engine.

Source code in src/diapason/engine/_stubs.py 
118
119
120
@abstractmethod
def list_models(self) -> List[str]:
    """Return identifiers of models available on this engine."""
health abstractmethod 
health() -> bool

Return True when the engine is reachable and healthy.

Source code in src/diapason/engine/_stubs.py 
122
123
124
@abstractmethod
def health(self) -> bool:
    """Return ``True`` when the engine is reachable and healthy."""
can_serve 
can_serve(model: str) -> bool

Return True if this engine can serve model.

Defaults to True: local engines accept any model id (whether a specific model is installed is a separate concern from engine selection). Engines that multiplex provider-specific clients (e.g. the cloud engine) override this so selection can skip an engine whose client for the model's provider isn't configured (see #532).

Source code in src/diapason/engine/_stubs.py 
126
127
128
129
130
131
132
133
134
135
def can_serve(self, model: str) -> bool:
    """Return ``True`` if this engine can serve *model*.

    Defaults to ``True``: local engines accept any model id (whether a
    specific model is *installed* is a separate concern from engine
    selection). Engines that multiplex provider-specific clients (e.g.
    the cloud engine) override this so selection can skip an engine whose
    client for the model's provider isn't configured (see #532).
    """
    return True
close 
close() -> None

Release resources (HTTP clients, connections, threads, etc.).

Source code in src/diapason/engine/_stubs.py 
137
138
def close(self) -> None:
    """Release resources (HTTP clients, connections, threads, etc.)."""
prepare 
prepare(model: str) -> None

Optional warm-up hook called before the first request.

Source code in src/diapason/engine/_stubs.py 
140
141
def prepare(self, model: str) -> None:
    """Optional warm-up hook called before the first request."""

Functions:

looks_like_context_length_error 

looks_like_context_length_error(text: str) -> bool

True when text reads like a context-window overflow error.

The single shared heuristic for recognizing vendor context-overflow phrasings — used by the engine layer (typing upstream 400s), agent error classification, and the server stream bridge, so a new vendor phrasing only ever needs to be added here.

Source code in src/diapason/engine/_base.py 
48
49
50
51
52
53
54
55
56
57
def looks_like_context_length_error(text: str) -> bool:
    """True when *text* reads like a context-window overflow error.

    The single shared heuristic for recognizing vendor context-overflow
    phrasings — used by the engine layer (typing upstream 400s), agent error
    classification, and the server stream bridge, so a new vendor phrasing
    only ever needs to be added here.
    """
    low = (text or "").lower()
    return any(marker in low for marker in CONTEXT_LENGTH_MARKERS)

messages_to_dicts 

messages_to_dicts(
    messages: Sequence[Message],
) -> List[Dict[str, Any]]

Convert Message objects to OpenAI-format dicts.

Source code in src/diapason/engine/_base.py 
 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
def messages_to_dicts(messages: Sequence[Message]) -> List[Dict[str, Any]]:
    """Convert ``Message`` objects to OpenAI-format dicts."""
    out: List[Dict[str, Any]] = []
    for m in messages:
        d: Dict[str, Any] = {"role": m.role.value, "content": m.content}
        if m.name:
            d["name"] = m.name
        if m.tool_calls:
            d["tool_calls"] = [
                {
                    "id": tc.id,
                    "type": "function",
                    "function": {
                        "name": tc.name,
                        "arguments": tc.arguments,
                    },
                }
                for tc in m.tool_calls
            ]
        if m.tool_call_id:
            d["tool_call_id"] = m.tool_call_id
        # Vision: forward base64 images to the engine. Ollama's /api/chat
        # accepts an "images" array on a message; text messages skip this.
        if getattr(m, "images", None):
            d["images"] = list(m.images)
        out.append(d)
    return out

discover_engines 

discover_engines(
    config: DiapasonConfig,
) -> List[Tuple[str, InferenceEngine]]

Probe registered engines and return [(key, instance)] for healthy ones.

Results are sorted with the config default engine first.

Source code in src/diapason/engine/_discovery.py 
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
193
194
195
def discover_engines(config: DiapasonConfig) -> List[Tuple[str, InferenceEngine]]:
    """Probe registered engines and return ``[(key, instance)]`` for healthy ones.

    Results are sorted with the config default engine first.
    """
    _maybe_register_mining_sidecar_engine()

    # Probe engines concurrently: each health() does a blocking network
    # check with its own timeout, so a serial loop costs the SUM of all
    # probe timeouts (dead localhost ports especially). Running them in
    # threads collapses that to roughly the slowest single probe. The
    # healthy.sort() below normalizes order, so completion order is
    # irrelevant and the result is identical to the serial version (#263).
    keys = list(EngineRegistry.keys())

    def _probe(key: str) -> Tuple[str, InferenceEngine] | None:
        try:
            engine = _make_engine(key, config)
            if engine.health():
                return (key, engine)
        except Exception as exc:
            logger.debug("Engine %r failed during discovery: %s", key, exc)
        return None

    healthy: List[Tuple[str, InferenceEngine]] = []
    if keys:
        with ThreadPoolExecutor(max_workers=len(keys)) as pool:
            for result in pool.map(_probe, keys):
                if result is not None:
                    healthy.append(result)

    default_key = config.engine.default

    def sort_key(item: Tuple[str, Any]) -> Tuple[int, str]:
        return (0 if item[0] == default_key else 1, item[0])

    healthy.sort(key=sort_key)
    return healthy

discover_models 

discover_models(
    engines: List[Tuple[str, InferenceEngine]],
) -> Dict[str, List[str]]

Call list_models() on each engine and return a dict.

Source code in src/diapason/engine/_discovery.py 
198
199
200
201
202
203
204
205
206
207
208
209
def discover_models(
    engines: List[Tuple[str, InferenceEngine]],
) -> Dict[str, List[str]]:
    """Call ``list_models()`` on each engine and return a dict."""
    result: Dict[str, List[str]] = {}
    for key, engine in engines:
        try:
            result[key] = engine.list_models()
        except Exception as exc:
            logger.debug("Failed to list models for engine %r: %s", key, exc)
            result[key] = []
    return result

get_engine 

get_engine(
    config: DiapasonConfig,
    engine_key: str | None = None,
    model: str | None = None,
) -> Tuple[str, InferenceEngine] | None

Get a specific engine by key, or the default with fallback.

When model is given, an engine is selected only if it can actually serve that model (engine.can_serve(model)). This stops the cloud fallback from being chosen — when the local engine is down — for a model whose provider client is missing, which otherwise surfaces as a confusing "OpenAI client not available" instead of a helpful "start your local engine" message (see #532). When model is None selection stays model-agnostic (unchanged behaviour).

Returns (key, engine_instance) or None if no engine is available.

Source code in src/diapason/engine/_discovery.py 
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
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 get_engine(
    config: DiapasonConfig,
    engine_key: str | None = None,
    model: str | None = None,
) -> Tuple[str, InferenceEngine] | None:
    """Get a specific engine by key, or the default with fallback.

    When *model* is given, an engine is selected only if it can actually
    serve that model (``engine.can_serve(model)``). This stops the cloud
    fallback from being chosen — when the local engine is down — for a model
    whose provider client is missing, which otherwise surfaces as a confusing
    "OpenAI client not available" instead of a helpful "start your local
    engine" message (see #532). When *model* is ``None`` selection stays
    model-agnostic (unchanged behaviour).

    Returns ``(key, engine_instance)`` or ``None`` if no engine is available.
    """

    def _usable(engine: InferenceEngine) -> bool:
        return engine.health() and (model is None or engine.can_serve(model))

    # Build an ordered list of keys to try, then fall back to full discovery.
    keys_to_try: list[str] = []
    if engine_key:
        keys_to_try.append(engine_key)

    default_key = config.engine.default
    if default_key and default_key not in keys_to_try:
        keys_to_try.append(default_key)

    for key in keys_to_try:
        if not EngineRegistry.contains(key):
            continue
        try:
            engine = _make_engine(key, config)
            if _usable(engine):
                return (key, engine)
        except Exception as exc:
            logger.debug("Engine %r health check failed: %s", key, exc)

    # Fallback to the first healthy engine that can serve the model.
    for key, engine in discover_engines(config):
        if model is None or engine.can_serve(model):
            return (key, engine)
    return None