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/openjarvis/engine/multi.py 
26
27
28
29
def __init__(self, engines: list[tuple[str, InferenceEngine]]) -> None:
    self._engines = engines
    self._model_map: Dict[str, InferenceEngine] = {}
    self._refresh_map()
Functions
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/openjarvis/engine/multi.py 
103
104
105
106
107
108
109
110
111
112
113
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 for chunk in engine.stream_full(messages, model=model, **kwargs):
        yield chunk