Source code for PsychiatryNLPKit.analysis.Density

"""Semantic density analysis for PsychiatryNLPKit.

Measures dimensional properties of semantic spaces derived from token embeddings,
and the rate at which propositions are expressed in text. Schizophrenia patients
can show high or low semantic density, and lower intrinsic dimensionality
indicates more redundant speech.
"""

import logging

import numpy as np
import torch
from skdim.id import MLE
from sklearn.decomposition import PCA

logger = logging.getLogger(__name__)

# POS tags that carry one proposition each
EN_PROPOSITION_TAGS: dict[str, set[str]] = {
    "verb": {"VB", "VBD", "VBG", "VBN", "VBP", "VBZ", "VERB"},
    "adjective": {"JJ", "JJR", "JJS", "ADJ"},
    "adverb": {"RB", "RBR", "RBS", "ADV"},
    "preposition": {"IN", "ADP"},
    "conjunction": {"CC", "WRB", "CCONJ", "SCONJ"},
    "relative": {"WDT", "WP", "WP$"},
}

FR_PROPOSITION_TAGS: dict[str, set[str]] = {
    "verb": {"VERB"},
    "adjective": {"ADJ"},
    "adverb": {"ADV"},
    "preposition": {"ADP"},
    "conjunction": {"CCONJ", "SCONJ"},
}

# Determiner tags; non-article determiners contribute a proposition, while
# a, an, and the do not.
EN_DETERMINER_TAGS: set[str] = {"DT", "PDT"}
FR_DETERMINER_TAGS: set[str] = {"DET", "PRON"}

# Non-article determiners and quantifiers contribute a proposition
EN_QUANTIFIERS: set[str] = {
    "every",
    "each",
    "all",
    "both",
    "few",
    "many",
    "most",
    "several",
    "some",
    "any",
    "no",
    "this",
    "that",
    "these",
    "those",
    "another",
}
FR_QUANTIFIERS: set[str] = {"chaque", "tout", "toute", "tous", "toutes"}

# Copular verbs contribute only when they predicate a noun phrase.
EN_COPULA_LEMMAS: set[str] = {"be"}
FR_COPULA_LEMMAS: set[str] = {"être"}

# Auxiliary verbs fold into the following main verb.
EN_AUXILIARY_LEMMAS: set[str] = {"have", "do"}
FR_AUXILIARY_LEMMAS: set[str] = {"avoir"}

# Tags that can begin the noun-phrase complement of a copula.
EN_NOUN_TAGS: set[str] = {
    "CD",
    "DT",
    "NN",
    "NNP",
    "NNPS",
    "NNS",
    "PDT",
    "POS",
    "PRP",
    "PRP$",
    "WDT",
    "WP",
    "WP$",
}
FR_NOUN_TAGS: set[str] = {"DET", "NOUN", "NUM", "PRON", "PROPN"}

# Subordinating "that" introduces a complement clause and does not itself
# carry an idea; demonstrative "that" still counts as a determiner.
EN_COMPLEMENTIZERS: set[str] = {"that"}


[docs] def pca_density_metrics( token_embedding_vectors: dict[str, torch.Tensor], sections: list[str] | None = None, ) -> dict[str, dict[str, float]]: """PCA-based density metrics per section. PCA is applied in **embedding space** (tokens as rows, embedding dimensions as columns) so the metrics reflect how many embedding-space directions are needed to explain the semantic variance across tokens. During statistical analysis, total paragraph length must be controlled. Notes: Theoretical basis - While healthy controls tend to have moderately compressible semantic space, schizophrenia patients can have high or low semantic density (Palominos et al., 2025). Args: token_embedding_vectors: Dict mapping section names to token-level embedding tensors (from ``TextData.token_embedding_vectors``). sections: Sections to process. ``None`` processes all sections in the dict. Returns: Dict mapping section names to a metric dict with keys ``"Ncomp_90"``, ``"Pcomp_90"``, and ``"ExVar_2"``. Empty sections receive ``float("nan")`` for all metrics. References: Palominos, C., Stein, F., Kircher, T., Ayesa-Arriola, R., Palaniyappan, L., Homan, P., Sommer, I. E., & Hinzen, W. (2025). Lexical meaning is lower dimensional in psychosis. Scientific Reports, 16(1), 859. https://doi.org/10.1038/s41598-025-30443-1 """ if sections is None: sections = list(token_embedding_vectors.keys()) results: dict[str, dict[str, float]] = {} for sec in sections: if sec not in token_embedding_vectors: logger.warning("Section %s not found in embeddings, skipping", sec) continue vectors = token_embedding_vectors[sec] n_tokens = vectors.shape[0] # Handle edge case: fewer than 2 tokens. if n_tokens < 2: results[sec] = { "Ncomp_90": float("nan"), "Pcomp_90": float("nan"), "ExVar_2": float("nan"), } continue # Move to CPU for sklearn. X = vectors.detach().cpu().numpy() # PCA in embedding space: tokens are samples, embedding dims are features. n_components = min(n_tokens, X.shape[1]) pca = PCA(n_components=n_components) pca.fit(X) explained_variance_ratio = np.asarray( pca.explained_variance_ratio_, dtype=float ) cumulative_variance = np.cumsum(explained_variance_ratio) ncomp_90 = int(np.searchsorted(cumulative_variance, 0.9, side="left") + 1) pcomp_90 = ncomp_90 / n_tokens exvar_2 = float( explained_variance_ratio[: min(2, len(explained_variance_ratio))].sum() ) results[sec] = { "Ncomp_90": float(ncomp_90), "Pcomp_90": float(pcomp_90), "ExVar_2": exvar_2, } return results
[docs] def intrinsic_dimensionality_density( token_embedding_vectors: dict[str, torch.Tensor], sections: list[str] | None = None, k: int | None = None, ) -> dict[str, dict[str, float]]: """Estimate intrinsic dimensionality using MLE (Levina & Bickel, 2004). Intrinsic dimensionality quantifies the local geometric complexity of the semantic space formed by token embeddings. Lower values indicate more redundant speech. Notes: Theoretical basis - Lower intrinsic dimensionality indicates more redundant speech (Palominos et al., 2025). Args: token_embedding_vectors: Dict mapping section names to token-level embedding tensors (from ``TextData.token_embedding_vectors``). sections: Sections to process. ``None`` processes all sections in the dict. k: Number of neighbors for MLE estimator. Defaults to ``min(10, n_samples - 1)``. Returns: Dict mapping section names to a metric dict with key ``"ID_MLE"``. Empty sections receive ``float("nan")``. References: Palominos, C., Stein, F., Kircher, T., Ayesa-Arriola, R., Palaniyappan, L., Homan, P., Sommer, I. E., & Hinzen, W. (2025). Lexical meaning is lower dimensional in psychosis. Scientific Reports, 16(1), 859. https://doi.org/10.1038/s41598-025-30443-1 """ if sections is None: sections = list(token_embedding_vectors.keys()) results: dict[str, dict[str, float]] = {} for sec in sections: if sec not in token_embedding_vectors: logger.warning("Section %s not found in embeddings, skipping", sec) continue vectors = token_embedding_vectors[sec] n_tokens = vectors.shape[0] # Handle edge case: fewer than 2 tokens. if n_tokens < 2: results[sec] = {"ID_MLE": float("nan")} continue # Move to CPU for skdim. X = vectors.detach().cpu().numpy() k_param = k if k is not None else min(10, n_tokens - 1) id_estimator = MLE(K=k_param) intrinsic_dim = float(id_estimator.fit_transform(X)) results[sec] = {"ID_MLE": intrinsic_dim} return results
def _f_ratio_split(weights: torch.Tensor) -> int: """Return the size of the high-weight group selected by F-ratio partitioning. Rank-ordered weights are partitioned into two groups at each possible split point; the F-ratio is the ratio of between-group to within-group variance. The split maximizing this ratio separates the meaning components (high weights) from the redundant words (low weights). Ties prefer larger high groups so that varied sentences retain most of their words. """ # Handle edge cases. n = weights.shape[0] if n == 0: return 0 if n == 1: return 1 # calculate global statistics. sorted_w, _ = torch.sort(weights, descending=True) csum = torch.cat( [torch.zeros(1, device=weights.device), torch.cumsum(sorted_w, dim=0)] ) csq = torch.cat( [torch.zeros(1, device=weights.device), torch.cumsum(sorted_w**2, dim=0)] ) total_sum = csum[n].item() total_sq = csq[n].item() mean_all = total_sum / n # Iterate over all possible splits and compute F-ratio. best_split, best_f = 1, -1.0 for split in range(1, n): hi_sum = csum[split].item() hi_sq = csq[split].item() hi_mean = hi_sum / split lo_sum = total_sum - hi_sum lo_mean = lo_sum / (n - split) between = ( split * (hi_mean - mean_all) ** 2 + (n - split) * (lo_mean - mean_all) ** 2 ) within = ( (hi_sq - hi_sum**2 / split) + (total_sq - hi_sq) - (lo_sum**2 / (n - split)) ) f = float("inf") if within <= 1e-12 else between / within if f > best_f or (f == best_f and split > best_split): best_f, best_split = f, split return best_split def _sentence_density( words: torch.Tensor, learning_rate: float, max_iterations: int, tau_iteration: int, ) -> tuple[float, int, int]: """Compute the semantic density of a single sentence by vector unpacking. Returns a ``(density, meaning_components, content_words)`` tuple. Identical word vectors are collapsed into a single meaning component, and sentences whose words all share one meaning have density ``1 / content_words``. """ # Normalize word vectors. words = torch.nn.functional.normalize(words, p=2, dim=1) n_words = words.shape[0] # Sum word vectors to get sentence vector. sentence_vec = torch.nn.functional.normalize(words.sum(dim=0), p=2, dim=0) # Remove identical word vectors. unique_words = torch.unique(words, dim=0) if unique_words.shape[0] == 1: return 1.0 / n_words, 1, n_words # Initialize weights. weights = torch.full( (unique_words.shape[0],), 1.0 / unique_words.shape[0], device=words.device ) # Run gradient descent. for _ in range(max_iterations): estimate = weights @ unique_words gradient = (estimate - sentence_vec) @ unique_words.T if float(torch.norm(gradient)) < 1e-6: break weights = weights - learning_rate * gradient # Prune low weights; the threshold grows with training iterations. threshold = tau_iteration / max_iterations weights[weights < threshold] = 0.0 # Calculate F-ratio on remaining weights. components = _f_ratio_split(weights[weights > 0.0]) return components / n_words, components, n_words
[docs] def vector_unpacking_density( content_word_embedding_vectors: dict[str, list[torch.Tensor]], sections: list[str] | None = None, learning_rate: float = 0.01, max_iterations: int = 5000, tau_iteration: int = 100, ) -> dict[str, dict[str, float]]: """Semantic density measured by vector unpacking (Rezaii et al., 2019). A sentence is represented by the normalized sum of its content-word embeddings, then decomposed into a linear combination of those word embeddings learned by gradient descent. The number of meaning components (word embeddings with high learned weights, selected by F-ratio partitioning) divided by the number of content words gives the sentence density; the section density is the mean over its sentences. Notes: Theoretical basis - Low semantic density predicts conversion to psychosis in clinical high-risk individuals and correlates negatively with negative symptoms (Rezaii et al., 2019). Args: content_word_embedding_vectors: Dict mapping section names to lists of per-sentence Word2Vec embedding tensors for content words (from ``TextData.content_word_embedding_vectors``). Only content words enter the density estimate; function words are excluded upstream. sections: Sections to process. ``None`` processes all sections in the dict. learning_rate: Gradient descent learning rate for the weight updates. Defaults to ``0.01``. max_iterations: Maximum number of gradient descent iterations per sentence. Defaults to ``5000``, matching the reference. tau_iteration: Number of iterations used to set the weight pruning threshold as ``tau_iteration / max_iterations``. Defaults to ``100``, matching the reference. Returns: Dict mapping section names to a metric dict with keys ``"semantic_density"`` (mean of sentence densities, where each sentence density is the number of meaning components m_j divided by the number of content words n_j), ``"semantic_density_std"`` (standard deviation across sentences), ``"mean_meaning_components"`` (mean number of components m_j), and ``"mean_content_words"`` (mean number of content words n_j, useful as a poverty-of-speech control). Sections with no analyzable sentences receive ``float("nan")`` for all metrics. References: Rezaii, N., Walker, E., & Wolff, P. (2019). A machine learning approach to predicting psychosis using semantic density and latent content analysis. Schizophrenia, 5(1), 9. https://doi.org/10.1038/s41537-019-0077-9 """ if sections is None: sections = list(content_word_embedding_vectors.keys()) results: dict[str, dict[str, float]] = {} for sec in sections: if sec not in content_word_embedding_vectors: logger.warning("Section %s not found in embeddings, skipping", sec) continue densities: list[float] = [] components: list[int] = [] word_counts: list[int] = [] for sentence in content_word_embedding_vectors[sec]: if sentence.shape[0] < 2: continue density, m_j, n_j = _sentence_density( sentence, learning_rate=learning_rate, max_iterations=max_iterations, tau_iteration=tau_iteration, ) densities.append(density) components.append(m_j) word_counts.append(n_j) # Handle edge case with no analyzable sentences. if not densities: results[sec] = { "semantic_density": float("nan"), "semantic_density_std": float("nan"), "mean_meaning_components": float("nan"), "mean_content_words": float("nan"), } continue results[sec] = { "semantic_density": float(np.mean(densities)), "semantic_density_std": ( float(np.std(densities)) if len(densities) > 1 else float("nan") ), "mean_meaning_components": float(np.mean(components)), "mean_content_words": float(np.mean(word_counts)), } return results
def _validate_language(lang: str) -> None: """Raise ValueError if lang is not "en" or "fr".""" if lang not in ("en", "fr"): raise ValueError(f"Unsupported language: {lang}. Must be one of 'en' or 'fr'.") def _next_content_tag(sent_tags: list[tuple[str, str, str]], index: int) -> str | None: """Return the tag of the next token that is neither punctuation nor an adverb.""" for word, _, tag in sent_tags[index + 1 :]: if not any(ch.isalnum() for ch in word): continue if tag in {"RB", "RBR", "RBS", "ADV"}: continue return tag return None def _count_propositions( sent_tags: list[tuple[str, str, str]], lang: str ) -> tuple[int, int]: """Count propositions and words in a POS-tagged sentence. Returns a ``(propositions, words)`` tuple. Propositions follow the CPIDR rule set: each verb, adjective, adverb, preposition, conjunction, relative pronoun, and non-article determiner contributes one idea, and modal verbs count only through their negation. Auxiliary "be", "have", and "do" fold into the following main verb, and the copula contributes only when it predicates a noun phrase. Punctuation is excluded from both counts. """ if lang == "en": tag_sets = EN_PROPOSITION_TAGS quantifiers = EN_QUANTIFIERS determiner_tags = EN_DETERMINER_TAGS copula_lemmas = EN_COPULA_LEMMAS auxiliary_lemmas = EN_AUXILIARY_LEMMAS noun_tags = EN_NOUN_TAGS complementizers = EN_COMPLEMENTIZERS else: tag_sets = FR_PROPOSITION_TAGS quantifiers = FR_QUANTIFIERS determiner_tags = FR_DETERMINER_TAGS copula_lemmas = FR_COPULA_LEMMAS auxiliary_lemmas = FR_AUXILIARY_LEMMAS noun_tags = FR_NOUN_TAGS complementizers = set() idea_tags = set() for tags in tag_sets.values(): idea_tags |= tags linking_lemmas = copula_lemmas | auxiliary_lemmas propositions = 0 words = 0 for index, (word, lemma, tag) in enumerate(sent_tags): # Skip punctuation tokens. if not any(ch.isalnum() for ch in word): continue words += 1 normalized = ((lemma or word) or "").lower() if tag in idea_tags: if tag in tag_sets["verb"] and normalized in linking_lemmas: next_tag = _next_content_tag(sent_tags, index) if next_tag in tag_sets["verb"]: # Auxiliary use; the main verb carries the idea. continue if normalized in copula_lemmas and next_tag not in noun_tags: # The complement carries the idea, not the copula. continue if tag in {"IN", "SCONJ"} and normalized in complementizers: # Complementizer "that" adds no idea beyond the embedded clause. continue propositions += 1 elif normalized in quantifiers and tag in determiner_tags: propositions += 1 return propositions, words
[docs] def propositional_idea_density( pos_tags: dict[str, list[list[tuple[str, str, str]]]], sections: list[str] | None = None, lang: str = "en", ) -> dict[str, dict[str, float]]: """Propositional idea density per section (Hill et al., 2021). Counts elementary predications (propositions) in text and reports their rate per 100 words. Following the Computerized Propositional Idea Density Rater (CPIDR) (Brown et al., 2008), propositions correspond roughly to verbs, adjectives, adverbs, prepositions, conjunctions, relative pronouns, and non-article determiners. Auxiliary "be", "have", and "do" fold into the following main verb, the copula contributes only when it predicates a noun phrase, and modal verbs count only through their negation. Notes: Theoretical basis - Lower propositional idea density is associated with an increased risk of Alzheimer's disease (Snowdon et al., 1996), and declines with age and in males (Hill et al., 2021). Args: pos_tags: Dict mapping section names to lists of POS-tagged sentences. Each sentence is a list of ``(word, lemma, tag)`` tuples. sections: Sections to process. ``None`` processes all sections in *pos_tags*. lang: Language code ("en" or "fr"). Determines the POS tag scheme (Penn Treebank for English, Universal Dependencies for French). Returns: Dict mapping section names to a metric dict with keys ``"propositional_idea_density"`` (propositions per 100 words), ``"proposition_count"`` (total propositions), and ``"word_count"`` (total words, useful as a poverty-of-speech control). Empty sections receive ``float("nan")`` for all metrics. References: Brown, C., Snodgrass, T., Kemper, S. J., Herman, R., & Covington, M. A. (2008). Automatic measurement of propositional idea density from part-of-speech tagging. Behavior Research Methods, 40(2), 540–545. https://doi.org/10.3758/brm.40.2.540 Hill, E., Alty, J., Bartlett, L., Goldberg, L., Park, M., Yeom, S., & Vickers, J. (2021). Automated analysis of propositional idea density in older adults. Cortex, 145, 264–272. https://doi.org/10.1016/j.cortex.2021.09.018 Snowdon, D. A., Kemper, S. J., Mortimer, J. A., Greiner, L. H., Wekstein, D. R., & Markesbery, W. R. (1996). Linguistic ability in early life and cognitive function and Alzheimer's disease in late life: Findings from the Nun Study. JAMA, 275(7), 528–532. https://doi.org/10.1001/jama.1996.03530270044029 """ _validate_language(lang) if sections is None: sections = list(pos_tags.keys()) results: dict[str, dict[str, float]] = {} for sec in sections: if sec not in pos_tags: logger.warning("Section %s not found in pos_tags, skipping", sec) continue tags = pos_tags[sec] if not tags: results[sec] = { "propositional_idea_density": float("nan"), "proposition_count": float("nan"), "word_count": float("nan"), } continue total_propositions = 0 total_words = 0 for sent in tags: props, words = _count_propositions(sent, lang) total_propositions += props total_words += words density = ( 100.0 * total_propositions / total_words if total_words else float("nan") ) results[sec] = { "propositional_idea_density": density, "proposition_count": float(total_propositions), "word_count": float(total_words), } return results
__all__: list[str] = [ "pca_density_metrics", "intrinsic_dimensionality_density", "vector_unpacking_density", "propositional_idea_density", ]