Adapt Keras¶

Adapt Keras — mvp.adapt_keras

Cluster: ML & Optimisation | Type: component | MCP Tools: 35

Overview¶

Keras deep-learning component for building, training, evaluating, and exporting Sequential or Functional API neural networks. Layers are specified as a list of LayerConfig objects (Dense, Conv2D, LSTM, Dropout, Flatten); the component infers input shape from the data, attaches the appropriate output layer for classification or regression, compiles with the requested optimizer and loss, and returns per-epoch training history. Missing Keras/backend dependencies are surfaced as loud degradation (ok=false, degraded=true, stable error_code, install guidance), not cosmetic success.

There are two entry points with different persistence behavior:

• AdaptKerasBlock is a lightweight stateless helper. Each infer() call builds a fresh model, so operation="predict" checks architecture and shape compatibility but does not reuse weights from a previous operation="train" call.
• AdaptKerasMCPBlock is the production-oriented path for train-then-predict workflows. It uses SQLite metadata plus ModelLifecycle weight persistence, so fit, predict, evaluate, export, and load operate on named models.
• Advanced users can call backend_capabilities plus explicit native operations (native_model_define, native_compile, native_fit, native_evaluate, native_predict, native_export, native_load). These operations are config-based, allowlisted, resource-limited, path-sandboxed, audited, and do not execute arbitrary Python or unsafe deserialization.

When to use:

• Building or smoke-testing a custom neural network on tabular or sequential data from within a pipeline step
• Evaluating a deep-learning approach alongside scikit-learn baselines for model selection
• Running predictions with a previously trained named model via AdaptKerasMCPBlock
• Prototyping LSTM or convolutional architectures without writing boilerplate Keras code

Example:

from mvp.adapt_keras import AdaptKerasBlock, KerasInput, LayerConfig

block = AdaptKerasBlock(name="keras")
result = block.infer(KerasInput(
    operation="train",
    layers=[LayerConfig(layer_type="dense", units=32, activation="relu"),
            LayerConfig(layer_type="dropout", dropout_rate=0.2)],
    X=[[0.1, 0.2], [0.3, 0.4], [0.5, 0.6]],
    y=[0, 1, 0],
    epochs=10,
    n_classes=2,
))
# result.value.final_accuracy → training accuracy; result.value.history → per-epoch metrics

For a real train-then-predict workflow, use the MCP block with a stable model name:

from mvp.adapt_keras.keras_mcp.keras_block import AdaptKerasMCPBlock
from mvp.adapt_keras.keras_mcp.schema import MCPKerasInput

block = AdaptKerasMCPBlock(db_path="keras.sqlite")

block.infer(MCPKerasInput(
    op="model_define",
    name="churn_mlp",
    arch_json='{"layers":[{"layer_type":"dense","units":32,"activation":"relu"}]}',
))

block.infer(MCPKerasInput(
    op="fit",
    name="churn_mlp",
    X_json="[[0.1,0.2],[0.3,0.4],[0.5,0.6]]",
    y_json="[0,1,0]",
    epochs=10,
))

predicted = block.infer(MCPKerasInput(
    op="predict",
    name="churn_mlp",
    X_json="[[0.2,0.3]]",
))

Works well with: adapt_pytorch, adapt_sklearn, align_evals

Public API¶

AdaptKerasBlock(AIBlock[KerasInput, KerasOutput, None])¶

Keras model builder and trainer.

Methods:

infer(data: KerasInput) -> Result[KerasOutput]¶

capability_plan(op: str, params: dict) -> CapabilityResolution¶

heavy_dependency_status() -> list[dict]¶

LayerConfig(BaseModel)¶

Configuration for a single Keras layer.

KerasInput(BaseModel)¶

Input to AdaptKerasBlock.

KerasOutput(BaseModel)¶

Output from AdaptKerasBlock.

AdaptKerasMCPBlock(AIBlock[MCPKerasInput, MCPKerasOutput, dict])¶

Full-featured Keras block with SQLite persistence and ModelLifecycle.

Methods:

infer(data: MCPKerasInput) -> Result[MCPKerasOutput]¶

MCPKerasInput(BaseModel)¶

MCPKerasOutput(BaseModel)¶

KerasStore¶

Sync SQLite store with 4 tables for Keras metadata.

Constructor:

Methods:

upsert_model(name: str, arch: dict, api_type: str, backend: str, n_parameters: int) -> int¶

get_model(name: str) -> dict[str, Any] | None¶

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

delete_model(name: str) -> bool¶

upsert_experiment(name: str, model_name: str, dataset_name: str, history: dict, final_loss: float, final_accuracy: float, epochs: int, backend: str, notes: str) -> int¶

get_experiment(name: str) -> dict[str, Any] | None¶

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

upsert_dataset(name: str, X: list, y: list, shape: list) -> int¶

get_dataset(name: str) -> dict[str, Any] | None¶

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

delete_dataset(name: str) -> bool¶

upsert_export(name: str, model_name: str, fmt: str, path: str) -> int¶

get_export(name: str) -> dict[str, Any] | None¶

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

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

MCP Tools¶