Skip to content

Index

engine

Inference Engine primitive — LLM runtime management.

Classes

EngineConnectionError

Bases: Exception

Raised when an engine is unreachable.

InferenceEngine

Bases: ABC

Base class for all inference engine backends.

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

Functions
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/openjarvis/engine/_stubs.py 
42
43
44
45
46
47
48
49
50
51
52
@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/openjarvis/engine/_stubs.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
@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
list_models abstractmethod 
list_models() -> List[str]

Return identifiers of models available on this engine.

Source code in src/openjarvis/engine/_stubs.py 
68
69
70
@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/openjarvis/engine/_stubs.py 
72
73
74
@abstractmethod
def health(self) -> bool:
    """Return ``True`` when the engine is reachable and healthy."""
close 
close() -> None

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

Source code in src/openjarvis/engine/_stubs.py 
76
77
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/openjarvis/engine/_stubs.py 
79
80
def prepare(self, model: str) -> None:
    """Optional warm-up hook called before the first request."""

Functions

messages_to_dicts 

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

Convert Message objects to OpenAI-format dicts.

Source code in src/openjarvis/engine/_base.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
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
        out.append(d)
    return out

discover_engines 

discover_engines(config: JarvisConfig) -> 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/openjarvis/engine/_discovery.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def discover_engines(config: JarvisConfig) -> List[Tuple[str, InferenceEngine]]:
    """Probe registered engines and return ``[(key, instance)]`` for healthy ones.

    Results are sorted with the config default engine first.
    """
    healthy: List[Tuple[str, InferenceEngine]] = []
    for key in EngineRegistry.keys():
        try:
            engine = _make_engine(key, config)
            if engine.health():
                healthy.append((key, engine))
        except Exception as exc:
            logger.debug("Engine %r failed during discovery: %s", key, exc)
            continue

    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/openjarvis/engine/_discovery.py 
66
67
68
69
70
71
72
73
74
75
76
77
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: JarvisConfig, engine_key: str | None = None) -> Tuple[str, InferenceEngine] | None

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

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

Source code in src/openjarvis/engine/_discovery.py 
 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
def get_engine(
    config: JarvisConfig, engine_key: str | None = None
) -> Tuple[str, InferenceEngine] | None:
    """Get a specific engine by key, or the default with fallback.

    Returns ``(key, engine_instance)`` or ``None`` if no engine is available.
    """
    if engine_key:
        if EngineRegistry.contains(engine_key):
            try:
                engine = _make_engine(engine_key, config)
                if engine.health():
                    return (engine_key, engine)
            except Exception as exc:
                logger.debug("Engine %r health check failed: %s", engine_key, exc)
        return None

    # Try default first
    default_key = config.engine.default
    if EngineRegistry.contains(default_key):
        try:
            engine = _make_engine(default_key, config)
            if engine.health():
                return (default_key, engine)
        except Exception as exc:
            logger.debug("Default engine %r health check failed: %s", default_key, exc)

    # Fallback to any healthy engine
    healthy = discover_engines(config)
    return healthy[0] if healthy else None