From a286cf4cebd801ca33bbe7cae83c73bf75f09425 Mon Sep 17 00:00:00 2001 From: Melvin Strobl Date: Mon, 19 Oct 2026 23:23:25 +0200 Subject: [PATCH] Assistant node, and the Control tab as one conversation (openpanel V5) run_turn takes its tools, system prompt and tool caller; the chat passes the MCP tools as before. The assistant node offers one tool per wired target, reads its other inputs as context, streams its reply a sentence at a time, warms the provider on `listening` and says so when the provider is not there. Completions share the per-server HTTP client. wren replies when nothing reads `unhandled`, stt shows a note when nothing was heard. /ai/control lists one conversation per entry node with the messages to watch. --- backend/fluksio/ai/agent.py | 137 ++++---- backend/fluksio/ai/client.py | 46 +-- backend/fluksio/api/routes/ai.py | 96 ++++-- backend/fluksio/flow/controller.py | 7 + backend/fluksio/flow/nodes/__init__.py | 2 + backend/fluksio/flow/nodes/assistant.py | 350 +++++++++++++++++++++ backend/fluksio/flow/nodes/speech.py | 15 +- backend/fluksio/flow/nodes/wren.py | 14 +- backend/tests/flow/test_assistant.py | 133 ++++++++ backend/tests/flow/test_node_types.py | 5 + backend/tests/flow/test_wren.py | 16 +- docs/code/agents.md | 19 +- docs/interface/index.md | 5 +- docs/reference/node-types.md | 56 +++- frontend/src/client/schemas.gen.ts | 39 ++- frontend/src/client/sdk.gen.ts | 9 +- frontend/src/client/types.gen.ts | 15 +- frontend/src/components/Chat/ChatPanel.tsx | 36 ++- frontend/src/components/Chat/useControl.ts | 117 ++++--- frontend/src/components/Flow/FlowNode.tsx | 2 + frontend/src/components/Flow/liveStore.ts | 2 + 21 files changed, 937 insertions(+), 184 deletions(-) create mode 100644 backend/fluksio/flow/nodes/assistant.py create mode 100644 backend/tests/flow/test_assistant.py diff --git a/backend/fluksio/ai/agent.py b/backend/fluksio/ai/agent.py index 14ad091..83ba0f6 100644 --- a/backend/fluksio/ai/agent.py +++ b/backend/fluksio/ai/agent.py @@ -1,9 +1,11 @@ """The turn: the model, the tools, and the rounds between them. -The tools are the MCP server's own, called in-process rather than over the -endpoint: same names, same schemas, same REST calls underneath. A tool added -for external agents is there for the assistant the moment it exists, and the -assistant can do nothing an external agent could not. +The turn takes its tools, its system prompt and the function that calls a +tool from whoever runs it. The chat passes the MCP server's own, called +in-process rather than over the endpoint: same names, same schemas, same REST +calls underneath. A tool added for external agents is there for the chat the +moment it exists, and it can do nothing an external agent could not. The +assistant node passes the targets wired to it, and nothing else. The transcript is a list of chat-completion messages, mutated in place by :func:`run_turn` and persisted by the caller. It is also what the browser @@ -14,7 +16,7 @@ from __future__ import annotations import json import logging -from collections.abc import AsyncIterator +from collections.abc import AsyncIterator, Awaitable, Callable from typing import Any import httpx @@ -83,6 +85,23 @@ def _arguments(raw: Any) -> dict[str, Any]: return parsed if isinstance(parsed, dict) else {} +#: Runs one tool call and answers with what the model reads back. +type ToolCaller = Callable[[str, dict[str, Any]], Awaitable[str]] + + +def mcp_caller(authorization: str) -> ToolCaller: + """The MCP tools, each call made as the person chatting.""" + + async def call(name: str, arguments: dict[str, Any]) -> str: + token = caller_authorization.set(authorization) + try: + return await _call_tool(name, arguments) + finally: + caller_authorization.reset(token) + + return call + + async def _call_tool(name: str, arguments: dict[str, Any]) -> str: """Run one tool; whatever goes wrong is text the model can read.""" try: @@ -114,8 +133,11 @@ async def run_turn( messages: list[dict[str, Any]], text: str, *, - flow: str | None, - authorization: str, + tools: list[dict[str, Any]], + system: str, + call_tool: ToolCaller, + extra: dict[str, Any] | None = None, + timeout: float | httpx.Timeout = 120.0, ) -> AsyncIterator[dict[str, Any]]: """One user turn, yielding progress events as it goes. @@ -125,56 +147,53 @@ async def run_turn( part-way leaves a transcript the provider still accepts. """ messages.append({"role": "user", "content": text}) - tools = await tool_specs() - system = {"role": "system", "content": system_prompt(flow)} - token = caller_authorization.set(authorization) - try: - for _ in range(config.max_rounds): - choice: dict[str, Any] = {} - try: - async for part in client.stream_completion( - api_base=config.api_base, - api_key=config.api_key, - model=config.model, - messages=[system, *messages], - tools=tools, - ): - if isinstance(part, str): - yield {"type": "delta", "content": part} - else: - choice = part - except (client.ProviderError, httpx.HTTPError) as exc: - yield {"type": "error", "detail": f"The provider answered {exc}"} - return - content = choice.get("content") or "" - calls = choice.get("tool_calls") or [] - assistant: dict[str, Any] = {"role": "assistant", "content": content} - if content: - yield {"type": "message", "content": content} - if not calls: - messages.append(assistant) - return - assistant["tool_calls"] = calls - answers: list[dict[str, Any]] = [] - for call in calls: - function = call.get("function") or {} - name = str(function.get("name", "")) - arguments = _arguments(function.get("arguments")) - yield {"type": "tool_call", "name": name, "arguments": arguments} - result = await _call_tool(name, arguments) - answers.append( - { - "role": "tool", - "tool_call_id": call.get("id", ""), - "content": result, - } - ) - yield {"type": "tool_result", "name": name, **summarize(result)} + head = {"role": "system", "content": system} + for _ in range(config.max_rounds): + choice: dict[str, Any] = {} + try: + async for part in client.stream_completion( + api_base=config.api_base, + api_key=config.api_key, + model=config.model, + messages=[head, *messages], + tools=tools, + timeout=timeout, + extra=extra, + ): + if isinstance(part, str): + yield {"type": "delta", "content": part} + else: + choice = part + except (client.ProviderError, httpx.HTTPError) as exc: + yield {"type": "error", "detail": f"The provider answered {exc}"} + return + content = choice.get("content") or "" + calls = choice.get("tool_calls") or [] + assistant: dict[str, Any] = {"role": "assistant", "content": content} + if content: + yield {"type": "message", "content": content} + if not calls: messages.append(assistant) - messages.extend(answers) - yield { - "type": "error", - "detail": f"Stopped after {config.max_rounds} rounds of tool calls.", - } - finally: - caller_authorization.reset(token) + return + assistant["tool_calls"] = calls + answers: list[dict[str, Any]] = [] + for call in calls: + function = call.get("function") or {} + name = str(function.get("name", "")) + arguments = _arguments(function.get("arguments")) + yield {"type": "tool_call", "name": name, "arguments": arguments} + result = await call_tool(name, arguments) + answers.append( + { + "role": "tool", + "tool_call_id": call.get("id", ""), + "content": result, + } + ) + yield {"type": "tool_result", "name": name, **summarize(result)} + messages.append(assistant) + messages.extend(answers) + yield { + "type": "error", + "detail": f"Stopped after {config.max_rounds} rounds of tool calls.", + } diff --git a/backend/fluksio/ai/client.py b/backend/fluksio/ai/client.py index 58efbb3..544d328 100644 --- a/backend/fluksio/ai/client.py +++ b/backend/fluksio/ai/client.py @@ -118,26 +118,32 @@ async def stream_completion( model: str, messages: list[dict[str, Any]], tools: list[dict[str, Any]], - timeout: float = 120.0, + timeout: float | httpx.Timeout = 120.0, + extra: dict[str, Any] | None = None, ) -> AsyncIterator[str | dict[str, Any]]: """``POST {api_base}/chat/completions`` streamed. Yields each piece of the answer's text as it arrives, then the whole message. The timeout is per chunk, so a long answer is not cut off. + ``extra`` is added to the request as it is, for what one provider + understands and another does not, such as Ollama's ``keep_alive``. """ - body: dict[str, Any] = {"model": model, "messages": messages, "stream": True} + body: dict[str, Any] = { + **(extra or {}), + "model": model, + "messages": messages, + "stream": True, + } if tools: body["tools"] = tools assembler = Assembler() - async with ( - httpx.AsyncClient(timeout=timeout) as http, - http.stream( - "POST", - api_base.rstrip("/") + "/chat/completions", - json=body, - headers=_headers(api_key), - ) as response, - ): + async with _client(api_base).stream( + "POST", + api_base.rstrip("/") + "/chat/completions", + json=body, + headers=_headers(api_key), + timeout=timeout, + ) as response: if response.is_error: await response.aread() raise ProviderError(_detail(response)) @@ -160,24 +166,24 @@ async def list_models(api_base: str, api_key: str, timeout: float = 15.0) -> lis return sorted(str(m["id"]) for m in data if isinstance(m, dict) and "id" in m) -#: One client per speech server, so its connection stays open between calls +#: One client per server, so its connection stays open between calls #: (`parrot serve` keeps an idle one 600 s). A new connection over a lossy #: link can cost a second, which a spoken reply cannot spare. -_speech: dict[str, httpx.AsyncClient] = {} +_clients: dict[str, httpx.AsyncClient] = {} -def _speech_client(api_base: str) -> httpx.AsyncClient: - """The client for this speech server. Used from the engine's loop only.""" - http = _speech.get(api_base) +def _client(api_base: str) -> httpx.AsyncClient: + """The client for this server. Used from the engine's loop only.""" + http = _clients.get(api_base) if http is None: limits = httpx.Limits(keepalive_expiry=300) - http = _speech[api_base] = httpx.AsyncClient(limits=limits) + http = _clients[api_base] = httpx.AsyncClient(limits=limits) return http async def warm(api_base: str) -> None: """Open the connection to a speech server ahead of the call that needs it.""" - await _speech_client(api_base).get(api_base.rstrip("/") + "/models", timeout=10) + await _client(api_base).get(api_base.rstrip("/") + "/models", timeout=10) async def transcribe( @@ -190,7 +196,7 @@ async def transcribe( timeout: float = 30.0, ) -> str: """``POST {api_base}/audio/transcriptions``: a whole clip, and what it says.""" - response = await _speech_client(api_base).post( + response = await _client(api_base).post( api_base.rstrip("/") + "/audio/transcriptions", files={"file": (filename, audio, media_type)}, data={"model": model, "response_format": "json"}, @@ -214,7 +220,7 @@ async def speak( MP3 rather than WAV: a seventh of the bytes, which over a slow link to the server is most of the wait for a sentence. """ - response = await _speech_client(api_base).post( + response = await _client(api_base).post( api_base.rstrip("/") + "/audio/speech", json={"model": model, "input": text, "voice": voice, "response_format": "mp3"}, timeout=timeout, diff --git a/backend/fluksio/api/routes/ai.py b/backend/fluksio/api/routes/ai.py index 0cae07b..cc520ef 100644 --- a/backend/fluksio/api/routes/ai.py +++ b/backend/fluksio/api/routes/ai.py @@ -12,7 +12,7 @@ import json import logging import uuid from collections.abc import AsyncIterator -from typing import Any +from typing import Any, Literal import httpx from fastapi import APIRouter, Depends, HTTPException, Request @@ -29,7 +29,7 @@ from fluksio.api.deps import ( ) from fluksio.core.config import settings from fluksio.flow.messages import qualify -from fluksio.flow.schemas import NAME_PATTERN +from fluksio.flow.schemas import NAME_PATTERN, NodeDef from fluksio.mcp import server as tools logger = logging.getLogger(__name__) @@ -77,15 +77,28 @@ class ChatItem(BaseModel): error: str | None = None +type WatchKind = Literal["partial", "text", "reply", "actions"] + + +class ControlWatch(BaseModel): + """A message the conversation shows, as one kind of line.""" + + msg: str + kind: WatchKind + #: The node whose values on it count. Empty counts every value, as the + #: prompt does: a command typed here or spoken to a device. + node: str = "" + + class ControlNode(BaseModel): - """A wren node the chat can talk to: where it sends and where it reads.""" + """A conversation the chat can hold: where a command goes, and what to show.""" #: Qualified, ``flow.node``. id: str flow: str title: str prompt_msg: str - actions_msg: str + watch: list[ControlWatch] class TurnBody(BaseModel): @@ -173,33 +186,75 @@ def read_ai_status(_: CurrentUser) -> Any: return AIStatus(available=True, model=config.model) +#: The node types a command can be typed to. +_TALKERS = ("wren", "assistant") + + +def _wired(flow: str, node: NodeDef, port: str) -> str: + """The message a node's port is wired to, qualified, or empty.""" + for spec in [*node.requires, *node.provides]: + if spec.port == port and spec.name: + return qualify(flow, spec.name) + return "" + + @router.get("/control", response_model=list[ControlNode]) def read_control_nodes(_: CurrentUser, controller: FlowControllerDep) -> Any: - """The wren nodes of the running live flows, for the chat's Control tab. + """The conversations of the running live flows, for the chat's Control tab. - Only those with both ends wired: the chat publishes the command on the - prompt's message and reads the answer off the actions'. + One per wren or assistant node a command reaches first. A node whose + prompt is another's ``unhandled`` answers in that one's conversation, + after it. The conversation shows the prompt, the words heard so far (the + ``partial`` of a ``stt`` node writing the prompt), and each node's + ``reply`` and ``actions``, so typed and spoken commands are one. """ found = [] for flow in controller.store.read_all(): if flow.mode != "live" or flow.name in controller.disabled: continue - for node in flow.nodes: - if node.type != "wren": + talkers = [ + node + for node in flow.nodes + if node.type in _TALKERS and _wired(flow.name, node, "prompt") + ] + passed = {_wired(flow.name, node, "unhandled") for node in talkers} + for node in talkers: + prompt = _wired(flow.name, node, "prompt") + if prompt in passed: continue - ends = { - spec.port: qualify(flow.name, spec.name) - for spec in [*node.requires, *node.provides] - if spec.port in ("prompt", "actions") and spec.name - } - if len(ends) == 2: + watch = [ControlWatch(msg=prompt, kind="text")] + watch += [ + ControlWatch(msg=partial, kind="partial", node=f"{flow.name}.{stt.id}") + for stt in flow.nodes + if stt.type == "stt" + and _wired(flow.name, stt, "text") == prompt + and (partial := _wired(flow.name, stt, "partial")) + ] + chain = [node] + for talker in chain: + unhandled = _wired(flow.name, talker, "unhandled") + chain += [ + other + for other in talkers + if unhandled + and _wired(flow.name, other, "prompt") == unhandled + and other not in chain + ] + answers: tuple[WatchKind, ...] = ("reply", "actions") + watch += [ + ControlWatch(msg=msg, kind=kind, node=f"{flow.name}.{talker.id}") + for talker in chain + for kind in answers + if (msg := _wired(flow.name, talker, kind)) + ] + if any(w.kind in answers for w in watch): found.append( ControlNode( id=f"{flow.name}.{node.id}", flow=flow.name, - title=node.title or node.id, - prompt_msg=ends["prompt"], - actions_msg=ends["actions"], + title=f"{flow.title or flow.name} · {node.title or node.id}", + prompt_msg=prompt, + watch=watch, ) ) return found @@ -280,8 +335,9 @@ def chat( config, messages, body.message, - flow=body.flow, - authorization=f"Bearer {token}", + tools=await agent.tool_specs(), + system=agent.system_prompt(body.flow), + call_tool=agent.mcp_caller(f"Bearer {token}"), ): yield _sse(event) except Exception as exc: # noqa: BLE001 - the stream is the error channel diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index 7115779..f921b7f 100644 --- a/backend/fluksio/flow/controller.py +++ b/backend/fluksio/flow/controller.py @@ -41,6 +41,7 @@ from fluksio.flow.executor import ExecutionService from fluksio.flow.messages import MessageSpec, flow_of, qualify from fluksio.flow.nodes import ( RESERVED_SETTINGS, + AssistantNode, ChangeNode, DebounceNode, DelayNode, @@ -385,6 +386,12 @@ NODE_TYPES: dict[str, NodeType] = { cls=WrenNode, params_schema=_schema_of(WrenNode), ), + "assistant": NodeType( + title="Assistant", + description="Answer a request with the AI provider, acting on what is wired to it.", + cls=AssistantNode, + params_schema=_schema_of(AssistantNode), + ), "stt": NodeType( title="Speech to text", description="Turn a microphone's audio into text, through a speech server.", diff --git a/backend/fluksio/flow/nodes/__init__.py b/backend/fluksio/flow/nodes/__init__.py index 02cffb5..a087629 100644 --- a/backend/fluksio/flow/nodes/__init__.py +++ b/backend/fluksio/flow/nodes/__init__.py @@ -4,6 +4,7 @@ Split by the outside world each one talks to. Importing from ``fluksio.flow.nodes`` keeps working, which is what every caller does. """ +from fluksio.flow.nodes.assistant import AssistantNode from fluksio.flow.nodes.base import RESERVED_SETTINGS, Node from fluksio.flow.nodes.debounce import DebounceNode from fluksio.flow.nodes.delay import DelayNode @@ -23,6 +24,7 @@ from fluksio.flow.nodes.websocket import WebSocketNode from fluksio.flow.nodes.wren import WrenNode __all__ = [ + "AssistantNode", "ChangeNode", "DebounceNode", "DelayNode", diff --git a/backend/fluksio/flow/nodes/assistant.py b/backend/fluksio/flow/nodes/assistant.py new file mode 100644 index 0000000..97bfe93 --- /dev/null +++ b/backend/fluksio/flow/nodes/assistant.py @@ -0,0 +1,350 @@ +"""Assistant: a request answered by the instance's language model. + +The request arrives on ``prompt``, typically what a wren node could not do. +Every output but ``reply`` and ``actions`` is a target the model may set, +offered as one tool named after its port and typed by its message, so the +wiring is the allowlist, as for wren: the MCP authoring tools are never +offered. Every other input is read as context ("outside_temperature: 7.5"). + +The answer goes out on ``reply`` a sentence at a time while the model writes +it, so a ``tts`` node speaks the first before the last is written. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import dataclasses +import json +import logging +import re +import threading +from collections.abc import Coroutine, Iterable +from concurrent.futures import Future +from typing import TYPE_CHECKING, Any + +import httpx +from pydantic import BaseModel, ConfigDict, Field + +from fluksio.ai import agent, client +from fluksio.ai import config as ai_config +from fluksio.flow.messages import DType, MessageSpec +from fluksio.flow.nodes.base import Node + +if TYPE_CHECKING: + from fastapi import FastAPI + +logger = logging.getLogger(__name__) + +REPLY = "reply" +ACTIONS = "actions" +#: An optional input: true warms the provider while the request is spoken. +LISTENING = "listening" + +#: Said when there is no provider, or it does not answer. +UNAVAILABLE = "The assistant isn't available right now." +#: Said when the model ends a turn without a word. +DONE, CANT = "Done.", "Sorry, I can't do that." + +#: A server that is not there is given up on after a few seconds, while a +#: model that has to be loaded first takes about ten. +TIMEOUT = httpx.Timeout(15.0, connect=3.0) +#: Rounds of tool calls one request may take. +ROUNDS = 3 + +#: A target's type, as its tool's parameter. +_SCHEMAS: dict[DType, dict[str, str]] = { + DType.BOOL: {"type": "boolean"}, + DType.INT: {"type": "integer"}, + DType.FLOAT: {"type": "number"}, + DType.STR: {"type": "string"}, +} + +_SYSTEM = """You are the voice of a home: someone spoke to a panel on the wall, \ +and you answer out loud. + +- To change something, call its tool. The tools are all you can change. +- After a change, say what you did in one short sentence. +- If no tool can do what was asked, or you did not understand it, say so in a \ +few words, such as "Sorry, I can't do that." +- Answer questions from the readings below. +- Speak plainly: one or two short sentences, no lists, no markdown, no emoji. +""" + +_END = re.compile(r"(?<=[.!?])\s+") + + +def cut(text: str) -> tuple[list[str], str]: + """The sentences ``text`` has finished, and the one still being written.""" + *done, rest = _END.split(text) + return [s.strip() for s in done if s.strip()], rest + + +class AssistantNode(Node): + """Answer the request on ``prompt``, acting only on the targets wired to it. + + Inputs: ``prompt`` (the request, which wakes the node), ``listening`` + (optional; turning true opens the connection and loads the model while + the request is still spoken) and any other, read as context. + + Outputs: ``reply``, the answer a sentence at a time, ``actions``, a list + of records, one per value set, and any number of targets, each a tool. + When the provider is not set up or does not answer, the reply says so. + """ + + class Params(BaseModel): + model_config = ConfigDict(extra="allow") + + model: str = Field( + default="", description="The model to ask. Empty asks the instance's." + ) + extra_body: dict[str, Any] = Field( + default_factory=lambda: { + "temperature": 0, + "reasoning_effort": "none", + "keep_alive": "30m", + }, + description=( + "Added to every request as it is. The default answers a request " + "the same way each time, turns thinking off and keeps the model " + "loaded for 30 minutes on Ollama; drop what a provider refuses." + ), + ) + + __slots__ = ("cfg", "_listening", "_lock", "_loop") + + # A tool call is a command to a device, as wren's are. + idempotent = False + + def __init__( + self, + requires: MessageSpec | Iterable[MessageSpec] = (), + provides: MessageSpec | Iterable[MessageSpec] = (), + params: dict[str, Any] | None = None, + name: str | None = None, + ): + self.cfg = self.Params.model_validate(params or {}) + self._lock = threading.Lock() + #: What ``listening`` held at the last run, to tell its change apart. + self._listening: bool | None = None + self._loop: asyncio.AbstractEventLoop | None = None + inputs = self._normalize_ports(requires) + outputs = self._normalize_ports(provides) + typed = {"prompt": DType.STR, LISTENING: DType.BOOL} + for spec in inputs: + if spec.port in typed and spec.dtype is not typed[spec.port]: + raise ValueError(f"'{spec.port}' is a {typed[spec.port].value}.") + if not any(spec.port == "prompt" for spec in inputs): + raise ValueError("assistant needs a 'prompt' input: the request.") + for spec in outputs: + if spec.port == REPLY and spec.dtype is not DType.STR: + raise ValueError(f"'{REPLY}' is a str.") + if spec.port == ACTIONS and ( + spec.dtype is not DType.LIST or spec.item_dtype is not DType.RECORD + ): + raise ValueError(f"'{ACTIONS}' is a list of records.") + if spec.port not in (REPLY, ACTIONS) and spec.dtype not in _SCHEMAS: + raise ValueError( + f"'{spec.port}' is {spec.dtype.value}; a target is bool, int, " + "float or str." + ) + super().__init__( + self._answer, + requires=[ + spec.model_copy(update={"trigger": spec.port in ("prompt", LISTENING)}) + for spec in inputs + ], + provides=outputs, + params=params, + name=name or "assistant", + ) + + async def start(self, app: FastAPI | None = None) -> None: + self._loop = asyncio.get_running_loop() + names = [spec.name for spec in self.input_ports if spec.port == LISTENING] + if names and self._pipeline is not None: + held = await asyncio.to_thread(self._pipeline.state.get_present, names) + self._listening = held.get(names[0]) + + def _targets(self) -> list[MessageSpec]: + return [ + spec + for spec in self.output_ports + if spec.name and spec.port not in (REPLY, ACTIONS) + ] + + def _tools(self) -> list[dict[str, Any]]: + return [ + { + "type": "function", + "function": { + "name": spec.port, + "description": f"Set the {spec.port.replace('_', ' ')}.", + "parameters": { + "type": "object", + "properties": {"value": _SCHEMAS[spec.dtype]}, + "required": ["value"], + }, + }, + } + for spec in self._targets() + ] + + def _system(self, context: dict[str, Any]) -> str: + """The instructions, then what the inputs read and the targets hold.""" + targets = self._targets() + held = ( + self._pipeline.state.get_present([spec.name for spec in targets]) + if self._pipeline is not None + else {} + ) + lines = [_SYSTEM, "Readings:"] + lines += [f"{port}: {json.dumps(value)}" for port, value in context.items()] + lines += ["", "What the tools set, now:"] + lines += [ + f"{spec.port}: {json.dumps(held[spec.name])}" + for spec in targets + if spec.name in held + ] + return "\n".join(lines) + + def _config(self) -> ai_config.AIConfig | None: + config = ai_config.load() + if config is None or not config.ready: + return None + return dataclasses.replace( + config, model=self.cfg.model or config.model, max_rounds=ROUNDS + ) + + def _call(self, work: Coroutine[Any, Any, Any]) -> Future[Any]: + """Run on the engine's loop, where the provider's connection is kept.""" + if self._loop is None: + work.close() + raise RuntimeError(f"{self.name} only runs in a live flow") + return asyncio.run_coroutine_threadsafe(work, self._loop) + + def _answer( + self, + params: dict[str, Any], + prompt: str = "", + listening: bool | None = None, + **context: Any, + ) -> dict[str, Any] | None: + with self._lock: + was, self._listening = self._listening, listening + config = self._config() + if listening != was: + # Not a request: the device started or stopped listening. + if listening and config is not None: + self._call(self._warm(config)) + return None + if not prompt.strip(): + return None + actions: list[dict[str, Any]] = [] + problem = self._call( + self._turn(config, prompt, self._system(context), actions) + ).result() + if problem: + self.report_health("degraded", problem) + elif self._health != "ok": + self.report_health("ok") + wired = {spec.port for spec in self.output_ports if spec.name} + return {ACTIONS: actions} if ACTIONS in wired else None + + async def _warm(self, config: ai_config.AIConfig) -> None: + """A request for one token: the connection opens and the model loads. + + Nothing more is worth sending ahead: a model with a sliding window, + such as Gemma, reads the whole prompt again on every request anyway. + """ + with contextlib.suppress(Exception): + async for _ in client.stream_completion( + api_base=config.api_base, + api_key=config.api_key, + model=config.model, + messages=[{"role": "user", "content": "Hello."}], + tools=[], + timeout=60.0, + extra={**self.cfg.extra_body, "max_tokens": 1}, + ): + pass + + async def _turn( + self, + config: ai_config.AIConfig | None, + prompt: str, + system: str, + actions: list[dict[str, Any]], + ) -> str: + """Ask, act and say. Returns what went wrong, or an empty string.""" + said = False + + async def say(text: str) -> None: + nonlocal said + said = True + await asyncio.to_thread(self._say, text) + + if config is None: + await say(UNAVAILABLE) + return "no AI provider is set up" + rest, problem = "", "" + try: + async for event in agent.run_turn( + config, + [], + prompt, + tools=self._tools(), + system=system, + call_tool=self._caller(actions), + extra=self.cfg.extra_body, + timeout=TIMEOUT, + ): + if event["type"] == "delta": + done, rest = cut(rest + event["content"]) + for sentence in done: + await say(sentence) + elif event["type"] == "message" and rest.strip(): + await say(rest.strip()) + rest = "" + elif event["type"] == "error": + problem = str(event["detail"]) + except Exception as exc: # noqa: BLE001 - said, then reported as the health + problem = f"{type(exc).__name__}: {exc}" + if problem: + logger.warning("'%s' got no answer: %s", self.id, problem) + if not said: + await say(UNAVAILABLE if problem else DONE if actions else CANT) + return problem + + def _caller(self, actions: list[dict[str, Any]]) -> agent.ToolCaller: + """Set the target a call names, and record it in ``actions``.""" + specs = {spec.port: spec for spec in self._targets()} + + async def call(name: str, arguments: dict[str, Any]) -> str: + spec = specs.get(name) + if spec is None: + return json.dumps({"error": f"There is no tool named {name}."}) + value = arguments.get("value") + if spec.dtype is DType.FLOAT and type(value) is int: + value = float(value) + try: + spec.check(value) + except TypeError as exc: + return json.dumps({"error": str(exc)}) + await asyncio.to_thread(self.emit, {name: value}) + actions.append( + { + "message": spec.name, + "value": value, + "applied": True, + "delay_s": None, + "call": f"{name}({json.dumps(value)})", + } + ) + return json.dumps({name: value}) + + return call + + def _say(self, text: str) -> None: + if any(spec.port == REPLY and spec.name for spec in self.output_ports): + self.emit({REPLY: text}) diff --git a/backend/fluksio/flow/nodes/speech.py b/backend/fluksio/flow/nodes/speech.py index 4b76f04..f061552 100644 --- a/backend/fluksio/flow/nodes/speech.py +++ b/backend/fluksio/flow/nodes/speech.py @@ -41,6 +41,9 @@ SPOKEN = 256 #: How long after its estimated end a sentence still counts as playing: the #: device fetches it before it starts. PLAYING_SLACK_S = 1.0 +#: The partial for an utterance that came back empty, often a wake word that +#: was not one, or was lost: shown on the caption, never spoken. +NOTHING_HEARD = "Didn't catch that." def _check(specs: list[MessageSpec], ports: dict[str, DType], node: str) -> None: @@ -152,8 +155,9 @@ class SttNode(_Speech): One realtime session per utterance, opened when ``listening`` turns true or at the first chunk, and committed when it turns false. A clip goes to the server whole. Either way the node returns at once, since a device's - next frame waits for it, and publishes once the server answers. An - utterance that fails publishes nothing and is logged. + next frame waits for it, and publishes once the server answers. Nothing + heard, or an utterance lost and logged, is a short note on ``partial`` + alone. """ __slots__ = ("_heard", "_listening", "_lock", "_session", "_tasks") @@ -268,12 +272,15 @@ class SttNode(_Speech): words = text.strip() if final else words + text if final and words: await asyncio.to_thread(self._say, partial=words, text=words) + elif final: + await asyncio.to_thread(self._say, partial=NOTHING_HEARD) elif words: await asyncio.to_thread(self._say, partial=words) except Exception as exc: logger.warning( "'%s' lost an utterance: %s: %s", self.id, type(exc).__name__, exc ) + await asyncio.to_thread(self._say, partial=NOTHING_HEARD) finally: # Ended without a commit, by a timeout or a failure: the next chunk # opens a session of its own rather than feeding this dead one. @@ -288,9 +295,11 @@ class SttNode(_Speech): ) except Exception as exc: logger.warning("'%s' lost a clip: %s: %s", self.id, type(exc).__name__, exc) - return + text = "" if text.strip(): await asyncio.to_thread(self._say, partial=text.strip(), text=text.strip()) + else: + await asyncio.to_thread(self._say, partial=NOTHING_HEARD) class TtsNode(_Speech): diff --git a/backend/fluksio/flow/nodes/wren.py b/backend/fluksio/flow/nodes/wren.py index 80e1d71..7b4a60a 100644 --- a/backend/fluksio/flow/nodes/wren.py +++ b/backend/fluksio/flow/nodes/wren.py @@ -71,6 +71,8 @@ INPUTS = { ACTIONS = "actions" REPLY = "reply" UNHANDLED = "unhandled" +#: The reply when nothing was done and nothing reads ``unhandled``. +CANT = "Sorry, I can't do that yet." _TARGET_TYPES = {DType.BOOL, DType.INT, DType.FLOAT, DType.STR} @@ -198,8 +200,9 @@ class WrenNode(Node): (otherwise read off the clock: before 07:00 or from 18:00). Outputs: ``actions``, a list of records, one per call and target, - ``reply``, a sentence saying what was applied ("Kitchen light on."), - ``unhandled``, the prompt when nothing was, and any number of targets + ``reply``, a sentence saying what was applied ("Kitchen light on."), or + that nothing was when no node reads ``unhandled``, ``unhandled``, the + prompt when nothing was, and any number of targets named ``_``. Targets of one area and kind are one device to the model; each takes what the call means for its type. A delayed call is held by the engine's work queue, like the Trigger @@ -401,4 +404,11 @@ class WrenNode(Node): writes[REPLY] = reply[0].upper() + reply[1:] + "." if not said and UNHANDLED in wired: writes[UNHANDLED] = prompt + if not said and REPLY in wired and not self._passed_on(): + writes[REPLY] = CANT return writes or None + + def _passed_on(self) -> bool: + """Whether a node reads ``unhandled``, such as an assistant.""" + name = next((s.name for s in self.output_ports if s.port == UNHANDLED), "") + return bool(name and self._pipeline and self._pipeline.consumes.get(name)) diff --git a/backend/tests/flow/test_assistant.py b/backend/tests/flow/test_assistant.py new file mode 100644 index 0000000..ba183df --- /dev/null +++ b/backend/tests/flow/test_assistant.py @@ -0,0 +1,133 @@ +"""The assistant node: the wiring is its allowlist, and it speaks as it writes.""" + +import asyncio +import json +from collections.abc import AsyncIterator +from typing import Any + +import httpx +import pytest + +from fluksio.ai import client +from fluksio.ai import config as ai_config +from fluksio.flow.messages import DType, MessageSpec +from fluksio.flow.nodes.assistant import UNAVAILABLE, AssistantNode, cut +from fluksio.flow.pipeline import Pipeline + +PROVIDER = ai_config.AIConfig(api_base="http://llm:11434/v1", api_key="k", model="m") + + +def _call(name: str, value: Any) -> dict[str, Any]: + arguments = json.dumps({"value": value}) + return {"id": name, "function": {"name": name, "arguments": arguments}} + + +def _node() -> AssistantNode: + node = AssistantNode( + requires=[ + MessageSpec(name="unhandled", port="prompt", dtype=DType.STR), + MessageSpec(name="outdoor", port="outside_temperature", dtype=DType.FLOAT), + ], + provides=[ + MessageSpec(name="kitchen_light", dtype=DType.BOOL), + MessageSpec( + name="setpoint", port="living_room_setpoint", dtype=DType.FLOAT + ), + MessageSpec(name="reply", dtype=DType.STR), + MessageSpec(name="actions", dtype=DType.LIST), + ], + ) + node.assign_flow("voice", "assistant") + Pipeline([node]) + return node + + +def _ask(node: AssistantNode, prompt: str) -> dict[str, Any] | None: + async def scenario() -> dict[str, Any] | None: + await node.start() + inputs = {"voice.unhandled": prompt, "voice.outdoor": 7.5} + return await asyncio.to_thread(node.execute, inputs) + + return asyncio.run(scenario()) + + +def test_a_sentence_is_cut_once_it_ends(): + assert cut("Done. It is 7.5 degr") == (["Done."], "It is 7.5 degr") + assert cut("Lights off! Anything else? ") == (["Lights off!", "Anything else?"], "") + + +def test_only_wired_targets_are_tools_and_each_sentence_goes_out_as_written( + monkeypatch: pytest.MonkeyPatch, +): + log: list[Any] = [] + seen: list[dict[str, Any]] = [] + rounds = iter( + [ + # The model reaches for a tool nothing is wired to, as well. + [ + _call("kitchen_light", False), + _call("delete_flow", "demo"), + _call("living_room_setpoint", 22), + ], + ["Kitchen light off", " and heating to 22. It is ", "7.5 degrees outside."], + ] + ) + + async def completion(**kwargs: Any) -> AsyncIterator[str | dict[str, Any]]: + seen.append(kwargs) + step = next(rounds) + if isinstance(step[0], dict): + yield {"role": "assistant", "content": None, "tool_calls": step} + return + for piece in step: + log.append(piece) + yield piece + yield {"role": "assistant", "content": "".join(step)} + + monkeypatch.setattr(ai_config, "load", lambda: PROVIDER) + monkeypatch.setattr(client, "stream_completion", completion) + monkeypatch.setattr(AssistantNode, "emit", lambda self, values: log.append(values)) + + out = _ask(_node(), "turn off the kitchen light and make it warmer") + + tools = {tool["function"]["name"] for tool in seen[0]["tools"]} + assert tools == {"kitchen_light", "living_room_setpoint"} + system = seen[0]["messages"][0]["content"] + assert "outside_temperature: 7.5" in system + assert seen[0]["extra"]["reasoning_effort"] == "none" + answers = [m["content"] for m in seen[1]["messages"] if m["role"] == "tool"] + assert "error" in json.loads(answers[1]) + # Set as called, and the first sentence said before the second is written. + assert log == [ + {"kitchen_light": False}, + {"living_room_setpoint": 22.0}, + "Kitchen light off", + " and heating to 22. It is ", + {"reply": "Kitchen light off and heating to 22."}, + "7.5 degrees outside.", + {"reply": "It is 7.5 degrees outside."}, + ] + assert out is not None + assert [a["message"] for a in out["voice.actions"]] == [ + "voice.kitchen_light", + "voice.setpoint", + ] + + +def test_no_provider_or_one_that_does_not_answer_is_said( + monkeypatch: pytest.MonkeyPatch, +): + said: list[dict[str, Any]] = [] + monkeypatch.setattr(AssistantNode, "emit", lambda self, values: said.append(values)) + + async def refused(**kwargs: Any) -> AsyncIterator[str | dict[str, Any]]: + raise httpx.ConnectError("refused") + yield "" + + monkeypatch.setattr(client, "stream_completion", refused) + for provider in (None, PROVIDER): + monkeypatch.setattr(ai_config, "load", lambda provider=provider: provider) + node = _node() + _ask(node, "what is the weather tomorrow") + assert said.pop() == {"reply": UNAVAILABLE} + assert node._health == "degraded" diff --git a/backend/tests/flow/test_node_types.py b/backend/tests/flow/test_node_types.py index d6be659..4d12752 100644 --- a/backend/tests/flow/test_node_types.py +++ b/backend/tests/flow/test_node_types.py @@ -105,6 +105,11 @@ FIXTURES: dict[str, dict] = { "requires": [MessageSpec(name="prompt", dtype=DType.STR)], "provides": [MessageSpec(name="kitchen_light", dtype=DType.BOOL)], }, + "assistant": { + "params": {"model": "gemma4:12b"}, + "requires": [MessageSpec(name="unhandled", port="prompt", dtype=DType.STR)], + "provides": [MessageSpec(name="kitchen_light", dtype=DType.BOOL)], + }, "stt": { "params": {}, "requires": [ diff --git a/backend/tests/flow/test_wren.py b/backend/tests/flow/test_wren.py index 18a8b03..87a98e0 100644 --- a/backend/tests/flow/test_wren.py +++ b/backend/tests/flow/test_wren.py @@ -10,7 +10,8 @@ from typing import Any import pytest from fluksio.flow.messages import DType, MessageSpec -from fluksio.flow.nodes.wren import WrenNode, _value +from fluksio.flow.nodes.base import Node +from fluksio.flow.nodes.wren import CANT, WrenNode, _value from fluksio.flow.pipeline import Pipeline @@ -184,8 +185,15 @@ def test_a_reply_says_what_was_applied_and_the_rest_is_unhandled(fake_wren): assert out == {"house.kitchen_light": True, "house.reply": "Kitchen light on."} - # Nothing wired for the call, or nothing to call at all. + # Nothing wired for the call, or nothing to call at all. Nothing reads + # what is left, so the node says so itself. + asked = "what is the weather tomorrow" for calls in ([FakeCall("lights.on", {"id": "garage:light"})], []): fake_wren.answer = (calls, 0.99) - out = node.execute({"house.command": "what is the weather tomorrow"}) - assert out == {"house.unhandled": "what is the weather tomorrow"} + out = node.execute({"house.command": asked}) + assert out == {"house.unhandled": asked, "house.reply": CANT} + + # An assistant reading it answers instead. + unhandled = MessageSpec(name="house.unhandled", port="unhandled", dtype=DType.STR) + Pipeline([node, Node(lambda params, unhandled: None, requires=[unhandled])]) + assert node.execute({"house.command": asked}) == {"house.unhandled": asked} diff --git a/docs/code/agents.md b/docs/code/agents.md index 9686821..7386563 100644 --- a/docs/code/agents.md +++ b/docs/code/agents.md @@ -67,11 +67,16 @@ works without it. ## Control -A live [Wren node](../reference/node-types.md#wren) with both `prompt` and -`actions` wired adds a **Control** tab to the chat. It uses the node's local -model and needs no assistant provider. +A live [Wren](../reference/node-types.md#wren) or +[Assistant](../reference/node-types.md#assistant) node with its `prompt` and a +`reply` or `actions` wired adds a **Control** tab to the chat. Wren uses its +local model, the Assistant node the provider above. -Choose a node, type a command, and inspect the reported actions: target -message, value, confidence, and whether each action was applied. -The node's `apply` and `min_confidence` settings decide whether it writes -to devices. Control conversations stay in the browser and clear on reload. +Choose a conversation and type a command. It goes out on the node's prompt, +which is where a `stt` node writes what a device heard, so typed and spoken +commands are one conversation: the words heard so far, the command, each +reply, and the reported actions with their target message, value and whether +each was applied. A node whose prompt is another's `unhandled` answers in that +node's conversation. Wren's `apply` and `min_confidence` settings decide +whether it writes to devices. Control conversations stay in the browser and +clear on reload. diff --git a/docs/interface/index.md b/docs/interface/index.md index 7482f83..1ba6ee5 100644 --- a/docs/interface/index.md +++ b/docs/interface/index.md @@ -45,8 +45,9 @@ broker topic, URL or other external target share a neuron; computation nodes appear individually. Messages connect them, and activity makes them pulse. A problem keeps the affected neuron's label visible. A -[`wren` node](../reference/node-types.md#wren) also adds a chat button for its -[Control tab](../code/agents.md#control). +[`wren`](../reference/node-types.md#wren) or +[`assistant`](../reference/node-types.md#assistant) node also adds a chat +button for its [Control tab](../code/agents.md#control). ### Flows and dashboards diff --git a/docs/reference/node-types.md b/docs/reference/node-types.md index 612a58a..ee7af2c 100644 --- a/docs/reference/node-types.md +++ b/docs/reference/node-types.md @@ -398,7 +398,9 @@ applied. Keep `apply=false` while checking how the model interprets commands. `reply` is one sentence saying what was applied, such as "Kitchen light on." or "Garage light on in 5 minutes.". When nothing was applied, `unhandled` -carries the prompt instead, for something else to answer. +carries the prompt instead, for something else to answer, such as an +[Assistant](#assistant). When no node reads `unhandled`, `reply` says "Sorry, +I can't do that yet." instead, so a spoken command never goes unanswered. A call with a delay ("in 10m") is held by the engine's work queue, like the Debounce node's second value: it survives a restart where the queue is in @@ -415,6 +417,53 @@ They are reported but cannot receive writes unless wired. The chat's [Control tab](../code/agents.md#control) sends commands by hand. +### Assistant + +**`assistant`** — answer a request with the instance's AI provider, acting only +on what is wired to it. + +| Input | Type | Notes | +|---|---|---| +| `prompt` | `str` | the request, such as a Wren node's `unhandled`; it runs the node | +| `listening` | `bool` | optional; turning true opens the connection and loads the model while the request is spoken | +| any other | any | read as context when a request arrives, as `outside_temperature: 12.5` | + +| Output | Type | Notes | +|---|---|---| +| `reply` | `str` | the answer, one sentence at a time while the model writes it | +| `actions` | `list` | one record per value set: `message`, `value`, `applied`, `delay_s`, `call` | +| any other | `bool`, `int`, `float` or `str` | a **target**: one tool, named after its port, taking a `value` of its type | + +| Setting | Default | Notes | +|---|---|---| +| `model` | empty | the model to ask; empty asks the instance's | +| `extra_body` | `{"temperature": 0, "reasoning_effort": "none", "keep_alive": "30m"}` | added to every request as it is | + +It asks the provider set under **Settings → Assistant**. As for Wren, the +wiring is the allowlist: the model is offered the targets and nothing else, +never the tools the chat's assistant uses. Its instructions list the context +inputs and what each target holds now. A target is set the moment the model +calls its tool, and a request may take three rounds of tool calls. + +Every request ends in a reply: + +- When no tool fits or the request makes no sense, the model says so, as + "Sorry, I can't do that.". +- When no provider is set up, the connection is refused, it takes over 3 + seconds to connect or 15 seconds to answer, the reply is "The assistant + isn't available right now." and the node reads degraded. A model that has to + be loaded first can take about 10 seconds. + +The default `extra_body` is for Ollama. `temperature` makes the same request +do the same thing each time. `reasoning_effort` turns thinking off through +Ollama's OpenAI-compatible API, which ignores Ollama's own `think`. +`keep_alive` keeps the model loaded for 30 minutes after the last request +instead of 5. Drop a field the provider refuses. + +In a voice flow, `stt` writes the prompt of a Wren node, Wren's `unhandled` is +the assistant's prompt, both `reply` outputs go to `tts`, and the device's +`listening` warms the assistant. + ### Speech to text **`stt`** — turn a microphone's audio into text, through a speech server. @@ -435,8 +484,9 @@ utterance is one realtime transcription session (`{url}/v1/realtime`), opened when `listening` turns true or at the first chunk and committed when it turns false. A clip is sent whole to `{url}/v1/audio/transcriptions`. The node returns at once and publishes when the server answers, since a device's next -chunk waits for it. An utterance that fails, or gets no chunk for 10 seconds, -publishes nothing and is logged. +chunk waits for it. An utterance that comes back empty, fails, or gets no +chunk for 10 seconds publishes "Didn't catch that." on `partial` alone, so a +caption shows it and nothing speaks it. A lost one is logged. `listening` has to hold a value before a chunk wakes the node. Declare it as a flow input with the initial value `false` when a device may never send it, diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts index e6a96e0..c954d93 100644 --- a/frontend/src/client/schemas.gen.ts +++ b/frontend/src/client/schemas.gen.ts @@ -606,15 +606,39 @@ export const ControlNodeSchema = { type: 'string', title: 'Prompt Msg' }, - actions_msg: { - type: 'string', - title: 'Actions Msg' + watch: { + items: { + '$ref': '#/components/schemas/ControlWatch' + }, + type: 'array', + title: 'Watch' } }, type: 'object', - required: ['id', 'flow', 'title', 'prompt_msg', 'actions_msg'], + required: ['id', 'flow', 'title', 'prompt_msg', 'watch'], title: 'ControlNode', - description: 'A wren node the chat can talk to: where it sends and where it reads.' + description: 'A conversation the chat can hold: where a command goes, and what to show.' +} as const; + +export const ControlWatchSchema = { + properties: { + msg: { + type: 'string', + title: 'Msg' + }, + kind: { + '$ref': '#/components/schemas/WatchKind' + }, + node: { + type: 'string', + title: 'Node', + default: '' + } + }, + type: 'object', + required: ['msg', 'kind'], + title: 'ControlWatch', + description: 'A message the conversation shows, as one kind of line.' } as const; export const DTypeSchema = { @@ -4937,6 +4961,11 @@ export const WaitingNodeSchema = { title: 'WaitingNode' } as const; +export const WatchKindSchema = { + type: 'string', + enum: ['partial', 'text', 'reply', 'actions'] +} as const; + export const WebPushKeySchema = { properties: { key: { diff --git a/frontend/src/client/sdk.gen.ts b/frontend/src/client/sdk.gen.ts index 23b8061..788d34b 100644 --- a/frontend/src/client/sdk.gen.ts +++ b/frontend/src/client/sdk.gen.ts @@ -84,10 +84,13 @@ export class AiService { /** * Read Control Nodes - * The wren nodes of the running live flows, for the chat's Control tab. + * The conversations of the running live flows, for the chat's Control tab. * - * Only those with both ends wired: the chat publishes the command on the - * prompt's message and reads the answer off the actions'. + * One per wren or assistant node a command reaches first. A node whose + * prompt is another's ``unhandled`` answers in that one's conversation, + * after it. The conversation shows the prompt, the words heard so far (the + * ``partial`` of a ``stt`` node writing the prompt), and each node's + * ``reply`` and ``actions``, so typed and spoken commands are one. * @returns ControlNode Successful Response * @throws ApiError */ diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index fa2fc59..ab2a7bd 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -149,14 +149,23 @@ export type ConnectorStatePublic = { }; /** - * A wren node the chat can talk to: where it sends and where it reads. + * A conversation the chat can hold: where a command goes, and what to show. */ export type ControlNode = { id: string; flow: string; title: string; prompt_msg: string; - actions_msg: string; + watch: Array; +}; + +/** + * A message the conversation shows, as one kind of line. + */ +export type ControlWatch = { + msg: string; + kind: WatchKind; + node?: string; }; /** @@ -1560,6 +1569,8 @@ export type WaitingNode = { seconds: number; }; +export type WatchKind = 'partial' | 'text' | 'reply' | 'actions'; + export type WebPushKey = { key: string; }; diff --git a/frontend/src/components/Chat/ChatPanel.tsx b/frontend/src/components/Chat/ChatPanel.tsx index dc46e40..60924a5 100644 --- a/frontend/src/components/Chat/ChatPanel.tsx +++ b/frontend/src/components/Chat/ChatPanel.tsx @@ -34,9 +34,9 @@ export const aiStatusQueryOptions = () => }) /** - * The wren nodes the Control tab can talk to. Under the flows prefix: the - * socket invalidates it on every rebuild, so a node published anywhere shows - * up without a reload. + * The conversations the Control tab can hold, one per wren or assistant node a + * command reaches first. Under the flows prefix: the socket invalidates it on + * every rebuild, so a node published anywhere shows up without a reload. */ export const controlNodesQueryOptions = () => queryOptions({ @@ -45,8 +45,8 @@ export const controlNodesQueryOptions = () => }) /** - * What the panel is for: building flows with the assistant, or commanding a - * wren node in one. A tab is there when there is something behind it, and the + * What the panel is for: building flows with the assistant, or commanding the + * wren and assistant nodes in one. A tab is there when there is something behind it, and the * strip only draws once there is a choice to make. */ const TABS: readonly (readonly [ChatTab, string])[] = [ @@ -103,7 +103,7 @@ function fromTranscript(transcript: ChatItem[]): Item[] { } /** - * The way into the Control tab from Home, there only while a wren node is. + * The way into the Control tab from Home, there only while a conversation is. * Shaped like the sidebar trigger, which it mirrors on a phone. */ export function ControlTrigger({ className }: { className?: string }) { @@ -371,7 +371,7 @@ export function ChatPanel() { {shown.length === 0 && !waiting ? (

{controlling - ? "Type a command, like turn on the lights in the kitchen. What it did comes back here." + ? "Type a command, like turn on the kitchen light. What was said to the flow and what it did show up here." : "Describe a flow to build, a change to make, or ask what a failure means. Drafts land in the editor for you to publish."}

) : null} @@ -407,6 +407,18 @@ function Message({ item }: { item: Item }) { ) + case "heard": + return ( +
+ {item.text} +
+ ) + case "reply": + return ( +
+ {item.text} +
+ ) case "error": return
{item.text}
case "actions": @@ -431,7 +443,7 @@ function Message({ item }: { item: Item }) { } } -/** A wren node's answer: each call, where it went, and whether it was applied. */ +/** What a node set: each call, where it went, and whether it was applied. */ function Actions({ actions }: { actions: Action[] }) { if (actions.length === 0) { return ( @@ -462,9 +474,11 @@ function Actions({ actions }: { actions: Action[] }) { : " (nothing wired for it)"} ))} -
- confidence {actions[0].confidence.toFixed(2)} -
+ {typeof actions[0].confidence === "number" ? ( +
+ confidence {actions[0].confidence.toFixed(2)} +
+ ) : null} ) } diff --git a/frontend/src/components/Chat/useControl.ts b/frontend/src/components/Chat/useControl.ts index f792609..9555ad4 100644 --- a/frontend/src/components/Chat/useControl.ts +++ b/frontend/src/components/Chat/useControl.ts @@ -1,61 +1,96 @@ import { useEffect, useState } from "react" import { ApiError, type ControlNode, MessagesService } from "@/client" -import { - liveStore, - useLiveValue, - useNodeFailure, -} from "@/components/Flow/liveStore" +import { liveStore } from "@/components/Flow/liveStore" -/** One call the node answered with, as its `actions` output records it. */ +/** One value a node set, as its `actions` output records it. */ export type Action = { message: string value: unknown applied: boolean delay_s: number | null call: string - confidence: number + /** Wren's certainty; the assistant has none. */ + confidence?: number } export type ControlItem = - | { kind: "user" | "error"; text: string } + | { kind: "user" | "heard" | "reply" | "error"; text: string } | { kind: "actions"; actions: Action[] } -/** How long a command may go unanswered before the chat says so. */ -const TIMEOUT_MS = 15_000 +/** The line each kind of message makes. */ +const LINE = { partial: "heard", text: "user", reply: "reply" } as const + +/** How long a typed command may go unanswered before the chat says so. */ +const TIMEOUT_MS = 20_000 + +/** The next item list with `item` added, merged into the line it continues. */ +function added(items: ControlItem[], item: ControlItem): ControlItem[] { + const last = items[items.length - 1] + // The words heard so far grow in one line, which the final text replaces. + if (last?.kind === "heard" && (item.kind === "heard" || item.kind === "user")) + return [...items.slice(0, -1), item] + // A reply arrives a sentence at a time. + if (last?.kind === "reply" && item.kind === "reply") + return [ + ...items.slice(0, -1), + { ...last, text: `${last.text} ${item.text}` }, + ] + return [...items, item] +} /** - * Talking to a wren node: a command goes out on its prompt's message, and the - * answer is the next value of its actions' message — or its next failure. - * - * The answer is told apart by the engine's own timestamps, taken before the - * command leaves, so no clock on this side is compared with the server's. The - * history lives here only, and a reload starts it over. + * One conversation with a flow's wren and assistant nodes: what was heard or + * typed, what each answered and what it set, as the engine publishes it. A + * typed command goes out on the prompt's message, which is where a spoken one + * arrives too, so both run through the same flow and show up the same way. + * The history lives here only, and a reload starts it over. */ export function useControl(node: ControlNode | undefined) { const [items, setItems] = useState([]) - // What "newer" means for this command: the last answer and the last failure - // before it was sent. - const [pending, setPending] = useState<{ - answer: number - failure: number - } | null>(null) - const answer = useLiveValue(node?.actions_msg) - const failure = useNodeFailure(node?.id ?? "") + const [pending, setPending] = useState(false) useEffect(() => { - if (!pending) return - if ((answer?.ts ?? 0) > pending.answer) { - const actions = Array.isArray(answer?.value) - ? (answer.value as Action[]) - : [] - setItems((prev) => [...prev, { kind: "actions", actions }]) - setPending(null) - } else if (failure && failure.ts > pending.failure) { - setItems((prev) => [...prev, { kind: "error", text: failure.error }]) - setPending(null) + if (!node) return + const replies = node.watch.some((w) => w.kind === "reply") + const add = (item: ControlItem) => setItems((prev) => added(prev, item)) + const stops = node.watch.map((watch) => { + let last = liveStore.getValue(watch.msg) + return liveStore.subscribe(`value:${watch.msg}`, () => { + const value = liveStore.getValue(watch.msg) + if (!value || value === last || value.prior) return + last = value + if (watch.node && value.source?.id !== watch.node) return + if (watch.kind === "actions") { + const actions = Array.isArray(value.value) + ? (value.value as Action[]) + : [] + // With a reply to come, an empty list says nothing. + if (actions.length > 0 || !replies) add({ kind: "actions", actions }) + if (!replies) setPending(false) + } else if (typeof value.value === "string" && value.value) { + add({ kind: LINE[watch.kind], text: value.value }) + if (watch.kind === "reply") setPending(false) + } + }) + }) + const answering = node.watch.filter((w) => w.kind !== "partial" && w.node) + for (const id of new Set(answering.map((w) => w.node ?? ""))) { + let since = liveStore.getFailure(id)?.ts ?? 0 + stops.push( + liveStore.subscribe(`failure:${id}`, () => { + const failure = liveStore.getFailure(id) + if (!failure || failure.ts <= since) return + since = failure.ts + add({ kind: "error", text: failure.error }) + setPending(false) + }), + ) } - }, [pending, answer, failure]) + return () => { + for (const stop of stops) stop() + } + }, [node]) useEffect(() => { if (!pending) return @@ -67,18 +102,14 @@ export function useControl(node: ControlNode | undefined) { text: "No answer yet. Check that its flow is running.", }, ]) - setPending(null) + setPending(false) }, TIMEOUT_MS) return () => clearTimeout(timer) }, [pending]) const send = async (text: string) => { if (!node) return - setItems((prev) => [...prev, { kind: "user", text }]) - setPending({ - answer: liveStore.getValue(node.actions_msg)?.ts ?? 0, - failure: liveStore.getFailure(node.id)?.ts ?? 0, - }) + setPending(true) try { await MessagesService.publishMessage({ name: node.prompt_msg, @@ -94,9 +125,9 @@ export function useControl(node: ControlNode | undefined) { text: typeof detail === "string" ? detail : String(error), }, ]) - setPending(null) + setPending(false) } } - return { items, busy: pending !== null, send, clear: () => setItems([]) } + return { items, busy: pending, send, clear: () => setItems([]) } } diff --git a/frontend/src/components/Flow/FlowNode.tsx b/frontend/src/components/Flow/FlowNode.tsx index fc90041..8d4d64c 100644 --- a/frontend/src/components/Flow/FlowNode.tsx +++ b/frontend/src/components/Flow/FlowNode.tsx @@ -2,6 +2,7 @@ import { Handle, type NodeProps, Position } from "@xyflow/react" import { Bell, Bird, + Bot, Braces, Bug, Cable, @@ -65,6 +66,7 @@ const NODE_ICONS = { file: FileText, ntfy: Bell, wren: Bird, + assistant: Bot, stt: Mic, tts: Volume2, // The glyph the Devices tab goes by. diff --git a/frontend/src/components/Flow/liveStore.ts b/frontend/src/components/Flow/liveStore.ts index 0e3fc10..fe95206 100644 --- a/frontend/src/components/Flow/liveStore.ts +++ b/frontend/src/components/Flow/liveStore.ts @@ -126,6 +126,8 @@ function subscribeKey(key: string, listener: Listener) { } export const liveStore = { + /** Told of every change to one key, such as `value:`, as it happens. */ + subscribe: subscribeKey, setValue(name: string, value: LiveValue) { values.set(name, value) notify(`value:${name}`)