1 files changed, 264 insertions, 0 deletions
diff --git a/worldalign/natural_families.py b/worldalign/natural_families.py
new file mode 100644
index 0000000..a13a1d9
--- /dev/null
+++ b/worldalign/natural_families.py
@@ -0,0 +1,264 @@
+"""Do factor families survive real language?
+
+The synthetic derivation found colour, count, and size families because
+templated phrases place exactly one member of each family in a fixed
+slot, so family members never co-occur. Real descriptions break every
+part of that: free word order, stacked adjectives, synonyms, and phrases
+that mention no attribute at all. This measures how much of the
+mutual-exclusivity signal survives, on Visual Genome region descriptions,
+using no lexicon and no labels.
+
+Families are recovered as low-co-occurrence, high-context-similarity
+groups: two words of one factor rarely modify the same head, and when
+they do appear they appear in the same distributional company. That is
+the paradigmatic relation of distributional semantics, and the greedy
+exclusivity pass of the synthetic pipeline is its degenerate case.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import re
+from collections import Counter, defaultdict
+from pathlib import Path
+
+import numpy as np
+
+from .common import read_json, seed_everything, write_json
+
+# Reference sets, used only to score the recovered families, never to find
+# them. Membership is checked after the fact.
+REFERENCE = {
+    "colour": {
+        "black", "white", "red", "blue", "green", "yellow", "brown", "gray",
+        "grey", "orange", "purple", "pink", "tan", "beige", "silver", "gold",
+        "golden", "dark", "light",
+    },
+    "number": {
+        "one", "two", "three", "four", "five", "six", "seven", "eight",
+        "nine", "ten", "a", "an", "the", "some", "many",
+    },
+    "size": {"small", "large", "big", "little", "tiny", "huge", "tall", "short", "long"},
+    "material": {"wooden", "metal", "plastic", "glass", "brick", "stone", "leather"},
+}
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--nodes", default="artifacts/vg_5k/text_nodes.jsonl")
+    parser.add_argument("--field", default="region_closed")
+    parser.add_argument("--min-count", type=int, default=200)
+    parser.add_argument("--max-vocabulary", type=int, default=300)
+    parser.add_argument("--context-window", type=int, default=2)
+    parser.add_argument("--exclusivity", type=float, default=0.25,
+                        help="Maximum observed-over-expected co-occurrence for "
+                             "two words to count as mutually exclusive.")
+    parser.add_argument(
+        "--method",
+        choices=["exclusivity", "distributional"],
+        default="distributional",
+        help="Mutual exclusivity is a template artifact; real paradigms are "
+        "found by distributional similarity, of which it is a special case.",
+    )
+    parser.add_argument("--components", type=int, default=64)
+    parser.add_argument("--clusters", type=int, default=30)
+    parser.add_argument("--min-family", type=int, default=3)
+    parser.add_argument("--seed", type=int, default=0)
+    parser.add_argument(
+        "--output", default="artifacts/vg_5k/natural_families.json"
+    )
+    return parser.parse_args()
+
+
+def load_phrases(path: Path, field: str) -> list[list[str]]:
+    phrases = []
+    for line in path.read_text(encoding="utf-8").splitlines():
+        if not line.strip():
+            continue
+        record = json.loads(line)
+        for phrase in record[field]:
+            tokens = re.findall(r"[a-z]+", phrase.lower())
+            if tokens:
+                phrases.append(tokens)
+    return phrases
+
+
+def statistics(
+    phrases: list[list[str]], vocabulary: set[str], window: int
+) -> tuple[Counter, Counter, dict[str, Counter]]:
+    unigram: Counter = Counter()
+    pair: Counter = Counter()
+    context: dict[str, Counter] = defaultdict(Counter)
+    for tokens in phrases:
+        present = [token for token in tokens if token in vocabulary]
+        for token in set(present):
+            unigram[token] += 1
+        for first in set(present):
+            for second in set(present):
+                if first < second:
+                    pair[(first, second)] += 1
+        for index, token in enumerate(tokens):
+            if token not in vocabulary:
+                continue
+            for offset in range(1, window + 1):
+                for neighbour_index in (index - offset, index + offset):
+                    if 0 <= neighbour_index < len(tokens):
+                        context[token][tokens[neighbour_index]] += 1
+    return unigram, pair, context
+
+
+def exclusivity_ratio(
+    first: str, second: str, unigram: Counter, pair: Counter, total: int
+) -> float:
+    """Observed co-occurrence over the independent expectation."""
+    expected = unigram[first] * unigram[second] / max(total, 1)
+    key = (first, second) if first < second else (second, first)
+    return pair[key] / max(expected, 1e-9)
+
+
+def context_similarity(first: Counter, second: Counter) -> float:
+    keys = set(first) | set(second)
+    a = np.array([first[k] for k in keys], dtype=float)
+    b = np.array([second[k] for k in keys], dtype=float)
+    a /= max(np.linalg.norm(a), 1e-9)
+    b /= max(np.linalg.norm(b), 1e-9)
+    return float(a @ b)
+
+
+def build_families(
+    words: list[str],
+    unigram: Counter,
+    pair: Counter,
+    context: dict[str, Counter],
+    total: int,
+    args: argparse.Namespace,
+) -> list[list[str]]:
+    """Greedy paradigmatic grouping: exclusive and distributionally alike."""
+    families: list[list[str]] = []
+    for word in words:
+        best_family, best_score = None, 0.0
+        for family in families:
+            ratios = [
+                exclusivity_ratio(word, member, unigram, pair, total)
+                for member in family
+            ]
+            if max(ratios) > args.exclusivity:
+                continue
+            similarity = float(
+                np.mean(
+                    [context_similarity(context[word], context[member])
+                     for member in family]
+                )
+            )
+            if similarity > best_score:
+                best_family, best_score = family, similarity
+        if best_family is not None and best_score > 0.15:
+            best_family.append(word)
+        else:
+            families.append([word])
+    return [family for family in families if len(family) >= args.min_family]
+
+
+def distributional_families(
+    words: list[str], context: dict[str, Counter], args: argparse.Namespace
+) -> list[list[str]]:
+    """Word classes from context vectors: positive PMI, truncated SVD, k-means.
+
+    The standard recipe of distributional word-class induction. Words of
+    one factor modify the same heads and so share company, whether or not
+    they exclude each other -- real colour terms co-occur freely.
+    """
+    from sklearn.cluster import KMeans
+
+    features = sorted({key for word in words for key in context[word]})
+    index = {key: position for position, key in enumerate(features)}
+    matrix = np.zeros((len(words), len(features)))
+    for row, word in enumerate(words):
+        for key, value in context[word].items():
+            matrix[row, index[key]] = value
+    total = matrix.sum()
+    row_sum = matrix.sum(1, keepdims=True)
+    column_sum = matrix.sum(0, keepdims=True)
+    expected = row_sum * column_sum / max(total, 1e-9)
+    pmi = np.log(np.maximum(matrix, 1e-9) / np.maximum(expected, 1e-9))
+    pmi[matrix == 0] = 0.0
+    pmi = np.maximum(pmi, 0.0)
+    left, values, _ = np.linalg.svd(pmi, full_matrices=False)
+    embedding = left[:, : args.components] * values[: args.components]
+    embedding /= np.linalg.norm(embedding, axis=1, keepdims=True).clip(1e-9)
+    labels = KMeans(args.clusters, n_init=10, random_state=args.seed).fit_predict(
+        embedding
+    )
+    families: dict[int, list[str]] = defaultdict(list)
+    for word, label in zip(words, labels):
+        families[int(label)].append(word)
+    return [family for family in families.values() if len(family) >= args.min_family]
+
+
+def label_family(family: list[str]) -> tuple[str, float]:
+    best_name, best_purity = "unlabelled", 0.0
+    for name, reference in REFERENCE.items():
+        purity = sum(1 for word in family if word in reference) / len(family)
+        if purity > best_purity:
+            best_name, best_purity = name, purity
+    return best_name, best_purity
+
+
+def main() -> None:
+    args = parse_args()
+    seed_everything(args.seed)
+    phrases = load_phrases(Path(args.nodes), args.field)
+    counts = Counter(token for tokens in phrases for token in set(tokens))
+    vocabulary = [
+        word
+        for word, count in counts.most_common(args.max_vocabulary)
+        if count >= args.min_count
+    ]
+    unigram, pair, context = statistics(phrases, set(vocabulary), args.context_window)
+    if args.method == "distributional":
+        families = distributional_families(vocabulary, context, args)
+    else:
+        families = build_families(
+            vocabulary, unigram, pair, context, len(phrases), args
+        )
+
+    described = []
+    for family in sorted(families, key=len, reverse=True):
+        name, purity = label_family(family)
+        described.append(
+            {
+                "size": len(family),
+                "label": name,
+                "purity": purity,
+                "words": sorted(family, key=lambda w: -unigram[w])[:14],
+            }
+        )
+    report = {
+        "protocol": (
+            "Families are recovered from region descriptions alone by "
+            "mutual exclusivity plus distributional similarity. Reference "
+            "word sets are read only to label the result."
+        ),
+        "phrases": len(phrases),
+        "vocabulary": len(vocabulary),
+        "families": described,
+        "recovered_labels": Counter(item["label"] for item in described),
+    }
+    for item in described[:12]:
+        print(
+            json.dumps(
+                {
+                    "label": item["label"],
+                    "purity": round(item["purity"], 2),
+                    "size": item["size"],
+                    "words": item["words"][:10],
+                }
+            )
+        )
+    write_json(args.output, report)
+    print(f"Wrote {args.output}")
+
+
+if __name__ == "__main__":
+    main()