Terminal chat

// A streaming terminal chat with history. No dependencies; Node 22.
// Needs TOKENHUB_BASE_URL, TOKENHUB_API_KEY and TOKENHUB_MODEL.
// Run: node --experimental-strip-types chat.ts   (type /exit or press Ctrl-D to quit)
import { createInterface } from "node:readline/promises";
import { readChatStream, StreamError } from "./sse.ts";

const base = (process.env.TOKENHUB_BASE_URL ?? "").replace(/\/+$/, "");
const key = process.env.TOKENHUB_API_KEY ?? "";
const model = process.env.TOKENHUB_MODEL ?? "";
if (!base || !key || !model) throw new Error("set TOKENHUB_BASE_URL, TOKENHUB_API_KEY and TOKENHUB_MODEL");

/** Keep the last N messages so a long session cannot grow the request without bound. */
const MAX_HISTORY = 20;
const history: { role: "user" | "assistant"; content: string }[] = [];
const rl = createInterface({ input: process.stdin, output: process.stdout });

for (;;) {
  let line: string;
  try {
    line = (await rl.question("you> ")).trim();
  } catch {
    break; // Ctrl-D
  }
  if (line === "/exit") break;
  if (line === "") continue;
  history.push({ role: "user", content: line });
  history.splice(0, Math.max(0, history.length - MAX_HISTORY));

  const res = await fetch(`${base}/chat/completions`, {
    method: "POST",
    headers: { Authorization: `Bearer ${key}`, "Content-Type": "application/json" },
    body: JSON.stringify({ model, messages: history, stream: true, max_tokens: 1000 }),
  });
  let answer = "";
  process.stdout.write("bot> ");
  try {
    const stream = readChatStream(res);
    let step = await stream.next();
    while (!step.done) {
      answer += step.value;
      process.stdout.write(step.value);
      step = await stream.next();
    }
    process.stdout.write("\n");
    const usage = step.value;
    if (usage) console.log(`[${usage.prompt_tokens} in, ${usage.completion_tokens} out; request ${res.headers.get("x-generation-id")}]`);
    history.push({ role: "assistant", content: answer });
  } catch (err) {
    // A failed or truncated answer is not kept in the history: the next turn would build on it.
    history.pop();
    const kind = err instanceof StreamError ? err.kind : "unknown";
    console.error(`\n[${kind}] ${(err as Error).message}`);
  }
}
rl.close();

"""A streaming terminal chat with history, standard library only. Python 3.10+.

Needs TOKENHUB_BASE_URL, TOKENHUB_API_KEY and TOKENHUB_MODEL.
Run: python3 terminal_chat.py   (type /exit or press Ctrl-D to quit)
"""
import json
import os
import urllib.error
import urllib.request

BASE = os.environ["TOKENHUB_BASE_URL"].rstrip("/")
KEY = os.environ["TOKENHUB_API_KEY"]
MODEL = os.environ["TOKENHUB_MODEL"]
MAX_HISTORY = 20  # keep the request bounded


class StreamError(Exception):
    pass


def stream_chat(messages):
    """Yields text deltas; returns usage. Raises StreamError on an error frame or truncation."""
    body = json.dumps({"model": MODEL, "messages": messages, "stream": True, "max_tokens": 1000})
    req = urllib.request.Request(f"{BASE}/chat/completions", data=body.encode(), method="POST", headers={
        "Authorization": f"Bearer {KEY}", "Content-Type": "application/json"})
    usage = None
    try:
        resp = urllib.request.urlopen(req, timeout=300)
    except urllib.error.HTTPError as err:
        raise StreamError(f"HTTP {err.code}: {err.read()[:500].decode(errors='replace')}") from None
    with resp:
        data_lines = []
        for raw in resp:
            line = raw.decode().rstrip("\r\n")
            if line.startswith(":"):
                continue  # keep-alive comment
            if line.startswith("data:"):
                data_lines.append(line[5:].lstrip())
                continue
            if line != "" or not data_lines:
                continue
            data, data_lines = "\n".join(data_lines), []
            if data == "[DONE]":
                return usage
            chunk = json.loads(data)
            if chunk.get("error"):
                raise StreamError(chunk["error"].get("message", "stream failed"))
            if chunk.get("usage"):
                usage = chunk["usage"]  # the last chunk: usage and an empty choices list
            choices = chunk.get("choices") or []
            if choices:
                text = (choices[0].get("delta") or {}).get("content")
                if text:
                    yield text
        if data_lines == ["[DONE]"]:  # a terminator with no blank line after it
            return usage
    raise StreamError("the stream ended without data: [DONE]")


def main():
    history = []
    while True:
        try:
            line = input("you> ").strip()
        except EOFError:
            break
        if line == "/exit":
            break
        if not line:
            continue
        history.append({"role": "user", "content": line})
        history[:] = history[-MAX_HISTORY:]
        print("bot> ", end="", flush=True)
        answer = ""
        stream = stream_chat(history)
        try:
            while True:
                text = next(stream)
                answer += text
                print(text, end="", flush=True)
        except StopIteration as stop:
            print()
            if stop.value:
                print(f"[{stop.value['prompt_tokens']} in, {stop.value['completion_tokens']} out]")
            history.append({"role": "assistant", "content": answer})
        except StreamError as err:
            history.pop()  # do not build the next turn on a failed answer
            print(f"\n[error] {err}")


if __name__ == "__main__":
    main()