Skip to content

Adapt Pytorch

Adapt PyTorch — mvp.adapt_pytorch

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

Overview

PyTorch MLP training and inference block that constructs a configurable multi-layer perceptron, trains it with Adam or SGD, and runs forward-pass predictions without requiring manual PyTorch session management. Layer widths are specified as a list of hidden dimensions; the block infers input and output sizes from the data, supports classification and regression tasks in the direct block API, and reports per-epoch training losses. PyTorch is an optional dependency; when it is missing, operations return an install-required error instead of pretending to train.

The MCP variant is designed for first-user workflows: define or register data, train, evaluate, predict, inspect, and export through a SQLite-backed tool surface. After train, the MCP block saves a latest state-dict checkpoint and later predict, predict_proba, embed, evaluate, gradient_check, profile, checkpoint_save, export_torchscript, and export_onnx load that latest checkpoint when it is available.

The MCP variant also exposes a constrained backend-native surface: native_capabilities, native_describe, native_validate, native_call, and native_artifacts. These tools are not unrestricted PyTorch or Python execution. They are allow-listed, bounded, audited, CPU-by-default operations for capability discovery, metadata-only inspection, dependency checks, tensor-shape preview, device reporting, module summaries, export capability checks, forward-preview planning, and artifact metadata listing.

All MCP outputs include canonical envelope fields: completion_state, warning_card, evidence, request_id, task_id, run_id, and error, while retaining legacy degraded and degradation_reason. Required backend absence returns blocked-escalated; fallback/stub paths return qualified-draft; verified non-degraded outputs return verified.

MVP lifecycle caveat

adapt_pytorch is useful for pilot workflows and lightweight neural-network experiments, but it is not a full model lifecycle platform. The MCP path keeps a single latest checkpoint per model name plus explicit checkpoint records; it does not yet provide rich checkpoint selection, model registry versioning, dataset lineage, approval gates, rollback policy, or production model monitoring. Use it to make train-predict-export workflows functional and inspectable, then graduate important models to a dedicated ML registry or deployment pipeline.

When to use:

  • Training a lightweight MLP on tabular data as part of a solver or evaluation pipeline
  • Comparing PyTorch against Keras or scikit-learn baselines within the same workflow
  • Running inference on new samples with the latest MCP-trained checkpoint for a named model
  • Prototyping custom network architectures before moving to a full Keras or Hugging Face setup

Example:

from mvp.adapt_pytorch import AdaptPyTorchBlock, TorchInput

block = AdaptPyTorchBlock(name="pytorch")
result = block.infer(TorchInput(
    operation="train",
    hidden_dims=[64, 32],
    task="classification",
    n_classes=3,
    X=[[1.0, 2.0], [3.0, 4.0], [5.0, 6.0]],
    y=[0, 1, 2],
    epochs=20,
    lr=0.01,
))
# result.value.final_loss → training loss; result.value.training_losses → per-epoch list

Works well with: adapt_keras, adapt_sklearn, align_evals

Persistence and Checkpoints

The direct AdaptPyTorchBlock API is stateless: each infer() call builds a fresh model. It is useful for simple build/train/predict calls inside one pipeline step, but it does not persist trained weights between direct API calls.

The MCP AdaptPyTorchMCPBlock stores metadata in SQLite and writes a latest state-dict checkpoint after every successful train. The checkpoint is named internally as <model_name>__latest and is also exposed through the checkpoints table. Downstream MCP operations automatically attempt to load that latest checkpoint for the same model name and report this in metadata.loaded_checkpoint.

For production use, treat latest as a convenience, not as a deployment contract. If a model matters, call checkpoint_save with an explicit checkpoint_name, record the dataset and run context separately, and validate the exported artifact before serving it.

_LazyMCP

Methods:

run(**kwargs)

Public API

AdaptPyTorchBlock(AIBlock[TorchInput, TorchOutput, None])

PyTorch MLP block: build / train / predict.

Field Type Default
name str 'adapt_pytorch'
resource_bounds ResourceBounds \| None None
usage ResourceUsage field(default_factory=ResourceUsage)

Methods:

infer(data: TorchInput) -> Result[TorchOutput]

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

heavy_dependency_status() -> list[dict]

TorchInput(BaseModel)

Input to AdaptPyTorchBlock.

Field Type Default
operation Literal['build', 'train', 'predict'] 'build'
hidden_dims list[int] Field(default_factory=lambda: [64, 32])
task Literal['classification', 'regression'] 'classification'
n_classes int 2
optimizer Literal['adam', 'sgd'] 'adam'
lr float 0.01
epochs int 10
batch_size int 16
device str ''
X list[list[float]] Field(default_factory=list)
y list[float \| int] Field(default_factory=list)

TorchOutput(BaseModel)

Output from AdaptPyTorchBlock.

Field Type Default
operation str required
model_summary str required
training_losses list[float] Field(default_factory=list)
final_loss float 0.0
predictions list[float] Field(default_factory=list)
n_parameters int 0
epochs_trained int 0
task str ''
device str ''
degraded bool False
degradation_reason str ''

AdaptPyTorchMCPBlock(AIBlock[MCPPyTorchInput, MCPPyTorchOutput, dict])

Full-featured PyTorch block with SQLite persistence.

Field Type Default
name str 'adapt_pytorch_mcp'
db_path str ':memory:'
resource_bounds ResourceBounds \| None None
usage ResourceUsage field(default_factory=ResourceUsage)
state dict field(default_factory=lambda: {'models': {}, 'datasets': {}})
agentic_planner SchedulePlanner \| None None

Methods:

infer(data: MCPPyTorchInput) -> Result[MCPPyTorchOutput]

MCPPyTorchInput(BaseModel)

Field Type Default
op Literal['model_define', 'model_list', 'model_info', 'model_delete', 'train', 'evaluate', 'tune', 'transfer_learn', 'lr_schedule', 'predict', 'predict_proba', 'embed', 'export_torchscript', 'export_onnx', 'checkpoint_save', 'checkpoint_load', 'gradient_check', 'profile', 'dataset_register', 'get_info', 'recommend_schedule', 'list_patterns', 'native_capabilities', 'native_describe', 'native_validate', 'native_call', 'native_artifacts'] required
name str ''
arch_json str '{}'
dataset_name str ''
X_json str '[]'
y_json str '[]'
epochs int 10
batch_size int 32
optimizer str 'adam'
lr float 0.001
loss str 'cross_entropy'
device str ''
layer_name str ''
export_path str ''
checkpoint_name str ''
tune_budget int 10
search_space_json str '{}'
backbone str 'resnet18'
freeze_backbone bool True
lr_schedule_type str 'step'
lr_schedule_params_json str '{}'
profile_steps int 10
jit_mode str 'trace'
limit int 20
notes str ''
val_split float 0.0
patience int 0
request_id str ''
task_id str ''
run_id str ''
native_action str ''
native_payload dict[str, Any] Field(default_factory=dict)
native_artifact_id str ''
allow_gpu bool False

MCPPyTorchOutput(BaseModel)

Field Type Default
op str required
ok bool True
name str ''
device str ''
message str ''
summary str ''
models list[dict[str, Any]] Field(default_factory=list)
datasets list[dict[str, Any]] Field(default_factory=list)
checkpoints list[dict[str, Any]] Field(default_factory=list)
history dict[str, list[float]] Field(default_factory=dict)
final_loss float 0.0
final_accuracy float 0.0
predictions list[Any] Field(default_factory=list)
probabilities list[list[float]] Field(default_factory=list)
embeddings list[list[float]] Field(default_factory=list)
n_parameters int 0
count int 0
metadata dict[str, Any] Field(default_factory=dict)
tune_results list[dict[str, Any]] Field(default_factory=list)
export_path str ''
gradient_norms dict[str, float] Field(default_factory=dict)
profile_results list[dict[str, Any]] Field(default_factory=list)
degraded bool False
degradation_reason str ''
completion_state str 'qualified-draft'
warning_card dict[str, Any] Field(default_factory=dict)
evidence list[dict[str, Any]] Field(default_factory=list)
request_id str ''
task_id str ''
run_id str ''
error dict[str, Any] \| None None

_LazyMCP

Methods:

run(**kwargs)

MCP Tools

Operation Source
model_define pytorch_mcp
model_list pytorch_mcp
model_info pytorch_mcp
model_delete pytorch_mcp
train pytorch_mcp
evaluate pytorch_mcp
tune pytorch_mcp
transfer_learn pytorch_mcp
lr_schedule pytorch_mcp
predict pytorch_mcp
predict_proba pytorch_mcp
embed pytorch_mcp
export_torchscript pytorch_mcp
export_onnx pytorch_mcp
checkpoint_save pytorch_mcp
checkpoint_load pytorch_mcp
gradient_check pytorch_mcp
profile pytorch_mcp
dataset_register pytorch_mcp
get_info pytorch_mcp
recommend_schedule pytorch_mcp
list_patterns pytorch_mcp
native_capabilities pytorch_mcp
native_describe pytorch_mcp
native_validate pytorch_mcp
native_call pytorch_mcp
native_artifacts pytorch_mcp