Adapt Bias¶

adapt_bias — Self-evolving inductive bias system with MLflow integration.

Cluster: Uncategorised | Type: component | MCP Tools: 33

Overview¶

Behavioral parameter-bias tuning orchestrator for bounded inductive-bias experiments. Observes system performance, proposes targeted parameter mutations for component-declared evolvable_params, verifies them against grounding and safety layers, and persists approved parameter versions with rollback support. It does not perform general fairness-bias detection unless a downstream component exposes those parameters and benchmarks, and it does not modify implementation files.

When to use:

• Tuning bounded component parameters based on workflow performance
• Running controlled evolution experiments on specific components
• Auditing the current bias landscape across all registered components

When not to use:

• As the first workflow for a new non-technical user
• Without a human reviewer approving the mutation token
• For regulated-domain outputs without a separate human review and compliance process

Example:

from mvp.adapt_bias import BiasEvolutionBlock, BiasInput

block = BiasEvolutionBlock(name="bias_evo")
result = block.infer(BiasInput(op="snapshot"))
# result.value.snapshot → dict of all component biases

Works well with: align_csf, formal_methods, adapt_memory, grounding

Operations¶

Configuration¶

See BiasEvolutionConfig for all configuration fields. Key environment variables:

BIAS_EVOLUTION_ENABLED=true          # Master feature flag
BIAS_EVOLUTION_CONFIRM_TOKEN=...     # Required for evolve/apply/rollback
BIAS_EVOLUTION_TIER=premium          # Or trialing for time-limited access
BIAS_DB_PATH=~/.bias_mcp/bias.db     # SQLite fallback path
BIAS_EVOLUTION_AUTO_EVOLVE=false     # Auto-run on workflow completion
BIAS_EVOLUTION_MAX_ITERATIONS=5      # Max proposals per cycle
BIAS_EVOLUTION_SAFETY_LAYERS=4       # Required safety layers

BIAS_EVOLUTION_CONFIRM_TOKEN is deliberately separate from API keys and license tier. Keep it in deployment configuration, not in prompts or reusable workflow definitions. Public MCP/REST callers must provide the matching confirm_token for mutating operations.

If MLflow is installed, get_store() uses the MLflow-backed store. Otherwise it falls back to SQLite at BIAS_DB_PATH or ~/.bias_mcp/bias.db, which is suitable for single-user local MCP pilots.

Public API¶

ABResult¶

Result of an A/B comparison between two bias configurations.

BiasEvolutionBlock(AIBlock['BiasInput', 'BiasOutput', dict])¶

Orchestrator for self-evolving inductive biases.

Methods:

infer(data: BiasInput) -> Result[BiasOutput]¶

Dispatch to the appropriate operation handler.

bias() -> dict[str, Any]¶

BiasInput(BaseModel)¶

Input for BiasEvolutionBlock operations.

BiasOutput(BaseModel)¶

Output from BiasEvolutionBlock operations.

BiasStoreABC(ABC)¶

Contract for bias-parameter storage backends.

Methods:

is_available() -> bool¶

Return True if this backend's dependencies are installed.

record_version(component: str, version: int, bias: dict[str, Any], metadata: dict[str, Any] | None = None) -> str¶

Persist a new bias version and return its record id.

get_versions(component: str, limit: int = 50) -> list[dict[str, Any]]¶

Return the most recent limit versions for component.

get_version(component: str, version: int) -> dict[str, Any] | None¶

Return a single version record, or None if not found.

record_evolution(component: str, op: str, input_data: dict[str, Any], output_data: dict[str, Any], metrics: dict[str, float] | None = None, artifacts: dict[str, Any] | None = None) -> str¶

Log an evolution operation and return its record id.

get_evolution_log(component: str = '', op: str = '', limit: int = 50) -> list[dict[str, Any]]¶

Query evolution history, optionally filtered by component/op.

record_grounding(component: str, candidate: dict[str, Any], report: dict[str, Any], passed: bool, artifacts: dict[str, Any] | None = None) -> str¶

Log a grounding check result and return its record id.

get_grounding_log(component: str = '', limit: int = 50) -> list[dict[str, Any]]¶

Query grounding history, optionally filtered by component.

promote(component: str, version: int, reason: str) -> dict[str, Any]¶

Promote a version to the next lifecycle stage.

get_stage_status(component: str) -> dict[str, Any]¶

Return the current lifecycle stage for component.

text_search(query: str, limit: int = 20) -> list[dict[str, Any]]¶

Full-text search across stored records.

count_all() -> dict[str, int]¶

Return counts of records by category.

clear_all() -> None¶

Delete all records (use with caution).

compare_runs(run_id_a: str, run_id_b: str) -> dict[str, Any]¶

Compare two runs and return a diff summary.

update_run_metrics(run_id: str, metrics: dict[str, float]) -> None¶

Update metrics on an existing run. Best-effort — may be a no-op.

AdaptBiasMCPBlock(AIBlock[MCPBiasInput, MCPBiasOutput, dict])¶

MCP-tier bias evolution block with SQLite persistence.

Methods:

infer(data: MCPBiasInput) -> Result[MCPBiasOutput]¶

MCPBiasInput(BaseModel)¶

MCPBiasOutput(BaseModel)¶

SqliteBiasStore(BiasStoreABC)¶

SQLite-backed store for the adapt_bias MCP sub-package.

Constructor:

Methods:

is_available() -> bool¶

record_version(component: str, version: int, bias: dict[str, Any], metadata: dict[str, Any] | None = None) -> str¶

Record a new bias version. Returns the assigned id.

get_versions(component: str, limit: int = 50) -> list[dict[str, Any]]¶

List all versions for a component, newest first.

get_version(component: str, version: int) -> dict[str, Any] | None¶

Get a specific version. Returns None if not found.

record_evolution(component: str, op: str, input_data: dict[str, Any], output_data: dict[str, Any], metrics: dict[str, float] | None = None, artifacts: dict[str, Any] | None = None) -> str¶

Log an evolution operation. Returns the assigned id.

get_evolution_log(component: str = '', op: str = '', limit: int = 50) -> list[dict[str, Any]]¶

Query evolution log, optionally filtered by component and/or op.

update_run_metrics(run_id: str, metrics: dict[str, float]) -> None¶

Merge additional metrics into an existing evolution_log row.

record_grounding(component: str, candidate: dict[str, Any], report: dict[str, Any], passed: bool, artifacts: dict[str, Any] | None = None) -> str¶

Log a grounding check result. Returns the assigned id.

get_grounding_log(component: str = '', limit: int = 50) -> list[dict[str, Any]]¶

Query grounding log, optionally filtered by component.

clear_all() -> None¶

Clear all tables.

count_all() -> dict[str, int]¶

text_search(query: str, limit: int = 10) -> list[dict[str, Any]]¶

Search across evolution_log by component + op text.

promote(component: str, version: int, reason: str) -> dict[str, Any]¶

get_stage_status(component: str) -> dict[str, Any]¶

compare_runs(run_id_a: str, run_id_b: str) -> dict[str, Any]¶

Functions¶

ab_compare(component_name: str, baseline_bias: dict[str, Any], candidate_bias: dict[str, Any], benchmark_cases: list[dict[str, Any]]) -> ABResult¶

Run the same benchmark suite with baseline vs candidate bias.

get_store(db_path: str | None = None, tracking_uri: str | None = None) -> BiasStoreABC¶

Create the best available bias store backend.

MCP Tools¶

REST Endpoints (7)¶

Exposed at /api/v1/bias/*. See the Bias Evolution API Reference for request/response schemas.