First learned result

Build a small process-output predictor. Supply completed observations, let Adapt-1 discover an input projection, and query again 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 data is a synthetic fixture, not a real physical process or a benchmark. No database, simulator, SDK, or additional Python package is needed.

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 your actual responses so you can inspect admission, predictions, and retained state. 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 three frames and saves the complete responses. Read the saved prediction and support fields for your 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; a stored event alone is not proof of learner admission. 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. It is a limited retained-use check, not proof that every subsystem is frozen. See Choose a learning setup for the full evaluation boundary.

What you supplied, and what Adapt-1 formed

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

No input path list, predictive rule, formula, or answer for the query is supplied to Adapt-1. There is no reward call because this example learns an observable target, not the utility of a selected action.

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. Keep evaluation records separate before ingestion; successful integration does not establish real-world prediction quality.

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

Discovery

Operations

API reference

Run the example

Read your actual responses

What you supplied, and what Adapt-1 formed

Use your own records

Other prediction and Discovery patterns

Learn from an executed decision

​Run the example

​Read your actual responses

​What you supplied, and what Adapt-1 formed

​Use your own records

Other prediction and Discovery patterns

Learn from an executed decision

Run the example

Read your actual responses

What you supplied, and what Adapt-1 formed

Use your own records