m2w-gemma4-e2b-pointer / code /m2w_form_demo.py
senpaicq's picture
Add LoRA adapter, pointer head, model card, code and results
a882958 verified
Raw History Blame Contribute Delete
23.1 kB
#!/usr/bin/env python3
"""Form doldurma demosu: eğitilmiş Gemma (LoRA + kafa) tek başına, canlı sayfada.
Her alan için bir adım: sayfa -> (tam sayfa ekran görüntüsü + DOM adayları) -> Gemma bir eleman seçer -> eylem uygulanır -> doğrulanır.
Girdi eğitimdekiyle aynı biçimde kurulur: aday çıkarma (m2w_prep.cand_text/rank) -> kural süzgeci (m2w_filter, K=160) -> tree biçimi (m2w_format).
İki çalışma biçimi (aynı formda ikisi de koşulur):
yonlendirmeli : her adımda GEÇMİŞ doğru eylemlerle verilir (hata birikmez) -> adım doğruluğu
serbest : modelin kendi seçimi uygulanır, geçmiş kendi eylemlerinden oluşur -> gerçek sonuç (alanlar doğru dolu mu)
İki aday kolu:
sadik : aday öznitelikleri veri setindekiyle aynı (<label> bağı eklenmez)
etiketli : <label> ile bağlı alanlara etiket metni aria_label olarak eklenir (erişilebilirlik adı gibi)
Güvenlik: sayfa gezintileri engellenir, hiçbir yerde gönder tuşuna kasıtlı basılmaz.
Kullanım: python m2w_form_demo.py --run egitim_gorselli_buyuk_... --out sonuc_form_demo
python m2w_form_demo.py --predictor tfidf --out deneme (model gerekmez; sadece sözcük eşleştirme tabanı)
"""
import argparse
import http.server
import io
import json
import math
import os
import re
import socketserver
import threading
import time
from collections import defaultdict
import numpy as np
from PIL import Image, ImageDraw
import m2w_filter as F
import m2w_format as FM
import m2w_form_sayfalari as S
from m2w_prep import cand_text, rank
OPS = {"CLICK": 0, "TYPE": 1, "SELECT": 2}
JS_ADAY = r"""
() => {
document.querySelectorAll('[data-m2w]').forEach(e => e.removeAttribute('data-m2w'));
const INTER = new Set(['a','button','input','select','textarea','option','label','summary']);
const EXTRA = new Set(['li','img','svg','i','span','div','td','tr','th','h1','h2','h3','h4','p']);
const NOTEXT = new Set(['img','svg','i']);
const TXT = ['role','type','aria-label','placeholder','title','name','alt','value','id','aria-checked','aria-expanded',
'aria-disabled','aria-selected','aria-haspopup'];
const BOOL = ['checked','disabled','readonly','required','selected'];
const sx = window.scrollX, sy = window.scrollY;
const out = [];
let k = 0;
for (const el of document.querySelectorAll('body *')) {
const tag = el.tagName.toLowerCase();
const inter = INTER.has(tag), extra = EXTRA.has(tag);
if (!inter && !extra) continue;
const r = el.getBoundingClientRect();
if (r.width <= 0 || r.height <= 0) continue;
const cs = getComputedStyle(el);
if (cs.visibility === 'hidden' || cs.display === 'none') continue;
if (tag === 'input' && el.type === 'hidden') continue;
const at = {};
for (const key of TXT) {
const v = el.getAttribute(key);
if (v !== null && String(v).trim()) at[key.replace(/-/g, '_')] = String(v).trim().slice(0, 120);
}
for (const key of BOOL) if (el.hasAttribute(key)) at[key] = 'true';
let inner = '';
try { inner = (el.innerText !== undefined ? el.innerText : el.textContent) || ''; } catch (e) {}
inner = inner.replace(/\s+/g, ' ').trim().slice(0, 200);
if (extra) {
const direct = Array.from(el.childNodes).some(n => n.nodeType === 3 && n.textContent.trim());
const lab = at.aria_label || at.title || at.alt || at.placeholder || at.value;
if (NOTEXT.has(tag) ? !lab : !(direct || lab)) continue;
}
let etiket = '';
if (el.labels && el.labels.length) etiket = Array.from(el.labels).map(l => (l.innerText || '').replace(/\s+/g, ' ').trim()).join(' ');
el.setAttribute('data-m2w', String(k));
out.push({k: k, tag: tag, at: at, inner: inner, etiket: etiket, bbox: [r.left + sx, r.top + sy, r.width, r.height]});
k += 1;
}
return out;
}
"""
JS_SNAP = """() => Array.from(document.querySelectorAll('input,textarea,select')).map(e =>
(e.type === 'checkbox' || e.type === 'radio') ? 'c' + e.checked : (e.tagName === 'SELECT' ? 's' + e.selectedIndex : 'v' + e.value))"""
JS_IDX = "e => Array.from(document.querySelectorAll('input,textarea,select')).indexOf(e)"
JS_LABELS = "e => Array.from(e.labels || []).flatMap(l => [l, ...l.querySelectorAll('[data-m2w]')]).map(x => x.getAttribute('data-m2w'))"
JS_SELTEXT = "e => e.selectedOptions && e.selectedOptions[0] ? e.selectedOptions[0].textContent.trim() : ''"
# ----------------------------------------------------------------------------- yardımcılar
def softmax(x, T=1.0):
x = np.asarray(x, dtype=np.float64) / T
x = x - x.max()
e = np.exp(x)
return e / e.sum()
def repr_of_cand(text, op, value):
"""Mind2Web 'action_reprs' biçimi: [rol] isim -> TYPE: değer"""
role = text.split(" ", 1)[0]
rest = re.sub(r"(\s\[[^\]]*\])+$", "", text[len(role):].strip())
name = rest[1:-1] if rest.startswith('"') and rest.endswith('"') and len(rest) >= 2 else rest
tail = f" -> {op}" + (f": {value}" if op in ("TYPE", "SELECT") else "")
return f"[{role}] {name}{tail}"
def repr_of_field(f):
role = S.ROLE_OF_KIND[f["kind"]]
op = S.OP_OF_KIND[f["kind"]]
tail = f" -> {op}" + (f": {f['value']}" if op in ("TYPE", "SELECT") else "")
return f"[{role}] {f['label']}{tail}"
def serve(directory):
class H(http.server.SimpleHTTPRequestHandler):
def __init__(self, *a, **k):
super().__init__(*a, directory=directory, **k)
def log_message(self, *a):
pass
srv = socketserver.TCPServer(("127.0.0.1", 0), H)
threading.Thread(target=srv.serve_forever, daemon=True).start()
return srv, f"http://127.0.0.1:{srv.server_address[1]}"
def aday_listesi(page, task, history, arm, K=160):
"""Canlı sayfadan Mind2Web biçiminde aday listesi. Dönen: (süzülmüş aday listesi, ham liste)."""
raw = page.evaluate(JS_ADAY)
cs = []
for r in raw:
at = dict(r["at"])
if arm == "etiketli" and r.get("etiket") and "aria_label" not in at:
at["aria_label"] = r["etiket"][:120]
cs.append({"bid": str(r["k"]), "tag": r["tag"], "at": at, "inner": r["inner"], "bbox": r["bbox"],
"text": cand_text({"tag": r["tag"], "attrs": at}, r["inner"][:100])})
if not cs:
return [], raw
query = task + " " + " ".join(re.sub(r"->.*", "", h) for h in history[-3:])
order = rank(query, [c["text"] for c in cs], [c["bbox"] for c in cs], True)
cs = [cs[i] for i in order]
cs = [cs[i] for i in F.select(cs, K)]
for c in cs:
c["text"] = FM.fmt_tree(c)
return cs, raw
def eylem(page, h, f, tag):
"""Playwright ile eylemi uygular. Hata olursa mesajı döndürür, yoksa ''."""
try:
k = f["kind"]
if k in ("text", "textarea"):
h.click(timeout=2500)
page.keyboard.type(f["value"], delay=0)
elif k == "select":
if tag == "select":
h.select_option(label=f["value"], timeout=2500)
else:
h.click(timeout=2500)
else:
h.click(timeout=2500)
page.wait_for_timeout(60)
return ""
except Exception as e:
return type(e).__name__
def alan_degeri_dogru(h, f):
try:
k = f["kind"]
if k in ("text", "textarea"):
return h.input_value() == f["value"]
if k == "select":
return h.evaluate(JS_SELTEXT) == f["value"]
return bool(h.is_checked())
except Exception:
return False
# ----------------------------------------------------------------------------- tahmin ediciler
class TfidfTaban:
"""Model yok: alan etiketini aday metinleriyle sözcük benzerliğiyle eşleştirir (bağlam için taban çizgisi; hedef etiketi doğrudan bilir)."""
ad = "tfidf_etiket_tabani"
T = 1.0
def __call__(self, ex, png, field):
from sklearn.feature_extraction.text import TfidfVectorizer
texts = [c["text"] for c in ex["cands"]]
try:
v = TfidfVectorizer(token_pattern=r"[A-Za-z0-9]{2,}", lowercase=True)
M = v.fit_transform(texts + [field["label"]])
sc = (M[:-1] @ M[-1].T).toarray().ravel()
except ValueError:
sc = np.zeros(len(texts))
return np.append(sc * 10.0, 0.5), np.zeros(3)
class Gemma:
ad = "gemma_lora"
def __init__(self, run_dir, shots_dir, temperature=1.0):
import torch
import m2w_train_lora as T
import m2w_train_lora_gorsel as G
from peft import PeftModel
from transformers import AutoModelForMultimodalLM, AutoProcessor, AutoTokenizer
self.torch, self.T, self.G = torch, T, G
self.temp = temperature
self.T = temperature
if not os.path.isdir(run_dir): # Hugging Face depo kimliği verildiyse indir
from huggingface_hub import snapshot_download
run_dir = snapshot_download(run_dir)
cfg = os.path.join(run_dir, "m2w_config.json")
cfg = cfg if os.path.exists(cfg) else os.path.join(run_dir, "sonuc.json")
a = json.load(open(cfg))["args"]
lora_dir = os.path.join(run_dir, "lora")
lora_dir = lora_dir if os.path.isdir(lora_dir) else run_dir # HF düzeninde adaptör kökte durur
self.T_scale = 1.0
T.CAND_TOK = a.get("max_cand_tok", T.MAX_CAND_TOK); G.MAX_SEQ_V = a.get("max_seq", 0)
self.layer = a.get("layer", 30)
self.device = "cuda" if torch.cuda.is_available() else "cpu"
self.tok = AutoTokenizer.from_pretrained(T.MODEL_ID, revision=T.REV)
try:
proc = AutoProcessor.from_pretrained(T.MODEL_ID, revision=T.REV)
except Exception:
proc = AutoProcessor.from_pretrained("google/gemma-4-E2B-it")
os.makedirs(shots_dir, exist_ok=True)
self.shots = shots_dir
self.sayfa = G.Sayfa(proc, self.tok, shots_dir, a.get("strip", 1280), a.get("max_strips", 5), a.get("img_tokens", 280), False)
base = AutoModelForMultimodalLM.from_pretrained(T.MODEL_ID, revision=T.REV, dtype=torch.bfloat16).to(self.device)
self.model = PeftModel.from_pretrained(base, lora_dir).to(self.device).eval()
try:
d = self.model.config.get_text_config().hidden_size
except Exception:
d = 1536
self.head = T.Head(d).to(self.device)
self.head.load_state_dict(torch.load(os.path.join(run_dir, "head.pt"), map_location=self.device)); self.head.eval()
self.n = 0
print(f"model yüklendi | aday satırı sınırı {T.CAND_TOK} token | toplam sınır {G.MAX_SEQ_V} | katman {self.layer} | {self.device}", flush=True)
def __call__(self, ex, png, field):
torch, G = self.torch, self.G
self.n += 1
uid = f"demo_{self.n}"
ex["action_uid"] = uid
with open(os.path.join(self.shots, uid + ".png"), "wb") as f:
f.write(png)
try:
with torch.no_grad():
order = list(range(len(ex["cands"])))
b, spans, qidx, _ = G.collate_v(self.tok, self.sayfa, ex, order, self.device)
with torch.autocast("cuda", dtype=torch.bfloat16, enabled=self.device == "cuda"):
h = G.run_model_v(self.model, b, self.layer)
logits, opl = self.head(h, spans, qidx)
return logits[0].float().cpu().numpy(), opl[0].float().cpu().numpy()
finally:
try:
os.remove(os.path.join(self.shots, uid + ".png"))
except OSError:
pass
# ----------------------------------------------------------------------------- bir form koşusu
def ornek_resmi(png, raw, tidx, cands, j, path):
"""Hedef (yeşil) ve tahmin (kırmızı) kutusunu çizer, hedef çevresini keser."""
try:
im = Image.open(io.BytesIO(png)).convert("RGB")
d = ImageDraw.Draw(im)
tb = raw[int(tidx)]["bbox"] if tidx is not None and tidx.isdigit() and int(tidx) < len(raw) else None
if tb:
d.rectangle([tb[0], tb[1], tb[0] + tb[2], tb[1] + tb[3]], outline=(0, 170, 0), width=4)
if j is not None and j < len(cands):
pb = cands[j]["bbox"]
d.rectangle([pb[0], pb[1], pb[0] + pb[2], pb[1] + pb[3]], outline=(220, 0, 0), width=3)
cy = (tb[1] + tb[3] / 2) if tb else im.size[1] / 2
top = int(max(0, min(cy - 450, im.size[1] - 900))); bot = min(im.size[1], top + 900)
im.crop((0, top, im.size[0], bot)).save(path)
except Exception as e:
print(" örnek resmi yazılamadı:", type(e).__name__)
def form_kos(browser, base, spec, pred, arm, mode, args, imgdir=None, img_budget=None):
url = spec["url"] if spec.get("remote") else base + spec["url"]
ctx = browser.new_context(viewport={"width": 1280, "height": 720})
durum = {"izin": True}
def engelle(route, request):
if (not durum["izin"]) and request.is_navigation_request():
route.abort()
else:
route.continue_()
ctx.route("**/*", engelle)
page = ctx.new_page()
page.on("dialog", lambda d: d.dismiss())
page.set_default_timeout(8000)
recs, extra = [], {"form": spec["name"]}
try:
page.goto(url, wait_until="load", timeout=20000)
except Exception as e:
ctx.close()
return recs, {"form": spec["name"], "yuklenemedi": type(e).__name__}
page.wait_for_timeout(300)
durum["izin"] = False
task = S.task_text(spec)
hist, before = [], page.evaluate(JS_SNAP)
hedef_idx = set(); n_bulunamadi = 0
for fi, f in enumerate(spec["fields"]):
try:
h_true = page.query_selector(f["css"])
except Exception:
h_true = None
if h_true is None:
n_bulunamadi += 1
continue
t0 = time.time()
cands, raw = aday_listesi(page, task, hist, arm, args.K)
try:
tidx = h_true.get_attribute("data-m2w")
labs = [x for x in h_true.evaluate(JS_LABELS) if x]
hedef_idx.add(h_true.evaluate(JS_IDX))
except Exception:
tidx, labs = None, []
png = page.screenshot(full_page=True)
t1 = time.time()
bids = [c["bid"] for c in cands]
ex = {"action_uid": "x", "task": task, "history": list(hist), "cands": cands, "pos_idx": [], "op": "CLICK"}
rec = {"form": spec["name"], "alan": f["label"], "tur": f["kind"], "stil": f.get("stil", spec.get("stil")), "kol": arm, "mod": mode,
"n_aday": len(cands), "listede": tidx in bids}
if not cands:
rec.update(dogru=False, gevsek=False, bos=True, conf=0.0, op_ok=False, null=True)
recs.append(rec); continue
logits, opl = pred(ex, png, f)
t2 = time.time()
n = len(cands)
p = softmax(logits, pred.T)
j = int(p.argmax())
isnull = j == n
bid = None if isnull else cands[j]["bid"]
op_pred = {v: k for k, v in OPS.items()}[int(np.argmax(opl))] if opl is not None and np.any(opl) else None
rec.update(conf=float(p[j]), conf_null=float(p[n]), null=isnull, dogru=bool(bid is not None and bid == tidx),
gevsek=bool(bid is not None and (bid == tidx or bid in labs)),
op_ok=(op_pred == S.OP_OF_KIND[f["kind"]]) if op_pred else None,
tahmin=None if isnull else cands[j]["text"], hedef=next((c["text"] for c in cands if c["bid"] == tidx), None),
sira_hedef=(bids.index(tidx) if tidx in bids else None), t_aday=t1 - t0, t_model=t2 - t1)
recs.append(rec)
if imgdir and img_budget is not None and arm == "sadik" and mode == "yonlendirmeli" and pred.ad == "gemma_lora":
kat = "dogru" if rec["gevsek"] else "yanlis"
if img_budget.get(kat, 0) > 0:
img_budget[kat] -= 1
ornek_resmi(png, raw, tidx, cands, None if isnull else j,
os.path.join(imgdir, f"{kat}_{spec['name']}_{fi}.png"))
# ---- eylem
if mode == "yonlendirmeli":
eylem(page, h_true, f, h_true.evaluate("e => e.tagName.toLowerCase()"))
hist.append(repr_of_field(f))
else:
if isnull:
continue
ph = page.query_selector(f'[data-m2w="{bid}"]')
if ph is None:
continue
hata = eylem(page, ph, f, cands[j]["tag"])
if hata:
rec["eylem_hata"] = hata
hist.append(repr_of_cand(cands[j]["text"], S.OP_OF_KIND[f["kind"]], f["value"]))
# ---- sonuç denetimi
after = page.evaluate(JS_SNAP)
stray = 0
for i, (a, b) in enumerate(zip(before, after)):
if a != b and i not in hedef_idx and not (b.startswith("c") and a.startswith("c")):
stray += 1
ok = 0; tot = 0
for f in spec["fields"]:
try:
h = page.query_selector(f["css"])
except Exception:
h = None
if h is None:
continue
tot += 1
ok += alan_degeri_dogru(h, f)
extra.update(alan_toplam=tot, alan_dogru=ok, kirlenen_alan=stray, hedef_bulunamadi=n_bulunamadi,
tam_basari=bool(tot and ok == tot and stray == 0))
ctx.close()
return recs, extra
# ----------------------------------------------------------------------------- özet
def risk_kapsama(conf, ok, cover=(100, 80, 60, 40, 20)):
o = np.argsort(-np.asarray(conf), kind="stable")
ok = np.asarray(ok, dtype=float)[o]
out = {}
for c in cover:
k = max(1, int(math.ceil(len(ok) * c / 100)))
out[f"acc@{c}%"] = round(float(ok[:k].mean()), 3)
return out
def auroc(conf, ok):
try:
from sklearn.metrics import roc_auc_score
if len(set(ok)) < 2:
return None
return round(float(roc_auc_score(ok, conf)), 3)
except Exception:
return None
def ozetle(recs, extras):
res = {}
for key in sorted({(r["kol"], r["mod"]) for r in recs}):
rs = [r for r in recs if (r["kol"], r["mod"]) == key]
d = {"n_adim": len(rs),
"dogru": round(float(np.mean([r["dogru"] for r in rs])), 3),
"gevsek": round(float(np.mean([r["gevsek"] for r in rs])), 3),
"hedef_listede": round(float(np.mean([r["listede"] for r in rs])), 3),
"null_orani": round(float(np.mean([r["null"] for r in rs])), 3)}
oks = [r["op_ok"] for r in rs if r.get("op_ok") is not None]
if oks:
d["islem_dogru"] = round(float(np.mean(oks)), 3)
cf = [r["conf"] for r in rs]
ok = [int(r["gevsek"]) for r in rs]
d["gevsek_risk_kapsama"] = risk_kapsama(cf, ok); d["gevsek_auroc"] = auroc(cf, ok)
d["ort_guven"] = round(float(np.mean(cf)), 3)
d["aday_ortanca"] = int(np.median([r["n_aday"] for r in rs]))
tm = [r["t_model"] for r in rs if "t_model" in r]
if tm:
d["sn_model_adim"] = round(float(np.mean(tm)), 2); d["sn_aday_ve_ekran_adim"] = round(float(np.mean([r["t_aday"] for r in rs if "t_aday" in r])), 2)
for ad, anahtar in (("tur", "tur"), ("stil", "stil")):
g = defaultdict(list)
for r in rs:
g[r[anahtar]].append(r["gevsek"])
d[f"gevsek_{ad}_kirilimi"] = {k: (len(v), round(float(np.mean(v)), 3)) for k, v in sorted(g.items(), key=lambda x: str(x[0]))}
g = defaultdict(list)
for r in rs:
g[r["form"]].append(r["gevsek"])
d["gevsek_form_kirilimi"] = {k: (len(v), round(float(np.mean(v)), 2)) for k, v in g.items()}
if key[1] == "serbest":
ex = [e for e in extras if e.get("kol") == key[0] and e.get("mod") == "serbest" and "alan_toplam" in e]
if ex:
d["serbest_alan_dogru"] = round(sum(e["alan_dogru"] for e in ex) / max(1, sum(e["alan_toplam"] for e in ex)), 3)
d["serbest_tam_form"] = f'{sum(e["tam_basari"] for e in ex)}/{len(ex)}'
d["serbest_kirlenen_alan_toplam"] = sum(e["kirlenen_alan"] for e in ex)
res[f"{key[0]}/{key[1]}"] = d
return res
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--run", default="", help="eğitilmiş koşu klasörü (lora/, head.pt, sonuc.json)")
ap.add_argument("--predictor", default="gemma", choices=["gemma", "tfidf"])
ap.add_argument("--out", default="sonuc_form_demo")
ap.add_argument("--arms", nargs="+", default=["sadik", "etiketli"])
ap.add_argument("--modes", nargs="+", default=["yonlendirmeli", "serbest"])
ap.add_argument("--K", type=int, default=160)
ap.add_argument("--temperature", type=float, default=1.0)
ap.add_argument("--no_remote", action="store_true")
ap.add_argument("--only", nargs="+", default=[], help="yalnız bu form adları")
ap.add_argument("--shots", default="demo_ekranlar")
args = ap.parse_args()
os.makedirs(args.out, exist_ok=True)
imgdir = os.path.join(args.out, "ornek_resimler"); os.makedirs(imgdir, exist_ok=True)
pagedir = os.path.join(args.out, "sayfalar")
specs = S.make_forms(pagedir)
if not args.no_remote:
specs += S.REMOTE
if args.only:
specs = [s for s in specs if s["name"] in args.only]
pred = Gemma(args.run, args.shots, args.temperature) if args.predictor == "gemma" else TfidfTaban()
pred.T = args.temperature
from playwright.sync_api import sync_playwright
srv, base = serve(pagedir)
all_recs, extras = [], []
budget = {"dogru": 6, "yanlis": 10}
with sync_playwright() as pw:
browser = pw.chromium.launch(headless=True, args=["--no-sandbox"])
for spec in specs:
for arm in args.arms:
for mode in args.modes:
t0 = time.time()
recs, ex = form_kos(browser, base, spec, pred, arm, mode, args, imgdir, budget)
ex.update(kol=arm, mod=mode)
all_recs += recs; extras.append(ex)
if "yuklenemedi" in ex:
print(f" {spec['name']:22s} {arm:8s} {mode:13s} SAYFA YÜKLENEMEDİ ({ex['yuklenemedi']})", flush=True)
break
ok = np.mean([r["gevsek"] for r in recs]) if recs else float("nan")
print(f" {spec['name']:22s} {arm:8s} {mode:13s} adım {len(recs):2d} | gevşek doğru {ok:.2f} | "
f"alan doğru {ex.get('alan_dogru')}/{ex.get('alan_toplam')} | kirlenen {ex.get('kirlenen_alan')} | {time.time() - t0:.0f}s", flush=True)
browser.close()
srv.shutdown()
ozet = ozetle(all_recs, extras)
json.dump({"ozet": ozet, "predictor": pred.ad, "adimlar": all_recs, "formlar": extras, "args": vars(args)},
open(os.path.join(args.out, "form_demo_sonuc.json"), "w"), indent=1, ensure_ascii=False, default=str)
for k, v in ozet.items():
print("\n==", k, "==")
for kk, vv in v.items():
print(f" {kk}: {vv}")
print("\nyazıldı:", args.out)
if __name__ == "__main__":
main()