Streamed answers

// stream.mjs: answers questions with a language model on your own server and streams each answer,
// as it is written, to the member who asked: to every tab and device of that user. A member can
// stop an answer, and a page that lost its connection gets what it missed when it is back. Any
// server with an OpenAI-compatible chat API works: Ollama, llama.cpp, vLLM. We use Gemma 4 with
// Ollama. Node 22+. npm install, then npm start (reads .env).
//
// A connection may send 10 client events a second, so the text goes out in pieces: every 125 ms
// the answer that has waited longest gets one event with all it has written since.
//
// Its key (dashboard, New API key): rights subscribe and client_events, kind presence,
// channels: prefix presence-ask-.
import { Pikoman, TokenSigner } from "@pikoman/ws";

const ROOMS = (process.env.ROOMS || "presence-ask-lobby").split(",");
const AI_URL = process.env.AI_URL || "http://127.0.0.1:11434/v1"; // Ollama's OpenAI-compatible API
const AI_MODEL = process.env.AI_MODEL || "gemma4:12b";
const AT_ONCE = 2; // answers the model writes at the same time
const SEND_EVERY = 125; // ms between two events from the bot: 8 a second, under the limit of 10
const MAX_PIECE = 6000; // characters in one event, well under the 12 KiB an event may hold
const KEEP_FOR = 10 * 60_000; // a finished answer is kept this long, for pages that come back

const pikoman = new Pikoman(process.env.PIKOMAN_APP, {
  ws_address: process.env.PIKOMAN_WS || "wss://next.pikoman.se",
});
const auth = TokenSigner.fromSecret(process.env.PIKOMAN_API_KEY, { name: "Assistant" }, 3);

/**
 * Every answer, by `<room>/<id>`: who asked, the text so far, how much of it has gone out, and
 * whether it is done. `pieces` holds where each event ended, so a page that comes back after
 * piece 4 gets everything after it.
 */
const answers = new Map();
const waiting = []; // answers for the model, oldest first
let writing = 0;

/** Streams one answer from the model, adding to `answer.text` as it comes. */
async function write(answer) {
  const response = await fetch(`${AI_URL}/chat/completions`, {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      ...(process.env.AI_KEY ? { Authorization: `Bearer ${process.env.AI_KEY}` } : {}),
    },
    body: JSON.stringify({
      model: AI_MODEL,
      stream: true,
      messages: [
        { role: "system", content: "Answer the user's question clearly and briefly." },
        { role: "user", content: answer.question },
      ],
    }),
    signal: answer.stop.signal,
  });
  if (!response.ok) throw new Error(`${AI_URL}: ${response.status}`);
  // Server-sent events: lines of "data: {json}", and "data: [DONE]" at the end.
  const decoder = new TextDecoder();
  let buffer = "";
  for await (const chunk of response.body) {
    buffer += decoder.decode(chunk, { stream: true });
    let end;
    while ((end = buffer.indexOf("\n")) >= 0) {
      const line = buffer.slice(0, end).trim();
      buffer = buffer.slice(end + 1);
      if (!line.startsWith("data:")) continue;
      const data = line.slice(5).trim();
      if (data === "[DONE]") return;
      answer.text += JSON.parse(data).choices?.[0]?.delta?.content ?? "";
    }
  }
}

function work() {
  while (writing < AT_ONCE && waiting.length) {
    const answer = waiting.shift();
    if (answer.done) continue; // stopped before the model got to it
    writing++;
    write(answer)
      .catch((e) => {
        if (!answer.stop.signal.aborted) {
          console.error("model:", e.message);
          answer.error = "The model did not answer.";
        }
      })
      .finally(() => {
        answer.done = true;
        answer.finishedAt = Date.now();
        writing--;
        work();
      });
  }
}

/** Sends the next piece of one answer to the user who asked: every tab and device of theirs. */
function sendPiece(answer) {
  const text = answer.text.slice(answer.sent, answer.sent + MAX_PIECE);
  const last = answer.done && answer.sent + text.length === answer.text.length;
  answer.sent += text.length;
  answer.pieces.push(answer.sent);
  const piece = { id: answer.id, seq: answer.pieces.length, text };
  if (last) {
    piece.done = true;
    if (answer.stopped) piece.stopped = true;
    if (answer.error) piece.error = answer.error;
    answer.finished = true;
  }
  answer.room.sendTo(answer.to, "answer", piece);
}

// One event every SEND_EVERY ms, for the answer that has waited longest and has something new:
// all it has written since its last piece. Never more than 10 events a second.
setInterval(() => {
  if (!pikoman.isConnected()) return; // the text waits until the bot is back
  for (const [key, answer] of answers) {
    if (answer.finished) {
      if (Date.now() - answer.finishedAt > KEEP_FOR) answers.delete(key);
      continue;
    }
    if (answer.sent < answer.text.length || answer.done) {
      answers.delete(key);
      answers.set(key, answer); // to the back of the line
      sendPiece(answer);
      return;
    }
  }
}, SEND_EVERY);

for (const name of ROOMS) {
  const room = pikoman.listenPresence(name, auth);

  // client-ask {id, text}: a question. The answer goes to the user who asked, on every tab.
  room.onClient("ask", (from, data, info) => {
    const id = String(data?.id ?? "").slice(0, 64);
    const question = String(data?.text ?? "").trim().slice(0, 2000);
    if (!id || !question || room.isBot(from)) return;
    const to = info?.userId ? { userId: info.userId } : from;
    const key = `${name}/${id}`;
    if (answers.has(key)) return; // asked twice: the first answer stands
    const busy = [...answers.values()].filter((a) => !a.done && a.owner === (info?.userId ?? from));
    if (busy.length || waiting.length > 100) {
      room.sendTo(to, "answer", { id, seq: 1, text: "", done: true, error: "Busy: ask again in a moment." });
      return;
    }
    const answer = {
      id, room, to, owner: info?.userId ?? from, question,
      text: "", sent: 0, pieces: [], done: false, finished: false, stop: new AbortController(),
    };
    answers.set(key, answer);
    waiting.push(answer);
    work();
  });

  // client-stop {id}: the model stops writing; what it wrote so far goes out, marked stopped.
  room.onClient("stop", (from, data, info) => {
    const answer = answers.get(`${name}/${String(data?.id ?? "")}`);
    if (!answer || answer.owner !== (info?.userId ?? from) || answer.done) return;
    answer.stopped = true;
    answer.done = true;
    answer.finishedAt = Date.now();
    answer.stop.abort();
  });

  // client-resume {id, after}: a page that was away gets the pieces after `after`, to it alone, as
  // many as fit in one event. With `more`, it asks again from the new `seq`.
  room.onClient("resume", (from, data, info) => {
    const answer = answers.get(`${name}/${String(data?.id ?? "")}`);
    if (!answer || answer.owner !== (info?.userId ?? from)) return;
    const after = Math.max(0, Math.min(Math.floor(Number(data?.after)) || 0, answer.pieces.length));
    const start = after === 0 ? 0 : answer.pieces[after - 1];
    let seq = after;
    while (seq < answer.pieces.length && (seq === after || answer.pieces[seq] - start <= MAX_PIECE)) seq++;
    const caught = { id: answer.id, seq, text: answer.text.slice(start, seq === 0 ? 0 : answer.pieces[seq - 1]), from: after };
    if (seq < answer.pieces.length) caught.more = true;
    else if (answer.finished) {
      caught.done = true;
      if (answer.stopped) caught.stopped = true;
      if (answer.error) caught.error = answer.error;
    }
    room.sendTo(from, "answer", caught);
  });

  room.refused((code) => console.error(`${name}: ${code}`));
}

console.log(`Answering in ${ROOMS.join(", ")} with ${AI_MODEL}`);
pikoman.revoked(() => {
  console.error("The bot's key was deleted.");
  process.exit(1);
});

"""stream.py: answers questions with a language model on your own server and streams each answer,
as it is written, to the member who asked: to every tab and device of that user, in Python. A
member can stop an answer, and a page that lost its connection gets what it missed when it is
back. Any server with an OpenAI-compatible chat API works: Ollama, llama.cpp, vLLM. We use Gemma 4
with Ollama. Python 3.10+. pip install -r requirements.txt, then python stream.py (reads .env).

A connection may send 10 client events a second, so the text goes out in pieces: every 125 ms the
answer that has waited longest gets one event with all it has written since.

Its key (dashboard, New API key): rights subscribe and client_events, kind presence,
channels: prefix presence-ask-.
"""

import asyncio
import json
import threading
import time
import urllib.request

from pikoman_bot import Bot, env

ROOMS = env("ROOMS", "presence-ask-lobby").split(",")
AI_URL = env("AI_URL", "http://127.0.0.1:11434/v1")  # Ollama's OpenAI-compatible API
AI_MODEL = env("AI_MODEL", "gemma4:12b")
AT_ONCE = 2  # answers the model writes at the same time
SEND_EVERY = 0.125  # seconds between two events from the bot: 8 a second, under the limit of 10
MAX_PIECE = 6000  # characters in one event, well under the 12 KiB an event may hold
KEEP_FOR = 600  # seconds a finished answer is kept, for pages that come back

bot = Bot("Assistant", access=3)
writing = asyncio.Semaphore(AT_ONCE)
answers = {}  # "<room>/<id>" -> Answer, the one that has waited longest first


class Answer:
    """Who asked, the text so far, how much of it has gone out, and where each piece ended."""

    def __init__(self, answer_id, room, to, owner, question):
        self.id, self.room, self.to, self.owner, self.question = answer_id, room, to, owner, question
        self.text = ""
        self.sent = 0
        self.pieces = []  # where each event ended, so a page that comes back gets what follows
        self.done = False
        self.finished = False
        self.finished_at = 0.0
        self.stopped = False
        self.error = None
        self.stop = threading.Event()


def stream_model(answer, add):
    """Streams one answer from the model (in a thread: urllib waits), calling add(text) as it comes."""
    body = json.dumps({
        "model": AI_MODEL,
        "stream": True,
        "messages": [
            {"role": "system", "content": "Answer the user's question clearly and briefly."},
            {"role": "user", "content": answer.question},
        ],
    }).encode()
    headers = {"Content-Type": "application/json"}
    if env("AI_KEY"):
        headers["Authorization"] = "Bearer " + env("AI_KEY")
    request = urllib.request.Request(f"{AI_URL}/chat/completions", data=body, headers=headers, method="POST")
    with urllib.request.urlopen(request, timeout=60) as response:
        # Server-sent events: lines of "data: {json}", and "data: [DONE]" at the end.
        for raw in response:
            if answer.stop.is_set():
                return
            line = raw.decode().strip()
            if not line.startswith("data:"):
                continue
            data = line[5:].strip()
            if data == "[DONE]":
                return
            choices = json.loads(data).get("choices") or [{}]
            text = (choices[0].get("delta") or {}).get("content") or ""
            if text:
                add(text)


def append(answer, text):
    answer.text += text


async def job(answer):
    async with writing:
        if answer.done:
            return  # stopped before the model got to it
        loop = asyncio.get_running_loop()
        try:
            await asyncio.to_thread(stream_model, answer, lambda text: loop.call_soon_threadsafe(append, answer, text))
        except Exception as e:  # noqa: BLE001: the next question may go better
            if not answer.stop.is_set():
                print("model:", e)
                answer.error = "The model did not answer."
    if not answer.done:
        answer.done = True
        answer.finished_at = time.time()


async def send_piece(answer):
    """Sends the next piece of one answer to the user who asked: every tab and device of theirs."""
    text = answer.text[answer.sent:answer.sent + MAX_PIECE]
    last = answer.done and answer.sent + len(text) == len(answer.text)
    piece = {"id": answer.id, "seq": len(answer.pieces) + 1, "text": text}
    if last:
        piece["done"] = True
        if answer.stopped:
            piece["stopped"] = True
        if answer.error:
            piece["error"] = answer.error
    try:
        await answer.room.send_to(answer.to, "answer", piece)
    except Exception as e:  # noqa: BLE001: the connection is down; the text waits for it
        return print("send:", e)
    answer.sent += len(text)
    answer.pieces.append(answer.sent)
    if last:
        answer.finished = True


async def sender():
    """One event every SEND_EVERY seconds, for the answer that has waited longest and has something new."""
    while True:
        await asyncio.sleep(SEND_EVERY)
        if bot.id is None:
            continue  # not connected: the text waits until the bot is back
        now = time.time()
        for key in list(answers):
            answer = answers[key]
            if answer.finished:
                if now - answer.finished_at > KEEP_FOR:
                    del answers[key]
                continue
            if answer.sent < len(answer.text) or answer.done:
                answers[key] = answers.pop(key)  # to the back of the line
                await send_piece(answer)
                break


def assistant(room):
    pending = set()  # tasks still with the model, kept so they are not collected

    def owner_of(sender, info):
        return info.get("user_id") or sender

    def find(data):
        answer_id = str((data if isinstance(data, dict) else {}).get("id", ""))
        return answers.get(f"{room.name}/{answer_id}")

    @room.on("ask")
    async def ask(sender, data, info):
        """client-ask {id, text}: a question. The answer goes to the user who asked, on every tab."""
        data = data if isinstance(data, dict) else {}
        answer_id = str(data.get("id", ""))[:64]
        question = str(data.get("text", "")).strip()[:2000]
        if not answer_id or not question or room.is_bot(sender):
            return
        owner = owner_of(sender, info)
        to = {"user_id": info["user_id"]} if info.get("user_id") else sender
        key = f"{room.name}/{answer_id}"
        if key in answers:
            return  # asked twice: the first answer stands
        busy = any(not a.done and a.owner == owner for a in answers.values())
        if busy or len(pending) > 100:
            return await room.send_to(to, "answer", {"id": answer_id, "seq": 1, "text": "", "done": True, "error": "Busy: ask again in a moment."})
        answer = answers[key] = Answer(answer_id, room, to, owner, question)
        task = asyncio.create_task(job(answer))
        pending.add(task)
        task.add_done_callback(pending.discard)

    @room.on("stop")
    def stop(sender, data, info):
        """client-stop {id}: the model stops writing; what it wrote so far goes out, marked stopped."""
        answer = find(data)
        if not answer or answer.owner != owner_of(sender, info) or answer.done:
            return
        answer.stopped = True
        answer.done = True
        answer.finished_at = time.time()
        answer.stop.set()

    @room.on("resume")
    async def resume(sender, data, info):
        """client-resume {id, after}: the pieces after `after`, to that page alone, as many as fit."""
        answer = find(data)
        if not answer or answer.owner != owner_of(sender, info):
            return
        try:
            after = int(float(data.get("after") or 0))
        except (TypeError, ValueError):
            after = 0
        after = max(0, min(after, len(answer.pieces)))
        start = 0 if after == 0 else answer.pieces[after - 1]
        seq = after
        while seq < len(answer.pieces) and (seq == after or answer.pieces[seq] - start <= MAX_PIECE):
            seq += 1
        caught = {"id": answer.id, "seq": seq, "text": answer.text[start:answer.pieces[seq - 1] if seq else 0], "from": after}
        if seq < len(answer.pieces):
            caught["more"] = True
        elif answer.finished:
            caught["done"] = True
            if answer.stopped:
                caught["stopped"] = True
            if answer.error:
                caught["error"] = answer.error
        await room.send_to(sender, "answer", caught)


for name in ROOMS:
    assistant(bot.join(name))

print(f"Answering in {', '.join(ROOMS)} with {AI_MODEL}")
bot.start(sender())

// page.js: ask the bot, and show its answer as it is written, on every tab and device of the user.
// A page that lost its connection asks for what it missed when it is back. Trust only the bot:
// room.isBot(from) is set by the server, never by a member.
// `pikoman` is your connection and `room` the presence channel your page joined:
// pikoman.listenPresence("presence-ask-lobby"). `show(id, text, piece)` puts the answer on the
// page, as text, never as HTML; `piece.done` says it is finished (`stopped`, `error`).
export function assistant(pikoman, room, show) {
  const answers = new Map(); // id -> { seq, text, done }
  let away = false;

  room.onClient("answer", (from, piece) => {
    if (!room.isBot(from) || typeof piece?.id !== "string") return;
    let answer = answers.get(piece.id);
    if (!answer) {
      // Asked on another tab of yours: shown here too, from the start.
      answer = { seq: 0, text: "", done: false };
      answers.set(piece.id, answer);
    }
    // A piece as it is written follows the one before; a catch-up says where it starts.
    const start = piece.from ?? piece.seq - 1;
    if (start !== answer.seq) {
      if (start > answer.seq) room.send("resume", { id: piece.id, after: answer.seq }); // a gap
      return;
    }
    answer.text += String(piece.text ?? "");
    answer.seq = piece.seq;
    if (piece.more) room.send("resume", { id: piece.id, after: answer.seq });
    if (piece.done) answer.done = true;
    show(piece.id, answer.text, piece);
  });

  // Back after a lost connection: ask for the rest of every answer not finished.
  pikoman.offline(() => (away = true));
  room.members(() => {
    if (!away) return;
    away = false;
    for (const [id, answer] of answers) {
      if (!answer.done) room.send("resume", { id, after: answer.seq });
    }
  });

  return {
    /** Asks a question; the answer comes to `show` under the id this returns. */
    ask(text) {
      const id = crypto.randomUUID();
      answers.set(id, { seq: 0, text: "", done: false });
      room.send("ask", { id, text });
      return id;
    },
    /** Stops an answer; what the model wrote so far stays, marked stopped. */
    stop(id) {
      room.send("stop", { id });
    },
  };
}

# Copy to .env and fill in. Never commit .env.
# Your app id, app-…
PIKOMAN_APP=
# The bot's own API key, sk_k_….… (rights subscribe and client_events, prefix presence-ask-)
PIKOMAN_API_KEY=
# The rooms to answer in, comma-separated (default presence-ask-lobby)
ROOMS=
# Your model server's OpenAI-compatible API (default http://127.0.0.1:11434/v1, Ollama) and model (default gemma4:12b)
AI_URL=
AI_MODEL=
# A key for the model server, if it wants one (sent as Authorization: Bearer)
AI_KEY=
# Another server than wss://next.pikoman.se, for tests
PIKOMAN_WS=

{
  "name": "pikoman-stream-bot",
  "version": "1.0.0",
  "private": true,
  "type": "module",
  "description": "Streams answers from a language model on your own server to the member who asked, in a Pikoman room.",
  "license": "MIT",
  "scripts": {
    "start": "node --env-file=.env stream.mjs"
  },
  "engines": {
    "node": ">=22"
  },
  "dependencies": {
    "@pikoman/ws": "1.0.0-rc.11"
  }
}

websockets==15.0.1
cryptography==46.0.3

# unpack stream-bot.zip, then in the folder:
npm install
cp .env.example .env    # fill in PIKOMAN_APP and PIKOMAN_API_KEY
npm start

python -m venv .venv && . .venv/bin/activate    # Windows: .venv\Scripts\activate
pip install -r requirements.txt
python stream.py