Domain types

Domain types for cooperative-game explainability.

@domain domain @see [[knowledge/architecture/domain]]

Conversation and explanation enumerations for the text MVP.

@domain domain @see [[knowledge/architecture/domain]] @tested-by [[tests/domain/test_enums.py]]

class mllm_shap.domain.enums.Role(*values)

Bases: StrEnum

Participant role attached to a message or token.

USER = 'user'

End-user or human-provided input.

ASSISTANT = 'assistant'

Model-generated output.

SYSTEM = 'system'

System-level instruction or steering text.

mllm_shap.domain.enums.RoleName

A Role value as plugin options spell it.

alias of Literal[‘user’, ‘assistant’, ‘system’]

class mllm_shap.domain.enums.EmbeddingMode(*values)

Bases: StrEnum

Whether value functions use static or contextual embeddings.

STATIC = 'static'

Embeddings computed from final generated tokens only.

CONTEXTUAL = 'contextual'

Embeddings derived from model internal states.

Immutable conversation snapshots without backend or IO.

@domain domain @see [[knowledge/architecture/domain]] @tested-by [[tests/domain/test_conversation.py]]

class mllm_shap.domain.conversation.TextPart(text: str)

Bases: object

Text content of a message.

text: str

Text payload.

class mllm_shap.domain.conversation.AudioPart(source: Path | bytes, sample_rate: int)

Bases: object

Audio content of a message, held in memory or referenced by path.

source: Path | bytes

Encoded audio bytes, or the path of an audio file.

sample_rate: int

Sample rate of the audio in hertz.

data() → bytes

Return the encoded audio bytes, reading source when it is a path.

mllm_shap.domain.conversation.ContentPart = mllm_shap.domain.conversation.TextPart | mllm_shap.domain.conversation.AudioPart

One part of a message’s content.

class mllm_shap.domain.conversation.Message(role: Role, text: str | None = None, content: tuple[TextPart | AudioPart, ...] | None = None, speaker: str | None = None)

Bases: object

One message of a conversation: a role and ordered content parts.

Build text messages with Message(role=..., text=...) and multimodal messages with Message(role=..., content=(TextPart(...), AudioPart(...))).

role: Role

Participant role for this message.

content: tuple[TextPart | AudioPart, ...]

Ordered content parts of the message.

speaker: str | None

Name of the participant who wrote the message, when roles are shared.

property text: str

Return the text parts joined with single spaces.

with_text(text: str) → Message

Return a copy whose text parts are replaced by one part holding text.

Non-text parts keep their order; the new text part takes the position of the first text part (or comes first when the message had none).

with_audio(audio: AudioPart) → Message

Return a copy whose first audio part is replaced by audio.

Raises:

ValueError – If the message has no audio part.

class mllm_shap.domain.conversation.Turn(messages: tuple[Message, ...])

Bases: object

One conversational turn composed of ordered messages.

messages: tuple[Message, ...]

Ordered messages that make up this turn.

class mllm_shap.domain.conversation.PartPayload

Bases: TypedDict

JSON form of one content part.

type: str

"text" or "audio".

text: NotRequired[str]

Text of a text part.

path: NotRequired[str]

Audio file of an audio part, relative to the snapshot’s base_dir.

data: NotRequired[str]

Base64 audio bytes of an audio part, instead of path.

sample_rate: NotRequired[int]

Sample rate of an audio part in hertz.

class mllm_shap.domain.conversation.MessagePayload

Bases: TypedDict

JSON form of a message: text for one text part, else content.

role: str

Message role, one of "user", "assistant" and "system".

text: NotRequired[str]

Message text, when the message is a single text part.

content: NotRequired[list[PartPayload]]

Ordered content parts, for messages with audio or several parts.

speaker: NotRequired[str]

Speaker name; omitted when the message has none.

class mllm_shap.domain.conversation.TurnPayload

Bases: TypedDict

JSON form of a turn.

messages: list[MessagePayload]

Ordered messages of the turn.

class mllm_shap.domain.conversation.SnapshotPayload

Bases: TypedDict

JSON form of a text conversation snapshot.

turns: list[TurnPayload]

Ordered turns of the conversation.

snapshot_id: NotRequired[str]

Snapshot id when it is not the one derived from the content.

class mllm_shap.domain.conversation.ConversationSnapshot(turns: tuple[Turn, ...], snapshot_id: str)

Bases: object

Frozen conversation state used as the explainability input.

turns: tuple[Turn, ...]

Ordered turns that define the conversation under study.

snapshot_id: str

Stable identifier used for deduplication and run archives.

classmethod from_turns(turns: tuple[Turn, ...]) → Self

Build a snapshot with a stable content-derived identifier.

messages() → tuple[Message, ...]

Return every message in conversation order (turns flattened).

classmethod from_json(text: str, base_dir: Path | None = None) → Self

Parse {"turns": [{"messages": [{"role": ..., "text": ...}]}]}.

A message may add "speaker", and gives either "text" or a "content" list of {"type": "text", "text": ...} and {"type": "audio", "path" or "data": ..., "sample_rate": ...} parts; relative audio paths resolve against base_dir (default: the working directory) and "data" is base64. An optional top-level "snapshot_id" sets the id; without it the id is derived from the content, as in from_turns().

Raises:

• ValueError – If text is not valid JSON, a role is unknown, or a turn, the snapshot or its id is empty.

• TypeError – If the JSON does not have the documented shape.

to_json(indent: int | None = 2) → str

Serialize the snapshot in the from_json() format.

"snapshot_id" is written only when it differs from the id derived from the content, so from_json() restores the same id either way. Audio held in memory is written as base64 "data", audio files as "path".

Player indexing for cooperative-game explainability.

@domain domain @see [[knowledge/architecture/domain]] @tested-by [[tests/domain/test_players.py]]

class mllm_shap.domain.players.PlayerSpan(message_index: int, start: int, end: int, text: str, speaker: str | None = None, label: str | None = None, samples: tuple[int, int] | None = None)

Bases: object

Where one player sits in the conversation it was partitioned from.

message_index: int

Index of the message in conversation order (turns flattened).

start: int

Start character offset of the player inside Message.text.

end: int

End character offset (exclusive) of the player inside Message.text.

text: str

Text the player covers, Message.text[start:end].

speaker: str | None

Speaker of the message the player sits in.

label: str | None

Display label set by the partitioner; display_label falls back to the text.

samples: tuple[int, int] | None

Sample range [start, end) of an audio player in the message’s first AudioPart; None for text-only players.

property display_label: str

Return label, else the text prefixed with "<speaker>: ".

class mllm_shap.domain.players.PlayerSet(player_ids: tuple[str, ...], spans: tuple[PlayerSpan, ...] = (), hidden_text: tuple[int, ...] = ())

Bases: object

Ordered explainability players aligned with coalition masks.

player_ids: tuple[str, ...]

Stable player identifiers in coalition-mask order.

spans: tuple[PlayerSpan, ...]

Player locations in mask order, filled by the partitioner.

Absence policies and label helpers need them; estimators do not.

hidden_text: tuple[int, ...]

Message indices whose text parts absence policies drop from every coalition: a transcript used only to align the audio, which the model must not read.

property num_players: int

Return the number of players in the set.

Coalition masks over explainability players.

@domain domain @see [[knowledge/architecture/domain]] @tested-by [[tests/domain/test_coalition.py]]

class mllm_shap.domain.coalition.CoalitionMask(present: tuple[bool, ...])

Bases: object

Boolean presence mask aligned with a player set.

present: tuple[bool, ...]

Presence flags in PlayerSet order.

classmethod from_sequence(present: Sequence[bool]) → Self

Build a mask from any boolean sequence.

coalition_size() → int

Return the number of present players.

validate_against(player_set: PlayerSet) → None

Ensure the mask length matches the player set.

Cooperative-game domain types.

@domain domain @see [[knowledge/architecture/domain]] @tested-by [[tests/domain/test_game.py]]

class mllm_shap.domain.game.CooperativeGame(player_set: PlayerSet)

Bases: object

Player set and reference values for a characteristic function.

player_set: PlayerSet

Ordered explainability players that define the game.

Generation payloads scored by value functions.

@domain domain @see [[knowledge/architecture/domain]]

class mllm_shap.domain.generation.GenerationRecord(text: str, text_token_rows: tuple[tuple[int, ...], ...] = (), audio_token_rows: tuple[tuple[int, ...], ...] = (), logprobs: tuple[float, ...] | None = None, embedding: tuple[float, ...] = (), contextual_embedding: tuple[float, ...] = ())

Bases: object

Concrete generation output consumed by built-in value functions.

text: str

Generated text payload.

text_token_rows: tuple[tuple[int, ...], ...] = ()

Token rows hashed for TF-IDF similarity (multimodal text stream).

audio_token_rows: tuple[tuple[int, ...], ...] = ()

Token rows hashed for TF-IDF similarity (multimodal audio stream).

logprobs: tuple[float, ...] | None = None

Natural-log probability of each generated token; None when the backend did not return them.

embedding: tuple[float, ...] = ()

Static embedding vector for embedding-based value functions.

contextual_embedding: tuple[float, ...] = ()

Contextual embedding vector for U2-style scoring.

Estimand labels for attribution outputs and archive manifests.

@domain domain @see [[knowledge/architecture/domain]] @tested-by [[tests/domain/test_estimands.py]]

mllm_shap.domain.estimands.EstimandWire

Wire values used in manifests and configuration payloads.

alias of Literal[‘shapley’, ‘banzhaf’]

class mllm_shap.domain.estimands.Estimand(*values)

Bases: StrEnum

Supported cooperative-game value formulations.

SHAPLEY = 'shapley'

Shapley value with coalition weights k!(n-k-1)!/n!.

BANZHAF = 'banzhaf'

Banzhaf index with uniform coalition weights 1/2^(n-1).

mllm_shap.domain.estimands.estimand_to_wire(estimand: Estimand) → Literal['shapley', 'banzhaf']

Serialize an estimand enum member to its wire value.

Generation and runtime settings shared by backends, runtime and pipeline.

@domain domain @see [[knowledge/architecture/domain]]

class mllm_shap.domain.config.GenerationConfig(*, max_new_tokens: int = 128, temperature: float = 0.0, top_k: int = 1, compute_embeddings: bool = False)

Bases: BaseModel

Generation parameters for the base and coalition evaluations.

model_config = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

max_new_tokens: int

Maximum tokens generated per coalition evaluation.

temperature: float

Sampling temperature for model generation.

top_k: int

Top-k sampling cutoff for model generation.

compute_embeddings: bool

Ask backends that can (transformers) to attach model embeddings to records.

The pipeline turns it on when the value function declares needs_embeddings; it costs extra forward passes per coalition.

class mllm_shap.domain.config.DedupConfig(*, enabled: Literal['auto', 'on', 'off'] = 'auto')

Bases: BaseModel

Coalition deduplication options.

model_config = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

enabled: Literal['auto', 'on', 'off']

Deduplication mode for repeated coalition evaluations.

class mllm_shap.domain.config.KvCacheConfig(*, enabled: bool = False, max_entries: Annotated[int, Ge(ge=1)] = 8)

Bases: BaseModel

Prefix cache options for supported backends.

model_config = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

enabled: bool

Whether prefix/KV cache reuse is enabled when supported.

max_entries: int

Maximum prefix KV states a backend keeps; the oldest is evicted first.