First learned result

Build a process-output predictor from completed observations. Adapt-1 discovers an input projection from the submitted records. Queries supply current measurements with the output withheld.

Step	API call	What to check
Create the task	`POST /domains`	Numeric target and Discovery configuration
Read the empty state	`POST /domains/{domain_id}/query`	Initial result before observations
Send completed observations	`POST /domains/{domain_id}/events`	Learner admission for each event
Predict and reuse	`POST /domains/{domain_id}/query`	Prediction status, support, and retained state

The included synthetic fixture and Python standard library are sufficient to run the example. The configured support thresholds and projection limits are example parameters.

Run the example

Use Python 3.10 or later and an API key from the quickstart. Running this script creates one fresh Domain and sends the observations below. It never clears existing state.

Environment

export ADAPT1_API_KEY="YOUR_SECRET_TOKEN"
python3 first_learned_result.py

The requests use Transition Discovery. The script saves each request and response for inspection. Save this complete file as first_learned_result.py:

Copy the complete Python program

first_learned_result.py

import json
import os
import uuid
from pathlib import Path
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen

BASE = "https://rei-neuroadapt-api.reilabs.org/api/v1"
KEY = os.environ.get("ADAPT1_API_KEY")
if not KEY:
    raise SystemExit("Set ADAPT1_API_KEY in your server environment.")

DOMAIN = "first-result-" + uuid.uuid4().hex[:16]
LOG = Path(DOMAIN)
LOG.mkdir(exist_ok=False)
(LOG / "run.json").write_text(json.dumps({"domain_id": DOMAIN}))
print("Domain and local record:", DOMAIN)


def post(path, body, label):
    # Save intent before sending; never log the key or retry a write blindly.
    record = {"path": path, "request": body}
    destination = LOG / (label + ".json")
    destination.write_text(json.dumps(record, indent=2), encoding="utf-8")
    request = Request(
        BASE + path,
        data=json.dumps(body, allow_nan=False).encode("utf-8"),
        headers={"Authorization": f"Bearer {KEY}", "Content-Type": "application/json"},
        method="POST",
    )
    try:
        with urlopen(request, timeout=120) as response:
            result = json.load(response)
    except HTTPError as error:
        raise SystemExit(f"{label}: HTTP {error.code}. No retry sent; check the contract.") from None
    except (URLError, TimeoutError, json.JSONDecodeError, UnicodeDecodeError):
        raise SystemExit(
            f"{label}: no usable response. A write may have applied; "
            f"reconcile it before retrying. Intent saved in {destination}."
        ) from None
    if not isinstance(result, dict):
        raise SystemExit(f"{label}: expected a JSON object.")
    record["response"] = result
    destination.write_text(json.dumps(record, indent=2), encoding="utf-8")
    return result


def inspect(response):
    state = response.get("learning_state")
    subsystems = state.get("subsystems") if isinstance(state, dict) else None
    learner = subsystems.get("structured_transition") if isinstance(subsystems, dict) else None
    if not isinstance(learner, dict):
        raise SystemExit("Missing transition diagnostics; inspect the saved response.")
    print(json.dumps({
        "projection": learner.get("autonomous_projection"),
        "sample_count": learner.get("sample_count"),
        "prediction": response.get("transition_prediction"),
    }, indent=2))
    return learner


post("/domains", {
    "domain_id": DOMAIN,
    "session_id": "ignored",
    "schema": {"event_types": ["observation"]},
    "learning": {"enabled": True, "transition": {
        "enabled": True,
        "event_types": ["observation"],
        "targets": [{"path": "values.process.output", "type": "number"}],
        "required_support": 3,
        "numeric_model": "auto",
        "autonomous_projection": {
            "enabled": True,
            "minimum_observations": 3,
            "minimum_availability": 0.8,
            "maximum_input_paths": 8,
        },
    }},
}, "00-create")

route = f"/domains/{DOMAIN}"
query = {
    "session_id": "ignored",
    "question": "Predict the process output.",
    "context": {"values": {"process": {"temperature": 26.0, "pressure": 1.2}}},
    "return_fields": ["transition_prediction", "learning_state"],
    "update_memory_state": False,
    "allow_exploration": False,
}
print("BEFORE OBSERVATIONS")
inspect(post(route + "/query", query, "01-before"))

# Synthetic completed observations: temperature, pressure, observed output.
observations = [
    (20.0, 1.0, 50.0), (20.0, 1.2, 52.0), (20.0, 1.4, 54.0),
    (24.0, 1.0, 58.0), (24.0, 1.2, 60.0), (24.0, 1.4, 62.0),
    (28.0, 1.0, 66.0), (28.0, 1.2, 68.0), (28.0, 1.4, 70.0),
    (32.0, 1.0, 74.0), (32.0, 1.2, 76.0), (32.0, 1.4, 78.0),
]
for index, (temperature, pressure, output) in enumerate(observations):
    receipt = post(route + "/events", {
        "session_id": "ignored",
        "event_type": "observation",
        "values": {"process": {
            "temperature": temperature, "pressure": pressure, "output": output,
        }},
    }, f"02-event-{index:02d}")
    if receipt.get("learner_eligibility") is None:
        raise SystemExit("Missing learner_eligibility; inspect the event receipt.")
    print("Observation", index, json.dumps(receipt["learner_eligibility"]))

print("AFTER OBSERVATIONS")
after = post(route + "/query", query, "03-after")
learned = inspect(after)
projection = learned.get("autonomous_projection")
if not isinstance(projection, dict):
    raise SystemExit("Missing projection diagnostics; inspect the saved response.")
if projection.get("status") != "ready" or projection.get("input_source") != "discovered":
    raise SystemExit("Projection is not ready/discovered; inspect admission and configuration.")
prediction = after.get("transition_prediction")
if not isinstance(prediction, dict) or prediction.get("status") != "predicted":
    raise SystemExit("No supported prediction; inspect the saved abstention diagnostics.")

print("RETAINED-STATE READ")
reused = inspect(post(route + "/query", query, "04-reused"))
for field in ("sample_count", "model_version"):
    if learned.get(field) is None or reused.get(field) is None:
        print(field, "not exposed on both reads; unchanged state is not verified.")
    elif learned[field] != reused[field]:
        raise SystemExit(f"{field} changed between reads; investigate before frozen evaluation.")
print("Responses saved. Review the prediction and supporting state; no accuracy score is claimed.")

Read your actual responses

The script prints the before, after, and retained-use responses and saves them in full. Read the saved prediction and support fields for your run. The recorded projection trace shows admission and later prediction fields from a separate run.

Frame	What to inspect
Before observations	What the fresh Domain returns without task history
After observations	The discovered `input_paths`, sample diagnostics, and `transition_prediction`
Retained-state read	The same query after no intervening event or feedback writes

Early events may report projection_accumulating while Discovery forms a usable projection. They can be buffered and rebuilt when the projection becomes ready. Review learner_eligibility in each saved receipt to check whether the event was admitted to learning. Only consume predicted values when transition_prediction.status is predicted. Keep the returned prediction object and its support information. Extract the target value using the response contract for your configured target. The final comparison checks exposed transition sample counts and model versions. Other learner subsystems need their own frozen-state checks. See Choose a learning setup for the full evaluation boundary.

Task contract and learned result

Supplied by your application	Formed from admitted observations
Event type, numeric target, and task scope	Eligible transition input projection
Completed measurements and their outputs	Retained transition evidence and learner state
New measurements with the output withheld	Supported prediction or an explicit abstention

Discovery selects eligible inputs from the records; the application supplies the target path and type. Completed measurements provide the observed target directly through events.

Reconnect, rerun, or clean up?

Keep the printed Domain ID and the same authenticated owner to reconnect to retained state. Running this script again intentionally creates a new Domain. It never resets or deletes the old one. Remove a tutorial Domain only through an explicit, ID-scoped deletion after deciding its state is no longer needed. The local response files can contain application data; protect them accordingly.

Use your own records

Replace the fixture with completed application records and the query with current measurements. Keep the target unavailable at query time. A new target path or type requires a matching event contract. Reserve evaluation records before ingestion to measure prediction quality on separate cases.

Other prediction and Discovery patterns

Use categorical outputs, structural discovery, or before/after observations.

Learn from an executed decision

Bind the measured outcome to a policy and inspect the later preference.

Get started

Concepts

Guides

Planning

Discovery

Operations

API reference

Run the example

Read your actual responses

Task contract and learned result

Use your own records

Other prediction and Discovery patterns

Learn from an executed decision

​Run the example

​Read your actual responses

​Task contract and learned result

​Use your own records

Other prediction and Discovery patterns

Learn from an executed decision

Run the example

Read your actual responses

Task contract and learned result

Use your own records