| """Requ Classify Small: reads a company's website and labels it. |
| |
| Gives the industry (64 classes), a type for each menu item and heading (offering, team, activity, |
| audience, website, other), a customer-journey stage for each page, and sentences that show a known |
| business process. Needs numpy, onnxruntime, tokenizers and safetensors. |
| |
| python requ_classify.py < site.json |
| |
| site.json: {"name", "description", "pages": [{"url", "title", "text", "items": [{"text", "kind"}]}]} |
| where an item's kind is one of: menu, footer, h1, h2, h3, button. |
| """ |
| import json, os, re, sys |
| from urllib.parse import urlsplit |
|
|
| import numpy as np |
| from safetensors.numpy import load_file |
|
|
| HERE = os.path.dirname(os.path.abspath(__file__)) |
| KINDS = ['menu', 'footer', 'h1', 'h2', 'h3', 'button'] |
| SENT = re.compile(r'(?<=[.!?])\s+') |
|
|
|
|
| class Embedder: |
| """bge-small-en-v1.5 (quantised ONNX), CLS vector, normalised.""" |
|
|
| def __init__(self, folder): |
| import onnxruntime as ort |
| from tokenizers import Tokenizer |
| opts = ort.SessionOptions() |
| opts.intra_op_num_threads = int(os.environ.get('REQU_THREADS', '2')) |
| self.session = ort.InferenceSession(os.path.join(folder, 'model_quantized.onnx'), opts, providers=['CPUExecutionProvider']) |
| self.tok = Tokenizer.from_file(os.path.join(folder, 'tokenizer.json')) |
| self.tok.enable_truncation(max_length=256) |
| self.inputs = {i.name for i in self.session.get_inputs()} |
|
|
| def __call__(self, texts, batch=64): |
| out = [] |
| for start in range(0, len(texts), batch): |
| self.tok.enable_padding(length=None) |
| enc = self.tok.encode_batch(texts[start:start + batch]) |
| ids = np.array([e.ids for e in enc], dtype=np.int64) |
| mask = np.array([e.attention_mask for e in enc], dtype=np.int64) |
| feed = {'input_ids': ids, 'attention_mask': mask} |
| if 'token_type_ids' in self.inputs: |
| feed['token_type_ids'] = np.zeros_like(ids) |
| cls = self.session.run(None, feed)[0][:, 0] |
| out.append(cls / np.linalg.norm(cls, axis=1, keepdims=True)) |
| return np.vstack(out) if out else np.zeros((0, 384), dtype=np.float32) |
|
|
|
|
| def path_of(url): |
| return urlsplit(url).path.strip('/').replace('-', ' ').replace('/', ' / ') or 'home' |
|
|
|
|
| def site_text(site): |
| items = [i for p in site['pages'] for i in p.get('items', [])] |
| heads = [i['text'] for i in items if i['kind'] in ('h1', 'h2')][:25] |
| menu = [i['text'] for i in items if i['kind'] == 'menu'][:30] |
| home = site['pages'][0]['text'][:800] if site['pages'] else '' |
| return f"{site.get('name', '')}. {site.get('description', '')} Pages: {'; '.join(p.get('title', '') for p in site['pages'])}. Menu: {', '.join(menu)}. Headings: {', '.join(heads)}. {home}" |
|
|
|
|
| def site_pieces(site): |
| items = [i for p in site['pages'] for i in p.get('items', [])] |
| out = [f"{site.get('name', '')}. {site.get('description', '')}", |
| 'Menu: ' + ', '.join(i['text'] for i in items if i['kind'] == 'menu')[:600], |
| 'Headings: ' + ', '.join(i['text'] for i in items if i['kind'] in ('h1', 'h2', 'h3'))[:600]] |
| out += [f"{p.get('title', '')}. {p['text'][:500]}" for p in site['pages'][:6]] |
| return [x for x in out if len(x) > 12] |
|
|
|
|
| def sentences(page, limit=60): |
| out = [s.strip() for s in SENT.split(page['text']) if 6 <= len(s.split()) <= 45] |
| return out[:limit] |
|
|
|
|
| def softmax(z): |
| e = np.exp(z - z.max(axis=-1, keepdims=True)) |
| return e / e.sum(axis=-1, keepdims=True) |
|
|
|
|
| class RequClassify: |
| def __init__(self, folder=HERE): |
| self.embed = Embedder(os.path.join(folder, 'model')) |
| w = load_file(os.path.join(folder, 'heads.safetensors')) |
| cfg = json.load(open(os.path.join(folder, 'config.json'))) |
| self.heads = {n: (w[f'{n}.weight'], w[f'{n}.bias'], cfg['classes'][n]) for n in ('industry', 'item_type', 'page_stage')} |
| self.proof = (w['proof.weight'], w['proof.bias']) |
| self.memory = w['memory'].astype(np.float32) |
| self.memory_keys = np.array(cfg['memory_keys']) |
| self.vectors = w['library_vectors'] |
| self.t = cfg['thresholds'] |
| self.library = json.load(open(os.path.join(folder, 'library.json')))['industries'] |
|
|
| def head(self, name, X): |
| coef, bias, classes = self.heads[name] |
| return softmax(X @ coef.T + bias), classes |
|
|
| def __call__(self, site): |
| pages = site['pages'] |
| X = np.hstack([self.embed([site_text(site)])[0], self.embed(site_pieces(site)).mean(0)])[None, :] |
| p, classes = self.head('industry', X) |
| order = np.argsort(-p[0]) |
| industry = classes[order[0]] |
| result = {'industry': {'id': industry, 'p': round(float(p[0][order[0]]), 3), 'sure': bool(p[0][order[0]] >= self.t['industry_gate']), |
| 'top3': [[classes[i], round(float(p[0][i]), 3)] for i in order[:3]]}} |
| flat = [(pi, it) for pi, pg in enumerate(pages) for it in pg.get('items', [])] |
| result['items'] = [] |
| if flat: |
| E = self.embed([f"{it['text']} | {it['kind']} | {path_of(pages[pi]['url'])}" for pi, it in flat]) |
| K = np.array([[1.0 if it['kind'] == k else 0.0 for k in KINDS] for _, it in flat]) |
| q, tclasses = self.head('item_type', np.hstack([E, K])) |
| result['items'] = [{'page': pi, 'text': it['text'], 'kind': it['kind'], 'type': tclasses[int(r.argmax())], 'p': round(float(r.max()), 3)} for (pi, it), r in zip(flat, q)] |
| if pages: |
| s, sclasses = self.head('page_stage', self.embed([f"{path_of(pg['url'])}. {pg.get('title', '')}. {pg['text'][:400]}" for pg in pages])) |
| result['pageStages'] = [sclasses[int(r.argmax())] for r in s] |
| |
| |
| entry = self.library[industry] |
| P = self.vectors[entry['rows']] |
| keys = [pid if pid.startswith('s-') else f'{industry}:{pid}' for pid in entry['ids']] |
| found = [] |
| sents = [(pi, s) for pi, pg in enumerate(pages) for s in sentences(pg)] |
| if sents: |
| S = self.embed([s for _, s in sents]) |
| keep = (1 / (1 + np.exp(-(S @ self.proof[0].T + self.proof[1]))))[:, 0] |
| chosen = np.where(keep >= self.t['match_threshold'])[0] |
| if len(chosen): |
| S = S[chosen] |
| score = S @ P.T |
| simM = S @ self.memory.T |
| for j, k in enumerate(keys): |
| cols = np.where(self.memory_keys == k)[0] |
| if len(cols): |
| score[:, j] = 0.4 * score[:, j] + 0.6 * np.sort(simM[:, cols], axis=1)[:, -3:].mean(axis=1) |
| for row, i in enumerate(chosen): |
| o = np.argsort(-score[row]) |
| pi, s = sents[i] |
| clear = bool(score[row][o[0]] - score[row][o[1]] >= self.t['process_margin']) if len(o) > 1 else True |
| found.append({'process': entry['ids'][int(o[0])], 'clear': clear, 'page': pi, 'quote': s, 'score': round(float(keep[i]), 3)}) |
| found.sort(key=lambda e: -e['score']) |
| best = {} |
| for e in found: |
| best.setdefault(e['process'], []).append(e) |
| result['evidence'] = [x for v in best.values() for x in v[:2]] |
| return result |
|
|
|
|
| if __name__ == '__main__': |
| json.dump(RequClassify()(json.load(sys.stdin)), sys.stdout, indent=1) |
|
|
