Learning Layer¶

learning_layer — Unified T0-T3 learning layer with OODA-driven self-training loops.

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

Overview¶

Unified T0-T3 learning pipeline orchestrating OODA-driven self-training loops. Manages dataset construction from JSONL sources, tier-specific learning strategies (T0 baseline through T3 theory-building), failure classification, prompt evolution, skill generation, and harness artifact production.

Supports any LLM or coding agent backend via pluggable AgentFn callables. MCP, REST, and GUI surfaces can also use the built-in LLMBlock runner by passing agent_runner="llm" with model/scoring options. Includes a TheoryStore for persisting learned theories, LearningRunner for executing training loops, and an observable event system with JSONL/log/callback observers.

When to use:

• Training a component to improve accuracy on a domain-specific benchmark
• Running tiered learning loops (T0 baseline, T1 audit, T2 evolve, T3 theory-build)
• Building and evaluating datasets from JSONL problem sets

Dataset JSONL fields:

Each line should contain task text plus either an expected answer or a rubric. The loader accepts common aliases:

• Task text: input, prompt, question, goal, task, or description
• Expected answer: expected, expected_output, answer, target, ground_truth, reference, reference_answer, or output
• Optional scoring metadata: rubric, criteria, score_mode, score_threshold, domain

Scoring modes:

• normalized_exact -- strict normalized exact match; best for short deterministic answers
• contains -- passes when expected text appears in the answer
• token_f1 -- partial-credit token overlap for longer answers
• rubric / llm_judge -- asks the configured LLM to score against a rubric from 0.0 to 1.0

Production controls:

• Public MCP/REST/GUI calls enforce a workspace root for dataset import and harness import/export.
• Preflight limits can cap max_dataset_instances, max_estimated_llm_calls, and max_estimated_cost_usd.
• holdout=true / train_with_holdout splits a dataset, trains on the train split, and reports holdout accuracy.
• overwrite=true is required to replace an existing run ID.
• backend_health reports configured runner/model and live-smoke environment state.
• capabilities / learning_capabilities reports strategies, scoring modes, executable ops, backend probe evidence, promotion resolver status, and the current weakest link.
• Learning outputs preserve existing success, message, and data fields and add completion_state, warning_card, and evidence. States use the canonical trio directly: verified, qualified-draft, blocked-escalated.
• Harness export/import scans artifacts for dangerous code, network/file writes, prompt-injection phrases, and likely secret references.

Example:

from mvp.learning_layer import LearningLayerBlock, LearningInput

block = LearningLayerBlock()
block.register_agent_fn(my_solver_fn)
result = block.infer(LearningInput(
    op="train",
    dataset_name="benchmarks",
    strategy="T1",
))
# result.value -> LearningOutput with accuracy, theories, artifacts

Caveats and known limitations:

• learning_train and learning_evaluate can call a live LLM backend. Before a customer pilot or production workflow, verify the exact deployment machine, G6_WORKSPACE, model/provider, credentials, and environment by running:

$env:G6_WORKSPACE="C:\path\to\pilot-workspace"
$env:G6_LEARNING_LAYER_LIVE_SMOKE="1"
python -m pytest -q tests/mvp/learning_layer/test_live_llm_smoke.py

If this smoke test fails, treat the backend/model/workspace configuration as not launch-ready. The normal unit suite mocks LLM calls and does not prove the live provider path works. - Exact-match scoring is only appropriate for short deterministic answers. Use contains, token_f1, or a rubric/LLM judge for open-ended outputs. - Rubric/LLM-judge scoring costs an extra LLM call per evaluated task and is only as reliable as the rubric. Keep rubrics concrete and review borderline cases. - Training improves a harness, prompt, theory store, or generated artifacts. It does not fine-tune model weights. - Lift is not guaranteed. Saturated datasets, vague expected answers, or weak rubrics may show no improvement. - Use held-out or frozen validation data before trusting artifacts in production. Do not judge launch readiness only on the same examples used for training. - Public surfaces reject dataset and harness paths outside the configured workspace root. Set G6_WORKSPACE deliberately for deployments. - Preflight budgets are estimates; actual provider billing can still vary by model, prompt length, retries, and judge calls. - For durable learning, set a persistent LearningConfig.db_path; in-memory stores are useful for tests but do not survive process restart. - For durable REST/MCP process state, set LearningConfig.state_dir; otherwise loaded datasets and run summaries are process-local. - Harness import/export preserves theories and artifacts, but users should review generated prompts/skills before applying them to sensitive workflows. - Checkpoint resume relies on stable instance IDs. Changing the dataset between runs can invalidate resume assumptions.

Works well with: self_training, align_evals, token_budget

Public API¶

DatasetBuilder¶

Builds and manipulates datasets for the learning loop.

Methods:

from_jsonl(path: str, name: str | None = None) -> Dataset¶

from_generator(gen: ProceduralGenerator, n: int, name: str = 'generated') -> Dataset¶

split(dataset: Dataset, train_ratio: float = 0.8, seed: int | None = None) -> tuple[Dataset, Dataset]¶

sample(dataset: Dataset, n: int, seed: int | None = None) -> Dataset¶

merge(datasets: list[Dataset], name: str | None = None) -> Dataset¶

filter(dataset: Dataset, predicate: Callable[[TaskInstance], bool]) -> Dataset¶

FailureClassifier(Protocol)¶

Protocol for failure classification — users can provide custom implementations.

Methods:

classify(result: TaskResult) -> FailureAnalysis¶

DefaultClassifier¶

Heuristic-based failure classifier (zero LLM cost).

Constructor:

Methods:

classify(result: TaskResult) -> FailureAnalysis¶

LearningInput¶

LearningOutput¶

LearningLayerBlock(AIBlock['LearningInput', 'LearningOutput', dict])¶

Unified learning layer — T0-T3 OODA strategies, dataset construction, self-training loops.

Constructor:

Methods:

register_agent_fn(name: str, fn: AgentFn) -> None¶

register_dataset(dataset: Dataset) -> None¶

infer(data: LearningInput) -> Result[LearningOutput]¶

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

Read-only catalog of the agentic-pattern slugs this component applies.

LearningEvent¶

LearningObserver(Protocol)¶

Methods:

on_event(event: LearningEvent) -> None¶

LogObserver¶

Logs events via stdlib logging.

Constructor:

Methods:

on_event(event: LearningEvent) -> None¶

JsonlObserver¶

Appends events as JSONL to a file.

Constructor:

Methods:

on_event(event: LearningEvent) -> None¶

CallbackObserver¶

Wraps a user-provided callback function.

Constructor:

Methods:

on_event(event: LearningEvent) -> None¶

EventEmitter¶

Dispatches events to all registered observers.

Constructor:

Methods:

add_observer(observer: LearningObserver) -> None¶

emit(event_type: str, **data: Any) -> list[ObserverFailure]¶

Dispatch event_type to every observer.

ObservationResult¶

Raw observations from running tasks — the Observe phase output.

Methods:

n_passed() -> int¶

n_failed() -> int¶

Orientation¶

Mental model update — the Orient phase output.

Methods:

all_theories() -> list[Theory]¶

TempoDecision¶

Whether to continue cycling and at what pace.

PromptEvolver¶

Selects top theories by strength and builds a compact prompt addendum.

Constructor:

Methods:

build_addendum(task_meta: dict[str, Any] | None = None) -> PromptAddendum¶

record_applications(instance_id: str, theory_ids: list[str]) -> None¶

record_outcome(instance_id: str, theory_ids: list[str], passed: bool, same_failure: bool = False) -> None¶

LearningRunner¶

Drives OODA cycles through the selected strategy.

Constructor:

Methods:

run(strategy: LearningStrategy, dataset: Dataset, agent_fn: AgentFn, run_dir: str | None = None) -> LearningResult¶

resume(checkpoint_path: str, strategy: LearningStrategy, dataset: Dataset, agent_fn: AgentFn) -> LearningResult¶

TaskInstance¶

DatasetSplit¶

Methods:

to_instances() -> list[TaskInstance]¶

Dataset¶

AgentTrace¶

TaskResult¶

FailureAnalysis¶

Theory¶

Methods:

compute_strength() -> float¶

should_retire() -> bool¶

PromptAddendum¶

ConfigOverride¶

SkillDocument¶

GeneratedTool¶

OODACycleRecord¶

LearningResult¶

LearningCheckpoint¶

LearningConfig¶

SkillGenerator¶

Generates markdown skill guides when theories need escalation.

Constructor:

Methods:

should_generate(failure_code: str) -> bool¶

generate(theory: Theory, analyses: list[FailureAnalysis] | None = None) -> SkillDocument | None¶

get_deployable_skills() -> list[dict[str, str]]¶

LearningStrategy(Protocol)¶

Each tier implements the OODA cycle with different depth and learning rules.

Methods:

tier() -> str¶

observe(dataset: Dataset, agent_fn: AgentFn) -> ObservationResult¶

orient(observations: ObservationResult, store: TheoryStore) -> Orientation¶

decide(orientation: Orientation) -> list[HarnessArtifact]¶

act(artifacts: list[HarnessArtifact], agent_fn: AgentFn) -> AgentFn¶

check_tempo(history: list[ObservationResult]) -> TempoDecision¶

TheoryStore¶

SQLite-backed theory persistence with Bayesian strength tracking.

Constructor:

Methods:

close() -> None¶

seed_theories(seeds: list[dict[str, str]]) -> list[str]¶

create_theory(failure_code: str, trigger: str = 'always', instruction: str = '', rationale: str = '', initial_strength: float = 0.5, source_instances: list[str] | None = None) -> str¶

upsert_theory(theory: Theory) -> None¶

Insert or replace a serialized theory while preserving its id.

import_theories(theories: list[Theory]) -> int¶

get_active_theories() -> list[Theory]¶

get_theory(theory_id: str) -> Theory | None¶

find_by_failure_code(failure_code: str) -> list[Theory]¶

record_application(instance_id: str, theory_id: str) -> None¶

record_outcome(instance_id: str, theory_id: str, passed: bool, same_failure: bool = False) -> None¶

retire_theory(theory_id: str) -> None¶

update_source_instances(theory_id: str, instance_id: str) -> None¶

update_skill(theory_id: str, skill_content: str) -> None¶

update_tool(theory_id: str, tool_code: str, tool_name: str) -> None¶

get_application_count(failure_code: str) -> int¶

all_theories() -> list[Theory]¶

Functions¶

export_formal_experience_bundle(run_dirs: Iterable[str | Path], output_path: str | Path, forbidden_terms: Iterable[str] = (), deploy_db_path: str = '', allow_empty: bool = False) -> dict[str, Any]¶

Export answer-free formal T3 experience as a learning harness bundle.

export_harness(run_dir: str, output_path: str, store: TheoryStore, artifacts: list[HarnessArtifact], metadata: dict[str, Any] | None = None, review: ReviewRecord | None = None) -> str¶

import_harness(bundle_path: str, target_env: str = 'development') -> tuple[list[Theory], list[HarnessArtifact]]¶