Spaces:
Running
Running
| """PyBot in the browser: load the bot once, then answer turns and explain each reply. | |
| Ported from the PyBot Space's server (bot_runner.py's LOADER and traced(), app.py's answer_source()): | |
| - Random seeds are fixed the way the eval server fixes them, so replies match the benchmark run as | |
| closely as a browser allows. The server forks a fresh worker per request; here `reset()` restores the | |
| just-loaded random state and starts a new memory instead. | |
| - For each reply: the module that won, the ones that also proposed a reply, the bot files the winner | |
| called (sys.monitoring, one PY_START event per function), and the file holding the reply's text. | |
| """ | |
| import json | |
| import os | |
| import random | |
| import re | |
| import sys | |
| import time | |
| ROOT = "/pybot" | |
| SEED = 0 | |
| _norm = lambda s: re.sub(r"[^a-z0-9]", "", s.lower()) | |
| _rt = None | |
| _loaded_state = None | |
| _memory = {} | |
| _files = {} | |
| _sources = None | |
| def load(): | |
| """Import the bot (seconds); returns {"load_s", "modules", "handlers"}.""" | |
| global _rt, _loaded_state | |
| sys.path.insert(0, ROOT) | |
| # 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 load 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 as rt | |
| _rt = rt | |
| _loaded_state = random.getstate() | |
| _watch_handlers() | |
| return {"load_s": round(time.monotonic() - started, 2), "modules": len(rt._MODULES), "handlers": len(rt._HANDLERS)} | |
| def _watch_handlers(): | |
| """Wrap every respond() so a turn records the bot files each handler called.""" | |
| mon, tool = sys.monitoring, 5 | |
| bot_dir, seen = 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] | |
| def reset(): | |
| """A new conversation: empty memory and the just-loaded random state.""" | |
| global _memory | |
| _memory = {} | |
| random.setstate(_loaded_state) | |
| def turn(text): | |
| """One reply as JSON: {"reply", "module", "candidates", "files", "source", "ms"}.""" | |
| text = " ".join(str(text).split()) | |
| _files.clear() | |
| started = time.monotonic() | |
| decision = {} | |
| try: | |
| decision = _rt.turn(text, _memory) | |
| reply = " ".join(str(decision["reply"]).split()) or "Hello!" | |
| except Exception: # noqa: BLE001 - the fixed __main__'s rule | |
| reply = "Hello!" | |
| ms = round((time.monotonic() - started) * 1000) | |
| module = decision.get("module") | |
| files = _files.get(module, []) if module else [] | |
| return json.dumps({"reply": reply, "module": module, | |
| "candidates": [c["module"] for c in decision.get("candidates", [])], | |
| "files": files, "source": answer_source(reply, files), "ms": ms}) | |
| def answer_source(reply, used): | |
| """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.""" | |
| global _sources | |
| if _sources is None: # every bot file, letters and digits only (built by build_bundle.py) | |
| with open(os.path.join(ROOT, "norm.json"), encoding="utf-8") as handle: | |
| _sources = json.load(handle) | |
| 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 | |