inference_mask_onnx.py

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import json
import os

os.environ.setdefault("TRANSFORMERS_NO_TF", "1")
os.environ.setdefault("TRANSFORMERS_NO_FLAX", "1")
os.environ.setdefault("TRANSFORMERS_NO_TORCHVISION", "1")
os.environ["USE_TF"] = "0"
os.environ["USE_FLAX"] = "0"
os.environ["USE_TORCH"] = "1"

import numpy as np

from common import (
    boundary_label_thresholds_from_config,
    decode_token_presence_segments,
    label_max_span_tokens_from_config,
    label_min_nonspace_chars_from_config,
    label_names_from_config,
    load_onnx_session,
    run_onnx_all,
    token_extend_thresholds_from_config,
    token_label_thresholds_from_config,
)


def replacement(label: str) -> str:
    return f"[PII:{label}]"


def mask_text(text: str, spans: list[dict]) -> str:
    out = text
    for span in sorted(spans, key=lambda item: (item["start"], item["end"]), reverse=True):
        out = out[: span["start"]] + replacement(span["label"]) + out[span["end"] :]
    return out


def predict(text: str, session, tokenizer, config, min_score: float):
    encoded = tokenizer(text, return_offsets_mapping=True, return_tensors="np", truncation=True)
    offsets = [tuple(item) for item in encoded["offset_mapping"][0].tolist()]
    token_logits, start_logits, end_logits = run_onnx_all(session, encoded)
    token_scores = 1.0 / (1.0 + np.exp(-token_logits[0]))
    start_scores = 1.0 / (1.0 + np.exp(-start_logits[0]))
    end_scores = 1.0 / (1.0 + np.exp(-end_logits[0]))
    label_names = label_names_from_config(config)
    thresholds = token_label_thresholds_from_config(config, min_score)
    extend_thresholds = token_extend_thresholds_from_config(config)
    max_span_tokens = label_max_span_tokens_from_config(config)
    min_nonspace_chars = label_min_nonspace_chars_from_config(config)
    boundary_thresholds = boundary_label_thresholds_from_config(config)
    spans = decode_token_presence_segments(
        text,
        offsets,
        token_scores,
        label_names,
        min_score,
        thresholds,
        extend_thresholds,
        max_span_tokens,
        min_nonspace_chars,
        boundary_thresholds,
        start_scores=start_scores,
        end_scores=end_scores,
    )
    for span in spans:
        span["replacement"] = replacement(span["label"])
    return spans


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--model", required=True)
    parser.add_argument("--text", required=True)
    parser.add_argument("--min-score", type=float, default=0.5)
    parser.add_argument("--json", action="store_true")
    args = parser.parse_args()

    session, tokenizer, config = load_onnx_session(args.model, onnx_file="model_quantized.onnx", onnx_subfolder="onnx")
    spans = predict(args.text, session, tokenizer, config, args.min_score)
    result = {
        "model": args.model,
        "backend": "onnx_token_span_q8",
        "min_score": args.min_score,
        "spans": spans,
        "masked_text": mask_text(args.text, spans),
    }
    if args.json:
        print(json.dumps(result, indent=2, ensure_ascii=False))
    else:
        print(result["masked_text"])


if __name__ == "__main__":
    main()