PyBot: the story and a live chat that shows which files made each reply (pinned to 64eb2c84)

.gitattributes

@@ -33,3 +33,5 @@ saved_model/**/* filter=lfs diff=lfs merge=lfs -text
 *.zip filter=lfs diff=lfs merge=lfs -text
 *.zst filter=lfs diff=lfs merge=lfs -text
 *tfevents* filter=lfs diff=lfs merge=lfs -text
+site/fonts/LetterGothicCode-B.ttf filter=lfs diff=lfs merge=lfs -text
+site/fonts/LetterGothicCode-R.ttf filter=lfs diff=lfs merge=lfs -text

Dockerfile

@@ -0,0 +1,12 @@
+FROM python:3.12-slim
+RUN useradd -m -u 1000 user
+USER user
+WORKDIR /home/user/app
+ENV PATH="/home/user/.local/bin:$PATH" PYTHONUNBUFFERED=1
+COPY --chown=user requirements.txt ./
+RUN pip install --no-cache-dir --user -r requirements.txt
+COPY --chown=user app.py bot_runner.py ./
+COPY --chown=user bot ./bot
+COPY --chown=user site ./site
+EXPOSE 7860
+CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -1,10 +1,15 @@
 ---
 title: PyBot
-emoji: 👁
-colorFrom: yellow
-colorTo: red
+emoji: 🐍
+colorFrom: gray
+colorTo: gray
 sdk: docker
+app_port: 7860
 pinned: false
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+PyBot: a conversational bot without LLMs, written by a swarm of LLM agents. One page: the story and a live chat that shows
+which of PyBot's modules and files produced each reply. Pinned to commit `64eb2c844738` of `HuggingFaceSwarms/SwarmChat-v8`.
+
+`app.py` serves `site/` and `POST /api/chat` / `POST /api/trace`; `bot/bot-<sha12>.zip` holds the bot's source (`bot/**.py` only);
+`bot_runner.py` runs it in a fresh worker per request. Built by `build_pybot.py` in the swarmchat-release workspace.

app.py

@@ -0,0 +1,111 @@
+"""PyBot release Space: the story page (site/) and the PyBot demo's chat endpoint.
+
+The bot is one pinned commit's bot/**.py, shipped in bot/bot-<sha12>.zip, run by bot_runner (from SwarmChat-Demo):
+one loader process, and a fresh forked worker per request that replays the conversation's user lines, so visitors
+never share bot memory. Run only trusted bot code here and add no secrets: the worker is not a security sandbox.
+"""
+import re
+import sys
+import tempfile
+import threading
+import zipfile
+from pathlib import Path
+from typing import Literal
+
+from fastapi import FastAPI, HTTPException
+from fastapi.staticfiles import StaticFiles
+from pydantic import BaseModel, Field
+
+from bot_runner import Bot, user_lines
+
+HERE = Path(__file__).parent
+TREE = tempfile.TemporaryDirectory(prefix="swarmchat-source-")
+zipfile.ZipFile(next(HERE.glob("bot/bot-*.zip"))).extractall(TREE.name)
+app = FastAPI(title="PyBot")
+_bot, _bot_lock = None, threading.Lock()
+_slots = threading.BoundedSemaphore(4)  # conversations answered at once; cpu-basic has 2 vCPUs
+_norm = lambda s: re.sub(r"[^a-z0-9]", "", s.lower())
+# Every bot file, letters and digits only, to find which one holds a reply's text verbatim.
+SOURCES = {p.relative_to(TREE.name).as_posix(): _norm(p.read_text(errors="replace")) for p in Path(TREE.name, "bot").rglob("*.py")}
+
+
+def bot():
+    global _bot
+    with _bot_lock:
+        if _bot is None or not _bot.alive():
+            _bot = Bot(Path(TREE.name), seed=0, timeout=30, workers=2, log=sys.stderr)
+        return _bot
+
+
+threading.Thread(target=bot, daemon=True).start()  # load at boot, so the first visitor does not wait
+
+
+class Message(BaseModel):
+    role: Literal["user", "assistant"]
+    content: str = Field(min_length=1, max_length=2000)
+
+
+class ChatRequest(BaseModel):
+    messages: list[Message] = Field(min_length=1, max_length=60)
+
+
+@app.post("/api/chat")
+def chat(req: ChatRequest):
+    lines = user_lines([m.model_dump() for m in req.messages])
+    if not lines or req.messages[-1].role != "user":
+        raise HTTPException(400, "A user message is required.")
+    if sum(map(len, lines)) > 20000:
+        raise HTTPException(400, "This conversation is too long. Reload the page to start again.")
+    if not _slots.acquire(timeout=15):
+        raise HTTPException(503, "The bot is busy. Please try again in a moment.")
+    try:
+        result = bot().ask(lines)
+    finally:
+        _slots.release()
+    if result.get("status") != "ok" or len(result.get("replies", [])) != len(lines):
+        raise HTTPException(502, "The bot failed on that line. Try another one.")
+    return {"reply": result["replies"][-1]}
+
+
+def answer_source(reply: str, used: list[str]) -> str | None:
+    """The one file that holds the reply verbatim (its first 40+ letters/digits), preferring files the winner called;
+    None when the reply is composed, too short, or found in several files."""
+    text = _norm(reply)
+    for n in (60, 40):
+        key = text[:n]
+        if len(key) < 20:  # too short to point at one place
+            continue
+        hits = [f for f, body in SOURCES.items() if key in body]
+        hits = hits if len(hits) == 1 else [f for f in hits if f in used]
+        if len(hits) == 1:
+            return hits[0]
+    return None
+
+
+@app.post("/api/trace")
+def trace(req: ChatRequest):
+    """Like /api/chat, plus how the last reply was made: the module that won, the ones that also proposed a reply,
+    the bot files the winner called, and the file holding the reply's text."""
+    lines = user_lines([m.model_dump() for m in req.messages])
+    if not lines or req.messages[-1].role != "user":
+        raise HTTPException(400, "A user message is required.")
+    if sum(map(len, lines)) > 20000:
+        raise HTTPException(400, "This conversation is too long. Reload the page to start again.")
+    if not _slots.acquire(timeout=15):
+        raise HTTPException(503, "The bot is busy. Please try again in a moment.")
+    try:
+        result = bot().ask(lines, trace=True)
+    finally:
+        _slots.release()
+    if result.get("status") != "ok" or len(result.get("replies", [])) != len(lines) or not result.get("trace"):
+        raise HTTPException(502, "The bot failed on that line. Try another one.")
+    reply, info = result["replies"][-1], result["trace"]
+    return {"reply": reply, **info, "source": answer_source(reply, info["files"])}
+
+
+@app.get("/health")
+def health():
+    return {"ok": True}
+
+
+app.mount("/", StaticFiles(directory=HERE / "site", html=True), name="site")

bot/bot-64eb2c844738.zip

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:6e0650e67c8a4ab9935aac7f0f2f866c15daafed1e835d85ee4744d434a4a91d
+size 9528378

bot_runner.py

@@ -0,0 +1,195 @@
+# From Space HuggingFaceSwarms/SwarmChat-Demo (bot_runner.py, itself adapted from flock's run4 eval server),
+# plus a trace mode: which modules proposed a reply, which won, and the bot files the winner called.
+"""Serve a checkout of a swarm's chatbot as SwarmChat, an OpenAI-compatible endpoint for evals.
+
+    python experiments/run4/eval/swarmchat_server.py <chatbot checkout> --port 8000
+
+Loading the bot takes seconds (hundreds of modules), answering takes milliseconds. So one
+process loads the bot and forks a worker per conversation, and every conversation starts from
+the same just-loaded state. The same lines get the same replies whatever the order or the
+concurrency of the requests, in this run or the next: the hash seed, the global `random` state
+(restored in each worker) and the seeds of the bot's own random.Random() instances are fixed.
+Replies that read the clock or seed from an object id (a few modules use id(memory)) can still
+change: ids move with address randomisation, which HF Jobs do not let us turn off.
+
+POST /v1/chat/completions feeds the request's user messages to the bot in order (system and
+assistant messages are ignored: the bot only reads user lines) and returns its last reply.
+There is no sandbox: the bot runs as the server's user, with an empty environment, so run it
+somewhere disposable such as an HF Job.
+"""
+
+import json
+import os
+import signal
+import select
+import socket
+import subprocess
+import sys
+import tempfile
+from pathlib import Path
+
+LOADER = r'''
+import io, json, os, random, resource, runpy, signal, socket, sys, time
+root, seed, timeout, path, workers = sys.argv[1], int(sys.argv[2]), float(sys.argv[3]), sys.argv[4], int(sys.argv[5])
+sys.path.insert(0, root)
+status = os.fdopen(os.dup(1), "w")
+os.dup2(2, 1)  # stray prints from the bot go to the log
+listener = socket.socket(socket.AF_UNIX)
+listener.bind(path)
+listener.listen(256)
+# A worker writes a byte here when it takes a conversation; the loader forks one per byte.
+spawn_r, spawn_w = os.pipe()
+os.write(spawn_w, bytes(workers))
+
+
+def converse():
+    """A worker: take one connection, hold that conversation, send the result and exit."""
+    conn = listener.accept()[0]
+    os.write(spawn_w, b"\0")
+    random.setstate(loaded)  # random reseeds itself from the OS in every forked child
+    started = time.monotonic()
+    request = json.loads(conn.makefile("rb").readline())
+    lines, trace = (request, False) if isinstance(request, list) else (request["lines"], bool(request.get("trace")))
+    limit = int(timeout) + 1
+    signal.alarm(limit)
+    resource.setrlimit(resource.RLIMIT_CPU, (limit, limit + 1))
+    if trace:
+        result, out = "ok", {}
+        try:
+            out = traced(lines)
+        except BaseException as exc:
+            result = f"error {type(exc).__name__}"
+        conn.sendall((json.dumps({"replies": out.get("replies", []), "trace": out.get("trace"), "status": result,
+                                  "seconds": round(time.monotonic() - started, 3)}) + "\n").encode())
+        os._exit(0)
+    sys.stdin = io.StringIO("".join(line + "\n" for line in lines))
+    sys.stdout = buffer = io.StringIO()
+    result = "ok"
+    try:
+        runpy.run_module("bot", run_name="__main__")
+    except BaseException as exc:  # the fixed __main__ catches reply errors; this is the rest
+        result = f"error {type(exc).__name__}"
+    conn.sendall((json.dumps({"replies": buffer.getvalue().splitlines(), "status": result,
+                              "seconds": round(time.monotonic() - started, 3)}) + "\n").encode())
+    os._exit(0)
+
+
+def traced(lines):
+    """Replies like the fixed __main__ (one per line, whitespace-normalised, "Hello!" on errors), plus, for the last
+    line, the bot's own decision (bot.runtime.turn: candidates and winner) and the bot files each proposing handler called.
+    sys.monitoring records a function once and then disables that event, so tracing costs about nothing."""
+    import bot.runtime as rt
+    def say(text, fn):
+        try:
+            return " ".join(str(fn(text)).split()) or "Hello!"
+        except Exception:  # noqa: BLE001 - the fixed __main__'s rule
+            return "Hello!"
+    memory = {}
+    replies = [say(line, lambda t: rt.reply(t, memory)) for line in lines[:-1]]
+    mon, tool, bot_dir, seen, files = sys.monitoring, 5, os.path.join(root, "bot") + os.sep, set(), {}
+    mon.use_tool_id(tool, "pybot-trace")
+    def start(code, offset):
+        if code.co_filename.startswith(bot_dir):
+            seen.add("bot/" + code.co_filename[len(bot_dir):])
+        return mon.DISABLE
+    mon.register_callback(tool, mon.events.PY_START, start)
+    def watch(name, respond):
+        def run(text, mem):
+            seen.clear(); mon.restart_events(); mon.set_events(tool, mon.events.PY_START)
+            try:
+                return respond(text, mem)
+            finally:
+                mon.set_events(tool, 0); files[name] = sorted(seen)
+        return run
+    rt._HANDLERS = [(name, priority, watch(name, respond)) for name, priority, respond in rt._HANDLERS]
+    decision = {}
+    def last(text):
+        decision.update(rt.turn(text, memory))
+        return decision["reply"]
+    replies.append(say(lines[-1], last))
+    module = decision.get("module")
+    return {"replies": replies, "trace": {"module": module,
+            "candidates": [c["module"] for c in decision.get("candidates", [])],
+            "files": files.get(module, []) if module else []}}
+
+
+# Modules make their own random.Random(), which seeds itself from the OS: derive those seeds
+# from the run's seed and a counter instead, so every run starts in the same state.
+os_seeded, created = random.Random.seed, iter(range(1 << 62))
+def derived_seed(self, a=None, version=2):
+    os_seeded(self, f"{seed}:{next(created)}" if a is None else a, version)
+random.Random.seed = derived_seed
+random.seed = random._inst.seed
+random.seed(seed)
+started = time.monotonic()
+import bot.runtime
+loaded = random.getstate()
+print(json.dumps({"load_s": round(time.monotonic() - started, 2),
+                  "handlers": len(bot.runtime._HANDLERS)}), file=status, flush=True)
+# Keep `workers` forks waiting for a connection. The workers read the requests, never the
+# loader, and the kernel reaps them.
+signal.signal(signal.SIGCHLD, signal.SIG_IGN)
+while True:
+    os.read(spawn_r, 1)
+    if os.fork() == 0:
+        converse()
+'''
+
+
+class Bot:
+    """The loader process; ask() holds one conversation in a fresh worker."""
+
+    def __init__(self, tree: Path, *, seed: int, timeout: float, workers: int, log):
+        self.timeout = timeout
+        self.tmp = tempfile.TemporaryDirectory(prefix="swarmchat-")
+        self.path = os.path.join(self.tmp.name, "bot.sock")
+        # No -I: PYTHONHASHSEED must apply. The environment holds nothing else.
+        self.proc = subprocess.Popen(
+            [sys.executable, "-s", "-S", "-P", "-c", LOADER, str(tree), str(seed), str(timeout),
+             self.path, str(workers)],
+            cwd=self.tmp.name, env={"PYTHONHASHSEED": str(seed)}, stdout=subprocess.PIPE,
+            stderr=log, text=True, start_new_session=True)
+        if not select.select([self.proc.stdout], [], [], 90)[0]:
+            self.close()
+            raise RuntimeError("The bot did not load within 90 seconds.")
+        line = self.proc.stdout.readline()
+        if not line:
+            self.close()
+            raise RuntimeError("This commit could not load its bot; try another commit.")
+        self.ready = json.loads(line)
+
+    def ask(self, lines: list[str], trace: bool = False) -> dict:
+        with socket.socket(socket.AF_UNIX) as conn:
+            conn.settimeout(self.timeout + 5)
+            conn.connect(self.path)
+            conn.sendall((json.dumps({"lines": lines, "trace": True} if trace else lines) + "\n").encode())
+            try:
+                raw = conn.makefile("rb").readline()
+            except TimeoutError:
+                return {"replies": [], "status": "timeout"}
+        # The worker died without replying, e.g. on its CPU or wall-clock limit.
+        return json.loads(raw) if raw else {"replies": [], "status": "worker exited"}
+
+    def alive(self) -> bool:
+        return self.proc.poll() is None
+
+    def close(self):
+        if self.alive():
+            os.killpg(self.proc.pid, signal.SIGKILL)
+        self.proc.wait()
+        self.proc.stdout.close()
+        self.tmp.cleanup()
+
+
+def user_lines(messages) -> list[str]:
+    """User message texts as single bot input lines (the bot reads one line per turn)."""
+    lines = []
+    for m in messages:
+        if m.get("role") != "user":
+            continue
+        content = m.get("content") or ""
+        if isinstance(content, list):
+            content = " ".join(p.get("text", "") for p in content if p.get("type") == "text")
+        lines.append(" ".join(str(content).split()))
+    return lines
+

requirements.txt

@@ -0,0 +1,2 @@
+fastapi==0.115.6
+uvicorn==0.32.1

site/fonts/LetterGothicCode-B.ttf

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:4d982d2025de5f980cbbf0e62b2d7036ea0e04a31019531cd995d2b3e378a8e2
+size 112960

site/fonts/LetterGothicCode-R.ttf

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:4af8b279313345424c02d7a2030826a831cc9fb367ff2251f09504d2b6f59ee6
+size 110084