Skip to content

multi

multi

Multi-engine wrapper — routes requests to the right backend by model name.

Classes

MultiEngine 

MultiEngine(engines: list[tuple[str, InferenceEngine]])

Bases: InferenceEngine

Wraps multiple engines and routes by model name.

Models from each engine are discovered via list_models(). When generate() or stream() is called, the model name is looked up to find which engine owns it.

Source code in src/diapason/engine/multi.py 
27
28
29
30
def __init__(self, engines: list[tuple[str, InferenceEngine]]) -> None:
    self._engines = engines
    self._model_map: Dict[str, InferenceEngine] = {}
    self._refresh_map()
Methods:
stream_full async 
stream_full(
    messages: Sequence[Message],
    *,
    model: str,
    **kwargs: Any,
) -> AsyncIterator["StreamChunk"]

Delegate stream_full() to the engine that owns the model.

Source code in src/diapason/engine/multi.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
async def stream_full(
    self,
    messages: Sequence[Message],
    *,
    model: str,
    **kwargs: Any,
) -> AsyncIterator["StreamChunk"]:
    """Delegate stream_full() to the engine that owns the model."""
    engine = self._engine_for(model)
    async with aclosing(
        engine.stream_full(messages, model=model, **kwargs)
    ) as source:
        async for chunk in source:
            yield chunk