Spaces:
Running
Running
File size: 4,495 Bytes
b56ec7a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 | """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
|