Coalition masking and absence policies

Business context: LLM attribution depends on how absent prompt pieces are represented. Deletion, masking placeholders, and neutral fillers change model behavior, audit reproducibility, and the cost of coalition sampling at scale.

Goal: Walk through every public masking API — partitioning, absence policies, shared views, packed-mask codecs, mask-space projection, and plugin resolution.

Prerequisites: mllm-shap, Python 3.12.

Theory: See the masking theory page.

Expected output: Token players partitioned from a snapshot; delete / pad / neutral renders; stable mask hashes; MaskSpace projection; structural sharing across views; built-in and entry-point plugin resolution for all absence policies.

import random

import numpy as np

from mllm_shap import (
    CoalitionMask,
    ConversationSnapshot,
    ExplainConfig,
    Message,
    Role,
    Turn,
)
from mllm_shap.masking import (
    DeletePolicy,
    MaskBuilder,
    MaskCodec,
    MaskedSnapshot,
    MaskSpace,
    NeutralPolicy,
    PackedMask,
    PadPolicy,
    TokenPartitioner,
)
from mllm_shap.plugins import (
    PluginContext,
    PluginGroup,
    PluginRegistry,
    register_builtin_plugins,
)

Build a conversation snapshot

Business: Partitioning defines the unit of accountability — what legal and product teams treat as one explainable “player.”

Technical: TokenPartitioner splits each message into whitespace-delimited tokens and returns a PlayerSet: stable player ids (<snapshot>:<turn>:<message>:<token>) and one PlayerSpan per player with the message index, the character offsets and the covered text. Absence policies render coalitions from these spans, so text outside every span and the original whitespace are kept.

turn = Turn(
    messages=(
        Message(role=Role.USER, text="Who are you?"),
        Message(role=Role.ASSISTANT, text="I am helpful assistant."),
    )
)
snapshot = ConversationSnapshot.from_turns((turn,))

partitioner = TokenPartitioner()
players = partitioner.partition(snapshot)
mask = CoalitionMask.from_sequence((True, False, True, True, True, True, True))

print("partition:", partitioner.name)
print("snapshot_id:", snapshot.snapshot_id)
print("num_players:", players.num_players)
print("player_ids:", players.player_ids)
print(
    "spans:",
    [(span.message_index, span.start, span.end, span.text) for span in players.spans],
)
print("coalition mask:", mask.present)
mask.validate_against(players)

partition: tokens
snapshot_id: 1bbc3b933438f952
num_players: 7
player_ids: ('1bbc3b933438f952:0:0:0', '1bbc3b933438f952:0:0:1', '1bbc3b933438f952:0:0:2', '1bbc3b933438f952:0:1:0', '1bbc3b933438f952:0:1:1', '1bbc3b933438f952:0:1:2', '1bbc3b933438f952:0:1:3')
spans: [(0, 0, 3, 'Who'), (0, 4, 7, 'are'), (0, 8, 12, 'you?'), (1, 0, 1, 'I'), (1, 2, 4, 'am'), (1, 5, 12, 'helpful'), (1, 13, 23, 'assistant.')]
coalition mask: (True, False, True, True, True, True, True)

Compare absence policies

Business: The same coalition can produce different model inputs under delete, pad, or neutral semantics — a core variable in faithfulness studies and customer-facing explanations.

Technical: This coalition keeps every token except are; each policy renders that absence differently across both messages.

policies = {
    "delete": DeletePolicy(),
    "pad": PadPolicy(),
    "neutral": NeutralPolicy(),
}

for name, policy in policies.items():
    rendered = policy.apply(snapshot, players, mask)
    user_text = rendered.turns[0].messages[0].text
    assistant_text = rendered.turns[0].messages[1].text
    print(f"{name}: user={user_text!r} assistant={assistant_text!r}")

delete: user='Who you?' assistant='I am helpful assistant.'
pad: user='Who [MASK] you?' assistant='I am helpful assistant.'
neutral: user='Who ___ you?' assistant='I am helpful assistant.'

Custom policy parameters

Business: Teams often standardize on a tokenizer-specific mask token or redaction style required by policy.

Technical: PadPolicy.placeholder and NeutralPolicy.fill_char configure those renderings; in a run, set them through explanation.plugin_options (absence_policies/pad: {placeholder: ...}, see plugins_extending.ipynb). NeutralPolicy repeats the fill character once per character of the absent word.

custom_pad = PadPolicy(placeholder="<mask>")
custom_neutral = NeutralPolicy(fill_char="·")

print("pad:", custom_pad.apply(snapshot, players, mask).turns[0].messages[0].text)
print(
    "neutral:",
    custom_neutral.apply(snapshot, players, mask).turns[0].messages[0].text,
)

pad: Who <mask> you?
neutral: Who ··· you?

MaskedSnapshot views and MaskBuilder

Business: Large explain runs sample thousands of coalitions; structural sharing avoids copying full prompts and keeps infrastructure costs predictable.

Technical: MaskBuilder.view returns a lightweight MaskedSnapshot; render materializes only when needed. mask_hash supports deduplication keys.

builder = MaskBuilder(PadPolicy())
view: MaskedSnapshot = builder.view(snapshot, players, mask)

print("policy_name:", view.policy_name)
print("base_snapshot_id:", view.base_snapshot_id)
print("mask_hash:", view.mask_hash)
print("builder.policy_name:", builder.policy_name)

rendered_via_builder = builder.render(view)
rendered_direct = PadPolicy().apply(view.base, view.players, view.mask)
rendered_via_builder == rendered_direct

policy_name: pad
base_snapshot_id: 1bbc3b933438f952
mask_hash: 5788788556950189190
builder.policy_name: pad

True

MaskCodec — pack, unpack, and stable hash

Business: Stable coalition fingerprints let you deduplicate expensive model calls and prove two runs evaluated the same coalition.

Technical: MaskCodec packs boolean masks into little-endian bytes with shape-stable hashing.

packed: PackedMask = MaskCodec.pack(mask.present)
restored = MaskCodec.unpack(packed)
flat_hash = MaskCodec.hash_mask(mask.present)
row_hash = MaskCodec.hash_mask(np.asarray([mask.present], dtype=np.bool_))

print("packed bytes:", packed.words)
print("n_bits:", packed.n_bits)
print("roundtrip:", restored == mask.present)
print("hash stable across shapes:", flat_hash == row_hash)
print("matches view.mask_hash:", flat_hash == view.mask_hash)

packed bytes: b'}'
n_bits: 7
roundtrip: True
hash stable across shapes: True
matches view.mask_hash: True

MaskSpace — project feature splits to full masks

Business: System instructions and fixed boilerplate often stay present while only user content is explainable — matching how enterprises scope reviews.

Technical: MaskSpace.materialize maps a feature-level split back to a full-length mask with fixed positions always True.

space = MaskSpace(
    explainable_mask=(False, True, True, False, True),
    target_length=5,
)
feature_split = CoalitionMask.from_sequence((True, False, True))
full_mask = space.materialize(feature_split)

print("n_features:", space.n_features)
print("feature split:", feature_split.present)
print("full mask:", full_mask)

n_features: 3
feature split: (True, False, True)
full mask: (True, True, False, True, True)

Structural sharing across many coalitions

Business: Monte Carlo and exact explainers generate large mask batches; shared base snapshots keep memory flat as coalition count grows.

Technical: Many MaskedSnapshot views reference the same base object and snapshot_id.

builder = MaskBuilder(DeletePolicy())
rng = random.Random(0)
views = []
for _ in range(100):
    present = tuple(rng.choice((True, False)) for _ in range(players.num_players))
    if not any(present):
        present = (True, *present[1:])
    views.append(builder.view(snapshot, players, CoalitionMask.from_sequence(present)))

(
    len({view.base_snapshot_id for view in views}),
    all(view.base is snapshot for view in views),
)

(1, True)

Plugin registry — built-ins, entry points, and ExplainConfig

Business: Operations teams select partition and absence behaviour through config (players, absence_policy) without code changes — the same keys CI and compliance review.

Technical: register_builtin_plugins and load_entry_points resolve tokens, delete, pad, and neutral.

registry = PluginRegistry()
register_builtin_plugins(registry)
registry.load_entry_points(PluginGroup.PARTITIONS)
registry.load_entry_points(PluginGroup.ABSENCE_POLICIES)

print("partitions:", registry.names(PluginGroup.PARTITIONS))
print("absence policies:", registry.names(PluginGroup.ABSENCE_POLICIES))

default_context = PluginContext(
    config=ExplainConfig.model_validate({
        "backend": {"kind": "mock", "model_id": "stub"}
    }),
    registry=registry,
)
resolved = {
    name: registry.resolve(PluginGroup.ABSENCE_POLICIES, name, default_context).name
    for name in registry.names(PluginGroup.ABSENCE_POLICIES)
}
print("resolved absence policies:", resolved)

partitions: ('messages', 'sgpa_segments', 'tokens')
absence policies: ('delete', 'neutral', 'pad')
resolved absence policies: {'delete': 'delete', 'neutral': 'neutral', 'pad': 'pad'}

for absence_policy in ("delete", "pad", "neutral"):
    config = ExplainConfig.model_validate({
        "backend": {"kind": "mock", "model_id": "stub"},
        "explanation": {
            "players": "tokens",
            "absence_policy": absence_policy,
        },
    })
    context = PluginContext(config=config, registry=registry)
    partitioner = registry.resolve(
        PluginGroup.PARTITIONS,
        config.explanation.players,
        context,
    )
    policy = registry.resolve(
        PluginGroup.ABSENCE_POLICIES,
        config.explanation.absence_policy,
        context,
    )
    print(
        absence_policy,
        type(partitioner).__name__,
        type(policy).__name__,
        policy.name,
    )

delete TokenPartitioner DeletePolicy delete
pad TokenPartitioner PadPolicy pad
neutral TokenPartitioner NeutralPolicy neutral

Validation guards

Business: Misaligned masks or builder/policy mismatches can silently corrupt attribution studies; explicit guards surface failures before costly model calls.

Technical: CoalitionMask.validate_against, empty-coalition delete rendering (v(∅) → empty message text), and builder/policy mismatch checks.

short_mask = CoalitionMask.from_sequence((True, False, True))
try:
    short_mask.validate_against(players)
except ValueError as error:
    print("mask length guard:", error)

empty_mask = CoalitionMask.from_sequence((False,) * players.num_players)
empty_rendered = DeletePolicy().apply(snapshot, players, empty_mask)
print(
    "delete empty coalition text:",
    repr(empty_rendered.turns[0].messages[0].text),
)

mismatched_view = MaskedSnapshot(
    base=snapshot,
    players=players,
    mask=mask,
    policy_name="delete",
)
try:
    MaskBuilder(PadPolicy()).render(mismatched_view)
except ValueError as error:
    print("builder policy guard:", error)

mask length guard: coalition mask length 3 does not match player count 7
delete empty coalition text: ''
builder policy guard: masked view policy 'delete' does not match builder policy 'pad'