oxillama-py 0.1.2

Python bindings for OxiLLaMa LLM inference engine
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

"""Typed StreamingCallback protocol for OxiLLaMa streaming callbacks.

This module provides a runtime-checkable Protocol that type-checkers
(mypy, pyright) can validate against, plus a convenience type alias.
"""

from __future__ import annotations

from typing import Callable

try:
    from typing import Protocol, runtime_checkable
except ImportError:  # Python 3.7 fallback
    from typing_extensions import Protocol, runtime_checkable  # type: ignore[assignment]


@runtime_checkable
class StreamingCallback(Protocol):
    """Protocol for streaming token callbacks.

    Any callable that accepts ``(token: str, token_id: int, is_final: bool)``
    satisfies this protocol.

    Example::

        def my_callback(token: str, token_id: int, is_final: bool) -> None:
            print(token, end="", flush=True)

        assert isinstance(my_callback, StreamingCallback)  # True
    """

    def __call__(self, token: str, token_id: int, is_final: bool) -> None:
        """Invoked for each decoded token.

        Parameters
        ----------
        token:
            The decoded string for this token (may be multiple bytes).
        token_id:
            The integer vocabulary ID of the token.
        is_final:
            ``True`` only for the last token in the sequence (EOS or
            max-tokens reached).
        """
        ...


#: Convenience alias for a bare callable that matches the same signature.
TokenCallback = Callable[[str, int, bool], None]

__all__ = ["StreamingCallback", "TokenCallback"]