oxillama-py 0.1.3

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
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83

"""Jupyter / tqdm-compatible streaming callback helpers for OxiLLaMa."""
from __future__ import annotations

from typing import Any, List

__all__ = ["TqdmProgress", "CollectTokens"]


class TqdmProgress:
    """Stream tokens into a tqdm progress bar or any tqdm-like widget.

    Works in standard terminals (tqdm) and Jupyter notebooks
    (tqdm.notebook.tqdm / ipywidgets).  The progress bar shows the count
    of decoded tokens; the postfix shows the most-recently decoded piece.

    Example::

        from tqdm.auto import tqdm
        from oxillama_py.tqdm_helper import TqdmProgress

        bar = tqdm(desc="Generating", unit="tok")
        cb = TqdmProgress(bar)
        engine.generate_streaming(prompt, callback=cb)
        bar.close()
    """

    def __init__(self, pbar: Any, flush_on_newline: bool = True) -> None:
        """Initialise with an already-open tqdm instance.

        Args:
            pbar: An open ``tqdm`` instance (any variant: auto, notebook, …).
            flush_on_newline: If *True*, call ``pbar.refresh()`` whenever a
                newline character is encountered in the token stream so that
                Jupyter output repagination is triggered promptly.
        """
        self._pbar = pbar
        self._flush_on_newline = flush_on_newline
        self._tokens: List[str] = []

    def __call__(self, token: str) -> None:
        """Token-callback compatible with ``StreamingCallback`` protocol."""
        self._tokens.append(token)
        self._pbar.update(1)
        self._pbar.set_postfix_str(repr(token)[:30], refresh=False)
        if self._flush_on_newline and "\n" in token:
            self._pbar.refresh()

    @property
    def text(self) -> str:
        """The full decoded text seen so far."""
        return "".join(self._tokens)

    def reset(self) -> None:
        """Clear the accumulated token list and reset the progress bar counter."""
        self._tokens.clear()
        self._pbar.reset()


class CollectTokens:
    """Simple token collector — no progress display, just accumulate text.

    Useful when you want the full text after generation but still want a
    callback-compatible object.

    Example::

        col = CollectTokens()
        engine.generate_streaming(prompt, callback=col)
        print(col.text)
    """

    def __init__(self) -> None:
        self._tokens: List[str] = []

    def __call__(self, token: str) -> None:
        self._tokens.append(token)

    @property
    def text(self) -> str:
        return "".join(self._tokens)

    def reset(self) -> None:
        self._tokens.clear()