Pipeline

Explain pipeline types and orchestration.

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

Explain pipeline configuration schema.

GenerationConfig, DedupConfig and KvCacheConfig live in mllm_shap.domain.config so the runtime can use them without importing the pipeline; they are re-exported here.

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

class mllm_shap.pipeline.config.ArchiveConfig(*, path: str = './runs/{run_id}', flush_every: Annotated[int, Ge(ge=1)] = 50, on_exists: Literal['error', 'overwrite'] = 'error')

Bases: BaseModel

Run archive persistence options.

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

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

path: str

Filesystem path template for persisted run archives.

flush_every: int

Number of coalition records between archive flushes.

on_exists: Literal['error', 'overwrite']

What to do when the archive directory already holds a run.

class mllm_shap.pipeline.config.BackendConfig(*, kind: str, model_id: str, api_host: str | None = None, timeout_s: Annotated[float, Gt(gt=0)] = 120.0, max_retries: Annotated[int, Ge(ge=0)] = 3, options: dict[str, object] = {})

Bases: BaseModel

Backend selection and connection parameters.

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

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

kind: str

Registered backend plugin identifier.

model_id: str

Model name or repository id passed to the backend.

api_host: str | None

API host for HTTP-compatible backends.

Required for the api backend (there is no environment or localhost fallback); optional for lmstudio, which otherwise asks LM Studio for its local server. The api backend sends MLLM_SHAP_API_KEY as the bearer key to any host; OPENAI_API_KEY is used only when that is unset and the host is api.openai.com.

timeout_s: float

Per-request timeout in seconds for remote backends.

max_retries: int

Retries after a transient backend failure (0 disables retrying).

options: dict[str, object]

Backend-specific settings validated by the selected backend.

class mllm_shap.pipeline.config.BudgetConfig(*, fraction: Annotated[float | None, Gt(gt=0.0), Le(le=1.0)] = None, evaluations: Annotated[int | None, Strict(strict=True), Ge(ge=1)] = None)

Bases: BaseModel

Coalition budget: a fraction of the 2^n coalitions or an absolute count.

Set at most one of fraction and evaluations; with neither the budget is the full table (fraction: 1.0). Only the key that is set is serialised, so configurations without evaluations keep their run id.

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

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

fraction: float | None

Share of the 2^n coalitions evaluated; 1.0 when evaluations is unset.

evaluations: int | None

Coalition evaluations in total, empty and grand included, instead of fraction; raised to the estimator’s floor (2 for mc, 2n + 2 for complementary and neyman_cc, which also round down to whole pairs) and capped at 2^n.

class mllm_shap.pipeline.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.pipeline.config.ExplainConfig(*, backend: BackendConfig, generation: GenerationConfig = GenerationConfig(max_new_tokens=128, temperature=0.0, top_k=1, compute_embeddings=False), explanation: ExplanationConfig = ExplanationConfig(estimand=<Estimand.SHAPLEY: 'shapley'>, estimator='neyman_cc', value_fn='tfidf_cosine', normalizer='identity', players='tokens', absence_policy='delete', budget=BudgetConfig(fraction=1.0, evaluations=None), include_minimal_masks=False, neyman=NeymanConfig(use_standard_method=False, initial_fraction=None, initial_num_samples=None), max_inflight=2, max_players=24, archive=ArchiveConfig(path='./runs/{run_id}', flush_every=50, on_exists='error'), dedup=DedupConfig(enabled='auto'), kv_cache=KvCacheConfig(enabled=False, max_entries=8), embedding_mode=<EmbeddingMode.STATIC: 'static'>, seed=42, plugin_options={}))

Bases: BaseModel

Top-level explain pipeline configuration.

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

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

backend: BackendConfig

Backend selection and connection parameters.

generation: GenerationConfig

Generation parameters for base and coalition evaluations.

explanation: ExplanationConfig

Explanation algorithm and runtime controls.

class mllm_shap.pipeline.config.ExplanationConfig(*, estimand: Estimand = Estimand.SHAPLEY, estimator: str = 'neyman_cc', value_fn: str = 'tfidf_cosine', normalizer: str = 'identity', players: str = 'tokens', absence_policy: str = 'delete', budget: BudgetConfig = BudgetConfig(fraction=1.0, evaluations=None), include_minimal_masks: bool = False, neyman: NeymanConfig = NeymanConfig(use_standard_method=False, initial_fraction=None, initial_num_samples=None), max_inflight: Annotated[int, Ge(ge=1)] = 2, max_players: Annotated[int, Ge(ge=1), Le(le=62)] = 24, archive: ArchiveConfig = ArchiveConfig(path='./runs/{run_id}', flush_every=50, on_exists='error'), dedup: DedupConfig = DedupConfig(enabled='auto'), kv_cache: KvCacheConfig = KvCacheConfig(enabled=False, max_entries=8), embedding_mode: EmbeddingMode = EmbeddingMode.STATIC, seed: int = 42, plugin_options: dict[str, dict[str, object]] = {})

Bases: BaseModel

Explanation algorithm and runtime controls.

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

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

estimand: Estimand

Cooperative-game value targeted by aggregation.

estimator: str

Registered estimator plugin used to sample coalitions.

value_fn: str

Registered value-function plugin used to score outputs.

normalizer: str

Registered normalizer applied after estimand aggregation.

players: str

Registered partition plugin that defines explainability players.

absence_policy: str

Registered absence policy used to render masked snapshots.

budget: BudgetConfig

Estimator budget controls.

include_minimal_masks: bool

Whether minimal coalitions are included in estimator sampling.

neyman: NeymanConfig

Neyman-CC controls applied when the Neyman estimator is selected.

max_inflight: int

Maximum concurrent coalition evaluations.

max_players: int

Runs planning more than 2^max_players coalitions (fraction * 2^n, or evaluations) are refused. The ceiling of 62 keeps 2^n inside a signed 64-bit integer, which the samplers’ NumPy mask arithmetic needs.

archive: ArchiveConfig

Run archive persistence settings.

dedup: DedupConfig

Coalition deduplication settings.

kv_cache: KvCacheConfig

Prefix cache settings for supported backends.

embedding_mode: EmbeddingMode

Embedding mode used by embedding-based value functions.

seed: int

Random seed for reproducible coalition sampling.

plugin_options: dict[str, dict[str, object]]

Per-plugin options keyed by plugin name, passed as PluginContext.options.

class mllm_shap.pipeline.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.pipeline.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.

class mllm_shap.pipeline.config.NeymanConfig(*, use_standard_method: bool = False, initial_fraction: Annotated[float | None, Gt(gt=0.0), Le(le=1.0)] = None, initial_num_samples: Annotated[int | None, Ge(ge=1)] = None)

Bases: BaseModel

Neyman-CC estimator controls used when estimator is neyman_cc.

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

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

use_standard_method: bool

Whether phase-one sampling uses unstructured random coalitions.

initial_fraction: float | None

Fraction of the total budget used during phase-one initialization, in (0, 1].

initial_num_samples: int | None

Explicit phase-one pair budget overriding initial_fraction; at least 1.

mllm_shap.pipeline.config.explain_config_from_yaml(text: str) → ExplainConfig

Parse YAML text into a validated ExplainConfig.

Unknown keys and values of the wrong type or range raise pydantic’s ValidationError.

Raises:

ConfigError – If the YAML root is not a mapping (an empty document included) or settings are inconsistent with each other (for example estimator: exact with a partial budget).

mllm_shap.pipeline.config.explain_config_to_yaml(config: ExplainConfig) → str

Serialize a config to YAML with stable key ordering within sections.

Public explain pipeline entry point.

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

class mllm_shap.pipeline.runner.ExplainRunner(config: ExplainConfig, registry: PluginRegistry | None = None, telemetry: ObservabilitySink | None = None, progress: CoalitionProgress | None = None, backend: GenerativeBackend | None = None)

Bases: object

Run the explain pipeline for a conversation snapshot.

With an injected backend, explain_sync() runs every call on one event loop owned by the runner, so backends holding loop-bound resources (HTTP clients, locks) work across calls. close(), aclose() or a with / async with block close the backend and that loop.

property registry: PluginRegistry

Return the plugin registry used by this runner.

async explain(snapshot: ConversationSnapshot) → ExplainRunOutput

Execute the async explain pipeline.

async aclose() → None

Close the injected backend and the loop of explain_sync().

When explain_sync() already ran, the backend is closed on its own loop (in a worker thread), since its resources are bound to that loop.

close() → None

Close the injected backend and the loop of explain_sync().

Raises:

RuntimeError – If called from a running event loop before explain_sync() created its loop; use await     runner.aclose() there.

explain_sync(snapshot: ConversationSnapshot) → ExplainRunOutput

Run explain() to completion from synchronous code.

Without an injected backend each call uses a fresh event loop. With one, every call reuses a loop the runner creates on first use and closes in close() / aclose().

Raises:

RuntimeError – If called from a running event loop (for example Jupyter); use await runner.explain(snapshot) there.

reanalyze(archive_root: Path | str, strict: bool = False) → ExplainRunOutput

Rescore an archived run with the runner’s current explain settings.

Embedding value functions score hash embeddings of the archived text. When the archived run scored backend embeddings, this logs a WARNING, or with strict raises ReanalyzeMismatchError.

Explain pipeline result types.

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

class mllm_shap.pipeline.result.PlayerLabelPayload

Bases: TypedDict

JSON form of one player: its id and, when known, where it sits.

player_id: str

Player identifier in mask bit order.

text: str | None

Text the player covers; None when the output has no player spans.

message_index: int | None

Message index (turns flattened); None without player spans.

start: int | None

Start character offset in the message; None without player spans.

end: int | None

End character offset (exclusive); None without player spans.

speaker: str | None

Speaker of the player’s message; None when unset or unknown.

label: str | None

Display label (PlayerSpan.display_label); None without spans.

class mllm_shap.pipeline.result.ExplainRunPayload

Bases: TypedDict

JSON form of an explain or reanalyze run, keys in a stable order.

labels: list[PlayerLabelPayload]

One entry per player, aligned with values.

base_value: float

Payoff of the empty coalition v(empty).

grand_value: float

Payoff of the grand coalition v(N).

estimand: Literal['shapley', 'banzhaf']

Estimand used to aggregate the coalitions.

metrics: dict[str, int]

Scheduler counters (SchedulerMetrics).

manifest: RunManifestPayload

Run manifest, as written next to the archive.

perf: dict[str, float] | None

Wall-clock breakdown in milliseconds, when measured.

class mllm_shap.pipeline.result.ExplainResult(estimand: Estimand, values: tuple[float, ...], base_value: float = 0.0, grand_value: float = 0.0)

Bases: object

Attribution output for a single explain run.

estimand: Estimand

Estimand used to aggregate coalition samples.

values: tuple[float, ...]

Per-player attribution values.

base_value: float

Payoff of the empty coalition v(empty); Shapley values sum to v(N) - v(empty).

grand_value: float

Payoff of the grand coalition v(N), scored against the base generation.

class mllm_shap.pipeline.result.ExplainRunOutput(result: ExplainResult, metrics: SchedulerMetrics, run_id: str, manifest: RunManifest, perf: PerfSummary | None = None, player_set: PlayerSet | None = None)

Bases: object

Explain pipeline output including scheduler metrics.

result: ExplainResult

Aggregated and normalized attribution values.

metrics: SchedulerMetrics

Coalition scheduler counters for the run.

run_id: str

Stable identifier for the explain run.

manifest: RunManifest

Manifest metadata persisted with optional archives.

perf: PerfSummary | None

Optional wall-clock breakdown for the run.

player_set: PlayerSet | None

Players of the explained snapshot; None for reanalyzed runs.

to_dict() → ExplainRunPayload

Return a JSON-compatible payload with a stable key order.

Labels carry each player’s text and span when player_set is known (explain); after reanalyze only the archived player ids are available and the span fields are None.

Versioned run manifest persisted next to every run archive.

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

class mllm_shap.pipeline.manifest.EmbeddingSource(*values)

Bases: StrEnum

Which embeddings the value function of a run scored.

NONE = 'none'

The value function read no embeddings.

HASH = 'hash'

Hash embeddings derived from the generated text, as reanalyze scores.

BACKEND = 'backend'

Model embeddings computed by the backend; reanalyze cannot reproduce them.

class mllm_shap.pipeline.manifest.RunManifest(estimand: Estimand, run_id: str, snapshot_id: str = '', estimator: str = '', value_fn: str = '', normalizer: str = '', absence_policy: str = '', players: str = '', backend_kind: str = '', model_id: str = '', player_ids: tuple[str, ...]=(), embedding_source: EmbeddingSource = EmbeddingSource.NONE, embedding_model: str = '', config: ExplainConfig | None = None, mllm_shap_version: str = <factory>, schema_version: int = 4)

Bases: object

Top-level metadata persisted alongside a run archive.

Only estimand and run_id are required so hand-built archives stay simple; the explain pipeline fills every field.

estimand: Estimand

Estimand label recorded for every attribution output.

run_id: str

Stable identifier for the explain run.

snapshot_id: str

Identifier of the explained conversation snapshot.

estimator: str

Registered estimator plugin that sampled the coalitions.

value_fn: str

Registered value function that scored the coalitions.

normalizer: str

Registered normalizer applied after aggregation.

absence_policy: str

Registered absence policy used to render masked snapshots.

players: str

Registered partition plugin that defined the players.

backend_kind: str

Registered backend kind that generated the coalition outputs.

model_id: str

Backend model identifier that generated the coalition outputs.

player_ids: tuple[str, ...]

Player identifiers in mask bit order.

embedding_source: EmbeddingSource

Which embeddings the value function scored.

embedding_model: str

Backend model that computed the embeddings when embedding_source is "backend"; empty otherwise.

config: ExplainConfig | None

Explain configuration of the run, when the pipeline wrote the manifest.

mllm_shap_version: str

Package version that wrote the manifest.

schema_version: int

Manifest schema version.

to_dict() → RunManifestPayload

Serialize manifest fields to JSON-compatible primitives.

class mllm_shap.pipeline.manifest.RunManifestPayload

Bases: TypedDict

JSON-compatible manifest payload persisted with a run archive.

schema_version: int

Manifest schema version; only MANIFEST_SCHEMA_VERSION is accepted.

mllm_shap_version: str

Package version that wrote the manifest.

run_id: str

Stable identifier for the explain run.

snapshot_id: str

Identifier of the explained conversation snapshot.

estimand: Literal['shapley', 'banzhaf']

Estimand label recorded for every attribution output.

estimator: str

Registered estimator plugin that sampled the coalitions.

value_fn: str

Registered value function that scored the coalitions.

normalizer: str

Registered normalizer applied after aggregation.

absence_policy: str

Registered absence policy used to render masked snapshots.

players: str

Registered partition plugin that defined the players.

backend_kind: str

Registered backend kind that generated the coalition outputs.

model_id: str

Backend model identifier that generated the coalition outputs.

player_ids: list[str]

Player identifiers in mask bit order.

embedding_source: str

Which embeddings the run scored, as an EmbeddingSource value.

embedding_model: str

Model that computed backend embeddings; empty unless "backend".

config: dict[str, object]

Full explain configuration echo (ExplainConfig.model_dump(mode="json")).

mllm_shap.pipeline.manifest.parse_manifest(payload: Mapping[str, object]) → RunManifest

Parse a JSON manifest payload into a run manifest.

A missing embedding_source reads as "none" and a missing embedding_model as empty.

Raises:

• ManifestVersionError – If schema_version is missing or differs from MANIFEST_SCHEMA_VERSION.

• ArchiveCorruptError – If a field has the wrong type or value, the recorded config is invalid, or the manifest records a config but lacks the run fields the explain pipeline always writes.

Reanalyze archived coalition generations without backend calls.

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

mllm_shap.pipeline.reanalyze.reanalyze_sync(archive_root: Path, config: ExplainConfig, registry: PluginRegistry, *, manifest: RunManifest | None = None, telemetry: ObservabilitySink | None = None, strict: bool = False) → ExplainRunOutput

Rescore and re-aggregate an archived explain run without backend calls.

The archived estimator re-aggregates the coalitions, so reanalyze reproduces explain for the same value function. Archives store text only, so embedding value functions score hash embeddings of that text; when the archived run scored backend embeddings this logs a WARNING, or raises with strict.

Raises:

• ValueError – If the archive has no coalition rows or no base generation.

• ArchiveNotFoundError – If the manifest or the SQLite database is missing.

• ArchiveCorruptError – If the manifest or the database cannot be read.

• ManifestVersionError – If the archive was written by another release.

• ArchiveIncompleteError – If the archive was not written by ExplainRunner, its run did not finish, it lacks the grand or empty coalition, or a full-budget run (budget.fraction 1.0, which includes every exact run) lacks some of the 2^n coalitions.

• ReanalyzeMismatchError – If config uses a different estimator, absence policy or partition than the archived run, the value function needs logprobs and a row has none, or, with strict, the value function reads embeddings and the archived run scored backend embeddings.

• ConfigError – If explanation.plugin_options names a plugin outside the run or holds an unknown option.

Deletion curves: how fast a ranking of players destroys the answer.

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

class mllm_shap.pipeline.faithfulness.DeletionCurve(num_players: int, grand_value: float, removed: tuple[int, ...], values: tuple[float, ...])

Bases: object

Value of the coalition left after removing the top-k players, per k.

num_players: int

Number of players of the explained snapshot.

grand_value: float

v(N), the value with every player present.

removed: tuple[int, ...]

Numbers k of top-ranked players removed, ascending.

values: tuple[float, ...]

v(N - top_k) for each k in removed.

property drops: tuple[float, ...]

Return v(N) - v(N - top_k) for each k.

property area: float

Return the trapezoid area of values over k / n.

Lower is more faithful: the value falls sooner when the ranking puts the players that matter first.

async mllm_shap.pipeline.faithfulness.deletion_curve(config: ExplainConfig, snapshot: ConversationSnapshot, attributions: Sequence[float], ks: Sequence[int] | None = None, registry: PluginRegistry | None = None, backend: GenerativeBackend | None = None) → DeletionCurve

Score snapshot with its k highest-attributed players removed at once.

Players are ranked by attributions (ties keep player order) and each k in ks (default 0..n) removes the top k together, never greedily, using config’s partition, absence policy, backend and value function. Use a different value function than the explanation’s to avoid a circular check. A backend (for example the explained run’s loaded model) is reused and left open; without one the run builds and closes its own.

Raises:

ValueError – If attributions does not have one value per player, or a k is outside 0..n.

Explain pipeline orchestration.

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

class mllm_shap.pipeline.orchestrator.ExplainOrchestrator(context: ExplainContext, backend: GenerativeBackend)

Bases: object

Wire partition, sampling, generation, scoring, and aggregation.

async run() → ExplainRunOutput

Execute the configured explain pipeline and return attributions.

mllm_shap.pipeline.orchestrator.score_coalitions(value_fn: ValueFunction, base_record: GenerationRecord, coalition_records: Sequence[GenerationRecord]) → list[float]

Fit a corpus-dependent value_fn on all records, then score each coalition.

Shared state for one explain pipeline run.

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

class mllm_shap.pipeline.context.ExplainContext(config: ExplainConfig, snapshot: ConversationSnapshot, player_set: PlayerSet, registry: PluginRegistry, run_id: str, archive_root: Path | None = None, telemetry: ObservabilitySink | None = None, progress: CoalitionProgress | None = None)

Bases: object

Inputs resolved for one explain orchestrator run.

config: ExplainConfig

Validated explain configuration.

snapshot: ConversationSnapshot

Conversation under explanation.

player_set: PlayerSet

Explainability players derived from the snapshot.

registry: PluginRegistry

Plugin registry used to resolve pipeline components.

run_id: str

Stable identifier for archive paths and manifests.

archive_root: Path | None

Optional filesystem root for run archive persistence.

telemetry: ObservabilitySink | None

Optional observability sink for stage spans.

progress: CoalitionProgress | None

Optional sync sink for coalition scheduling progress.

plugin_context() → PluginContext

Return the context passed to plugin factories for this run.

resolved_telemetry() → ObservabilitySink

Return the configured telemetry sink or a no-op implementation.

resolved_progress() → CoalitionProgress

Return the configured progress sink or a no-op implementation.