Spaces:
Sleeping
Sleeping
| """A map of what the literature is missing, built from the indexed corpus alone. | |
| Retrieval answers "what exists". This module answers the complementary question: | |
| for a region of the embedding space, *why* is nothing there. Three kinds of absence | |
| are distinguishable from the fields the index already stores and are not the same | |
| thing at all: | |
| - **null-saturated** — the work was done and came back null or inconclusive. An idea | |
| landing here is not novel, it is refuted, and the refutations are usually | |
| uncited, which is exactly why a reader (or a language model summarising the top | |
| ten hits) concludes the question is open. | |
| - **dark** — trials completed, results never reported. Absence of a paper, not | |
| absence of evidence. | |
| - **gap** — no work between two populated regions. A combination the neighbouring | |
| literature implies but nobody has run. | |
| Regions are clusters of real documents, so they can only describe where work exists. | |
| Gaps therefore come from *interpolation*: the midpoint between two nearby region | |
| centroids is a point the literature implies but may never have occupied, and its | |
| cosine to the nearest real paper measures how unoccupied it is. Nothing here calls | |
| an LLM or Elasticsearch, and no score is a probability: every number is a count, a | |
| share of counts, or a cosine. | |
| """ | |
| from __future__ import annotations | |
| from collections import Counter | |
| import numpy as np | |
| MAP_VERSION = "gapmap-v1" | |
| # A region needs this many primary studies before its verdict mix means anything; | |
| # below it the honest label is "thin", not a finding about the science. | |
| MIN_REGION_ATTEMPTS = 5 | |
| # Rows per chunk when multiplying the corpus against the centroids. Bounds the | |
| # intermediate to a few tens of megabytes whatever the corpus size. | |
| ASSIGN_CHUNK = 50_000 | |
| # Share of attempts that must be null (or unreported) before the region is called | |
| # saturated (or dark). Both are deliberately majority-ish rather than "any". | |
| NULL_SATURATED_SHARE = 0.5 | |
| DARK_SHARE = 0.34 | |
| # Both directions must be well represented for a region to count as contested. | |
| CONTESTED_SHARE = 0.25 | |
| # Two regions are neighbours, and so worth interpolating between, when their | |
| # centroid cosine is in the top decile of all region pairs; below that the | |
| # midpoint is empty for the boring reason that the two literatures are unrelated. | |
| # A quantile rather than a fixed cosine because what counts as "close" is a | |
| # property of the embedding model and of how narrow the indexed corpus is: on | |
| # MiniLM, unrelated clinical topics sit around 0.1-0.3, so any constant either | |
| # admits everything in a single-disease index or nothing in a broad one. The | |
| # floor stops a corpus of entirely unrelated clusters from producing neighbours | |
| # by ranking alone. | |
| NEIGHBOUR_QUANTILE = 0.9 | |
| NEIGHBOUR_FLOOR = 0.35 | |
| # The band between two regions is open when it holds at most this share of the | |
| # papers the thinner of the two parent cores holds. A ratio, because an absolute | |
| # cosine cannot express "between": every midpoint is close to its own parents. | |
| GAP_OCCUPANCY = 0.1 | |
| # Below this share of readable outcomes a region says nothing about the science, | |
| # only about what the index could read. | |
| READABLE_SHARE = 0.1 | |
| REGION_LABELS = ("active", "contested", "null_saturated", "dark", "unread", "thin") | |
| NULL_BUCKETS = ("credible_null", "reported_null") | |
| def _vector(document: dict) -> np.ndarray | None: | |
| embedding = document.get("embedding") | |
| if not isinstance(embedding, (list, tuple)) or not embedding: | |
| return None | |
| vector = np.asarray(embedding, dtype=np.float32) | |
| return vector if np.isfinite(vector).all() and vector.any() else None | |
| def normalize(vectors: np.ndarray) -> np.ndarray: | |
| norms = np.linalg.norm(vectors, axis=-1, keepdims=True) | |
| return vectors / np.clip(norms, 1e-8, None) | |
| def seed_centroids(vectors: np.ndarray, k: int, seed: int = 0) -> np.ndarray: | |
| """k-means++ seeding over a cosine space, deterministic for a given seed.""" | |
| rng = np.random.default_rng(seed) | |
| count = len(vectors) | |
| k = max(1, min(k, count)) | |
| centroids = np.empty((k, vectors.shape[1]), dtype=np.float32) | |
| centroids[0] = vectors[rng.integers(count)] | |
| for index in range(1, k): | |
| distance = 1.0 - (vectors @ centroids[:index].T).max(axis=1) | |
| distance = np.clip(distance, 0.0, None) ** 2 | |
| total = distance.sum() | |
| pick = rng.integers(count) if total <= 0 else rng.choice(count, p=distance / total) | |
| centroids[index] = vectors[pick] | |
| return centroids | |
| def assign(vectors: np.ndarray, centroids: np.ndarray, chunk: int = ASSIGN_CHUNK) -> np.ndarray: | |
| """Nearest centroid per row, in chunks so a corpus-sized matrix is never built. | |
| ``vectors @ centroids.T`` over two million rows and eighty centroids is a | |
| 640 MB intermediate; the chunked loop is the same arithmetic in 16 MB pieces. | |
| """ | |
| if not len(vectors): | |
| return np.zeros(0, dtype=np.int64) | |
| labels = np.empty(len(vectors), dtype=np.int64) | |
| for start in range(0, len(vectors), chunk): | |
| stop = start + chunk | |
| labels[start:stop] = (vectors[start:stop] @ centroids.T).argmax(axis=1) | |
| return labels | |
| def _recentre(vectors: np.ndarray, labels: np.ndarray, centroids: np.ndarray) -> np.ndarray: | |
| """Move each centroid onto the mean direction of the rows assigned to it. | |
| Empty clusters keep their previous position rather than being reseeded, so a | |
| warm-started run cannot make the map jump between successive iterations. | |
| """ | |
| moved = centroids.copy() | |
| order = np.argsort(labels, kind="stable") | |
| bounds = np.searchsorted(labels[order], np.arange(len(centroids) + 1)) | |
| for index in range(len(centroids)): | |
| members = order[bounds[index] : bounds[index + 1]] | |
| if len(members): | |
| moved[index] = vectors[members].mean(axis=0) | |
| return normalize(moved) | |
| def kmeans_steps( | |
| vectors: np.ndarray, | |
| k: int, | |
| seed: int = 0, | |
| iterations: int = 40, | |
| centroids: np.ndarray | None = None, | |
| ): | |
| """Spherical k-means, yielding ``(centroids, labels)`` after every iteration. | |
| Yielding the intermediate states is what lets a caller show the map settling | |
| instead of a spinner: each yielded pair is a real assignment of the corpus, | |
| not an interpolation. Deterministic for a given seed and starting position, | |
| and stops early once the assignment stops changing. | |
| """ | |
| if not len(vectors): | |
| return | |
| if centroids is None: | |
| centroids = seed_centroids(vectors, k, seed) | |
| centroids = normalize(np.asarray(centroids, dtype=np.float32)) | |
| labels = np.full(len(vectors), -1, dtype=np.int64) | |
| for _ in range(max(1, iterations)): | |
| assigned = assign(vectors, centroids) | |
| settled = np.array_equal(assigned, labels) | |
| labels = assigned | |
| centroids = _recentre(vectors, labels, centroids) | |
| yield centroids, labels | |
| if settled: | |
| return | |
| def kmeans(vectors: np.ndarray, k: int, seed: int = 0, iterations: int = 40) -> np.ndarray: | |
| """Spherical k-means with k-means++ seeding, returning a label per row. | |
| Deterministic for a given seed so a map can be rebuilt and compared. Written | |
| out here rather than pulled from scikit-learn because the ingest extra that | |
| provides it is optional and the serving path must not depend on it. | |
| """ | |
| labels = np.zeros(len(vectors), dtype=np.int64) | |
| for _, labels in kmeans_steps(vectors, k, seed=seed, iterations=iterations): | |
| pass | |
| return labels | |
| class RunningClusters: | |
| """Mini-batch spherical k-means: centroids that exist before the scan ends. | |
| The full corpus cannot be clustered until it has all arrived, but a map that | |
| only appears at the end cannot be watched. Each batch is assigned to the | |
| current centroids and each centroid is moved towards the mean of what it | |
| just received, weighted by how much it has seen — the standard mini-batch | |
| update. These centroids are provisional and are replaced by full-corpus | |
| k-means once the scan finishes; nothing is reported as final until then. | |
| """ | |
| def __init__(self, k: int, seed: int = 0): | |
| self.k = max(1, k) | |
| self.seed = seed | |
| self.centroids: np.ndarray | None = None | |
| self.counts: np.ndarray | None = None | |
| def update(self, vectors: np.ndarray) -> np.ndarray | None: | |
| if not len(vectors): | |
| return self.centroids | |
| if self.centroids is None: | |
| if len(vectors) < self.k: | |
| return None # too few rows to seed k distinct centroids honestly | |
| self.centroids = normalize(seed_centroids(vectors, self.k, self.seed)) | |
| self.counts = np.zeros(len(self.centroids), dtype=np.float64) | |
| labels = assign(vectors, self.centroids) | |
| sums = np.zeros_like(self.centroids, dtype=np.float32) | |
| np.add.at(sums, labels, vectors) | |
| counts = np.bincount(labels, minlength=len(self.centroids)).astype(np.float64) | |
| assert self.counts is not None | |
| seen = self.counts + counts | |
| rate = np.divide(counts, seen, out=np.zeros_like(seen), where=seen > 0)[:, None] | |
| self.centroids = normalize( | |
| self.centroids * (1 - rate) | |
| + np.divide(sums, np.clip(counts, 1, None)[:, None]) * rate | |
| ) | |
| self.counts = seen | |
| return self.centroids | |
| def fit_projection(vectors: np.ndarray) -> tuple[np.ndarray, np.ndarray]: | |
| """A deterministic 2-D basis for drawing the map: PCA over centered vectors. | |
| Returns ``(mean, basis)`` so centroids and idea vectors can be projected | |
| through the same fit via :func:`project_with`. SVD is deterministic for a | |
| given input, but the sign of each component is arbitrary, so every axis is | |
| flipped until its largest-magnitude weight is positive — without that the | |
| same map could render mirrored between builds. | |
| """ | |
| mean = vectors.mean(axis=0, keepdims=True).astype(np.float32) | |
| _, _, vt = np.linalg.svd(vectors - mean, full_matrices=False) | |
| basis = np.zeros((2, vectors.shape[1]), dtype=np.float32) | |
| rows = min(2, len(vt)) # a single document yields only one axis | |
| basis[:rows] = vt[:rows] | |
| for component in basis: | |
| if component[np.argmax(np.abs(component))] < 0: | |
| component *= -1 | |
| return mean, basis | |
| def project_with(mean: np.ndarray, basis: np.ndarray, vectors: np.ndarray) -> np.ndarray: | |
| """Project ``vectors`` through a basis fitted by :func:`fit_projection`.""" | |
| return (np.atleast_2d(np.asarray(vectors, dtype=np.float32)) - mean) @ basis.T | |
| def neighbor_edges( | |
| vectors: np.ndarray, | |
| start: int = 0, | |
| neighbors: int = 3, | |
| floor: float = 0.0, | |
| chunk: int = 1024, | |
| ) -> list[tuple[int, int, float]]: | |
| """Cosine-nearest-neighbor edges among unit ``vectors``. | |
| Each row from ``start`` on is joined to its ``neighbors`` most similar rows | |
| anywhere in the matrix, provided the cosine reaches ``floor``; rows before | |
| ``start`` already had their turn on an earlier call, so a stream can call | |
| this once per frame with only the new rows and never repeats an edge. Pairs | |
| are returned once each as ``(i, j, cosine)`` with ``i < j``, indices being | |
| positions in ``vectors``. The degree cap is what keeps a dense literature | |
| from becoming one hub; the floor is what keeps an isolated paper isolated. | |
| """ | |
| total = len(vectors) | |
| if total < 2 or start >= total or neighbors < 1: | |
| return [] | |
| matrix = np.asarray(vectors, dtype=np.float32) | |
| edges: dict[tuple[int, int], float] = {} | |
| for begin in range(start, total, chunk): | |
| end = min(begin + chunk, total) | |
| scores = matrix[begin:end] @ matrix.T | |
| scores[np.arange(end - begin), np.arange(begin, end)] = -np.inf | |
| top = min(neighbors, total - 1) | |
| candidates = np.argpartition(-scores, top - 1, axis=1)[:, :top] | |
| for offset, row in enumerate(range(begin, end)): | |
| for column in candidates[offset].tolist(): | |
| cosine = float(scores[offset, column]) | |
| if cosine < floor: | |
| continue | |
| key = (row, column) if row < column else (column, row) | |
| edges[key] = cosine | |
| return [(i, j, cosine) for (i, j), cosine in edges.items()] | |
| def project(vectors: np.ndarray) -> np.ndarray: | |
| """2-D coordinates for ``vectors`` under a basis fitted on themselves.""" | |
| if not len(vectors): | |
| return np.zeros((0, 2), dtype=np.float32) | |
| mean, basis = fit_projection(vectors) | |
| return project_with(mean, basis, vectors) | |
| def combine(positive: list[np.ndarray], negative: list[np.ndarray]) -> np.ndarray | None: | |
| """Embedding arithmetic: a starting point plus added contexts, minus removed ones. | |
| ``start - context + context`` is the document-level analogue of word2vec | |
| analogy: the result points at the neighbourhood the combination implies, | |
| and the corpus decides whether anything actually sits there. The sum is | |
| renormalized because only direction carries meaning in a cosine space. | |
| ``None`` when the terms cancel out — a zero-length vector has no direction, | |
| and placing it would land somewhere meaningless. | |
| """ | |
| total = np.zeros_like(np.asarray(positive[0], dtype=np.float64)) | |
| for vector in positive: | |
| total += np.asarray(vector, dtype=np.float64) | |
| for vector in negative: | |
| total -= np.asarray(vector, dtype=np.float64) | |
| norm = np.linalg.norm(total) | |
| if norm < 1e-6: | |
| return None | |
| return (total / norm).astype(np.float32) | |
| def _bucket(document: dict) -> str: | |
| return str(document.get("query_bucket") or document.get("bucket") or "inconclusive") | |
| def _is_primary(document: dict) -> bool: | |
| """Reviews restate other people's results, so they cannot evidence a region.""" | |
| return document.get("is_review") is not True and document.get("record_kind", "study") == "study" | |
| def label_region(counts: Counter, attempts: int) -> str: | |
| """Name the kind of absence a region represents, most specific rule first. | |
| ``unread`` is the label for a region whose documents carry no readable | |
| outcome at all — registrations, retractions and unreadable reports. It exists | |
| because the fallback must not be ``active``: measured against the live index, | |
| every region held only ``inconclusive``/``failed``/``unreported`` documents, | |
| and calling those "active" would assert a literature that was never read. | |
| """ | |
| if attempts < MIN_REGION_ATTEMPTS: | |
| return "thin" | |
| nulls = sum(counts[bucket] for bucket in NULL_BUCKETS) | |
| effects, dark = counts["effect"], counts["unreported"] | |
| if dark / attempts >= DARK_SHARE: | |
| return "dark" | |
| if nulls / attempts >= NULL_SATURATED_SHARE: | |
| return "null_saturated" | |
| if min(nulls, effects) / attempts >= CONTESTED_SHARE: | |
| return "contested" | |
| if (nulls + effects) / attempts < READABLE_SHARE: | |
| return "unread" | |
| return "active" | |
| def _exemplars(documents: list[dict], limit: int = 3) -> list[dict]: | |
| ranked = sorted(documents, key=lambda doc: doc.get("cited_by_count") or 0, reverse=True) | |
| return [ | |
| { | |
| "id": doc.get("id", ""), | |
| "title": doc.get("title", ""), | |
| "year": doc.get("year"), | |
| "bucket": _bucket(doc), | |
| } | |
| for doc in ranked[:limit] | |
| ] | |
| def group(labels: np.ndarray, clusters: int) -> list[np.ndarray]: | |
| """Row indices per cluster. One sort rather than a scan per cluster: the | |
| obvious ``labels == index`` loop is O(rows × clusters), which is hours of | |
| Python over a corpus of millions.""" | |
| order = np.argsort(labels, kind="stable") | |
| bounds = np.searchsorted(labels[order], np.arange(clusters + 1)) | |
| return [order[bounds[index] : bounds[index + 1]] for index in range(clusters)] | |
| def describe_regions( | |
| documents: list[dict], vectors: np.ndarray, labels: np.ndarray, clusters: int | |
| ) -> list[dict]: | |
| """Describe each occupied cluster by what happened in the documents it holds. | |
| Cluster indices are the region ids, and an empty cluster yields no region, so | |
| a region keeps its identity between two runs that share centroids — which is | |
| what lets a partially built map be redrawn rather than replaced. | |
| """ | |
| described = [] | |
| for index, members in enumerate(group(labels, clusters)): | |
| if not len(members): | |
| continue | |
| rows = vectors[members] | |
| centroid = normalize(rows.mean(axis=0, keepdims=True))[0] | |
| held = [documents[row] for row in members.tolist()] | |
| primary = [doc for doc in held if _is_primary(doc)] | |
| counts = Counter(_bucket(doc) for doc in primary) | |
| years = [doc["year"] for doc in held if isinstance(doc.get("year"), int)] | |
| citations = [doc.get("cited_by_count") or 0 for doc in primary] | |
| described.append( | |
| { | |
| "id": int(index), | |
| "size": len(held), | |
| "attempts": len(primary), | |
| "centroid": [float(value) for value in centroid], | |
| "label": label_region(counts, len(primary)), | |
| "bucketCounts": dict(counts), | |
| "medianYear": int(np.median(years)) if years else None, | |
| "medianCitations": float(np.median(citations)) if citations else None, | |
| "coherence": float((rows @ centroid).mean()), | |
| "exemplars": _exemplars(held), | |
| } | |
| ) | |
| return described | |
| def build_regions( | |
| documents: list[dict], | |
| regions: int = 24, | |
| seed: int = 0, | |
| vectors: np.ndarray | None = None, | |
| ) -> dict: | |
| """Cluster embedded documents and describe each cluster by what happened in it. | |
| Documents without a stored vector are counted in ``skipped`` rather than | |
| dropped silently: a region map over a partly unembedded corpus describes the | |
| embedded part only, and saying so is the difference between a map and a claim. | |
| ``vectors`` lets a caller that already holds the corpus as one normalized | |
| matrix — the streaming build does — skip rebuilding it from the documents, | |
| which would cost a second copy of several gigabytes. | |
| """ | |
| corpus, matrix = _corpus(documents, vectors) | |
| skipped = len(documents) - len(corpus) | |
| if not len(matrix): | |
| return {"version": MAP_VERSION, "regions": [], "documents": 0, "skipped": skipped} | |
| clusters = max(1, min(regions, len(matrix))) | |
| labels = kmeans(matrix, clusters, seed=seed) | |
| return { | |
| "version": MAP_VERSION, | |
| "regions": describe_regions(corpus, matrix, labels, clusters), | |
| "documents": len(corpus), | |
| "skipped": skipped, | |
| } | |
| def _corpus( | |
| documents: list[dict], vectors: np.ndarray | None = None | |
| ) -> tuple[list[dict], np.ndarray]: | |
| """Documents paired with their unit vectors, from the dicts or from a matrix.""" | |
| if vectors is not None: | |
| return documents, vectors | |
| embedded = [(doc, vector) for doc in documents if (vector := _vector(doc)) is not None] | |
| if not embedded: | |
| return [], np.zeros((0, 0), dtype=np.float32) | |
| return ( | |
| [doc for doc, _ in embedded], | |
| normalize(np.vstack([vector for _, vector in embedded])), | |
| ) | |
| def nearest_neighbors( | |
| vector: np.ndarray, documents: list[dict], vectors: np.ndarray, limit: int = 1 | |
| ) -> list[dict]: | |
| """The ``limit`` closest sampled papers to a point, closest first. | |
| Callers show these so a bare cosine never has to stand alone: the list of | |
| real papers near a point is the legible form of "how occupied is this spot". | |
| """ | |
| similarity = vectors @ vector | |
| limit = max(1, min(limit, len(similarity))) | |
| # argpartition first: a full sort of a corpus-sized array to read five rows | |
| # off the top is seconds of wasted work on the live index. | |
| top = np.argpartition(-similarity, limit - 1)[:limit] | |
| order = top[np.argsort(-similarity[top])] | |
| return [ | |
| { | |
| "id": documents[index].get("id", ""), | |
| "title": documents[index].get("title", ""), | |
| "year": documents[index].get("year"), | |
| "bucket": _bucket(documents[index]), | |
| "cosine": float(similarity[index]), | |
| } | |
| for index in order | |
| ] | |
| def _nearest(vector: np.ndarray, documents: list[dict], vectors: np.ndarray) -> dict: | |
| return nearest_neighbors(vector, documents, vectors, limit=1)[0] | |
| def neighbour_threshold(centroids: np.ndarray) -> float: | |
| """Centroid cosine above which two regions count as adjacent literatures.""" | |
| pairs = centroids @ centroids.T | |
| upper = pairs[np.triu_indices(len(centroids), k=1)] | |
| if not len(upper): | |
| return NEIGHBOUR_FLOOR | |
| return max(NEIGHBOUR_FLOOR, float(np.quantile(upper, NEIGHBOUR_QUANTILE))) | |
| def find_gaps( | |
| documents: list[dict], | |
| regions: list[dict], | |
| limit: int = 10, | |
| neighbour_cosine: float | None = None, | |
| occupancy: float = GAP_OCCUPANCY, | |
| vectors: np.ndarray | None = None, | |
| ) -> list[dict]: | |
| """Bands between neighbouring regions that the indexed corpus barely occupies. | |
| Occupancy is counted, not measured by distance: the two regions' own members | |
| are given to whichever of the two centroids and their midpoint they are | |
| closest to, and the band is open when the midpoint keeps almost nothing while | |
| both parents keep plenty. Pairs whose midpoint is better described by some | |
| third region are skipped — that band is occupied, by a whole literature. | |
| Distance alone cannot express any of this: a midpoint is always near its own | |
| parents, so an absolute cosine threshold either rejects every gap or accepts | |
| every one, and a fixed-radius ball around each point holds nothing at all in | |
| a real high-dimensional corpus (measured against the live index: every such | |
| ball was empty, so no pair could ever have been a gap). | |
| ``support`` is the thinner parent core: a band between two substantial | |
| literatures is a combination many people were positioned to try and did not. | |
| The parent labels carry the warning that matters — an open band between two | |
| null-saturated regions is unexplored *because the neighbours failed*. | |
| """ | |
| corpus, vectors = _corpus(documents, vectors) | |
| if not corpus or len(regions) < 2: | |
| return [] | |
| centroids = normalize(np.vstack([region["centroid"] for region in regions]).astype(np.float32)) | |
| if neighbour_cosine is None: | |
| neighbour_cosine = neighbour_threshold(centroids) | |
| assigned = assign(vectors, centroids) | |
| gaps = [] | |
| for left in range(len(regions)): | |
| for right in range(left + 1, len(regions)): | |
| separation = float(np.clip(centroids[left] @ centroids[right], -1.0, 1.0)) | |
| if separation < neighbour_cosine: | |
| continue | |
| midpoint = normalize(((centroids[left] + centroids[right]) / 2)[None, :])[0] | |
| to_midpoint = centroids @ midpoint | |
| others = np.delete(to_midpoint, [left, right]) | |
| if len(others) and others.max() >= min(to_midpoint[left], to_midpoint[right]): | |
| continue # a third literature already sits in the band | |
| members = vectors[(assigned == left) | (assigned == right)] | |
| owner = ( | |
| members @ np.vstack([centroids[left], centroids[right], midpoint]).T | |
| ).argmax(axis=1) | |
| support = min(int((owner == 0).sum()), int((owner == 1).sum())) | |
| if not support: | |
| continue | |
| band = int((owner == 2).sum()) | |
| if band > occupancy * support: | |
| continue | |
| parents = [regions[left], regions[right]] | |
| gaps.append( | |
| { | |
| "regions": [region["id"] for region in parents], | |
| "parentLabels": [region["label"] for region in parents], | |
| "openness": 1.0 - band / support, | |
| "support": support, | |
| "band": band, | |
| "separation": separation, | |
| "nearest": _nearest(midpoint, corpus, vectors), | |
| "exemplars": [region["exemplars"][:1] for region in parents], | |
| "discouraged": any( | |
| region["label"] in {"null_saturated", "dark"} for region in parents | |
| ), | |
| } | |
| ) | |
| gaps.sort(key=lambda gap: (gap["openness"], gap["support"]), reverse=True) | |
| return gaps[:limit] | |
| def place( | |
| vector: list[float] | np.ndarray, | |
| documents: list[dict], | |
| regions: list[dict], | |
| gaps: list[dict] | None = None, | |
| nearest_limit: int = 5, | |
| vectors: np.ndarray | None = None, | |
| ) -> dict: | |
| """Locate an idea on the map: how redundant it is and what surrounds it. | |
| ``redundancy`` is the cosine to the single closest indexed paper. It is the | |
| honest headline: most proposals are a near-duplicate of something already | |
| published, and the nearest papers are shown so the claim can be checked by | |
| reading them. | |
| """ | |
| corpus, vectors = _corpus(documents, vectors) | |
| query = normalize(np.asarray(vector, dtype=np.float32)[None, :])[0] | |
| if not corpus: | |
| return { | |
| "redundancy": None, | |
| "nearest": None, | |
| "neighbors": [], | |
| "region": None, | |
| "nearestGap": None, | |
| } | |
| neighbors = nearest_neighbors(query, corpus, vectors, limit=nearest_limit) | |
| region = None | |
| if regions: | |
| centroids = normalize( | |
| np.vstack([entry["centroid"] for entry in regions]).astype(np.float32) | |
| ) | |
| similarity = centroids @ query | |
| position = int(similarity.argmax()) | |
| region = {**regions[position], "cosine": float(similarity[position])} | |
| region.pop("centroid", None) | |
| return { | |
| "redundancy": neighbors[0]["cosine"], | |
| "nearest": neighbors[0], | |
| "neighbors": neighbors[1:], | |
| "region": region, | |
| "nearestGap": _nearest_gap(query, regions, gaps or []), | |
| } | |
| def _nearest_gap(query: np.ndarray, regions: list[dict], gaps: list[dict]) -> dict | None: | |
| if not gaps or not regions: | |
| return None | |
| by_id = {region["id"]: region for region in regions} | |
| best, best_cosine = None, -1.0 | |
| for gap in gaps: | |
| parents = [by_id[identifier] for identifier in gap["regions"] if identifier in by_id] | |
| if len(parents) != 2: | |
| continue | |
| midpoint = normalize( | |
| np.vstack([parent["centroid"] for parent in parents]).astype(np.float32).mean( | |
| axis=0, keepdims=True | |
| ) | |
| )[0] | |
| cosine = float(midpoint @ query) | |
| if cosine > best_cosine: | |
| best, best_cosine = gap, cosine | |
| return {**best, "cosine": best_cosine} if best else None | |
| def calibrate( | |
| documents: list[dict], | |
| cutoff_year: int, | |
| regions: int = 24, | |
| seed: int = 0, | |
| vectors: np.ndarray | None = None, | |
| ) -> dict: | |
| """Rebuild the map as it would have looked at ``cutoff_year`` and score it since. | |
| The labels claim something checkable: work published after the cutoff should | |
| behave differently depending on the label its region carried before it. A | |
| null-saturated region that keeps producing nulls, and an active region that | |
| keeps producing effects, is the map earning its labels. This is a calibration | |
| record over one corpus and one cutoff, not a forecast and not an accuracy | |
| study: a region can be mislabelled and still look right if later authors | |
| simply copied the earlier design. | |
| """ | |
| corpus, matrix = _corpus(documents, vectors) | |
| dated = [ | |
| index for index, doc in enumerate(corpus) if isinstance(doc.get("year"), int) | |
| ] | |
| before = np.array( | |
| [index for index in dated if corpus[index]["year"] < cutoff_year], dtype=np.int64 | |
| ) | |
| after = np.array( | |
| [index for index in dated if corpus[index]["year"] >= cutoff_year], dtype=np.int64 | |
| ) | |
| past = [corpus[index] for index in before.tolist()] | |
| future = [corpus[index] for index in after.tolist()] | |
| built = build_regions( | |
| past, regions=regions, seed=seed, vectors=matrix[before] if len(before) else None | |
| ) | |
| if not built["regions"] or not future: | |
| return { | |
| "version": MAP_VERSION, | |
| "cutoffYear": cutoff_year, | |
| "past": len(past), | |
| "future": 0, | |
| "labels": [], | |
| } | |
| centroids = normalize( | |
| np.vstack([region["centroid"] for region in built["regions"]]).astype(np.float32) | |
| ) | |
| later: dict[int, list[dict]] = {region["id"]: [] for region in built["regions"]} | |
| corpus, vectors = _corpus(future, matrix[after] if len(after) else None) | |
| for document, vector in zip(corpus, vectors): | |
| later[built["regions"][int((centroids @ vector).argmax())]["id"]].append(document) | |
| rows: dict[str, dict] = {} | |
| for region in built["regions"]: | |
| arrivals = [doc for doc in later[region["id"]] if _is_primary(doc)] | |
| row = rows.setdefault( | |
| region["label"], | |
| {"label": region["label"], "regions": 0, "priorAttempts": 0, "laterPapers": 0, | |
| "laterNull": 0, "laterEffect": 0}, | |
| ) | |
| row["regions"] += 1 | |
| row["priorAttempts"] += region["attempts"] | |
| row["laterPapers"] += len(arrivals) | |
| row["laterNull"] += sum(_bucket(doc) in NULL_BUCKETS for doc in arrivals) | |
| row["laterEffect"] += sum(_bucket(doc) == "effect" for doc in arrivals) | |
| for row in rows.values(): | |
| row["laterNullShare"] = row["laterNull"] / row["laterPapers"] if row["laterPapers"] else None | |
| return { | |
| "version": MAP_VERSION, | |
| "cutoffYear": cutoff_year, | |
| "past": len(past), | |
| "future": len(corpus), | |
| "labels": sorted(rows.values(), key=lambda row: row["label"]), | |
| } | |