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:

The URL is: https://llm-test.stepping-stone.ch/v1/chat/completions

• apertus-ai/Apertus-v1.5-8B
• deepseek-ai/DeepSeek-V4-Flash-0731
• MiniMaxAI/MiniMax-M2.5
• NVIDIA/NVIDIA-Nemotron-3-Super-120B-A12B
• Qwen/Qwen3-Coder-Next
• Qwen/Qwen3.5-35B-A3B-FP8
• zai-org/GLM-5.3-Flash

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 or parsing.

#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
latest_usage = None
done = False

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

    payload = line[5:].strip()
    if payload == "[DONE]":
        done = 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):
        latest_usage = usage

if latest_usage is not None:
    label = "Usage" if done else "Last reported usage (stream incomplete)"
    print("\n\n{}:\n{}".format(
        label, json.dumps(latest_usage, indent=2)
    ), flush=True)

print(
    "\n[Stream complete]" if done else "\n[Stream ended without DONE]",
    flush=True,
)
'