| |
| """v5 online teach-train daemon — closes the live-teaching loop. |
| |
| Every cycle: |
| 1. Snapshot the live teach traces (/content/teach/trace_teach_*.jsonl) |
| 2. Rebuild dataset_teach from ALL traces (labels = the teacher's actions) |
| 3. Continue-train the student (heads + LoRA resume) for a bounded step budget |
| 4. Upload the new generation to HF t4_assets_v5/checkpoints/ (HEAD-verified) |
| |
| The teacher (main agent) keeps playing via teach_server; this daemon absorbs |
| whatever has been demonstrated so far. Cycle skips itself when no new steps. |
| """ |
| import json |
| import os |
| import shutil |
| import subprocess |
| import sys |
| import time |
| import glob |
|
|
| sys.path.insert(0, '/content/qwenjev') |
|
|
| REPO = 'tchbcb/qwenjev' |
| OUT = '/content/qwenjev/checkpoints/qwenjev-s1-hf-v5t' |
| TRACES = '/content/teach' |
| CYCLE_SECONDS = 300 |
| MIN_NEW_STEPS = 15 |
| STEPS_PER_CYCLE = 25 |
| TOKEN_FILE = '/content/qwenjev_work_v3/.hf_token' |
|
|
|
|
| def log(msg): |
| print('[daemon %s] %s' % (time.strftime('%H:%M:%S'), msg), flush=True) |
|
|
|
|
| def count_steps(): |
| n = 0 |
| for f in glob.glob(TRACES + '/trace_teach_*.jsonl'): |
| with open(f) as fh: |
| n += sum(1 for line in fh if line.strip()) |
| return n |
|
|
|
|
| def snapshot_traces(): |
| os.makedirs('/content/qwenjev/data/traces_teach', exist_ok=True) |
| for f in glob.glob('/content/qwenjev/data/traces_teach/ep*.jsonl'): |
| os.remove(f) |
| copied = [] |
| for i, src in enumerate(sorted(glob.glob(TRACES + '/trace_teach_*.jsonl'))): |
| game = os.path.basename(src).replace('trace_teach_', '').replace('.jsonl', '') |
| dst = '/content/qwenjev/data/traces_teach/ep%03d_%s.jsonl' % (i + 1, game) |
| shutil.copyfile(src, dst) |
| copied.append(dst) |
| return copied |
|
|
|
|
| def hf_token(): |
| if os.path.exists(TOKEN_FILE): |
| return open(TOKEN_FILE).read().strip() |
| return os.environ.get('HF_TOKEN', '') |
|
|
|
|
| def _lora_pt_name(): |
| """PEFT saves adapter_model.safetensors, but HF strips that extension on |
| this repo (v0.4.8 iron law) — present the SAME bytes as adapter_model.pt.""" |
| d = OUT + '/hf_lora' |
| pt = d + '/adapter_model.pt' |
| st = d + '/adapter_model.safetensors' |
| if not os.path.exists(pt) and os.path.exists(st): |
| shutil.copyfile(st, pt) |
| return pt |
|
|
|
|
| def upload_generation(): |
| from huggingface_hub import HfApi |
| api = HfApi(token=hf_token()) |
| |
| |
| try: |
| import tarfile |
| tb = '/content/teach_traces_snapshot.tgz' |
| with tarfile.open(tb, 'w:gz') as t: |
| for f in sorted(glob.glob(TRACES + '/trace_teach_*.jsonl')): |
| t.add(f, arcname=os.path.basename(f)) |
| api.upload_file(path_or_fileobj=tb, |
| path_in_repo='t4_assets_v5/traces/teach_traces.tgz', |
| repo_id=REPO) |
| log('traces snapshot uploaded (%d files)' % len(glob.glob(TRACES + '/trace_teach_*.jsonl'))) |
| except Exception as exc: |
| log('traces snapshot FAILED (non-fatal): %r' % exc) |
| uploads = [ |
| (OUT + '/qwenjev_s1.pt', 't4_assets_v5/checkpoints/qwenjev_s1.pt'), |
| (_lora_pt_name(), 't4_assets_v5/checkpoints/hf_lora/adapter_model.pt'), |
| (OUT + '/hf_lora/adapter_config.json', 't4_assets_v5/checkpoints/hf_lora/adapter_config.json'), |
| (OUT + '/stage_b_meta.json', 't4_assets_v5/checkpoints/stage_b_meta.json'), |
| ] |
| for local, remote in uploads: |
| if not os.path.exists(local): |
| log('skip missing %s' % local) |
| continue |
| api.upload_file(path_or_fileobj=local, path_in_repo=remote, repo_id=REPO) |
| log('uploaded %s' % remote) |
| |
| import urllib.request |
| for remote in ('t4_assets_v5/checkpoints/qwenjev_s1.pt', |
| 't4_assets_v5/checkpoints/hf_lora/adapter_model.pt'): |
| url = 'https://huggingface.co/%s/resolve/main/%s' % (REPO, remote) |
| try: |
| r = urllib.request.urlopen(urllib.request.Request(url, method='HEAD')) |
| log('HEAD ok %s (%s)' % (remote, r.status)) |
| except Exception as exc: |
| log('HEAD FAIL %s: %s' % (remote, exc)) |
|
|
|
|
| def train_cycle(): |
| |
| |
| src = OUT + '/hf_lora' |
| dst = OUT + '/stage_a_lora' |
| os.makedirs(dst, exist_ok=True) |
| _lora_pt_name() |
| if os.path.exists(src + '/adapter_model.pt'): |
| for f in ('adapter_model.pt', 'adapter_config.json'): |
| if os.path.exists(src + '/' + f): |
| shutil.copyfile(src + '/' + f, dst + '/' + f) |
| |
| shutil.copyfile(dst + '/adapter_model.pt', dst + '/adapter_model.safetensors') |
| r = subprocess.run( |
| ['python3', '-u', '-B', 'train/train_system1.py', '--config', 'configs/train_teach.yaml'], |
| cwd='/content/qwenjev', capture_output=True, text=True, timeout=3600) |
| log('trainer exit=%s' % r.returncode) |
| tail = [l for l in r.stdout.splitlines() if 'step=' in l or 'done' in l.lower()] |
| for l in tail[-3:]: |
| log(' ' + l[:110]) |
| if r.returncode != 0: |
| log('STDERR tail: ' + r.stderr[-400:]) |
| return r.returncode == 0 |
|
|
|
|
| def build_dataset(): |
| r = subprocess.run( |
| ['python3', '-m', 'data.build_dataset', '--in', 'data/traces_teach', |
| '--out', 'data/dataset_teach'], |
| cwd='/content/qwenjev', capture_output=True, text=True, timeout=600) |
| ok = r.returncode == 0 |
| log('build_dataset ok=%s %s' % (ok, r.stdout.strip()[-90:])) |
| return ok |
|
|
|
|
| def main(): |
| log('daemon up; baseline steps=%d' % count_steps()) |
| last = count_steps() |
| cycle = 0 |
| while True: |
| |
| |
| for _ in range(CYCLE_SECONDS // 5): |
| time.sleep(5) |
| now = count_steps() |
| if now - last < MIN_NEW_STEPS: |
| log('cycle %d skipped: only +%d new steps' % (cycle, now - last)) |
| continue |
| cycle += 1 |
| log('cycle %d: %d new steps (%d total) -> teaching the student' % (cycle, now - last, now)) |
| try: |
| snapshot_traces() |
| if not build_dataset(): |
| continue |
| if train_cycle(): |
| upload_generation() |
| last = now |
| except Exception as exc: |
| log('cycle %d FAILED: %r' % (cycle, exc)) |
|
|
|
|
| if __name__ == '__main__': |
| main() |
|
|
