Explain a conversation with ExplainRunner

Business context: a reviewer needs to see what was explained (the conversation and its players), how (the configuration), and whether the numbers are consistent (efficiency). It takes one conversation to a JSON result that can sit next to a ticket or an audit record.

Goal: the public explain workflow: messages, snapshot JSON, player spans, YAML configuration and its guards, explain / explain_sync, the efficiency check, the JSON result, every built-in estimator and estimand, and run monitoring.

Prerequisites: mllm-shap, Python 3.12. A rule-based stand-in plays the model, so every number is reproducible; text_explain_e2e.ipynb uses real backends.

Expected output: player spans, refused configurations, efficient exact Shapley values, an estimator and estimand table, and the stage events of a monitored run.

import asyncio
import json
import re

from pydantic import ValidationError

from mllm_shap import (
    AudioPart,
    CoalitionMask,
    ConfigError,
    ConversationSnapshot,
    DeletePolicy,
    ExplainConfig,
    ExplainRunner,
    ExplainRunOutput,
    GenerationRecord,
    InMemoryObservabilitySink,
    Message,
    MllmShapError,
    PluginContext,
    Role,
    TextPart,
    TokenPartitioner,
    Turn,
    __version__,
    configure_logging,
    explain_config_from_yaml,
    explain_config_to_yaml,
)
from mllm_shap.domain.config import GenerationConfig
from mllm_shap.value.generation_records import generation_record_from_text

A stand-in model

Business: a deterministic stand-in makes the walkthrough reproducible anywhere; a real backend changes one configuration key, not the workflow.

Technical: a backend is any object with a model_id property and an async generate(snapshot, generation) that returns a GenerationRecord, here built with generation_record_from_text. The bot confirms the refund, receipt and deadline when those words reach it; otherwise it asks for clarification.

class RefundBot:
    """Rule-based stand-in for a support model."""

    model_id = "refund-bot"

    async def generate(
        self, snapshot: ConversationSnapshot, generation: GenerationConfig
    ) -> GenerationRecord:
        """Answer from the keywords that survived masking."""
        _ = generation
        prompt = " ".join(message.text for message in snapshot.messages())
        words = set(re.findall(r"[a-z]+", prompt.lower()))
        parts = []
        if "refund" in words:
            parts.append("refund approved")
        if "receipt" in words:
            parts.append("receipt checked")
        if words & {"thirty", "days"}:
            parts.append("deadline met")
        reply = " ".join(parts) or "please clarify"
        return generation_record_from_text(reply)

The conversation

Business: the snapshot is the unit that gets explained and archived; its id comes from the content, so the same conversation gets the same id.

Technical: Message(role=..., text=...) holds one TextPart; Message(role=..., content=(...)) holds several parts, and Message.text joins the text parts. ConversationSnapshot.to_json / from_json use the command-line format and keep the id; messages with several parts or audio cannot be written.

SYSTEM = (
    "You are a support assistant for a store. Answer only from the question, in one "
    'short sentence. If it does not give enough information, say "Not enough '
    'information." Never invent policy details and never ask a question.'
)
snapshot = ConversationSnapshot.from_turns((
    Turn(
        messages=(
            Message(role=Role.SYSTEM, text=SYSTEM),
            Message(role=Role.USER, text="Refund within thirty days with receipt?"),
        )
    ),
))
print("snapshot_id:", snapshot.snapshot_id)
print(snapshot.to_json(indent=None))
restored = ConversationSnapshot.from_json(snapshot.to_json())
print("JSON round trip keeps the id:", restored.snapshot_id == snapshot.snapshot_id)

split = Message(
    role=Role.USER, content=(TextPart("Refund within"), TextPart("thirty days?"))
)
print("text of a two-part message:", repr(split.text))
voice = Message(
    role=Role.USER,
    content=(TextPart("Recording attached."), AudioPart(b"RIFF", sample_rate=16_000)),
)
try:
    ConversationSnapshot.from_turns((Turn(messages=(voice,)),)).to_json()
except ValueError as exc:
    print("to_json refuses:", exc)

snapshot_id: 84caf96b080cfbe6
{"turns": [{"messages": [{"role": "system", "text": "You are a support assistant for a store. Answer only from the question, in one short sentence. If it does not give enough information, say \"Not enough information.\" Never invent policy details and never ask a question."}, {"role": "user", "text": "Refund within thirty days with receipt?"}]}]}
JSON round trip keeps the id: True
text of a two-part message: 'Refund within thirty days?'

Configure the run in YAML

Business: the YAML file is the reviewable record of how attributions were produced; it is stored in the run manifest and hashed into the run_id.

Technical: explain_config_from_yaml validates the text into an ExplainConfig; explain_config_to_yaml writes it back with every default filled in. The run below injects the stand-in, so backend.kind is required but unused.

CONFIG_YAML = """
backend:
  kind: mock
  model_id: refund-bot
generation:
  max_new_tokens: 16
explanation:
  estimator: exact
  estimand: shapley
  value_fn: tfidf_cosine
  absence_policy: delete
  budget:
    fraction: 1.0
  archive:
    path: ""   # no archive here; see archive_reanalyze.ipynb
  seed: 0
  plugin_options:
    tokens: {roles: [user]}   # explain the user's words; the system prompt stays fixed
"""
config = explain_config_from_yaml(CONFIG_YAML)
print(explain_config_to_yaml(config))

backend:
  kind: mock
  model_id: refund-bot
  api_host: null
  timeout_s: 120.0
  max_retries: 3
  options: {}
generation:
  max_new_tokens: 16
  temperature: 0.0
  top_k: 1
  compute_embeddings: false
explanation:
  estimand: shapley
  estimator: exact
  value_fn: tfidf_cosine
  normalizer: identity
  players: tokens
  absence_policy: delete
  budget:
    fraction: 1.0
  include_minimal_masks: false
  neyman:
    use_standard_method: false
    initial_fraction: null
    initial_num_samples: null
  max_inflight: 2
  max_players: 24
  archive:
    path: ''
    flush_every: 50
    on_exists: error
  dedup:
    enabled: auto
  kv_cache:
    enabled: false
    max_entries: 8
  embedding_mode: static
  seed: 0
  plugin_options:
    tokens:
      roles:
      - user

Players and their spans

Business: players are the units that receive credit; reviewers need to map every value back to its exact text.

Technical: the tokens partitioner splits messages on whitespace into a PlayerSet: one id per player and one PlayerSpan (message index, character offsets, covered text). Policies and labels read these spans, so punctuation stays attached to its word (receipt?). The roles option keeps the system prompt out of the game: it is sent every time and gets no credit.

print("players if every role were explained:", TokenPartitioner().partition(snapshot).num_players)
options = config.explanation.plugin_options["tokens"]
player_set = TokenPartitioner(
    PluginContext(config=config, options=options)
).partition(snapshot)
for player_id, span in zip(player_set.player_ids, player_set.spans, strict=True):
    print(
        f"{player_id}  message {span.message_index} "
        f"[{span.start:>2}:{span.end:>2}]  {span.text!r}"
    )

players if every role were explained: 43
84caf96b080cfbe6:0:1:0  message 1 [ 0: 6]  'Refund'
84caf96b080cfbe6:0:1:1  message 1 [ 7:13]  'within'
84caf96b080cfbe6:0:1:2  message 1 [14:20]  'thirty'
84caf96b080cfbe6:0:1:3  message 1 [21:25]  'days'
84caf96b080cfbe6:0:1:4  message 1 [26:30]  'with'
84caf96b080cfbe6:0:1:5  message 1 [31:39]  'receipt?'

Validation guards

Business: a configuration that cannot produce the attribution it names should fail before any model call is paid for.

Technical: contradictory settings raise ConfigError: exact needs budget.fraction: 1.0, complementary and neyman_cc estimate Shapley values only, and neyman_cc does not take include_minimal_masks. Unknown keys and wrong types raise pydantic’s ValidationError. Every deliberate error derives from MllmShapError and a built-in exception (ConfigError is a ValueError); the command line exits with 2 for them (cli.ipynb).

def refused(yaml_text: str) -> None:
    """Print why a configuration is refused."""
    try:
        explain_config_from_yaml(yaml_text)
    except (ConfigError, ValidationError) as exc:
        reason = " / ".join(str(exc).splitlines()[:2])
        print(f"{type(exc).__name__}: {reason}")


BACKEND = "backend: {kind: mock, model_id: refund-bot}\n"
refused(BACKEND + "explanation: {estimator: exact, budget: {fraction: 0.5}}")
refused(BACKEND + "explanation: {estimator: complementary, estimand: banzhaf}")
refused(BACKEND + "explanation: {estimator: neyman_cc, include_minimal_masks: true}")
refused(BACKEND + "explanation: {estimatr: exact}")
refused("")
print(
    "ConfigError is a ValueError and an MllmShapError:",
    issubclass(ConfigError, ValueError) and issubclass(ConfigError, MllmShapError),
)

ConfigError: estimator 'exact' enumerates every coalition; set budget.fraction: 1.0
ConfigError: estimator 'complementary' estimates the shapley estimand only; use estimator 'mc' or 'exact' for banzhaf
ConfigError: estimator 'neyman_cc' does not support include_minimal_masks: true
ValidationError: 1 validation error for ExplainConfig / explanation.estimatr
ConfigError: explain config root must be a mapping, got NoneType
ConfigError is a ValueError and an MllmShapError: True

Run the explanation

Business: efficiency is a check a reviewer can do by hand: exact Shapley values add up to the change in the value function between the empty and the full prompt.

Technical: await runner.explain(snapshot) runs the pipeline and reuses the injected backend. Inside Jupyter’s event loop explain_sync refuses; from a worker thread it returns the same result. With tfidf_cosine a coalition’s value is the similarity of its answer to the full prompt’s answer, so grand_value is 1 and base_value (the “please clarify” answer) is 0. The metrics count 2^6 = 64 coalitions; the grand coalition reuses the base generation, the one cache hit.

runner = ExplainRunner(config, backend=RefundBot())
output = await runner.explain(snapshot)
result = output.result
for span, value in zip(output.player_set.spans, result.values, strict=True):
    print(f"{span.text:>9} {value:+.4f}")
print("base_value:", result.base_value, " grand_value:", result.grand_value)
print("sum of values:          ", round(sum(result.values), 6))
print("grand_value - base_value:", round(result.grand_value - result.base_value, 6))
print(output.metrics)
try:
    runner.explain_sync(snapshot)
except RuntimeError as exc:
    print("explain_sync inside Jupyter:", exc)


def explain_in_a_script() -> ExplainRunOutput:
    """What a script does: a blocking call; ``with`` closes the runner."""
    with ExplainRunner(config, backend=RefundBot()) as script_runner:
        return script_runner.explain_sync(snapshot)


scripted = await asyncio.to_thread(explain_in_a_script)
print(
    "explain_sync without an event loop gives the same result:",
    scripted.result == result,
)

   Refund +0.3435
   within +0.0000
   thirty +0.1565
     days +0.1565
     with +0.0000
 receipt? +0.3435
base_value: 0.0  grand_value: 1.0
sum of values:           1.0
grand_value - base_value: 1.0
SchedulerMetrics(requested=64, executed=63, deduplicated=0, cache_hits=1, kv_cache_hits=0, kv_cache_reused_tokens=0)
explain_sync inside Jupyter: explain_sync cannot run inside an event loop; use 'await runner.explain(...)'
explain_sync without an event loop gives the same result: True

What the model answers: the full prompt, then the prompt with one user word deleted.

backend = RefundBot()
full = await backend.generate(snapshot, config.generation)
print(f"{'full prompt':>16} -> {full.text}")
for index, span in enumerate(output.player_set.spans):
    mask = CoalitionMask.from_sequence(
        tuple(other != index for other in range(output.player_set.num_players))
    )
    absent = DeletePolicy().apply(snapshot, output.player_set, mask)
    answer = await backend.generate(absent, config.generation)
    print(f"{'without ' + span.text:>16} -> {answer.text}")

     full prompt -> refund approved receipt checked deadline met
  without Refund -> receipt checked deadline met
  without within -> refund approved receipt checked deadline met
  without thirty -> refund approved receipt checked deadline met
    without days -> refund approved receipt checked deadline met
    without with -> refund approved receipt checked deadline met
without receipt? -> refund approved deadline met

within and with never change the answer, so they are null players with value 0. Refund and receipt? each switch one phrase on and get the largest, equal shares. Either thirty or days triggers the deadline phrase, so they split its credit (symmetry). The two printed sums agree: exact Shapley values are efficient.

The JSON result

Business: a stable JSON payload can be attached to a ticket, diffed between releases and loaded by dashboards without mllm-shap.

Technical: ExplainRunOutput.to_dict() holds the values, one label per player, base_value, grand_value, the estimand, the counters, the manifest and the timings; the command line writes the same payload.

payload = output.to_dict()
print("keys:", list(payload))
print("label of player 3:", json.dumps(payload["labels"][3]))
manifest = payload["manifest"]
print({
    key: value for key, value in manifest.items() if key not in ("config", "player_ids")
})
print(
    "manifest records this package version:",
    manifest["mllm_shap_version"] == __version__,
)
print("JSON size:", len(json.dumps(payload)), "characters")

keys: ['values', 'labels', 'base_value', 'grand_value', 'estimand', 'metrics', 'manifest', 'perf']
label of player 3: {"player_id": "84caf96b080cfbe6:0:1:3", "text": "days", "message_index": 1, "start": 21, "end": 25, "speaker": null, "label": "days"}
{'schema_version': 4, 'mllm_shap_version': '0.6.0', 'run_id': '84caf96b080cfbe6-c8b90ed02a00', 'snapshot_id': '84caf96b080cfbe6', 'estimand': 'shapley', 'estimator': 'exact', 'value_fn': 'tfidf_cosine', 'normalizer': 'identity', 'absence_policy': 'delete', 'players': 'tokens', 'backend_kind': 'mock', 'model_id': 'refund-bot', 'embedding_source': 'none', 'embedding_model': ''}
manifest records this package version: True
JSON size: 2680 characters

Estimators, estimands and budgets

Business: exact enumeration costs 2^n model calls; approximate estimators trade accuracy for budget, and Banzhaf answers a different question than Shapley.

Technical: each row changes only explanation keys. budget.fraction is the share of the 2^n coalitions an estimator may request; complementary and Neyman-CC never go below 2n + 2 masks. Efficiency holds exactly only on the full table (exact Shapley, or any Shapley estimator at fraction: 1.0); sampled estimates converge to it, and Banzhaf values are not efficient by definition. max_players refuses oversized plans before any call.

def variant(**explanation: object) -> ExplainConfig:
    """Return ``config`` with some explanation keys replaced (validated)."""
    payload = config.model_dump(mode="json")
    payload["explanation"].update(explanation)
    return ExplainConfig.model_validate(payload)


ROWS = (
    ("exact", "shapley", 1.0),
    ("exact", "banzhaf", 1.0),
    ("mc", "shapley", 0.25),
    ("mc", "banzhaf", 0.25),
    ("complementary", "shapley", 0.25),
    ("neyman_cc", "shapley", 0.25),
    ("complementary", "shapley", 0.05),
)
print(
    f"{'estimator':13} {'estimand':8} {'fraction':>8} {'coalitions':>10} "
    f"{'sum':>7} {'v(N)-v(0)':>9}"
)
for estimator, estimand, fraction in ROWS:
    run = await ExplainRunner(
        variant(estimator=estimator, estimand=estimand, budget={"fraction": fraction}),
        backend=RefundBot(),
    ).explain(snapshot)
    gap = run.result.grand_value - run.result.base_value
    print(
        f"{estimator:13} {estimand:8} {fraction:8.2f} {run.metrics.requested:10d} "
        f"{sum(run.result.values):7.3f} {gap:9.3f}"
    )

try:
    await ExplainRunner(variant(max_players=5), backend=RefundBot()).explain(snapshot)
except ConfigError as exc:
    print("max_players guard:", exc)

estimator     estimand fraction coalitions     sum v(N)-v(0)
exact         shapley      1.00         64   1.000     1.000
exact         banzhaf      1.00         64   0.842     1.000
mc            shapley      0.25         16   1.340     1.000
mc            banzhaf      0.25         16   1.302     1.000
complementary shapley      0.25         16   1.000     1.000
neyman_cc     shapley      0.25         16   0.894     1.000
complementary shapley      0.05         14   1.000     1.000
max_players guard: snapshot has 6 players; budget.fraction 1.0 plans about 2^6.0 coalitions, more than the 2^5 allowed by explanation.max_players=5. Lower the budget, use coarser players, or raise max_players.

The exact Shapley row adds up to v(N) - v(∅); the exact Banzhaf row does not, because Banzhaf weights coalitions equally instead of splitting the total. The sampled rows land near the target. The last row shows the 2n + 2 floor: 14 coalitions although 5% of 64 would be 3. estimator_comparison.ipynb measures accuracy against budget.

Monitor a run

Business: long runs on paid APIs need progress and timing data for capacity planning and incident review.

Technical: progress= takes a CoalitionProgress (on_coalitions_planned, on_coalition_finished); Neyman-CC plans in two phases, so the planned total grows once. telemetry= takes an ObservabilitySink; InMemoryObservabilitySink keeps one span per stage. configure_logging(level="INFO") prints stage events to stderr (json=True for JSON lines). output.perf splits the wall-clock time into generation, scoring and aggregation.

class PrintProgress:
    """Print planned totals and every 32nd finished coalition."""

    def on_coalitions_planned(self, total: int) -> None:
        """Report the number of coalitions planned so far."""
        print("planned:", total)

    def on_coalition_finished(self, done: int, total: int) -> None:
        """Report progress every 32 coalitions and at the end."""
        if done % 32 == 0 or done == total:
            print(f"finished {done}/{total}")


configure_logging(level="INFO")
sink = InMemoryObservabilitySink()
monitored = ExplainRunner(
    variant(estimator="neyman_cc", budget={"fraction": 0.5}),
    telemetry=sink,
    progress=PrintProgress(),
    backend=RefundBot(),
)
monitored_output = await monitored.explain(snapshot)
configure_logging(level="WARNING")
print("stages:", [span.name for span in sink.spans])
print("perf fields:", list(monitored_output.to_dict()["perf"]))

planned: 32
finished 32/32
stages: ['orchestrator.generate_base', 'estimator.sample_masks', 'orchestrator.schedule', 'orchestrator.score', 'orchestrator.aggregate', 'orchestrator.run']
perf fields: ['total_ms', 'generation_ms', 'scoring_ms', 'aggregation_ms']

2026-10-13 09:35:21,867 INFO [mllm_shap.pipeline.orchestrator] explain run 84caf96b080cfbe6-c917fde971f9 started: 6 players, estimator neyman_cc, backend mock
2026-10-13 09:35:21,868 INFO [mllm_shap.pipeline.orchestrator] base generated in 0.1 ms
2026-10-13 09:35:21,869 INFO [mllm_shap.pipeline.orchestrator] 32 coalitions scheduled
2026-10-13 09:35:21,872 INFO [mllm_shap.pipeline.orchestrator] explain run 84caf96b080cfbe6-c917fde971f9 done in 4.5 ms (32 coalitions, 31 backend calls)