Pipeline configuration

Explain pipeline configuration schema.

class nlp_shap.pipeline.config.BackendConfig(*, kind: str, model_id: str, api_host: str | None = None, hub_search: str | None = None, quantization: str = 'Q4_K_M', auto_download: bool = True, serialize_generate: bool = True)

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

Optional API host for HTTP-compatible backends.

hub_search: str | None

Optional LM Studio repository search term; defaults to model_id.

quantization: str

GGUF quantization to download when the model is not already local.

auto_download: bool

Download missing models through the LM Studio repository API.

serialize_generate: bool

Serialize backend generate calls (needed for MLX LM Studio).

class nlp_shap.pipeline.config.GenerationConfig(*, max_new_tokens: int = 128, temperature: float = 0.0, top_k: int = 1, precompute_base: bool = True)

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.

precompute_base: bool

Whether to generate the grand-coalition reference before explaining.

class nlp_shap.pipeline.config.BudgetConfig(*, fraction: Annotated[float, Gt(gt=0.0), Le(le=1.0)] = 1.0)

Bases: BaseModel

Estimator budget controls replacing standard/limited split configs.

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

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

fraction: float

Fraction of the full coalition budget used by the estimator.

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

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.

class nlp_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 nlp_shap.pipeline.config.KvCacheConfig(*, enabled: bool = True)

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.

class nlp_shap.pipeline.config.NeymanConfig(*, use_standard_method: bool = False, initial_fraction: float | None = None, initial_num_samples: int | None = 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.

initial_num_samples: int | None

Explicit phase-one pair budget overriding initial_fraction.

class nlp_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), 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, archive: ArchiveConfig = ArchiveConfig(path='./runs/{run_id}', flush_every=50), dedup: DedupConfig = DedupConfig(enabled='auto'), kv_cache: KvCacheConfig = KvCacheConfig(enabled=True), embedding_mode: EmbeddingMode = EmbeddingMode.STATIC, seed: int = 42)

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.

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.

class nlp_shap.pipeline.config.ExplainConfig(*, backend: BackendConfig, generation: GenerationConfig = GenerationConfig(max_new_tokens=128, temperature=0.0, top_k=1, precompute_base=True), 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), include_minimal_masks=False, neyman=NeymanConfig(use_standard_method=False, initial_fraction=None, initial_num_samples=None), max_inflight=2, archive=ArchiveConfig(path='./runs/{run_id}', flush_every=50), dedup=DedupConfig(enabled='auto'), kv_cache=KvCacheConfig(enabled=True), embedding_mode=<EmbeddingMode.STATIC: 'static'>, seed=42))

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.

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

Parse YAML text into a validated ExplainConfig.

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

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

Explain pipeline result types.

class nlp_shap.pipeline.result.ExplainResult(estimand: Estimand, values: tuple[float, ...])

Bases: object

Attribution output for a single explain run.

estimand: Estimand

Estimand used to aggregate coalition samples.

values: tuple[float, ...]

Per-player attribution values.

class nlp_shap.pipeline.result.ExplainRunOutput(result: ExplainResult, metrics: SchedulerMetrics, run_id: str, manifest: RunManifest, perf: PerfSummary | 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.

Run archive manifest schema stubs.

class nlp_shap.pipeline.manifest.RunManifestPayload

Bases: TypedDict

JSON-compatible manifest payload persisted with a run archive.

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

Estimand label recorded for every attribution output.

run_id: str

Stable identifier for the explain run.

class nlp_shap.pipeline.manifest.RunManifest(estimand: Estimand, run_id: str)

Bases: object

Top-level metadata persisted alongside a run archive.

estimand: Estimand

Estimand label recorded for every attribution output.

run_id: str

Stable identifier for the explain run.

to_dict() → RunManifestPayload

Serialize manifest fields to JSON-compatible primitives.

nlp_shap.pipeline.manifest.parse_manifest(payload: RunManifestPayload) → RunManifest

Parse a typed manifest payload into a run manifest.

Public explain pipeline entry point.

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

Bases: object

Run the explain pipeline for a conversation snapshot.

async explain(snapshot: ConversationSnapshot) → ExplainRunOutput

Execute the async explain pipeline.

explain_sync(snapshot: ConversationSnapshot) → ExplainRunOutput

Execute the explain pipeline on the current event loop policy.

reanalyze(archive_root: Path | str) → ExplainRunOutput

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

Explain pipeline orchestration.

class nlp_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.