Analysis layer ([graph] extra)

The optional analysis pipeline: sentence embeddings → cosine kNN graph → (optional disparity backbone) → PageRank and community detection. Install it with uv add "lexograph[graph]". Everything here only produces the plain per-unit arrays the encode channels accept; the core never imports it.

lexograph.analyze.analyze_text(text, *, embeddings=None, k=5, community='louvain', n_clusters=10, backbone=False, min_alpha_ptile=0.5, seed=42)

Run the analysis pipeline and return per-sentence channel arrays.

Parameters:

Name	Type	Description	Default
text	str	The source text.	required
embeddings	FloatArray | None	Precomputed (N, D) embeddings to use. If None, the sentences are embedded with the default model (a model download).	None
k	int	Neighbours per node in the kNN graph.	5
community	CommunityMethod	Community method — "louvain" (default) or "kmeans".	'louvain'
n_clusters	int	Number of clusters for the "kmeans" method.	10
backbone	bool	If True, sparsify the kNN graph with the disparity filter before PageRank and community detection.	False
min_alpha_ptile	float	Disparity-filter threshold (used when backbone).	0.5
seed	int	Random seed for reproducibility.	42

Returns:

Name	Type	Description
An	Analysis	class:Analysis whose arrays align to the segmented sentences.

Raises:

Type	Description
ValueError	If embeddings is given but its length does not match the sentence count.

Example

Drive a walk and a semantic dotplot from the analysis (not run as a doctest — embedding downloads a model)::

from lexograph import text_walk, recurrence_plot, load_demo_text
from lexograph.analyze import analyze_text

a = analyze_text(load_demo_text())
walk = text_walk(load_demo_text(), colour=a.community,
                 colour_kind="categorical", size=a.size)
dots = recurrence_plot(load_demo_text(), distances=a.distances,
                       threshold=0.4)

Source code in lexograph/analyze/__init__.py

def analyze_text(
    text: str,
    *,
    embeddings: FloatArray | None = None,
    k: int = 5,
    community: CommunityMethod = "louvain",
    n_clusters: int = 10,
    backbone: bool = False,
    min_alpha_ptile: float = 0.5,
    seed: int = 42,
) -> Analysis:
    """Run the analysis pipeline and return per-sentence channel arrays.

    Args:
        text: The source text.
        embeddings: Precomputed ``(N, D)`` embeddings to use. If ``None``, the
            sentences are embedded with the default model (a model download).
        k: Neighbours per node in the kNN graph.
        community: Community method — ``"louvain"`` (default) or ``"kmeans"``.
        n_clusters: Number of clusters for the ``"kmeans"`` method.
        backbone: If ``True``, sparsify the kNN graph with the disparity filter
            before PageRank and community detection.
        min_alpha_ptile: Disparity-filter threshold (used when ``backbone``).
        seed: Random seed for reproducibility.

    Returns:
        An :class:`Analysis` whose arrays align to the segmented sentences.

    Raises:
        ValueError: If ``embeddings`` is given but its length does not match the
            sentence count.

    Example:
        Drive a walk and a semantic dotplot from the analysis (not run as a
        doctest — embedding downloads a model)::

            from lexograph import text_walk, recurrence_plot, load_demo_text
            from lexograph.analyze import analyze_text

            a = analyze_text(load_demo_text())
            walk = text_walk(load_demo_text(), colour=a.community,
                             colour_kind="categorical", size=a.size)
            dots = recurrence_plot(load_demo_text(), distances=a.distances,
                                   threshold=0.4)
    """
    units = split_sentences(text)
    n = len(units)
    if embeddings is None:
        embeddings = embed_sentences(units)
    elif len(embeddings) != n:
        msg = f"embeddings must have one row per sentence ({n}), got {len(embeddings)}"
        raise ValueError(msg)

    graph = knn_graph(embeddings, k=k)
    if backbone:
        graph = extract_backbone(graph, min_alpha_ptile=min_alpha_ptile)
    size = pagerank_scores(graph, n)
    labels = community_labels(
        graph,
        n,
        method=community,
        embeddings=embeddings,
        n_clusters=n_clusters,
        seed=seed,
    )
    distances = embedding_distances(embeddings)
    return Analysis(
        sentences=units,
        embeddings=embeddings,
        size=size,
        community=labels,
        distances=distances,
    )

lexograph.analyze.Analysis dataclass

The per-sentence channel arrays derived from a text.

Attributes:

Name	Type	Description
sentences	list[str]	The segmented sentences (length N).
embeddings	FloatArray	The (N, D) sentence embeddings.
size	FloatArray	PageRank centrality per sentence — the size channel.
community	ndarray	Community id per sentence — the colour channel.
distances	FloatArray	The (N, N) cosine distance matrix — pass it to recurrence_plot(distances=...) for a semantic dotplot.

Source code in lexograph/analyze/__init__.py

@dataclass(frozen=True, slots=True)
class Analysis:
    """The per-sentence channel arrays derived from a text.

    Attributes:
        sentences: The segmented sentences (length ``N``).
        embeddings: The ``(N, D)`` sentence embeddings.
        size: PageRank centrality per sentence — the size channel.
        community: Community id per sentence — the colour channel.
        distances: The ``(N, N)`` cosine distance matrix — pass it to
            ``recurrence_plot(distances=...)`` for a semantic dotplot.
    """

    sentences: list[str]
    embeddings: FloatArray
    size: FloatArray
    community: np.ndarray
    distances: FloatArray

lexograph.analyze.embeddings.embed_sentences(sentences, *, model_name=DEFAULT_MODEL, batch_size=64)

Embed sentences into L2-normalised vectors.

The model is downloaded on first use and cached by sentence-transformers. Because the embeddings are L2-normalised, a dot product equals cosine similarity.

Parameters:

Name	Type	Description	Default
sentences	Sequence[str]	The sentences to embed.	required
model_name	str	A sentence-transformers model id.	DEFAULT_MODEL
batch_size	int	Encoding batch size.	64

Returns:

Type	Description
FloatArray	An (N, D) float array of unit-norm sentence embeddings.

Example

Not run as a doctest (it would download the model)::

from lexograph import segment, load_demo_text
from lexograph.analyze.embeddings import embed_sentences

sentences = segment(load_demo_text())
embeddings = embed_sentences(sentences)

Source code in lexograph/analyze/embeddings.py

def embed_sentences(
    sentences: Sequence[str],
    *,
    model_name: str = DEFAULT_MODEL,
    batch_size: int = 64,
) -> FloatArray:
    """Embed sentences into L2-normalised vectors.

    The model is downloaded on first use and cached by ``sentence-transformers``.
    Because the embeddings are L2-normalised, a dot product equals cosine
    similarity.

    Args:
        sentences: The sentences to embed.
        model_name: A sentence-transformers model id.
        batch_size: Encoding batch size.

    Returns:
        An ``(N, D)`` float array of unit-norm sentence embeddings.

    Example:
        Not run as a doctest (it would download the model)::

            from lexograph import segment, load_demo_text
            from lexograph.analyze.embeddings import embed_sentences

            sentences = segment(load_demo_text())
            embeddings = embed_sentences(sentences)
    """
    # Imported by name so the heavy optional dependency is neither required to
    # import this module nor statically type-checked when it is absent.
    st: Any = importlib.import_module("sentence_transformers")

    model = st.SentenceTransformer(model_name)
    vectors = model.encode(
        list(sentences),
        batch_size=batch_size,
        convert_to_numpy=True,
        normalize_embeddings=True,
    )
    return np.asarray(vectors, dtype=float)

lexograph.analyze.graph.knn_graph(embeddings, *, k=5)

Build a weighted cosine k-nearest-neighbour graph over the embeddings.

Node i is sentence i; an edge carries the cosine similarity (1 - cosine distance) as its weight. The graph is undirected: the mutual edge keeps the stronger of the two directed similarities.

Parameters:

Name	Type	Description	Default
embeddings	FloatArray	An (N, D) array of sentence embeddings.	required
k	int	Neighbours per node (capped at N - 1).	5

Returns:

Type	Description
Graph	A networkx graph with nodes 0 .. N-1 and weighted edges.

Examples:

>>> import numpy as np
>>> emb = np.eye(4)
>>> g = knn_graph(emb, k=1)
>>> g.number_of_nodes()
4

Source code in lexograph/analyze/graph.py

def knn_graph(embeddings: FloatArray, *, k: int = 5) -> nx.Graph:
    """Build a weighted cosine k-nearest-neighbour graph over the embeddings.

    Node ``i`` is sentence ``i``; an edge carries the cosine **similarity**
    (``1 - cosine distance``) as its ``weight``. The graph is undirected: the
    mutual edge keeps the stronger of the two directed similarities.

    Args:
        embeddings: An ``(N, D)`` array of sentence embeddings.
        k: Neighbours per node (capped at ``N - 1``).

    Returns:
        A networkx graph with nodes ``0 .. N-1`` and weighted edges.

    Examples:
        >>> import numpy as np
        >>> emb = np.eye(4)
        >>> g = knn_graph(emb, k=1)
        >>> g.number_of_nodes()
        4
    """
    n = len(embeddings)
    graph = nx.Graph()
    graph.add_nodes_from(range(n))
    if n < 2:
        return graph
    k_eff = min(k, n - 1)
    adjacency = kneighbors_graph(
        np.asarray(embeddings, dtype=float),
        n_neighbors=k_eff,
        mode="distance",
        metric="cosine",
        include_self=False,
    ).tocoo()
    for i, j, distance in zip(
        adjacency.row, adjacency.col, adjacency.data, strict=True
    ):
        weight = 1.0 - float(distance)
        if weight <= 0.0:
            continue
        if graph.has_edge(int(i), int(j)):
            graph[int(i)][int(j)]["weight"] = max(
                graph[int(i)][int(j)]["weight"], weight
            )
        else:
            graph.add_edge(int(i), int(j), weight=weight)
    return graph

lexograph.analyze.graph.embedding_distances(embeddings)

Return the pairwise cosine distance matrix of the embeddings.

Ready to pass as distances to :func:lexograph.presets.recurrence.recurrence_plot for a semantic recurrence dotplot.

Parameters:

Name	Type	Description	Default
embeddings	FloatArray	An (N, D) array of sentence embeddings.	required

Returns:

Type	Description
FloatArray	An (N, N) cosine distance matrix with a zero diagonal.

Examples:

>>> import numpy as np
>>> d = embedding_distances(np.eye(3))
>>> d.shape
(3, 3)

Source code in lexograph/analyze/graph.py

def embedding_distances(embeddings: FloatArray) -> FloatArray:
    """Return the pairwise cosine distance matrix of the embeddings.

    Ready to pass as ``distances`` to
    :func:`lexograph.presets.recurrence.recurrence_plot` for a semantic
    recurrence dotplot.

    Args:
        embeddings: An ``(N, D)`` array of sentence embeddings.

    Returns:
        An ``(N, N)`` cosine distance matrix with a zero diagonal.

    Examples:
        >>> import numpy as np
        >>> d = embedding_distances(np.eye(3))
        >>> d.shape
        (3, 3)
    """
    return np.asarray(
        cosine_distances(np.asarray(embeddings, dtype=float)), dtype=float
    )

lexograph.analyze.graph.pagerank_scores(graph, n)

Return weighted PageRank as a per-sentence array (the size channel).

Parameters:

Name	Type	Description	Default
graph	Graph	A weighted sentence graph (e.g. from :func:knn_graph).	required
n	int	The total sentence count, so the result aligns to every sentence even if the graph has dropped some nodes.	required

Returns:

Type	Description
FloatArray	A length-n float array; entry i is the PageRank of node i (or
FloatArray	0.0 if the node is absent or the graph has no edges).

Examples:

>>> import networkx as nx
>>> g = nx.path_graph(3)
>>> for u, v in g.edges():
...     g[u][v]["weight"] = 1.0
>>> pagerank_scores(g, 3).shape
(3,)

Source code in lexograph/analyze/graph.py

def pagerank_scores(graph: nx.Graph, n: int) -> FloatArray:
    """Return weighted PageRank as a per-sentence array (the size channel).

    Args:
        graph: A weighted sentence graph (e.g. from :func:`knn_graph`).
        n: The total sentence count, so the result aligns to every sentence even
            if the graph has dropped some nodes.

    Returns:
        A length-``n`` float array; entry ``i`` is the PageRank of node ``i`` (or
        ``0.0`` if the node is absent or the graph has no edges).

    Examples:
        >>> import networkx as nx
        >>> g = nx.path_graph(3)
        >>> for u, v in g.edges():
        ...     g[u][v]["weight"] = 1.0
        >>> pagerank_scores(g, 3).shape
        (3,)
    """
    if graph.number_of_edges() == 0:
        return np.zeros(n, dtype=float)
    ranks = nx.pagerank(graph, weight="weight")
    return np.asarray([ranks.get(i, 0.0) for i in range(n)], dtype=float)

lexograph.analyze.graph.community_labels(graph, n, *, method='louvain', embeddings=None, n_clusters=10, seed=42)

Return a per-sentence community label (the colour channel).

Parameters:

Name	Type	Description	Default
graph	Graph	A weighted sentence graph (used by the Louvain method).	required
n	int	The total sentence count, so labels align to every sentence.	required
method	CommunityMethod	"louvain" (graph communities, the default) or "kmeans" (clusters the embeddings directly).	'louvain'
embeddings	FloatArray | None	Required for "kmeans"; the (N, D) embedding array.	None
n_clusters	int	Number of clusters for "kmeans" (capped at N).	10
seed	int	Random seed for reproducibility.	42

Returns:

Type	Description
ndarray	A length-n integer array of community ids. For Louvain, id 0 is
ndarray	the largest community; a node absent from the graph gets -1.

Raises:

Type	Description
ValueError	If method is "kmeans" and embeddings is None, or method is unrecognised.

Examples:

>>> import networkx as nx
>>> g = nx.path_graph(4)
>>> for u, v in g.edges():
...     g[u][v]["weight"] = 1.0
>>> labels = community_labels(g, 4)
>>> labels.shape
(4,)

Source code in lexograph/analyze/graph.py

def community_labels(
    graph: nx.Graph,
    n: int,
    *,
    method: CommunityMethod = "louvain",
    embeddings: FloatArray | None = None,
    n_clusters: int = 10,
    seed: int = 42,
) -> np.ndarray:
    """Return a per-sentence community label (the colour channel).

    Args:
        graph: A weighted sentence graph (used by the Louvain method).
        n: The total sentence count, so labels align to every sentence.
        method: ``"louvain"`` (graph communities, the default) or ``"kmeans"``
            (clusters the embeddings directly).
        embeddings: Required for ``"kmeans"``; the ``(N, D)`` embedding array.
        n_clusters: Number of clusters for ``"kmeans"`` (capped at ``N``).
        seed: Random seed for reproducibility.

    Returns:
        A length-``n`` integer array of community ids. For Louvain, id ``0`` is
        the largest community; a node absent from the graph gets ``-1``.

    Raises:
        ValueError: If ``method`` is ``"kmeans"`` and ``embeddings`` is ``None``,
            or ``method`` is unrecognised.

    Examples:
        >>> import networkx as nx
        >>> g = nx.path_graph(4)
        >>> for u, v in g.edges():
        ...     g[u][v]["weight"] = 1.0
        >>> labels = community_labels(g, 4)
        >>> labels.shape
        (4,)
    """
    if method == "kmeans":
        if embeddings is None:
            msg = "kmeans community detection requires embeddings"
            raise ValueError(msg)
        from sklearn.cluster import KMeans

        k = min(n_clusters, n)
        labels = KMeans(n_clusters=k, random_state=seed, n_init=10).fit_predict(
            np.asarray(embeddings, dtype=float)
        )
        return labels.astype(int)
    if method == "louvain":
        raw = cast(
            "list[set[int]]",
            nx.community.louvain_communities(graph, weight="weight", seed=seed),
        )
        communities = sorted(raw, key=len, reverse=True)
        label_of = {
            node: cid for cid, members in enumerate(communities) for node in members
        }
        return np.asarray([label_of.get(i, -1) for i in range(n)], dtype=int)
    msg = f"method must be 'louvain' or 'kmeans', got {method!r}"
    raise ValueError(msg)

lexograph.analyze.backbone.extract_backbone(graph, *, min_alpha_ptile=0.5, min_degree=1)

Return the disparity-filter backbone of a weighted graph.

The input is copied (never mutated): edges below min_alpha_ptile are dropped, then nodes whose degree falls below min_degree are pruned iteratively until stable.

Parameters:

Name	Type	Description	Default
graph	Graph	A weighted networkx graph.	required
min_alpha_ptile	float	Edges with an alpha percentile below this are removed.	0.5
min_degree	int	Nodes left with a degree below this are pruned (1 keeps any node that still has an edge).	1

Returns:

Type	Description
Graph	A new graph containing only the backbone.

Examples:

>>> import networkx as nx
>>> g = nx.path_graph(5)
>>> for u, v in g.edges():
...     g[u][v]["weight"] = float(v + 1)
>>> bb = extract_backbone(g, min_alpha_ptile=0.3)
>>> bb.number_of_nodes() <= g.number_of_nodes()
True

Source code in lexograph/analyze/backbone.py

def extract_backbone(
    graph: nx.Graph,
    *,
    min_alpha_ptile: float = 0.5,
    min_degree: int = 1,
) -> nx.Graph:
    """Return the disparity-filter backbone of a weighted graph.

    The input is copied (never mutated): edges below ``min_alpha_ptile`` are
    dropped, then nodes whose degree falls below ``min_degree`` are pruned
    iteratively until stable.

    Args:
        graph: A weighted networkx graph.
        min_alpha_ptile: Edges with an alpha percentile below this are removed.
        min_degree: Nodes left with a degree below this are pruned (``1`` keeps
            any node that still has an edge).

    Returns:
        A new graph containing only the backbone.

    Examples:
        >>> import networkx as nx
        >>> g = nx.path_graph(5)
        >>> for u, v in g.edges():
        ...     g[u][v]["weight"] = float(v + 1)
        >>> bb = extract_backbone(g, min_alpha_ptile=0.3)
        >>> bb.number_of_nodes() <= g.number_of_nodes()
        True
    """
    if graph.number_of_edges() == 0:
        return nx.Graph()

    result = copy.deepcopy(graph)
    apply_disparity_filter(result)
    result.remove_edges_from(
        [
            (u, v)
            for u, v, data in result.edges(data=True)
            if data.get("alpha_ptile", 0.0) < min_alpha_ptile
        ]
    )
    changed = True
    while changed:
        prune = [n for n in list(result.nodes()) if result.degree(n) < min_degree]
        changed = bool(prune)
        result.remove_nodes_from(prune)
    return result