Plugins: registry, context and options¶
Business context: every building block of a run (estimator, estimand, value function, normalizer, backend, partition, absence policy, renderer) is a named plugin. Teams can add their own model client or metric without forking the package, and the configuration still names every choice.
Goal: Cover the public plugin surface: the eight PluginGroups and default_registry(), resolving a plugin with a PluginContext, explanation.plugin_options for built-in plugins and their validation, registering a custom backend and a custom value function with its own options model, name conflicts (PluginConflictError, replace=True), unknown names (PluginNotFoundError), and packaging entry points.
Prerequisites: mllm-shap, Python 3.12. No model download.
Expected output: The registered names per group; a pad policy rendering a custom placeholder from the configuration; refused plugin_options with the group, plugin and key named; a run that uses a custom backend and value function selected purely by configuration.
import re
from typing import ClassVar
from mllm_shap import (
CoalitionMask,
ConfigError,
ConversationSnapshot,
ExplainConfig,
ExplainRunner,
GenerationRecord,
Message,
PluginConflictError,
PluginContext,
PluginGroup,
PluginNotFoundError,
PluginRegistry,
Role,
TokenPartitioner,
Turn,
default_registry,
)
from mllm_shap.domain.config import GenerationConfig
from mllm_shap.protocols.backend import GenerationResult
from mllm_shap.protocols.plugin import PluginOptions
from mllm_shap.value.generation_records import generation_record_from_text
The registry¶
Business: The list below is the vocabulary a configuration may use; mllm-shap plugins prints the same list.
Technical: default_registry() registers the built-in plugins of all eight groups and then loads third-party plugins from the mllm_shap.<group> packaging entry points (an entry point that reuses a built-in name is skipped). ExplainRunner uses it unless you pass registry=.
registry = default_registry()
for group in PluginGroup:
print(f"{group.value:26} {', '.join(registry.names(group))}")
mllm_shap.estimators complementary, exact, loo, mc, neyman_cc
mllm_shap.estimands banzhaf, shapley
mllm_shap.value_fns embedding_cosine, embedding_euclidean, logprob, reference_wer, tfidf_cosine
mllm_shap.normalizers abs_sum, identity, min_max, power_shift
mllm_shap.backends api, lmstudio, mock, mock_audio, transformers
mllm_shap.partitions messages, sgpa_segments, tokens
mllm_shap.absence_policies delete, neutral, pad
mllm_shap.renderers token_bar, token_text
Resolve a built-in plugin with its options¶
Business: House conventions, such as the mask token a tokenizer expects, live in the configuration instead of code.
Technical: registry.resolve(group, name, context) calls the plugin factory with a PluginContext holding the validated run configuration, the plugin’s options and the registry. Options come from explanation.plugin_options, keyed by <group>/<name> (for example absence_policies/pad) or by the bare plugin name when only one plugin of the run has it. Plugins declare their options in a PluginOptions model: pad takes placeholder, neutral takes fill_char, power_shift takes power.
config = ExplainConfig.model_validate({
"backend": {"kind": "mock", "model_id": "stub"},
"explanation": {
"estimator": "exact",
"absence_policy": "pad",
"normalizer": "power_shift",
"archive": {"path": ""},
"plugin_options": {
"absence_policies/pad": {"placeholder": "<unk>"},
"power_shift": {"power": 2.0},
"tokens": {"roles": ["user"]},
},
},
})
context = PluginContext(config=config, registry=registry)
pad = registry.resolve(PluginGroup.ABSENCE_POLICIES, "pad", context)
power_shift = registry.resolve(PluginGroup.NORMALIZERS, "power_shift", context)
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?"),
)
),
))
players = TokenPartitioner(
PluginContext(config=config, options=config.explanation.plugin_options["tokens"])
).partition(snapshot)
without_refund = CoalitionMask.from_sequence((False, True, True, True, True, True))
print("pad renders:", pad.apply(snapshot, players, without_refund).messages()[0].text)
print("power_shift power:", power_shift.power)
pad renders: 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.
power_shift power: 2.0
Refused options¶
Business: A typo in an option must not silently fall back to a default and change the reported numbers.
Technical: ExplainRunner checks every plugin_options entry before the first model call and raises ConfigError naming the group, the plugin and the key: an unknown option, a value of the wrong type, a key that names no plugin of the run, or a bare name used by several groups. reanalyze runs the same check.
async def options_refused(plugin_options: dict[str, dict[str, object]]) -> None:
"""Print why ``plugin_options`` is refused."""
payload = config.model_dump(mode="json")
payload["explanation"]["plugin_options"] = plugin_options
try:
await ExplainRunner(ExplainConfig.model_validate(payload)).explain(snapshot)
except ConfigError as exc:
print("ConfigError:", exc)
await options_refused({"absence_policies/pad": {"placehodler": "<unk>"}})
await options_refused({"power_shift": {"power": "two"}})
await options_refused({"absence_policies/neutral": {"fill_char": "*"}})
await options_refused({"tfidf": {}})
ConfigError: invalid plugin_options for absence_policies/pad: 'placehodler': Extra inputs are not permitted
ConfigError: invalid plugin_options for normalizers/power_shift: 'power': Input should be a valid number
ConfigError: plugin_options key 'absence_policies/neutral' matches no plugin of this run
ConfigError: plugin_options key 'tfidf' matches no plugin of this run; selected plugins: {'estimators': 'exact', 'estimands': 'shapley', 'value_fns': 'tfidf_cosine', 'normalizers': 'power_shift', 'backends': 'mock', 'partitions': 'tokens', 'absence_policies': 'pad'}
A custom backend and a custom value function¶
Business: A team plugs in its own model client and a metric that matches its product question (“does the answer still mention the policy terms?”), then selects both by name in YAML like any built-in.
Technical: A plugin class takes the PluginContext as its first constructor argument. The backend needs a model_id property and an async generate(snapshot, generation); the value function needs a name and score(base, candidate). Declaring options_model (a PluginOptions subclass) makes the registry validate the plugin’s options; context.option(key, default) reads one with a type check. registry.register(group, name, factory) adds a plugin to one registry; pass that registry to ExplainRunner.
class RefundBot:
"""Rule-based stand-in for a support model."""
def __init__(self, context: PluginContext) -> None:
"""Read the model id from the run configuration."""
self._model_id = context.config.backend.model_id
@property
def model_id(self) -> str:
"""Return the configured model id."""
return self._model_id
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)
class KeywordOptions(PluginOptions):
"""Options of the ``keyword_hits`` value function."""
keywords: list[str] = ["refund", "receipt"] # noqa: RUF012
class KeywordHits:
"""Value = number of required keywords the answer mentions."""
options_model: ClassVar[type[PluginOptions]] = KeywordOptions
def __init__(self, context: PluginContext) -> None:
"""Read the keyword list from ``plugin_options``."""
self._keywords = context.option("keywords", ["refund", "receipt"])
@property
def name(self) -> str:
"""Return the registered name."""
return "keyword_hits"
def score(self, base: GenerationResult, candidate: GenerationResult) -> float:
"""Count the keywords present in the candidate answer."""
_ = base
return float(sum(keyword in candidate.text for keyword in self._keywords))
custom = default_registry()
custom.register(PluginGroup.BACKENDS, "refund_bot", RefundBot)
custom.register(PluginGroup.VALUE_FNS, "keyword_hits", KeywordHits)
custom_config = ExplainConfig.model_validate({
"backend": {"kind": "refund_bot", "model_id": "refund-bot"},
"explanation": {
"estimator": "exact",
"value_fn": "keyword_hits",
"archive": {"path": ""},
"plugin_options": {
"keyword_hits": {"keywords": ["refund", "deadline"]},
"tokens": {"roles": ["user"]},
},
},
})
runner = ExplainRunner(custom_config, registry=custom)
output = await runner.explain(snapshot)
for span, value in zip(output.player_set.spans, output.result.values, strict=True):
print(f"{span.text:>9} {value:+.3f}")
print(
"base_value:", output.result.base_value, " grand_value:", output.result.grand_value
)
print("manifest:", output.manifest.backend_kind, output.manifest.value_fn)
Refund +1.000
within +0.000
thirty +0.500
days +0.500
with +0.000
receipt? +0.000
base_value: 0.0 grand_value: 2.0
manifest: refund_bot keyword_hits
What the model answers. The custom backend is a registry plugin too. The values above rank words by how the answer changes when each one is deleted:
context = PluginContext(config=custom_config, registry=custom)
backend = custom.resolve(PluginGroup.BACKENDS, "refund_bot", context)
policy = custom.resolve(
PluginGroup.ABSENCE_POLICIES, custom_config.explanation.absence_policy, context
)
full = await backend.generate(snapshot, custom_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 = policy.apply(snapshot, output.player_set, mask)
answer = await backend.generate(absent, custom_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
With the keywords refund and deadline, the full prompt scores 2 and the empty prompt 0. Refund alone switches the refund phrase on and gets 1; thirty and days share the deadline phrase and get 0.5 each; receipt? only changes a phrase this metric ignores, so it gets 0, like within and with. The values add up to 2, the difference between the grand and base values.
Name conflicts, unknown names and entry points¶
Business: two packages claiming the same plugin name must fail loudly; replacing a built-in must be a deliberate act.
Technical: registering a name twice raises PluginConflictError unless replace=True; an unregistered name raises PluginNotFoundError (a LookupError) before any model call. Installed packages publish plugins as entry points in their pyproject.toml: default_registry() (used by ExplainRunner and mllm-shap plugins) loads them, PluginRegistry.load_entry_points(group) loads one group, and an entry point that fails to import raises PluginLoadError.
[project.entry-points."mllm_shap.value_fns"]
keyword_hits = "my_package.values:KeywordHits"
try:
custom.register(PluginGroup.VALUE_FNS, "keyword_hits", KeywordHits)
except PluginConflictError as exc:
print("PluginConflictError:", exc)
custom.register(PluginGroup.VALUE_FNS, "keyword_hits", KeywordHits, replace=True)
print("replaced on purpose")
try:
custom.resolve(PluginGroup.VALUE_FNS, "bleu", PluginContext(config=custom_config))
except PluginNotFoundError as exc:
print("PluginNotFoundError:", exc)
own = PluginRegistry()
own.load_entry_points(PluginGroup.ESTIMANDS)
print("estimands from entry points:", own.names(PluginGroup.ESTIMANDS))
PluginConflictError: plugin 'keyword_hits' is already registered in group 'mllm_shap.value_fns'; pass replace=True to override it
replaced on purpose
PluginNotFoundError: unknown plugin 'bleu' in group 'mllm_shap.value_fns'
estimands from entry points: ('banzhaf', 'shapley')