Adapt Learning¶

adapt_learning — mvp.adapt_learning

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

Overview¶

Meta-learning algorithm selector that analyses a dataset's statistical properties — sample count, feature count, class imbalance, sparsity, linearity — and recommends the most suitable scikit-learn algorithm ranked by estimated performance, with complexity and reasoning included. Works without any external model or API call; all recommendations are derived from heuristic meta-feature rules and can be passed directly to adapt_sklearn or adapt_automl for execution.

When to use:

• Choosing an appropriate ML algorithm before committing to training in a resource-constrained pipeline
• Explaining algorithm selection decisions to downstream agents or human reviewers
• Routing tabular datasets to the right solver (classification / regression / clustering) automatically
• Building a two-phase pipeline: recommend with adapt_learning, then fit with adapt_sklearn

Example:

from mvp.adapt_learning import AdaptLearningBlock, MetaLearningInput

block = AdaptLearningBlock(name="learning")
result = block.infer(MetaLearningInput(
    X=[[1.0, 2.0, 3.0], [4.0, 5.0, 6.0], [7.0, 8.0, 9.0]],
    y=[0, 1, 0],
    task="classification",
    budget="small",
))
# result.value.recommendations[0].algorithm → e.g. "LogisticRegression"
# result.value.recommendations[0].reasoning → explanation string

Works well with: adapt_sklearn, adapt_automl, align_evals

Operational Notes¶

• AdaptLearningBlock is a deterministic rule-based selector. It is useful for fast algorithm triage, explainable recommendations, and routing into training components, but its estimated_score values are heuristic priors, not measured model accuracy.
• task="auto" uses simple target-shape heuristics. Empty y is treated as clustering, string labels are treated as classification, and low-cardinality numeric targets are treated as classification. Pass an explicit task when the target semantics are known.
• The MCP layer adds SQLite-backed memory for experiences, heuristics, algorithm outcomes, feedback patterns, and tool-use strengths. This memory can improve ranking through recorded outcomes, but it should be treated as local/project history rather than a global benchmark.
• MCP search requires a non-empty query. Blank queries return an empty result with an explanatory message instead of returning arbitrary records.
• MCP numeric controls are bounded for production safety: limit is 0-500, top_k is 0-100, score and worth are finite values from 0.0 to 1.0, and latency_ms must be finite and non-negative.
• AdaptLearningMCPBlock.aio_run() runs synchronous work in a thread pool. Calls are serialized per block instance to protect the shared SQLite connection during concurrent writes.
• status and info include available_ops metadata so MCP clients can discover the supported 20-operation contract without relying on undocumented ops or help operations.
• The recommend op is uplifted with an advisory LLM re-ranking layer over the deterministic blended-ranking floor: when an LLM backend is reachable (and not opted out via G6_LEARNING_AGENTIC_RUNTIME=0 or G6_DISABLE_LLM=1) the model may RE-RANK the same fixed candidate algorithms using the natural-language notes context; it can never introduce a new algorithm, train a model, or write the store, and degrades to the floor with honest degraded/degradation_reason evidence. list_patterns lists the applied agentic pattern cards.

Public API¶

RankedAlgorithmRecommendation¶

Validated ranked algorithm recommendation decision record.

Methods:

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

The top-ranked recommendation dict (empty dict if the ranking is empty).

top_algorithm() -> str¶

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

The advisory payload: the ranked list of recommendation dicts.

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

LearningDecisionError(ValueError)¶

The LLM did not produce a usable, validated algorithm recommendation.

LLMLearningRuntime¶

Provider-neutral algorithm-ranking runtime over G6's LLM caller.

Constructor:

Methods:

recommend(meta_features: dict[str, Any], candidates: list[dict[str, Any]], context: str = '') -> Result[dict]¶

LearningAlgorithmPlanner¶

Runtime-first facade with the deterministic floor as honest fallback.

Constructor:

Methods:

recommend(meta_features: dict[str, Any], candidates: list[dict[str, Any]], history: dict[str, float] | None = None, context: str = '') -> RankedAlgorithmRecommendation¶

AdaptLearningBlock(AIBlock[MetaLearningInput, MetaLearningOutput, None])¶

Meta-learning algorithm selector.

Methods:

infer(data: MetaLearningInput) -> Result[MetaLearningOutput]¶

MetaLearningInput(BaseModel)¶

Input to AdaptLearningBlock.

AlgorithmRecommendation(BaseModel)¶

A recommended algorithm with justification.

MetaLearningOutput(BaseModel)¶

Output from AdaptLearningBlock.

Methods:

to_sklearn_config(rank: int = 0) -> 'ModelConfig'¶

Convert the top recommendation to an adapt_sklearn ModelConfig.

AdaptLearningMCPBlock(AIBlock[MCPLearningInput, MCPLearningOutput, dict])¶

Full-featured meta-learning block with SQLite persistence.

Methods:

infer(data: MCPLearningInput) -> Result[MCPLearningOutput]¶

aio_run(data: MCPLearningInput) -> Result[MCPLearningOutput]¶

Async wrapper: runs the synchronous infer() in a thread pool to avoid blocking the event loop.

MCPLearningRecord(BaseModel)¶

MCPLearningInput(BaseModel)¶

MCPLearningOutput(BaseModel)¶

LearningStore¶

Sync SQLite learning store with 5 tables.

Constructor:

Methods:

get_schema_version() -> int¶

Return the stored schema version integer.

add_experience(task_id: str, description: str, what_worked: str, what_failed: str, task_type: str, tags: list[str]) -> str¶

query_experiences(query: str = '', task_type: str = '', limit: int = 50) -> list[dict[str, Any]]¶

upsert_heuristic(name: str, description: str, domain: str, worth: float, tags: list[str]) -> str¶

query_heuristics(query: str = '', domain: str = '', top_k: int = 5) -> list[dict[str, Any]]¶

update_heuristic_eval(name: str, success: bool, score: float, notes: str) -> bool¶

add_outcome(algorithm: str, task_type: str, score: float, notes: str) -> str¶

get_avg_scores_by_task(task_type: str = '') -> dict[str, float]¶

Return {algorithm: avg_score} from recorded outcomes, optionally filtered by task type.

query_outcomes(algorithm: str = '', task_type: str = '', limit: int = 50) -> list[dict[str, Any]]¶

add_error(task_type: str, algorithm: str, error_type: str, notes: str) -> str¶

get_error_patterns(min_frequency: int = 2) -> list[dict[str, Any]]¶

add_tool_use(tool_name: str, domain: str, success: bool, latency_ms: float) -> str¶

get_domain_strengths() -> list[dict[str, Any]]¶

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

TF-IDF search across experiences, heuristics, and outcomes. Results cached per write cycle.

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

Functions¶

agentic_planner_enabled(default_enabled: bool = True) -> bool¶

Decide whether the agentic algorithm-planner should be used.

validate_algorithm_recommendation(rec: Any, allowed: set[str]) -> dict¶

Reject any algorithm recommendation outside the canonical shape / vocabulary.

recommend_floor(candidates: list[dict[str, Any]], history: dict[str, float] | None = None, context: str = '') -> RankedAlgorithmRecommendation¶

Deterministic blended-ranking algorithm selector (the honest floor).

MCP Tools¶