cliving 0.1.0

Dynamic prototype memory classifier library

cliving

cliving est une librairie de classification incrémentale par mémoire dynamique.

Elle ne calcule pas les embeddings. Elle prend des vecteurs numériques en entrée et maintient une mémoire de prototypes supervisés qui évolue au fil du temps.

L'objectif est d'avoir un classifieur :

• adaptable immédiatement après correction utilisateur
• explicable
• peu coûteux à l'inférence
• sans retrain global

Idée générale

cliving fonctionne comme un système de classes vivantes.

Chaque classe n'est pas représentée par un unique centre figé. Elle possède une ou plusieurs zones locales :

• une zone = un prototype
• un prototype = un centre + une covariance diagonale régularisée + une confiance + un historique local

Quand un nouvel embedding arrive :

• il est comparé à tous les prototypes
• chaque prototype produit un score local
• les scores sont agrégés par classe
• la meilleure classe gagne, sauf si le cas est trop ambigu ou trop peu confiant

Quand un utilisateur corrige :

• si l'embedding ressemble à un prototype existant de la bonne classe, ce prototype est mis à jour
• sinon un nouveau prototype est créé

Quand on lance la maintenance :

• les prototypes trop vieux sont affaiblis
• certains prototypes proches peuvent fusionner
• certains prototypes trop diffus peuvent être coupés
• certains prototypes faibles peuvent être supprimés

Positionnement dans une stack ML

cliving se place après le calcul des embeddings.

Pipeline typique :

1. préparer les transactions
2. construire un embedding dense par transaction
3. appeler fit(X, y) une première fois
4. appeler predict(x) pour classer
5. appeler fix(x, y_true) lorsqu'un utilisateur corrige
6. appeler clean() périodiquement
7. appeler save() / load() pour persister la mémoire

API publique

La classe principale est cliving.Cliving.

Construction

Signature :

Cliving(
    *,
    abstain_label: str = "review_needed",
    acceptance_score: float | None = None,
    conflict_score: float | None = None,
    low_confidence_score: float | None = None,
    ambiguity_margin: float | None = None,
)

Paramètres principaux :

• abstain_label : classe renvoyée quand la prédiction doit être revue
• acceptance_score : seuil minimal pour rattacher une correction à un prototype existant
• conflict_score : seuil à partir duquel un prototype concurrent est considéré comme conflit local
• low_confidence_score : seuil minimal de confiance pour accepter une classe
• ambiguity_margin : marge minimale entre top-1 et top-2

Exemple :

from cliving import Cliving

model = Cliving(
    abstain_label="review_needed",
    acceptance_score=-7.0,
    conflict_score=-6.4,
    low_confidence_score=-2.2,
    ambiguity_margin=1.0,
)

Méthodes principales

fit

fit(
    X: np.ndarray | list[list[float]],
    y: np.ndarray | list[object],
) -> Cliving

Initialise complètement la mémoire à partir d'un jeu d'embeddings et de labels.

predict

predict(
    X: np.ndarray | list[float] | list[list[float]],
) -> PredictionResult | list[PredictionResult]

Retourne directement un résultat riche :

• si X est un seul embedding 1D : un PredictionResult
• si X est un batch 2D : une list[PredictionResult]

Les champs principaux de PredictionResult sont :

• predicted_class
• class_scores
• best_prototype_id
• best_prototype_class
• best_prototype_score
• margin
• ambiguous
• low_confidence
• review_needed

fix

fix(
    embedding: np.ndarray | list[float],
    class_id: str,
    metadata: dict[str, object] | None = None,
) -> UpdateEvent

Applique une correction utilisateur sur un embedding unique.

Le metadata est optionnel et sert au debug / audit local des prototypes.

clean

clean() -> dict[str, int]

Lance manuellement la maintenance et retourne un résumé du type :

{"merged": 1, "split_created": 0, "pruned": 2}

summary

summary() -> dict[str, list[dict[str, object]]]

Retourne un résumé léger de la mémoire par classe et prototype.

save

save(path: str | Path) -> Path

Sauvegarde l'instance via pickle.

load

load(path: str | Path) -> Cliving

Recharge une instance précédemment sauvegardée.

Exemple minimal

import numpy as np

from cliving import Cliving

X = np.array(
    [
        [0.2, 1.1, -0.4],
        [0.1, 1.0, -0.3],
        [4.0, -0.2, 0.7],
    ],
    dtype=float,
)
y = np.array(["groceries", "groceries", "salary"], dtype=object)

model = Cliving(
    abstain_label="review_needed",
    acceptance_score=-7.0,
    conflict_score=-6.4,
    low_confidence_score=-2.2,
    ambiguity_margin=1.0,
)

model.fit(X, y)

result = model.predict([0.15, 1.05, -0.35])
print(result.predicted_class)
print(result.class_scores)

event = model.fix([0.18, 1.07, -0.33], "groceries")
stats = model.clean()

Construire des embeddings simples pour une transaction

cliving ne calcule pas les embeddings, mais un pipeline sklearn simple suffit souvent très bien.

Supposons une transaction avec :

• label : texte court
• amount : numérique
• support : catégorielle, par exemple cheque, carte, virement

Exemple de pipeline simple :

import numpy as np
import pandas as pd

from sklearn.compose import ColumnTransformer
from sklearn.decomposition import TruncatedSVD
from sklearn.feature_extraction.text import HashingVectorizer
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import FunctionTransformer, OneHotEncoder, StandardScaler

from cliving import Cliving


def amount_features(frame: pd.DataFrame) -> np.ndarray:
    amount = pd.to_numeric(frame.iloc[:, 0], errors="coerce").fillna(0.0).to_numpy(dtype=float)
    return np.column_stack(
        [
            amount,
            np.abs(amount),
            np.log1p(np.abs(amount)),
            np.sign(amount),
        ]
    )


transactions = pd.DataFrame(
    [
        {"label": "carrefour city paris", "amount": -12.4, "support": "carte"},
        {"label": "monoprix courses", "amount": -48.2, "support": "carte"},
        {"label": "virement salaire acme", "amount": 2450.0, "support": "virement"},
        {"label": "cheque remboursement", "amount": 120.0, "support": "cheque"},
    ]
)

y = np.array(["groceries", "groceries", "salary", "refund"], dtype=object)

preprocess = ColumnTransformer(
    transformers=[
        (
            "text",
            HashingVectorizer(
                analyzer="char_wb",
                ngram_range=(3, 5),
                n_features=2**12,
                alternate_sign=False,
                lowercase=True,
                norm="l2",
            ),
            "label",
        ),
        ("support", OneHotEncoder(handle_unknown="ignore"), ["support"]),
        ("amount", FunctionTransformer(amount_features, validate=False), ["amount"]),
    ],
    remainder="drop",
)

embedding_pipeline = Pipeline(
    steps=[
        ("features", preprocess),
        ("svd", TruncatedSVD(n_components=16, random_state=0)),
        ("scale", StandardScaler()),
    ]
)

X = embedding_pipeline.fit_transform(transactions)

model = Cliving(
    abstain_label="review_needed",
    acceptance_score=-7.0,
    conflict_score=-6.4,
    low_confidence_score=-2.2,
    ambiguity_margin=1.0,
)
model.fit(X, y)

new_tx = pd.DataFrame(
    [
        {"label": "carrefour market", "amount": -16.3, "support": "carte"}
    ]
)

z = embedding_pipeline.transform(new_tx)[0]
result = model.predict(z)
print(result.predicted_class)

Pourquoi ce pipeline est adapté :

• le texte capture la similarité locale des libellés
• la variable catégorielle sépare des modes transactionnels simples
• le montant injecte un signal métier utile
• TruncatedSVD + StandardScaler produit un espace dense compact, adapté aux prototypes et à la distance de Mahalanobis

Ce que la librairie attend en entrée

Les embeddings doivent être :

• numériques
• de dimension fixe
• convertibles en numpy.ndarray

Formats usuels :

• np.ndarray 2D pour fit
• np.ndarray 1D pour predict ou fix sur un seul exemple
• list[list[float]]
• list[float]

Démo

La démo du projet utilise l’encodeur sklearn local pour produire des embeddings, mais cet encodeur ne fait pas partie de la librairie publiée.

uv run --extra demo python examples/demo_cliving.py

Scenario simulé

Le projet contient aussi une simulation plus longue pour visualiser le caractère "vivant" de la mémoire :

• une phase d'apprentissage initial sur un historique de transactions
• puis un run en ligne avec nouvelles transactions, confirmations positives, corrections utilisateur et clean() périodiques
• une première phase plus exploratoire, suivie d'une phase plus stable où certains motifs reviennent souvent

Dans cette simulation, cliving crée de nouveaux prototypes quand il découvre des zones encore mal couvertes, puis consolide progressivement la mémoire via les update, les merge et les clean(). On cherche donc à visualiser à la fois la montée en complexité de la mémoire et sa tendance à converger quand le flux devient plus répétitif.

Tests

uv run --extra dev pytest