diff --git a/backend/fluksio/alembic/versions/a3c7e5d91f42_run_for_node.py b/backend/fluksio/alembic/versions/a3c7e5d91f42_run_for_node.py new file mode 100644 index 0000000..8832a66 --- /dev/null +++ b/backend/fluksio/alembic/versions/a3c7e5d91f42_run_for_node.py @@ -0,0 +1,33 @@ +"""run.for_node + +A wren node's Train button submits wren's training flow as a run. The node +it was pressed on is what ties that run back to the node's panel. + +Revision ID: a3c7e5d91f42 +Revises: ccb21eacf5d5 +Create Date: 2026-10-24 + +""" + +import sqlalchemy as sa +import sqlmodel.sql.sqltypes +from alembic import op + +# revision identifiers, used by Alembic. +revision = "a3c7e5d91f42" +down_revision = "ccb21eacf5d5" +branch_labels = None +depends_on = None + + +def upgrade(): + op.add_column( + "run", + sa.Column( + "for_node", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=True + ), + ) + + +def downgrade(): + op.drop_column("run", "for_node") diff --git a/backend/fluksio/api/routes/flows.py b/backend/fluksio/api/routes/flows.py index d343fa9..44fbfcd 100644 --- a/backend/fluksio/api/routes/flows.py +++ b/backend/fluksio/api/routes/flows.py @@ -57,6 +57,7 @@ from fluksio.flow.devices import ( ) from fluksio.flow.events import event_bus from fluksio.flow.messages import MEDIA_FAMILIES, DType, flow_of, qualify +from fluksio.flow.nodes.wren import TRAIN_FLOW, WrenNode, pack_meta from fluksio.flow.pipeline import ValidationIssue from fluksio.flow.runs import RunRejected from fluksio.flow.schemas import ( @@ -84,7 +85,7 @@ from fluksio.flow.store import ( LibNotFound, StaleVersion, ) -from fluksio.models import Flavor, Message, Run +from fluksio.models import Flavor, Message, Run, RunArtifact, RunNode logger = logging.getLogger(__name__) @@ -159,6 +160,24 @@ class TriggerRequest(BaseModel): values: dict[str, Any] = {} +class Training(BaseModel): + """A wren node's latest training run, as its panel shows it.""" + + id: str + #: queued, running, ok, error, cancelled or abandoned. + status: str + #: What went wrong, when it failed. + error: str = "" + #: final_loss, n_params, exact_match, threshold and coverage, once it ended. + result: dict[str, Any] = {} + #: The trained model's digest, for the node's ``model`` setting. + model: str = "" + #: From the model's own record: the confidence it recommends, and its + #: score on the node's ``tests`` when there were any. + threshold: float | None = None + test_exact_match: float | None = None + + def _endpoints(controller: FlowController, definition: FlowDef) -> list[Endpoint]: """Everything wired into this flow from outside it.""" flow = definition.name @@ -1010,6 +1029,111 @@ def acknowledge_node_error( return Message(message=f"Cleared the failure on '{node_id}'") +# ----------------------------------------------------------------------------- +# Training a wren node +# ----------------------------------------------------------------------------- + + +def _node_def(controller: FlowController, name: str, node_id: str) -> NodeDef: + """A node as the editor shows it: from the draft when there is one.""" + node = next( + (n for n in _read_flow(controller, name).nodes if n.id == node_id), None + ) + if node is None: + raise HTTPException( + status_code=404, detail=f"No node named '{node_id}' in flow '{name}'" + ) + return node + + +@router.post("/{name}/nodes/{node_id}/train", response_model=Training, status_code=202) +async def train_node( + name: str, + node_id: str, + request: Request, + controller: FlowControllerDep, + user: CurrentUser, +) -> Any: + """Train a new model for a wren node, as a run of wren's training flow. + + It learns the node's targets, its ``examples`` and its ``tests``, as the + editor shows them. The node keeps running its model until the new one is + set as its ``model``. + """ + node = _node_def(controller, name, node_id) + if node.type != "wren": + raise HTTPException(status_code=400, detail=f"'{node_id}' is not a wren node") + try: + wren = WrenNode( + requires=[spec for spec in node.requires if spec.name], + provides=[spec for spec in node.provides if spec.name], + params=node.params, + ) + except (ValueError, RuntimeError) as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + service = getattr(request.app.state, "run_service", None) + if service is None: + raise HTTPException(status_code=503, detail="Runs are not available") + try: + run = await run_in_threadpool( + service.submit, + TRAIN_FLOW, + params={"installation": orjson.dumps(wren.installation()).decode()}, + cause="train", + actor=user.email, + no_cache=True, + for_node=f"{name}.{node_id}", + ) + except FlowNotFound: + raise HTTPException( + status_code=409, + detail=( + f"wren's training flow '{TRAIN_FLOW}' is not registered. Run " + "`fluksio sync wren.flow` from an environment with wren[train]." + ), + ) from None + except RunRejected as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc + return Training(id=run.id, status=run.status) + + +@router.get("/{name}/nodes/{node_id}/train", response_model=Training | None) +def read_training( + name: str, + node_id: str, + request: Request, + controller: FlowControllerDep, + session: SessionDep, +) -> Any: + """A wren node's latest training run, or null when it has none.""" + _node_def(controller, name, node_id) + run = session.exec( + select(Run) + .where(col(Run.flow) == TRAIN_FLOW, col(Run.for_node) == f"{name}.{node_id}") + .order_by(col(Run.created_at).desc()) + ).first() + if run is None: + return None + training = Training(id=run.id, status=run.status, result=run.result) + if run.status == "error": + failed = session.exec( + select(RunNode.error).where( + col(RunNode.run_id) == run.id, col(RunNode.error) != "" + ) + ).first() + training.error = failed or run.status_reason + # The model, once the run made one the store still has. + pack = session.get(RunArtifact, {"run_id": run.id, "name": f"{TRAIN_FLOW}.pack"}) + store = getattr(request.app.state, "artifact_store", None) + path = store.path(pack.digest) if pack is not None and store else None + if pack is not None and path is not None: + meta = pack_meta(path) + training.model = pack.digest + training.threshold = meta.get("threshold") + training.test_exact_match = meta.get("test_exact_match") + return training + + @router.get("/{name}/state", response_model=FlowStatePublic) def read_flow_state(name: str, controller: FlowControllerDep, hidden: HiddenDep) -> Any: """The last value seen on every message of this flow.""" diff --git a/backend/fluksio/flow/nodes/wren.py b/backend/fluksio/flow/nodes/wren.py index 6459b15..d4a51ba 100644 --- a/backend/fluksio/flow/nodes/wren.py +++ b/backend/fluksio/flow/nodes/wren.py @@ -46,6 +46,13 @@ def _wren() -> Any: return wren +def pack_meta(path: Any) -> dict[str, Any]: + """What a model records of its training: its targets, threshold and, + when it had tests, ``test_exact_match``.""" + found: dict[str, Any] = _wren().load_pack(path)["meta"] + return found + + #: The input the command arrives on, and the only one that wakes the node. PROMPT = "prompt" #: The output the calls are reported on, and the two that say what came of @@ -55,6 +62,8 @@ REPLY = "reply" UNHANDLED = "unhandled" #: The reply when nothing was done and nothing reads ``unhandled``. CANT = "Sorry, I can't do that yet." +#: wren's training flow, which wren registers: ``fluksio sync wren.flow``. +TRAIN_FLOW = "wren_train" _TARGET_TYPES = {DType.BOOL, DType.INT, DType.FLOAT, DType.STR} @@ -260,6 +269,17 @@ class WrenNode(Node): description="Your own words, and words heard wrong, in the model's: " '{"traverse": "living room light", "page too": "patio"}.', ) + examples: list[dict[str, Any]] = Field( + default_factory=list, + description="Training only: commands to learn, targets named by port, " + 'as {"state": {"user_location": "kitchen"}, "command": "lights on", ' + '"calls": [["kitchen_light", true]]}.', + ) + tests: list[dict[str, Any]] = Field( + default_factory=list, + description="Training only: commands a new model is scored on, like " + 'examples. "calls": [] is one it must pass on.', + ) __slots__ = ("cfg", "targets", "pack", "untrained") @@ -339,6 +359,15 @@ class WrenNode(Node): name=name or "wren", ) + def installation(self) -> dict[str, Any]: + """What wren's training flow learns from: the targets, each with a + value, and the ``examples`` and ``tests``.""" + return { + "targets": [t | {"value": _unset(t)} for t in self.targets.values()], + "examples": self.cfg.examples, + "tests": self.cfg.tests, + } + async def start(self, app: FastAPI | None = None) -> None: store = getattr(app.state, "artifact_store", None) if app else None self.pack = self._load(store) diff --git a/backend/fluksio/flow/runs.py b/backend/fluksio/flow/runs.py index 50fb47f..5f12f2b 100644 --- a/backend/fluksio/flow/runs.py +++ b/backend/fluksio/flow/runs.py @@ -951,6 +951,7 @@ class RunService: idempotency_key: str | None = None, parent_id: str | None = None, passed: list[str] | None = None, + for_node: str | None = None, ) -> Run: """Journal a run and wake an engine up for it. Never blocks on it. @@ -993,6 +994,7 @@ class RunService: actor=actor, idempotency_key=idempotency_key, parent_id=parent_id, + for_node=for_node, ) with Session(db_engine) as session: session.add(run) @@ -1249,6 +1251,7 @@ class RunService: # What its parent's caller chose. A parent from before that was # recorded falls back to the names its params carry. passed=source.passed, + for_node=source.for_node, ) def _flow_of(self, run_id: str) -> str | None: diff --git a/backend/fluksio/models.py b/backend/fluksio/models.py index 01b41bb..f8edf75 100644 --- a/backend/fluksio/models.py +++ b/backend/fluksio/models.py @@ -388,8 +388,11 @@ class Run(SQLModel, table=True): external_id: str | None = Field( default=None, unique=True, index=True, max_length=255 ) - #: Where it was asked for: api, cli, sdk, hook, sweep or import. + #: Where it was asked for: api, cli, sdk, hook, sweep, import or train. cause: str = Field(default="api", max_length=32) + #: The node it was asked for from, as `flow.node`: the wren node a + #: training run is for. Null for every other run. + for_node: str | None = Field(default=None, max_length=255) #: Re-execute every node, whatever the stage cache holds for it. no_cache: bool = False #: Run the unpublished working copy rather than what is published. Read diff --git a/backend/tests/api/routes/test_wren_train.py b/backend/tests/api/routes/test_wren_train.py new file mode 100644 index 0000000..af7e111 --- /dev/null +++ b/backend/tests/api/routes/test_wren_train.py @@ -0,0 +1,186 @@ +"""Train on a wren node: its wiring and settings go to wren's training flow. + +The model package is optional, so ``FakeWren`` (tests/flow/conftest.py) +stands in for it. +""" + +import json +from collections.abc import Generator +from datetime import UTC, datetime, timedelta + +import pytest +from fastapi.testclient import TestClient +from sqlmodel import Session + +from fluksio.core.config import settings +from fluksio.core.db import engine as db_engine +from fluksio.models import Run, RunArtifact +from tests.flow.conftest import FakeWren + +FLOW = "wren_house" +PREFIX = f"{settings.API_V1_STR}/flows/{FLOW}/nodes" +EXAMPLES = [{"command": "warmer please", "calls": [["living_room_heating", 22]]}] +TESTS = [ + {"command": "is it cold outside", "calls": [["outdoor_temperature"]]}, + {"command": "good night", "calls": []}, +] + + +@pytest.fixture(autouse=True) +def fake_wren(monkeypatch: pytest.MonkeyPatch) -> FakeWren: + fake = FakeWren() + monkeypatch.setattr("fluksio.flow.nodes.wren._wren", lambda: fake) + return fake + + +@pytest.fixture(scope="module", autouse=True) +def house( + client: TestClient, superuser_token_headers: dict[str, str] +) -> Generator[None, None, None]: + """Two wren nodes, so a run for one is not the other's, and a function.""" + prompt = {"name": "command", "dtype": "str", "port": "prompt"} + nodes = [ + { + "id": "control", + "type": "wren", + "params": { + "targets": { + "living_room_heating": {"range": [5, 30], "unit": "°C"}, + "outdoor_temperature": {"description": "temperature outside"}, + }, + "examples": EXAMPLES, + "tests": TESTS, + }, + "requires": [ + prompt, + {"name": "outdoor", "dtype": "float", "port": "outdoor_temperature"}, + ], + "provides": [ + {"name": "setpoint", "dtype": "float", "port": "living_room_heating"} + ], + }, + { + "id": "lights", + "type": "wren", + "requires": [prompt], + "provides": [{"name": "kitchen_light", "dtype": "bool"}], + }, + {"id": "sensor", "type": "python"}, + ] + url = f"{settings.API_V1_STR}/flows/{FLOW}" + saved = client.put( + url, headers=superuser_token_headers, json={"name": FLOW, "nodes": nodes} + ) + assert saved.status_code == 200, saved.text + yield + client.delete(url, headers=superuser_token_headers) + + +def test_train_submits_the_node_as_wren_s_installation( + client, superuser_token_headers, monkeypatch +): + """The draft's ports, targets, examples and tests, never from the cache.""" + seen: dict[str, object] = {} + + class Recorder: + def submit(self, name, **kwargs): + seen.update(kwargs, flow=name) + return Run(id="train-1", flow=name, created_at=datetime.now(UTC)) + + monkeypatch.setattr(client.app.state, "run_service", Recorder()) + answer = client.post(f"{PREFIX}/control/train", headers=superuser_token_headers) + + assert answer.status_code == 202, answer.text + assert answer.json()["id"] == "train-1" + assert (seen["flow"], seen["no_cache"], seen["cause"], seen["for_node"]) == ( + "wren_train", + True, + "train", + f"{FLOW}.control", + ) + installation = json.loads(seen["params"]["installation"]) # type: ignore[index] + assert installation["targets"] == [ + { + "name": "living_room_heating", + "type": "float", + "values": None, + "range": [5, 30], + "description": "", + "unit": "°C", + "readonly": False, + "value": 5.0, + }, + { + "name": "outdoor_temperature", + "type": "float", + "values": None, + "range": None, + "description": "temperature outside", + "unit": None, + "readonly": True, + "value": 0.0, + }, + ] + assert (installation["examples"], installation["tests"]) == (EXAMPLES, TESTS) + + +def test_train_says_how_to_register_a_missing_training_flow( + client, superuser_token_headers +): + answer = client.post(f"{PREFIX}/control/train", headers=superuser_token_headers) + + assert answer.status_code == 409 + assert "fluksio sync wren.flow" in answer.json()["detail"] + + +def test_only_a_wren_node_trains(client, superuser_token_headers): + answer = client.post(f"{PREFIX}/sensor/train", headers=superuser_token_headers) + assert answer.status_code == 400 + missing = client.post(f"{PREFIX}/nobody/train", headers=superuser_token_headers) + assert missing.status_code == 404 + + +def test_a_node_reads_its_own_latest_training( + client, superuser_token_headers, fake_wren +): + """With the model's digest and what its record says of it.""" + pack = client.app.state.artifact_store.put([b"weights"], name="wren.bin") + fake_wren.pack = {"meta": {"threshold": 0.93, "test_exact_match": 0.5}} + now = datetime.now(UTC) + with Session(db_engine) as session: + for run_id, node, ago in ( + ("train-old", "control", 3), + ("train-new", "control", 2), + ("train-lights", "lights", 1), + ): + session.add( + Run( + id=run_id, + flow="wren_train", + status="ok", + for_node=f"{FLOW}.{node}", + result={"exact_match": 0.98, "coverage": 0.9}, + created_at=now - timedelta(minutes=ago), + ) + ) + session.add( + RunArtifact( + run_id="train-new", name="wren_train.pack", digest=pack["digest"] + ) + ) + session.commit() + + training = client.get(f"{PREFIX}/control/train", headers=superuser_token_headers) + + assert training.status_code == 200 + assert training.json() == { + "id": "train-new", + "status": "ok", + "error": "", + "result": {"exact_match": 0.98, "coverage": 0.9}, + "model": pack["digest"], + "threshold": 0.93, + "test_exact_match": 0.5, + } + lights = client.get(f"{PREFIX}/lights/train", headers=superuser_token_headers) + assert lights.json()["id"] == "train-lights" diff --git a/docs/reference/node-types.md b/docs/reference/node-types.md index a7e4a0d..cc5d69c 100644 --- a/docs/reference/node-types.md +++ b/docs/reference/node-types.md @@ -378,6 +378,8 @@ name of its own, in the node panel's port name field or as | `min_confidence` | empty | the confidence a call needs; empty takes the model's own threshold | | `targets` | empty | by target port, what the model is told of it and how far it is trusted (below) | | `aliases` | empty | your own words, or words the recogniser hears wrong, in the model's: `{"traverse": "living room light", "page too": "patio"}` | +| `examples` | empty | for training only, commands to learn beyond the ones wren generates (below) | +| `tests` | empty | for training only, commands a new model is scored on (below) | Every field of a `targets` entry is optional: @@ -427,6 +429,34 @@ it runs without that warning only on ports named and described as there. A and every command goes on to `unhandled`. A model trained with wren goes into the store with `PUT /artifacts` (see the [API](../code/api.md#artifacts)). +**Train** in the node panel trains a new model for the node as a run of wren's +training flow, `wren_train`. It learns the targets wired to the node, as the +panel shows them with unpublished edits included, and its `examples`, and it +scores the new model on its `tests`. Both are lists in wren's installation +format, with targets named by port: + +```json +{"state": {"user_location": "kitchen"}, "command": "lights on please", + "calls": [["kitchen_light", true]]} +``` + +A call is `[target, value]`, `[target, value, delay_s]` or, for a read, +`[target]`. A test with `"calls": []` is a command the model must pass on. The +node never reads either setting when it runs, and wren refuses an example its +targets cannot answer before training starts. The panel follows the run and +shows the model's exact match, coverage and threshold, and its exact match on +the tests, once it ends. **Use this model** sets the new model as `model` in +the draft, and publishing puts it live. `POST` and `GET` on +`/flows/{name}/nodes/{id}/train` start a training run and read the latest one. + +Training needs wren's `train` extra, which brings JAX, where the run executes: +in **Modules** (see [Packages](../code/nodes.md#packages)) or on a worker. List +`wren[train] @ https://github.com/stroblme/wren/archive/main.tar.gz` there +(an archive, since the engine image has no git). The flow comes from +wren and is registered once with `fluksio sync wren.flow`, run from an +environment with `wren[train]` and fluksio. Until then Train answers that the +flow is missing. + The command is read in lower case without sentence marks and with number words as digits ("in five minutes" as "in 5 minutes"), as the model was trained. `aliases` are applied before, whole words in any case, and `unhandled` carries diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts index c954d93..a1b71b4 100644 --- a/frontend/src/client/schemas.gen.ts +++ b/frontend/src/client/schemas.gen.ts @@ -4505,6 +4505,61 @@ export const TokenRequestSchema = { title: 'TokenRequest' } as const; +export const TrainingSchema = { + properties: { + id: { + type: 'string', + title: 'Id' + }, + status: { + type: 'string', + title: 'Status' + }, + error: { + type: 'string', + title: 'Error', + default: '' + }, + result: { + additionalProperties: true, + type: 'object', + title: 'Result', + default: {} + }, + model: { + type: 'string', + title: 'Model', + default: '' + }, + threshold: { + anyOf: [ + { + type: 'number' + }, + { + type: 'null' + } + ], + title: 'Threshold' + }, + test_exact_match: { + anyOf: [ + { + type: 'number' + }, + { + type: 'null' + } + ], + title: 'Test Exact Match' + } + }, + type: 'object', + required: ['id', 'status'], + title: 'Training', + description: "A wren node's latest training run, as its panel shows it." +} as const; + export const TriggerRequestSchema = { properties: { values: { diff --git a/frontend/src/client/sdk.gen.ts b/frontend/src/client/sdk.gen.ts index 788d34b..6db2c46 100644 --- a/frontend/src/client/sdk.gen.ts +++ b/frontend/src/client/sdk.gen.ts @@ -3,7 +3,7 @@ import type { CancelablePromise } from './core/CancelablePromise'; import { OpenAPI } from './core/OpenAPI'; import { request as __request } from './core/request'; -import type { AiReadAiConfigResponse, AiUpdateAiConfigData, AiUpdateAiConfigResponse, AiDeleteAiConfigResponse, AiListAiModelsData, AiListAiModelsResponse, AiReadAiStatusResponse, AiReadControlNodesResponse, AiReadChatResponse, AiChatData, AiChatResponse, AiClearChatResponse, AlertsReadAlertsConfigResponse, AlertsSaveAlertsConfigData, AlertsSaveAlertsConfigResponse, AlertsReadWebpushKeyResponse, AlertsAddWebpushSubscriptionData, AlertsAddWebpushSubscriptionResponse, AlertsRemoveWebpushSubscriptionData, AlertsRemoveWebpushSubscriptionResponse, AlertsTestChannelData, AlertsTestChannelResponse, ArtifactsPutArtifactData, ArtifactsPutArtifactResponse, ArtifactsGetArtifactData, ArtifactsGetArtifactResponse, ArtifactsDeleteArtifactData, ArtifactsDeleteArtifactResponse, CloudReadStatusResponse, CloudEnrollData, CloudEnrollResponse, CloudAddRemoteUserData, CloudAddRemoteUserResponse, CloudReadMembersResponse, CloudLinkMemberData, CloudLinkMemberResponse, CloudRevokeMemberData, CloudRevokeMemberResponse, CloudDisconnectResponse, ConnectorsReadConnectorStateData, ConnectorsReadConnectorStateResponse, ConnectorsClearConnectorStateData, ConnectorsClearConnectorStateResponse, ConnectorsDeleteConnectorStateData, ConnectorsDeleteConnectorStateResponse, DashboardsReadDashboardsResponse, DashboardsReadDashboardData, DashboardsReadDashboardResponse, DashboardsCreateDashboardData, DashboardsCreateDashboardResponse, DashboardsSaveDashboardData, DashboardsSaveDashboardResponse, DashboardsDeleteDashboardData, DashboardsDeleteDashboardResponse, DashboardsGenerateResultsDashboardData, DashboardsGenerateResultsDashboardResponse, DashboardsPublishDashboardData, DashboardsPublishDashboardResponse, DashboardsDiscardDashboardDraftData, DashboardsDiscardDashboardDraftResponse, DashboardsDuplicateDashboardData, DashboardsDuplicateDashboardResponse, DashboardsRenameDashboardData, DashboardsRenameDashboardResponse, DevicesStartPairingData, DevicesStartPairingResponse, DevicesPollPairingData, DevicesPollPairingResponse, DevicesReadDevicesResponse, DevicesReadPendingResponse, DevicesPendingDeviceData, DevicesPendingDeviceResponse, DevicesCreateDeviceData, DevicesCreateDeviceResponse, DevicesReadDeviceData, DevicesReadDeviceResponse, DevicesSaveDeviceData, DevicesSaveDeviceResponse, DevicesDeleteDeviceData, DevicesDeleteDeviceResponse, DevicesReadIceData, DevicesReadIceResponse, DevicesPutReportData, DevicesPutReportResponse, DevicesApprovePairingData, DevicesApprovePairingResponse, DevicesUnpairDeviceData, DevicesUnpairDeviceResponse, FlavorsReadFlavorsResponse, FlavorsCreateFlavorData, FlavorsCreateFlavorResponse, FlavorsUpdateFlavorData, FlavorsUpdateFlavorResponse, FlavorsDeleteFlavorData, FlavorsDeleteFlavorResponse, FlowsReadFlowsResponse, FlowsReadNodeTypesResponse, FlowsReadGraphResponse, FlowsReadLibraryResponse, FlowsDeleteSharedNodeData, FlowsDeleteSharedNodeResponse, FlowsReadFlowData, FlowsReadFlowResponse, FlowsSaveFlowData, FlowsSaveFlowResponse, FlowsDeleteFlowData, FlowsDeleteFlowResponse, FlowsPublishFlowData, FlowsPublishFlowResponse, FlowsDiscardDraftData, FlowsDiscardDraftResponse, FlowsRenameFlowData, FlowsRenameFlowResponse, FlowsDuplicateFlowData, FlowsDuplicateFlowResponse, FlowsReadNodeSourceData, FlowsReadNodeSourceResponse, FlowsSaveNodeSourceData, FlowsSaveNodeSourceResponse, FlowsShareNodeData, FlowsShareNodeResponse, FlowsUnshareNodeData, FlowsUnshareNodeResponse, FlowsStartFlowData, FlowsStartFlowResponse, FlowsStopFlowData, FlowsStopFlowResponse, FlowsPauseFlowData, FlowsPauseFlowResponse, FlowsResumeFlowData, FlowsResumeFlowResponse, FlowsStepFlowData, FlowsStepFlowResponse, FlowsValidateFlowData, FlowsValidateFlowResponse, FlowsRunFlowData, FlowsRunFlowResponse, FlowsTriggerNodeData, FlowsTriggerNodeResponse, FlowsCancelNodeData, FlowsCancelNodeResponse, FlowsAcknowledgeNodeErrorData, FlowsAcknowledgeNodeErrorResponse, FlowsReadFlowStateData, FlowsReadFlowStateResponse, FlowsReadMessageHistoryData, FlowsReadMessageHistoryResponse, FlowsIssueSocketTicketResponse, LoginLoginAccessTokenData, LoginLoginAccessTokenResponse, LoginTestTokenResponse, LoginRecoverPasswordData, LoginRecoverPasswordResponse, LoginResetPasswordData, LoginResetPasswordResponse, LoginRecoverPasswordHtmlContentData, LoginRecoverPasswordHtmlContentResponse, MessagesReadMessagesResponse, MessagesPublishMessageData, MessagesPublishMessageResponse, MessagesReadMessageHistoryData, MessagesReadMessageHistoryResponse, ModulesReadModulesResponse, ModulesApplyModulesData, ModulesApplyModulesResponse, ModulesRefreshModulesResponse, OauthRegisterClientData, OauthRegisterClientResponse, OauthAuthorizeValidateData, OauthAuthorizeValidateResponse, OauthAuthorizeData, OauthAuthorizeResponse, OauthTokenData, OauthTokenResponse, OauthReadClientsResponse, OauthRevokeClientData, OauthRevokeClientResponse, ObservabilityReadSummaryResponse, ObservabilityReadTimeseriesData, ObservabilityReadTimeseriesResponse, ObservabilityReadFlowRollupsData, ObservabilityReadFlowRollupsResponse, ObservabilityReadNodeTypeTimingsData, ObservabilityReadNodeTypeTimingsResponse, ObservabilityReadRunsData, ObservabilityReadRunsResponse, ObservabilityReadEventsData, ObservabilityReadEventsResponse, ObservabilityReadDeadLettersData, ObservabilityReadDeadLettersResponse, ObservabilityFlushMetricsResponse, PrivateCreateUserData, PrivateCreateUserResponse, RunsCreateRunData, RunsCreateRunResponse, RunsCreateSweepData, RunsCreateSweepResponse, RunsImportRunsData, RunsImportRunsResponse, RunsReadRunsData, RunsReadRunsResponse, RunsDeleteRunsData, RunsDeleteRunsResponse, RunsReadOverviewResponse, RunsExportMetricsData, RunsExportMetricsResponse, RunsExportRunsData, RunsExportRunsResponse, RunsReadMetricNamesData, RunsReadMetricNamesResponse, RunsReadRunData, RunsReadRunResponse, RunsDeleteRunData, RunsDeleteRunResponse, RunsReadNodeOutputsData, RunsReadNodeOutputsResponse, RunsCancelRunData, RunsCancelRunResponse, RunsRetryRunData, RunsRetryRunResponse, RunsReadMetricsData, RunsReadMetricsResponse, RunsCompareMetricData, RunsCompareMetricResponse, SearchReadSearchIndexResponse, SecretsReadSecretsResponse, SecretsSaveSecretData, SecretsSaveSecretResponse, SecretsDeleteSecretData, SecretsDeleteSecretResponse, UsersReadUsersData, UsersReadUsersResponse, UsersCreateUserData, UsersCreateUserResponse, UsersReadUserMeResponse, UsersDeleteUserMeResponse, UsersUpdateUserMeData, UsersUpdateUserMeResponse, UsersUpdatePasswordMeData, UsersUpdatePasswordMeResponse, UsersRegisterUserData, UsersRegisterUserResponse, UsersReadUserByIdData, UsersReadUserByIdResponse, UsersUpdateUserData, UsersUpdateUserResponse, UsersDeleteUserData, UsersDeleteUserResponse, UtilsTestEmailData, UtilsTestEmailResponse, UtilsReadConfigResponse, UtilsHealthCheckResponse, UtilsHealthResponse, WorkersReadWorkersResponse, WorkersReadResourcesResponse, WorkersReadTokensResponse, WorkersIssueTokenData, WorkersIssueTokenResponse, WorkersRevokeTokenData, WorkersRevokeTokenResponse, WorkersReadRuntimeResponse } from './types.gen'; +import type { AiReadAiConfigResponse, AiUpdateAiConfigData, AiUpdateAiConfigResponse, AiDeleteAiConfigResponse, AiListAiModelsData, AiListAiModelsResponse, AiReadAiStatusResponse, AiReadControlNodesResponse, AiReadChatResponse, AiChatData, AiChatResponse, AiClearChatResponse, AlertsReadAlertsConfigResponse, AlertsSaveAlertsConfigData, AlertsSaveAlertsConfigResponse, AlertsReadWebpushKeyResponse, AlertsAddWebpushSubscriptionData, AlertsAddWebpushSubscriptionResponse, AlertsRemoveWebpushSubscriptionData, AlertsRemoveWebpushSubscriptionResponse, AlertsTestChannelData, AlertsTestChannelResponse, ArtifactsPutArtifactData, ArtifactsPutArtifactResponse, ArtifactsGetArtifactData, ArtifactsGetArtifactResponse, ArtifactsDeleteArtifactData, ArtifactsDeleteArtifactResponse, CloudReadStatusResponse, CloudEnrollData, CloudEnrollResponse, CloudAddRemoteUserData, CloudAddRemoteUserResponse, CloudReadMembersResponse, CloudLinkMemberData, CloudLinkMemberResponse, CloudRevokeMemberData, CloudRevokeMemberResponse, CloudDisconnectResponse, ConnectorsReadConnectorStateData, ConnectorsReadConnectorStateResponse, ConnectorsClearConnectorStateData, ConnectorsClearConnectorStateResponse, ConnectorsDeleteConnectorStateData, ConnectorsDeleteConnectorStateResponse, DashboardsReadDashboardsResponse, DashboardsReadDashboardData, DashboardsReadDashboardResponse, DashboardsCreateDashboardData, DashboardsCreateDashboardResponse, DashboardsSaveDashboardData, DashboardsSaveDashboardResponse, DashboardsDeleteDashboardData, DashboardsDeleteDashboardResponse, DashboardsGenerateResultsDashboardData, DashboardsGenerateResultsDashboardResponse, DashboardsPublishDashboardData, DashboardsPublishDashboardResponse, DashboardsDiscardDashboardDraftData, DashboardsDiscardDashboardDraftResponse, DashboardsDuplicateDashboardData, DashboardsDuplicateDashboardResponse, DashboardsRenameDashboardData, DashboardsRenameDashboardResponse, DevicesStartPairingData, DevicesStartPairingResponse, DevicesPollPairingData, DevicesPollPairingResponse, DevicesReadDevicesResponse, DevicesReadPendingResponse, DevicesPendingDeviceData, DevicesPendingDeviceResponse, DevicesCreateDeviceData, DevicesCreateDeviceResponse, DevicesReadDeviceData, DevicesReadDeviceResponse, DevicesSaveDeviceData, DevicesSaveDeviceResponse, DevicesDeleteDeviceData, DevicesDeleteDeviceResponse, DevicesReadIceData, DevicesReadIceResponse, DevicesPutReportData, DevicesPutReportResponse, DevicesApprovePairingData, DevicesApprovePairingResponse, DevicesUnpairDeviceData, DevicesUnpairDeviceResponse, FlavorsReadFlavorsResponse, FlavorsCreateFlavorData, FlavorsCreateFlavorResponse, FlavorsUpdateFlavorData, FlavorsUpdateFlavorResponse, FlavorsDeleteFlavorData, FlavorsDeleteFlavorResponse, FlowsReadFlowsResponse, FlowsReadNodeTypesResponse, FlowsReadGraphResponse, FlowsReadLibraryResponse, FlowsDeleteSharedNodeData, FlowsDeleteSharedNodeResponse, FlowsReadFlowData, FlowsReadFlowResponse, FlowsSaveFlowData, FlowsSaveFlowResponse, FlowsDeleteFlowData, FlowsDeleteFlowResponse, FlowsPublishFlowData, FlowsPublishFlowResponse, FlowsDiscardDraftData, FlowsDiscardDraftResponse, FlowsRenameFlowData, FlowsRenameFlowResponse, FlowsDuplicateFlowData, FlowsDuplicateFlowResponse, FlowsReadNodeSourceData, FlowsReadNodeSourceResponse, FlowsSaveNodeSourceData, FlowsSaveNodeSourceResponse, FlowsShareNodeData, FlowsShareNodeResponse, FlowsUnshareNodeData, FlowsUnshareNodeResponse, FlowsStartFlowData, FlowsStartFlowResponse, FlowsStopFlowData, FlowsStopFlowResponse, FlowsPauseFlowData, FlowsPauseFlowResponse, FlowsResumeFlowData, FlowsResumeFlowResponse, FlowsStepFlowData, FlowsStepFlowResponse, FlowsValidateFlowData, FlowsValidateFlowResponse, FlowsRunFlowData, FlowsRunFlowResponse, FlowsTriggerNodeData, FlowsTriggerNodeResponse, FlowsCancelNodeData, FlowsCancelNodeResponse, FlowsAcknowledgeNodeErrorData, FlowsAcknowledgeNodeErrorResponse, FlowsTrainNodeData, FlowsTrainNodeResponse, FlowsReadTrainingData, FlowsReadTrainingResponse, FlowsReadFlowStateData, FlowsReadFlowStateResponse, FlowsReadMessageHistoryData, FlowsReadMessageHistoryResponse, FlowsIssueSocketTicketResponse, LoginLoginAccessTokenData, LoginLoginAccessTokenResponse, LoginTestTokenResponse, LoginRecoverPasswordData, LoginRecoverPasswordResponse, LoginResetPasswordData, LoginResetPasswordResponse, LoginRecoverPasswordHtmlContentData, LoginRecoverPasswordHtmlContentResponse, MessagesReadMessagesResponse, MessagesPublishMessageData, MessagesPublishMessageResponse, MessagesReadMessageHistoryData, MessagesReadMessageHistoryResponse, ModulesReadModulesResponse, ModulesApplyModulesData, ModulesApplyModulesResponse, ModulesRefreshModulesResponse, OauthRegisterClientData, OauthRegisterClientResponse, OauthAuthorizeValidateData, OauthAuthorizeValidateResponse, OauthAuthorizeData, OauthAuthorizeResponse, OauthTokenData, OauthTokenResponse, OauthReadClientsResponse, OauthRevokeClientData, OauthRevokeClientResponse, ObservabilityReadSummaryResponse, ObservabilityReadTimeseriesData, ObservabilityReadTimeseriesResponse, ObservabilityReadFlowRollupsData, ObservabilityReadFlowRollupsResponse, ObservabilityReadNodeTypeTimingsData, ObservabilityReadNodeTypeTimingsResponse, ObservabilityReadRunsData, ObservabilityReadRunsResponse, ObservabilityReadEventsData, ObservabilityReadEventsResponse, ObservabilityReadDeadLettersData, ObservabilityReadDeadLettersResponse, ObservabilityFlushMetricsResponse, PrivateCreateUserData, PrivateCreateUserResponse, RunsCreateRunData, RunsCreateRunResponse, RunsCreateSweepData, RunsCreateSweepResponse, RunsImportRunsData, RunsImportRunsResponse, RunsReadRunsData, RunsReadRunsResponse, RunsDeleteRunsData, RunsDeleteRunsResponse, RunsReadOverviewResponse, RunsExportMetricsData, RunsExportMetricsResponse, RunsExportRunsData, RunsExportRunsResponse, RunsReadMetricNamesData, RunsReadMetricNamesResponse, RunsReadRunData, RunsReadRunResponse, RunsDeleteRunData, RunsDeleteRunResponse, RunsReadNodeOutputsData, RunsReadNodeOutputsResponse, RunsCancelRunData, RunsCancelRunResponse, RunsRetryRunData, RunsRetryRunResponse, RunsReadMetricsData, RunsReadMetricsResponse, RunsCompareMetricData, RunsCompareMetricResponse, SearchReadSearchIndexResponse, SecretsReadSecretsResponse, SecretsSaveSecretData, SecretsSaveSecretResponse, SecretsDeleteSecretData, SecretsDeleteSecretResponse, UsersReadUsersData, UsersReadUsersResponse, UsersCreateUserData, UsersCreateUserResponse, UsersReadUserMeResponse, UsersDeleteUserMeResponse, UsersUpdateUserMeData, UsersUpdateUserMeResponse, UsersUpdatePasswordMeData, UsersUpdatePasswordMeResponse, UsersRegisterUserData, UsersRegisterUserResponse, UsersReadUserByIdData, UsersReadUserByIdResponse, UsersUpdateUserData, UsersUpdateUserResponse, UsersDeleteUserData, UsersDeleteUserResponse, UtilsTestEmailData, UtilsTestEmailResponse, UtilsReadConfigResponse, UtilsHealthCheckResponse, UtilsHealthResponse, WorkersReadWorkersResponse, WorkersReadResourcesResponse, WorkersReadTokensResponse, WorkersIssueTokenData, WorkersIssueTokenResponse, WorkersRevokeTokenData, WorkersRevokeTokenResponse, WorkersReadRuntimeResponse } from './types.gen'; export class AiService { /** @@ -1796,6 +1796,56 @@ export class FlowsService { }); } + /** + * Train Node + * Train a new model for a wren node, as a run of wren's training flow. + * + * It learns the node's targets, its ``examples`` and its ``tests``, as the + * editor shows them. The node keeps running its model until the new one is + * set as its ``model``. + * @param data The data for the request. + * @param data.name + * @param data.nodeId + * @returns Training Successful Response + * @throws ApiError + */ + public static trainNode(data: FlowsTrainNodeData): CancelablePromise { + return __request(OpenAPI, { + method: 'POST', + url: '/api/v1/flows/{name}/nodes/{node_id}/train', + path: { + name: data.name, + node_id: data.nodeId + }, + errors: { + 422: 'Validation Error' + } + }); + } + + /** + * Read Training + * A wren node's latest training run, or null when it has none. + * @param data The data for the request. + * @param data.name + * @param data.nodeId + * @returns unknown Successful Response + * @throws ApiError + */ + public static readTraining(data: FlowsReadTrainingData): CancelablePromise { + return __request(OpenAPI, { + method: 'GET', + url: '/api/v1/flows/{name}/nodes/{node_id}/train', + path: { + name: data.name, + node_id: data.nodeId + }, + errors: { + 422: 'Validation Error' + } + }); + } + /** * Read Flow State * The last value seen on every message of this flow. diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index ab2a7bd..1366835 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -1471,6 +1471,21 @@ export type TokenRequest = { labels?: Array<(string)>; }; +/** + * A wren node's latest training run, as its panel shows it. + */ +export type Training = { + id: string; + status: string; + error?: string; + result?: { + [key: string]: unknown; + }; + model?: string; + threshold?: (number | null); + test_exact_match?: (number | null); +}; + export type TriggerRequest = { values?: { [key: string]: unknown; @@ -2112,6 +2127,20 @@ export type FlowsAcknowledgeNodeErrorData = { export type FlowsAcknowledgeNodeErrorResponse = (Message); +export type FlowsTrainNodeData = { + name: string; + nodeId: string; +}; + +export type FlowsTrainNodeResponse = (Training); + +export type FlowsReadTrainingData = { + name: string; + nodeId: string; +}; + +export type FlowsReadTrainingResponse = ((Training | null)); + export type FlowsReadFlowStateData = { name: string; }; diff --git a/frontend/src/components/Flow/NodePanel.tsx b/frontend/src/components/Flow/NodePanel.tsx index eb847ff..74e2f95 100644 --- a/frontend/src/components/Flow/NodePanel.tsx +++ b/frontend/src/components/Flow/NodePanel.tsx @@ -1,4 +1,5 @@ import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query" +import { Link } from "@tanstack/react-router" import { Activity, CornerDownRight, @@ -33,6 +34,8 @@ import { wiresAs, } from "@/components/Devices/ports" import { devicesQueryOptions } from "@/components/Devices/queries" +import { isLive, shortId } from "@/components/Runs/queries" +import { RunStatusBadge } from "@/components/Runs/RunStatus" import { Button } from "@/components/ui/button" import { Command, @@ -56,6 +59,7 @@ import useAuth from "@/hooks/useAuth" import useCustomToast from "@/hooks/useCustomToast" import { inCodeEditor, useShortcuts } from "@/lib/shortcuts" import { cn } from "@/lib/utils" +import { handleError } from "@/utils" import { qualify } from "./deriveEdges" import { DtypeValue } from "./FlowBoundary" import { @@ -1593,6 +1597,119 @@ function SharingSection({ ) } +/** A fraction as a percentage, or a dash when there is none. */ +const percent = (value: unknown) => + typeof value === "number" + ? value.toLocaleString(undefined, { + style: "percent", + maximumFractionDigits: 1, + }) + : "–" + +/** + * Train a new model for a wren node and offer it as the node's `model`. Using + * it is an edit like any other, so it waits in the draft until publishing. + */ +function TrainingSection({ + flow, + node, + onUse, +}: { + flow: string + node: NodeDef_Input + onUse: (model: string) => void +}) { + const queryClient = useQueryClient() + // Why it would not start, such as wren's training flow not being registered. + const [problem, setProblem] = useState(null) + const queryKey = ["wren-training", flow, node.id] + const { data: training } = useQuery({ + queryKey, + queryFn: () => FlowsService.readTraining({ name: flow, nodeId: node.id }), + // Training takes minutes, so only a run still going is read again. + refetchInterval: (query) => + query.state.data && isLive(query.state.data.status) ? 5_000 : false, + }) + const train = useMutation({ + mutationFn: () => FlowsService.trainNode({ name: flow, nodeId: node.id }), + onMutate: () => setProblem(null), + onSuccess: () => queryClient.invalidateQueries({ queryKey }), + onError: handleError.bind(setProblem), + }) + const running = training ? isLive(training.status) : false + const model = training?.model + const inUse = Boolean(model) && node.params?.model === model + + return ( +
+
+
+ Training + + Trains a new model on the targets wired here and on the examples and + tests, as a run of wren's training flow. The node keeps its model + until you use the new one and publish. + +
+ +
+ {problem ?

{problem}

: null} + {training ? ( +
+ + + {shortId(training.id)} + +
+ ) : null} + {training?.error ? ( +

{training.error}

+ ) : null} + {training && model ? ( + <> +
+ {[ + ["Exact match", percent(training.result?.exact_match)], + ["Coverage", percent(training.result?.coverage)], + ["Threshold", training.threshold?.toFixed(2) ?? "–"], + ["Test exact match", percent(training.test_exact_match)], + ].map(([label, value]) => ( +
+
{label}
+
{value}
+
+ ))} +
+ + + ) : null} +
+ ) +} + /** The value a fresh output returns, so the scaffold runs as written. */ const PLACEHOLDER: Record = { float: "0.0", @@ -2037,6 +2154,15 @@ function PanelBody({ {nodeType?.plugin && user?.is_superuser ? ( ) : null} + {node.type === "wren" ? ( + + editNode({ ...node, params: { ...node.params, model } }) + } + /> + ) : null} {hasSource ? (