# Direct Transformers Inference

These examples use this repository's Transformers 5.12.1-compatible implementation, not the browser Demo or its WebSocket protocol. Prepare the [compatible backend environment](https://github.com/fnlp-vision/sglang-omni-realtime/blob/main/docs/get_started/installation.md); a running SGLang server is not required for direct Python inference. Do not use the Demo CPU environment to load this checkpoint.

Download the current complete model repository. Sample paths below refer to media supplied by the caller. Runtime requirements are listed in the [Demo README](https://github.com/fnlp-vision/MOSS-VL-Realtime_Demo#compatibility-and-updates).

## Load the Model

```python
import torch
from transformers import AutoModelForCausalLM, AutoProcessor
from huggingface_hub import snapshot_download

checkpoint = snapshot_download("OpenMOSS-Team/MOSS-VL-Realtime-SGLANG")

processor = AutoProcessor.from_pretrained(
    checkpoint,
    trust_remote_code=True,
    frame_extract_num_threads=1,
)
model = AutoModelForCausalLM.from_pretrained(
    checkpoint,
    trust_remote_code=True,
    device_map="auto",
    dtype=torch.bfloat16,
    attn_implementation="eager",
)
model.eval()
```

This direct-Transformers example uses eager attention. The SGLang-Omni backend has its own attention configuration; do not confuse the two execution paths.


## Inference Examples

### Online Inference

<details>
<summary><b>Session-style Online Inference</b></summary>

The recommended direct API is `create_realtime_session(...)`. A service or application owns the video capture pipeline, converts camera, screen, or video-file input into PIL-compatible frames, and pushes each frame with a non-decreasing timestamp.

Common session operations:

- `session.push_frame(image, timestamp=...)` appends one visual frame.
- `session.push_prompt("...")` appends a user question while the stream is running.
- `session.push_prompt_frame(prompt, image, timestamp=...)` aligns a prompt with a specific frame.
- `session.poll_output(...)` or `session.stream_outputs(...)` returns incremental text chunks.

`system_prompt` and `initial_prompt` are tokenized as the initial system/user turns before the first frame arrives. Subsequent user turns can be appended with `push_prompt(...)` while the same session continues observing frames.

For complete real-time inference usage, including local-video replay and service deployment, see [`realtime_inference`](https://github.com/OpenMOSS/MOSS-VL/tree/main/realtime_inference) in the MOSS-VL GitHub repository.

```python
import time
from PIL import Image

session = model.create_realtime_session(
    processor,
    initial_prompt=(
        "As the video streams frame by frame, describe important changes as they happen. "
        "Stay silent when there is no relevant update."
    ),
    frame_queue_size=256,
    max_tokens_per_turn=12,
    max_new_tokens=4096,
    do_sample=False,
)

frame_paths = [
    "data/frame_0001.jpg",
    "data/frame_0002.jpg",
    "data/frame_0003.jpg",
]

try:
    session.start()

    for index, frame_path in enumerate(frame_paths):
        image = Image.open(frame_path).convert("RGB")
        session.push_frame(image, timestamp=index / 1.0)

        while True:
            chunk = session.poll_output(timeout=0.0)
            if chunk is None:
                break
            print(chunk, end="", flush=True)

        time.sleep(1.0)

    session.push_prompt("What changed in the latest frames?")

    # Realtime sessions stay alive waiting for future input, so use a bounded
    # drain window and close the session explicitly when the producer is done.
    drain_deadline = time.monotonic() + 5.0
    while time.monotonic() < drain_deadline:
        chunk = session.poll_output(timeout=0.1)
        if chunk is not None:
            print(chunk, end="", flush=True)
finally:
    session.close()
```

Frame timestamps are measured in seconds and must be non-decreasing within a session. The input producer can be a camera, screen capture, decoded video file, browser frame sampler, or any other source that yields images with timestamps.

</details>

<details>
<summary><b>Queue-style Online Inference</b></summary>

`online_generate(...)` is useful for backend systems that separate frame production and model inference through queues. It accepts dictionaries containing frames, prompts, events, reset controls, and stop controls.

```python
import queue
import threading
import time
from PIL import Image

input_queue = queue.Queue()
output_queue = queue.Queue()

worker = threading.Thread(
    target=model.online_generate,
    args=(processor, input_queue, output_queue),
    kwargs={
        "frame_queue_size": 256,
        "max_tokens_per_turn": 12,
        "max_new_tokens": 4096,
        "do_sample": False,
    },
    daemon=True,
)
worker.start()

input_queue.put({
    "initial_prompt": "Answer only when the streamed video provides enough evidence.",
})

input_queue.put({"frame": Image.open("data/frame_0001.jpg").convert("RGB"), "timestamp": 0.0})
input_queue.put({"frame": Image.open("data/frame_0002.jpg").convert("RGB"), "timestamp": 1.0})
input_queue.put({"prompt": "What is happening now?"})

# A quiet queue is not completion. This finite demo observes for up to 60 s;
# a live application uses its producer/session lifecycle to decide when to stop.
try:
    deadline = time.monotonic() + 60.0
    while time.monotonic() < deadline:
        try:
            chunk = output_queue.get(timeout=min(0.5, max(0.001, deadline - time.monotonic())))
        except queue.Empty:
            if not worker.is_alive():
                break
            continue
        print(chunk, end="", flush=True)
finally:
    input_queue.put({"stop_online_generate": True})
    worker.join(timeout=10.0)
    if worker.is_alive():
        raise RuntimeError("Inference worker did not stop within the shutdown timeout")
```

Each queue item can contain `frame` or `image`, `timestamp`, `prompt`, `frames`, `event`, `events`, `initial_prompt`, `system_prompt`, `generate_kwargs`, `reset_session`, or stop controls such as `stop_online_generate`.

</details>

### Offline Inference

MOSS-VL-Realtime also keeps the offline helper APIs for image and video prompts. For purely offline use, MOSS-VL-Instruct is usually the preferred checkpoint, but the realtime checkpoint can still process complete image and video inputs.

<details>
<summary><b>Single-video Offline Inference</b></summary>

```python
video_path = "data/example_video.mp4"
prompt = "Describe this video."

text = model.offline_video_generate(
    processor,
    prompt=prompt,
    video=video_path,
    shortest_edge=4096,
    longest_edge=16777216,
    video_max_pixels=201326592,
    patch_size=16,
    temporal_patch_size=1,
    merge_size=2,
    video_fps=1.0,
    min_frames=1,
    max_frames=256,
    num_extract_threads=4,
    image_mean=[0.5, 0.5, 0.5],
    image_std=[0.5, 0.5, 0.5],
    max_new_tokens=256,
    temperature=1.0,
    top_k=50,
    top_p=1.0,
    repetition_penalty=1.0,
    do_sample=False,
    vision_chunked_length=64,
)

print(text)
```

</details>

<details>
<summary><b>Batched Offline Inference</b></summary>

`offline_batch_generate` accepts independent image/video/text queries. Queries in the same batch should share the same `media_kwargs` and `generate_kwargs`.

```python
queries = [
    {
        "prompt": "Describe sample A.",
        "images": [],
        "videos": ["data/sample_a.mp4"],
        "media_kwargs": {
            "video_fps": 1.0,
            "min_frames": 8,
            "max_frames": 256,
        },
        "generate_kwargs": {
            "temperature": 1.0,
            "top_k": 50,
            "top_p": 1.0,
            "max_new_tokens": 256,
            "repetition_penalty": 1.0,
            "do_sample": False,
        },
    },
    {
        "prompt": "Describe sample B.",
        "images": [],
        "videos": ["data/sample_b.mp4"],
        "media_kwargs": {
            "video_fps": 1.0,
            "min_frames": 8,
            "max_frames": 256,
        },
        "generate_kwargs": {
            "temperature": 1.0,
            "top_k": 50,
            "top_p": 1.0,
            "max_new_tokens": 256,
            "repetition_penalty": 1.0,
            "do_sample": False,
        },
    },
]

with torch.no_grad():
    result = model.offline_batch_generate(
        processor,
        queries,
        vision_chunked_length=64,
    )

texts = [item["text"] for item in result["results"]]
print(texts)
```

</details>