AI on Demand: Test streaming

Overview

This pages describes some useful commands on how to test token streaming.

Usage

We assume you've received your API key from us in our usual manner.

For testing purposes the model selection for general purpose models is:

• apertus-ai/Apertus-v1.5-8B
• AI_on_demand:_MiniMaxAI/MiniMax-M2.5 MiniMaxAI/MiniMax-M2.5
• NVIDIA/NVIDIA-Nemotron-3-Super-120B-A12B
• Qwen/Qwen3-Coder-Next
• Qwen/Qwen3.5-35B-A3B-FP8

Usage - Inference with token-stream

Regular token streaming is difficult to read.

The expanded command processes the output into a far more digestible format on the console.

Initial tests confirm that structured responses remain unaffected by this parser, as seen in code generation examples.

Chat platforms like LibreChat include native mechanisms for handling token streaming.

#Modell auswählen
MODEL='NVIDIA/NVIDIA-Nemotron-3-Super-120B-A12B'

#Prompt eingeben
PROMPT='Hallo, bitte führe meine Prompt aus'

#Tokenlimit setzen
MAX_TOKENS=2000

#Inference Call
jq -n \
  --arg model "$MODEL" \
  --arg prompt "$PROMPT" \
  --argjson max_tokens "$MAX_TOKENS" '{
    model: $model,
    messages: [
      {
        role: "user",
        content: $prompt
      }
    ],
    stream: true,
    stream_options: {include_usage: true},
    max_tokens: $max_tokens
  }' |
curl --fail-with-body --no-buffer -sS \
  --connect-timeout 5 \
  --max-time 180 \
  https://llm-test.stepping-stone.ch/v1/chat/completions \
  -H "apikey: $STONEY_KEY" \
  -H 'Content-Type: application/json' \
  --data-binary @- |
tee /tmp/kong-model-stream.sse |
python3 -u -c '
import json
import sys

section = None

for line in sys.stdin:
    if not line.startswith("data:"):
        continue

    payload = line[5:].strip()
    if payload == "[DONE]":
        print("\n[Stream complete]", flush=True)
        continue

    try:
        event = json.loads(payload)
    except json.JSONDecodeError:
        continue

    if not isinstance(event, dict):
        continue

    if "error" in event:
        print("\nAPI error:", event["error"], flush=True)

    for choice in event.get("choices") or []:
        delta = choice.get("delta") or {}
        reasoning = delta.get("reasoning") or delta.get("reasoning_content")

        for label, text in (
            ("Reasoning", reasoning),
            ("Answer", delta.get("content")),
        ):
            if isinstance(text, str) and text:
                if section != label:
                    print("\n\n--- {} ---".format(label), flush=True)
                    section = label
                print(text, end="", flush=True)

        if choice.get("finish_reason"):
            print(
                "\n[Finish reason: {}]".format(choice["finish_reason"]),
                flush=True,
            )

    usage = event.get("usage")
    if isinstance(usage, dict):
        print("\n\nUsage:\n" + json.dumps(usage, indent=2), flush=True)
'