File size: 5,382 Bytes
6fd3740 | 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 126 127 128 129 130 | """Data audit (v3 S0): how much argument mass can be copied from the query, picked from options, or is a list."""
from __future__ import annotations
import os
from collections import Counter, defaultdict
from .schema import param_options, value_to_text
def query_offsets(tok, query: str) -> list[tuple[int, int]]:
"""Char span per query token, with a leading space (BPE 'Ġ') trimmed off."""
offs = tok(query, add_special_tokens=False, return_offsets_mapping=True).offset_mapping
out = []
for a, b in offs:
while a < b and query[a].isspace():
a += 1
out.append((a, b))
return out
def copy_span(tok, query: str, text: str, offs=None):
"""First case-insensitive, token-aligned occurrence of `text` in `query` -> (start_tok, end_tok) or None."""
text = str(text).strip()
if not text:
return None
offs = offs if offs is not None else query_offsets(tok, query)
starts = {a: i for i, (a, b) in reversed(list(enumerate(offs))) if b > a}
ends = {b: i for i, (a, b) in enumerate(offs) if b > a}
low, t = query.lower(), text.lower()
at = low.find(t)
while at != -1:
s, e = starts.get(at), ends.get(at + len(t))
if s is not None and e is not None and e >= s:
return s, e
at = low.find(t, at + 1)
return None
def _kind(v) -> str:
if isinstance(v, bool):
return "bool"
if isinstance(v, (int, float)):
return "num"
if isinstance(v, str):
return "str"
if isinstance(v, list):
return "list"
if isinstance(v, dict):
return "dict"
return "other"
def audit(tok, examples) -> dict:
"""Per-source counters over every gold argument value."""
stats: dict[str, Counter] = defaultdict(Counter)
for ex in examples:
c = stats[ex.source.split("_hardneg")[0].split("_neg")[0] if ex.source.startswith("xlam") else ex.source]
if not ex.calls:
continue
offs = query_offsets(tok, ex.query)
by_name = {t.name: t for t in ex.tools}
for call in ex.calls:
tool = by_name.get(call.name)
if tool is None:
continue
pmap = {p.name: p for p in tool.params}
for name, v in call.args.items():
p = pmap.get(name)
k = _kind(v)
c["args"] += 1
c[f"type_{k}"] += 1
n_tok = len(tok(value_to_text(v), add_special_tokens=False).input_ids)
if n_tok > 32:
c["over_32_tokens"] += 1
opts = param_options(p) if p else None
if opts is not None:
c["has_options"] += 1
if any(value_to_text(o).lower() == value_to_text(v).lower() for o in opts):
c["gold_in_options"] += 1
if k in ("str", "num"):
c[f"{k}_total"] += 1
if copy_span(tok, ex.query, value_to_text(v), offs) is not None:
c[f"{k}_copyable"] += 1
if k == "list":
c["list_items"] += len(v)
if len(v) > 7:
c["list_over_7"] += 1
for it in v:
c["list_item_total"] += 1
if len(tok(value_to_text(it), add_special_tokens=False).input_ids) > 8:
c["list_item_over_8_tokens"] += 1
if isinstance(it, (str, int, float)) and not isinstance(it, bool) and \
copy_span(tok, ex.query, value_to_text(it), offs) is not None:
c["list_item_copyable"] += 1
return stats
def _pct(a, b):
return f"{100 * a / b:.1f}%" if b else "-"
def report(stats: dict) -> str:
rows = ["| source | args | str copyable | num copyable | bool/enum (options) | gold in options | list args | "
"list items copyable | item >8 tok | list >7 | value >32 tok |", "|" + "---|" * 11]
for src in sorted(stats):
c = stats[src]
if not c["args"]:
continue
rows.append(f"| {src} | {c['args']} | {_pct(c['str_copyable'], c['str_total'])} of {c['str_total']} | "
f"{_pct(c['num_copyable'], c['num_total'])} of {c['num_total']} | "
f"{_pct(c['has_options'], c['args'])} | {_pct(c['gold_in_options'], c['has_options'])} | "
f"{_pct(c['type_list'], c['args'])} | {_pct(c['list_item_copyable'], c['list_item_total'])} | "
f"{_pct(c['list_item_over_8_tokens'], c['list_item_total'])} | {c['list_over_7']} | "
f"{_pct(c['over_32_tokens'], c['args'])} |")
return "\n".join(rows)
if __name__ == "__main__":
from .data import load_jsonl
from .features import load_tokenizer
tok = load_tokenizer()
parts = []
for f in ("train_v2.jsonl", "test.jsonl", "bfcl.jsonl"):
st = audit(tok, load_jsonl(os.path.join("data/prepared", f)))
parts.append(f"## {f}\n\n{report(st)}\n")
out = "# v3 data audit (S0)\n\nShares are over gold argument values. 'copyable' = the value text occurs in the " \
"query, case-insensitively, on token boundaries.\n\n" + "\n".join(parts)
open("docs/results/audit.md", "w", encoding="utf-8").write(out)
print(out)
|