File size: 5,879 Bytes
795f737 | 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 123 124 125 | """Read-only offline grounding diagnostic; never executes downloaded HTML."""
import argparse
from collections import Counter
from dataclasses import asdict
import hashlib
from html.parser import HTMLParser
import json
from pathlib import Path
import random
import torch
from .features import encode
from .policy import LearnedPolicy
from .state import Element
from .train import logits
class TextIndex(HTMLParser):
def __init__(self):
super().__init__(convert_charrefs=True)
self.stack = []
self.nodes = {}
def handle_starttag(self,tag,attrs):
attrs = dict(attrs)
node = dict(tag=tag,attrs=attrs,text=[])
if attrs.get('backend_node_id'):
self.nodes[attrs['backend_node_id']] = node
if tag not in {'area','base','br','col','embed','hr','img','input','link','meta','param','source','track','wbr'}:
self.stack.append(node)
def handle_endtag(self,tag):
for index in range(len(self.stack)-1,-1,-1):
if self.stack[index]['tag'] == tag:
del self.stack[index:]
break
def handle_data(self,data):
if any(node['tag'] in {'script','style'} for node in self.stack):
return
for node in self.stack:
if sum(map(len,node['text'])) < 512:
node['text'].append(data[:512])
def normalize_task(task):
for step_index, step in enumerate(task['actions']):
positives = step['pos_candidates']
if not positives:
yield None
continue
parser = TextIndex()
parser.feed(step['cleaned_html'])
entries = [(candidate,True) for candidate in positives] + [(candidate,False) for candidate in step['neg_candidates']]
rng = random.Random(int(hashlib.sha256(step['action_uid'].encode()).hexdigest(),16))
rng.shuffle(entries) # Positive-first ordering must not leak the answer.
elements, targets = [], []
seen = set()
for candidate,positive in entries:
ident = str(candidate['backend_node_id'])
if ident in seen:
continue
seen.add(ident)
attrs = json.loads(candidate['attributes'])
node = parser.nodes.get(ident,{})
html_attrs = node.get('attrs',{})
attrs = {**html_attrs,**attrs}
tag = candidate['tag'].lower()
role = attrs.get('role') or {'button':'button','a':'link','input':'textbox',
'textarea':'textbox','select':'combobox'}.get(tag,'generic')
if tag=='input':
role = {'checkbox':'checkbox','radio':'radio','submit':'button','button':'button'}.get(attrs.get('type'),role)
if role=='generic' and str(attrs.get('is_clickable','')).lower() in {'true','1'}:
role='button'
sensitive = attrs.get('type')=='password' or 'cc-' in attrs.get('autocomplete','')
name = attrs.get('aria-label') or attrs.get('placeholder') or attrs.get('title') or ' '.join(node.get('text',[]))
name = ' '.join(str(name).split())[:512]
index = len(elements)
elements.append(asdict(Element(f'e{index}',role,'[REDACTED]' if sensitive else name,
str(index),enabled='disabled' not in attrs,sensitive=sensitive)))
if positive:
targets.append(index)
if len(targets)!=1:
yield None
continue
operation = {'CLICK':'C','TYPE':'T','SELECT':'O'}.get(step['operation']['op'])
if operation is None:
yield None
continue
yield dict(goal=task['confirmed_task'],elements=elements,action=operation,target=targets[0],
history=task.get('action_reprs',[])[:step_index])
def main():
parser = argparse.ArgumentParser()
parser.add_argument('--source',required=True)
parser.add_argument('--checkpoint',default='models/v000-mean')
parser.add_argument('--output',default='reports/mind2web-smoke-v000.json')
args = parser.parse_args()
torch.set_num_threads(2)
tasks = json.loads(Path(args.source).read_text(encoding='utf-8'))
all_rows = [row for task in tasks for row in normalize_task(task)]
rows = [row for row in all_rows if row is not None]
policy = LearnedPolicy(args.checkpoint)
inputs,actions,targets,_ = encode(rows,policy.vocab)
a,t = logits(policy.model,inputs)
report = dict(source='osunlp/Mind2Web',revision='6314166657eec4aa0e22c00f8d801e609ce8e80f',
file=Path(args.source).name,source_sha256=hashlib.sha256(Path(args.source).read_bytes()).hexdigest(),
license='CC-BY-4.0',attribution='Deng et al., Mind2Web: Towards a Generalist Agent for the Web, 2023, arXiv:2306.06070',
checkpoint=args.checkpoint,tasks=len(tasks),websites=len({task['website'] for task in tasks}),
total_steps=len(all_rows),scorable_steps=len(rows),unscorable_steps=len(all_rows)-len(rows),
operations=dict(Counter(row['action'] for row in rows)),
candidate_recall=float((targets>=0).float().mean()),
action_accuracy=float((a.argmax(-1)==actions).float().mean()),
target_accuracy=float((t.argmax(-1)==targets).float().mean()),
joint_accuracy=float(((a.argmax(-1)==actions)&(t.argmax(-1)==targets)).float().mean()),
browser_execution=False,training_on_source=False,
limitations='Small training-shard diagnostic, NOT official held-out benchmark. Approximate HTML names/roles; no history; 24-token goal truncation. No raw page or goal text persisted in report.')
Path(args.output).parent.mkdir(parents=True,exist_ok=True)
Path(args.output).write_text(json.dumps(report,indent=2),encoding='utf-8')
print(json.dumps(report,indent=2))
if __name__=='__main__':
main()
|