PowerOn/gateway
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

grapheditor enhanced para set for ai

Browse source
This commit is contained in:
ValueOn AG 2026-04-19 01:31:57 +02:00
parent 7d27ddf6b5
commit 1ffe521ad8
2 changed files with 317 additions and 17 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

31
modules/features/graphicalEditor/routeFeatureGraphicalEditor.py
View file

@ -757,12 +757,31 @@ async def _runEditorAgent(
"for the user — you must NEVER execute the workflow or any of its actions. "
"Even when the user says 'create a workflow that sends an email', you build the "
"graph (e.g. add an email node, connect it) — you do NOT actually send an email. "
"Use these workflow tools to mutate the graph: "
"readWorkflowGraph, listAvailableNodeTypes, addNode, removeNode, connectNodes, "
"setNodeParameter, validateGraph. "
"Always read the current graph and list available node types first, then plan the "
"smallest set of mutations, then apply them. Respond concisely in the user's "
"language and confirm what you changed in the graph."
"\n\nGraph-mutating tools: readWorkflowGraph, listAvailableNodeTypes, "
"describeNodeType, addNode, removeNode, connectNodes, setNodeParameter, "
"autoLayoutWorkflow, validateGraph. "
"Connection discovery (for parameters of frontendType='userConnection'): listConnections."
"\n\nMandatory build sequence:"
"\n1. readWorkflowGraph — understand current state."
"\n2. listAvailableNodeTypes — find candidate node ids."
"\n3. For EACH node type you plan to add: call describeNodeType(nodeType=...) "
"to learn its requiredParameters, allowedValues and ports. Never skip this "
"step — guessing parameters leaves the user with empty config cards."
"\n4. If any required parameter has frontendType='userConnection' (e.g. "
"email.checkEmail.connectionReference), call listConnections and pick the "
"connectionId that matches the user's intent (or ask the user if none clearly fits)."
"\n5. addNode with parameters={...} containing AT LEAST every requiredParameter "
"filled with a sensible value (use the user's request, the parameter "
"description, sane defaults, or — for required user-connection fields — "
"an actual connectionId). Do NOT pass position; the layout step handles it."
"\n6. connectNodes — wire the nodes consistent with port schemas from describeNodeType."
"\n7. autoLayoutWorkflow — call exactly once as the LAST graph-mutating step so the "
"canvas shows a readable top-down layout instead of overlapping boxes."
"\n8. validateGraph — sanity check, then answer the user."
"\n\nIf a required parameter cannot be filled from the user's request and has "
"no safe default, ask the user once for that specific value (e.g. recipient "
"address, target language, prompt text) instead of leaving the field blank. "
"Respond concisely in the user's language and list what you changed in the graph."
)
editorConfig = AgentConfig(

303
modules/serviceCenter/services/serviceAgent/workflowTools.py
View file

@ -3,7 +3,8 @@
"""
Workflow Toolbox - AI-assisted graph manipulation tools for the GraphicalEditor.
Tools: readWorkflowGraph, addNode, removeNode, connectNodes, setNodeParameter,
listAvailableNodeTypes, validateGraph, listWorkflowHistory, readWorkflowMessages.
listAvailableNodeTypes, describeNodeType, autoLayoutWorkflow,
validateGraph, listWorkflowHistory, readWorkflowMessages.
Conventions enforced here (matches coreTools / actionToolAdapter):
- Every ``ToolResult(...)`` provides ``toolCallId`` and ``toolName`` (pydantic
@ -144,14 +145,32 @@ async def _addNode(params: Dict[str, Any], context: Any) -> ToolResult:
nodeId = params.get("nodeId") or str(uuid.uuid4())[:8]
title = params.get("title", "")
nodeParams = params.get("parameters", {}) or {}
position = params.get("position") or {"x": len(nodes) * 200, "y": 100}
# Frontend stores positions as TOP-LEVEL ``x`` / ``y`` on the node
# (see ``fromApiGraph`` / ``toApiGraph``). Accept either explicit
# ``x`` / ``y`` or a ``position={x,y}`` shape from the model and
# always persist as top-level ``x`` / ``y``. Fallback puts new
# nodes in a horizontal stripe so the user sees them even before
# ``autoLayoutWorkflow`` runs.
position = params.get("position") or {}
x = params.get("x")
if x is None:
x = position.get("x") if isinstance(position, dict) else None
if x is None:
x = 40 + len(nodes) * 260
y = params.get("y")
if y is None:
y = position.get("y") if isinstance(position, dict) else None
if y is None:
y = 40
newNode = {
"id": nodeId,
"type": nodeType,
"title": title,
"parameters": nodeParams,
"position": position,
"x": x,
"y": y,
}
nodes.append(newNode)
graph["nodes"] = nodes
@ -287,8 +306,55 @@ def _coerceLabel(rawLabel: Any, fallback: str) -> str:
return fallback
def _summarizeNodeForCatalog(n: Dict[str, Any]) -> Dict[str, Any]:
"""Compact summary used in ``listAvailableNodeTypes`` — small but
informative enough that the model can pick the right type and knows
whether ``describeNodeType`` is worth a follow-up call."""
nodeId = n.get("id") or ""
paramsList = n.get("parameters") or []
requiredCount = sum(1 for p in paramsList if isinstance(p, dict) and p.get("required"))
return {
"id": nodeId,
"category": n.get("category"),
"label": _coerceLabel(n.get("label"), nodeId),
"description": _coerceLabel(n.get("description"), ""),
"paramCount": len(paramsList),
"requiredParamCount": requiredCount,
"usesAi": bool(((n.get("meta") or {}).get("usesAi"))),
}
def _summarizeParameter(p: Dict[str, Any]) -> Dict[str, Any]:
"""Reduce a node parameter spec to just what the AI needs to fill it."""
out: Dict[str, Any] = {
"name": p.get("name"),
"type": p.get("type"),
"required": bool(p.get("required")),
"frontendType": p.get("frontendType"),
"description": _coerceLabel(p.get("description"), ""),
}
if "default" in p:
out["default"] = p.get("default")
feOpts = p.get("frontendOptions")
if isinstance(feOpts, dict):
# Expose enum-style choices ("options") so the model sticks to allowed values.
if isinstance(feOpts.get("options"), list):
out["allowedValues"] = feOpts.get("options")
if p.get("frontendType") == "userConnection":
out["hint"] = (
"Call listConnections to discover available connections; pass the "
"connectionId here. Required before this node can run."
)
return out
async def _listAvailableNodeTypes(params: Dict[str, Any], context: Any) -> ToolResult:
"""List all available node types for the flow builder."""
"""List all available node types for the flow builder (compact catalog).
Returns ``id``, ``category``, ``label``, short ``description``, and the
parameter counts. To learn HOW to fill a node's parameters use
``describeNodeType(nodeType=...)`` — that returns the full schema.
"""
name = "listAvailableNodeTypes"
try:
from modules.features.graphicalEditor.nodeDefinitions import STATIC_NODE_TYPES
@ -296,18 +362,188 @@ async def _listAvailableNodeTypes(params: Dict[str, Any], context: Any) -> ToolR
for n in STATIC_NODE_TYPES:
if not isinstance(n, dict):
continue
nodeId = n.get("id") or ""
nodeTypes.append({
"id": nodeId,
"category": n.get("category"),
"label": _coerceLabel(n.get("label"), nodeId),
})
nodeTypes.append(_summarizeNodeForCatalog(n))
return _ok(name, {"nodeTypes": nodeTypes, "count": len(nodeTypes)})
except Exception as e:
logger.exception("listAvailableNodeTypes failed: %s", e)
return _err(name, str(e))
async def _describeNodeType(params: Dict[str, Any], context: Any) -> ToolResult:
"""Return the full schema for a single node type so the AI can fill
``addNode.parameters`` correctly (which fields are required, what types,
default values, allowed enum values, what each port expects/produces).
This is the canonical way to discover required parameters before
calling ``addNode`` — without it the model guesses ``parameters={}``
and the user gets empty configuration cards.
"""
name = "describeNodeType"
try:
nodeType = params.get("nodeType") or params.get("id")
if not nodeType:
return _err(name, "nodeType required")
from modules.features.graphicalEditor.nodeDefinitions import STATIC_NODE_TYPES
target: Dict[str, Any] = {}
for n in STATIC_NODE_TYPES:
if isinstance(n, dict) and n.get("id") == nodeType:
target = n
break
if not target:
return _err(name, f"Unknown nodeType '{nodeType}' — call listAvailableNodeTypes first")
rawParams = target.get("parameters") or []
parameters = [
_summarizeParameter(p) for p in rawParams if isinstance(p, dict)
]
def _portList(portsDict: Any) -> List[Dict[str, Any]]:
if not isinstance(portsDict, dict):
return []
out: List[Dict[str, Any]] = []
for idx, spec in sorted(portsDict.items(), key=lambda kv: int(kv[0]) if str(kv[0]).isdigit() else 0):
if not isinstance(spec, dict):
continue
entry: Dict[str, Any] = {"index": int(idx) if str(idx).isdigit() else idx}
if "schema" in spec:
entry["schema"] = spec.get("schema")
if "accepts" in spec:
entry["accepts"] = spec.get("accepts")
out.append(entry)
return out
meta = target.get("meta") or {}
return _ok(name, {
"id": target.get("id"),
"category": target.get("category"),
"label": _coerceLabel(target.get("label"), target.get("id") or ""),
"description": _coerceLabel(target.get("description"), ""),
"usesAi": bool(meta.get("usesAi")),
"inputs": int(target.get("inputs") or 0),
"outputs": int(target.get("outputs") or 0),
"inputPorts": _portList(target.get("inputPorts")),
"outputPorts": _portList(target.get("outputPorts")),
"parameters": parameters,
"requiredParameters": [p["name"] for p in parameters if p.get("required")],
})
except Exception as e:
logger.exception("describeNodeType failed: %s", e)
return _err(name, str(e))
# Geometry constants — MUST match the frontend (FlowCanvas.tsx) so the
# server-side auto-layout produces the exact same coordinates the user
# would get by clicking "Arrange" in the UI.
_NODE_WIDTH = 200
_NODE_HEIGHT = 72
_LAYOUT_V_GAP = 80
_LAYOUT_H_GAP = 60
_LAYOUT_START_X = 40
_LAYOUT_START_Y = 40
def _computeAutoLayout(
nodes: List[Dict[str, Any]],
connections: List[Dict[str, Any]],
) -> List[Dict[str, Any]]:
"""Topological-layer layout — port of ``computeAutoLayout`` in FlowCanvas.tsx.
Arranges nodes top-to-bottom in layers (one layer per BFS step from the
sources). Disconnected nodes are appended as extra single-node layers,
same as the frontend. Returns a NEW node list with updated top-level
``x``/``y``; legacy ``position`` keys are stripped to avoid two
competing sources of truth.
"""
if not nodes:
return nodes
nodeIds = {n.get("id") for n in nodes if n.get("id")}
inDegree: Dict[str, int] = {nid: 0 for nid in nodeIds if nid}
children: Dict[str, List[str]] = {nid: [] for nid in nodeIds if nid}
for c in connections or []:
src = c.get("source")
tgt = c.get("target")
if src in inDegree and tgt in inDegree:
inDegree[tgt] = inDegree[tgt] + 1
children[src].append(tgt)
layers: List[List[str]] = []
layerOf: Dict[str, int] = {}
queue: List[str] = [nid for nid, deg in inDegree.items() if deg == 0]
while queue:
batch = list(queue)
queue = []
layerIdx = len(layers)
layers.append(batch)
for nid in batch:
layerOf[nid] = layerIdx
for childId in children.get(nid, []):
inDegree[childId] = inDegree[childId] - 1
if inDegree[childId] == 0:
queue.append(childId)
# Cycles: append remaining nodes as their own layers (matches frontend).
for n in nodes:
nid = n.get("id")
if nid and nid not in layerOf:
layerIdx = len(layers)
layers.append([nid])
layerOf[nid] = layerIdx
out: List[Dict[str, Any]] = []
for n in nodes:
nid = n.get("id")
layer = layerOf.get(nid, 0) if nid else 0
siblings = layers[layer] if 0 <= layer < len(layers) else [nid]
idxInLayer = siblings.index(nid) if nid in siblings else 0
new = dict(n)
new["x"] = _LAYOUT_START_X + idxInLayer * (_NODE_WIDTH + _LAYOUT_H_GAP)
new["y"] = _LAYOUT_START_Y + layer * (_NODE_HEIGHT + _LAYOUT_V_GAP)
# Strip legacy ``position`` so frontend never sees two coordinates.
new.pop("position", None)
out.append(new)
return out
async def _autoLayoutWorkflow(params: Dict[str, Any], context: Any) -> ToolResult:
"""Re-arrange all nodes of the workflow into a clean top-down layered layout.
Same algorithm as the editor's "Arrange" button — call this after you
finished adding/connecting nodes so the user doesn't see an unreadable
pile of overlapping boxes.
"""
name = "autoLayoutWorkflow"
try:
workflowId, instanceId = _resolveIds(params, context)
if not workflowId or not instanceId:
return _err(name, "workflowId and instanceId required (and not present in agent context)")
iface = _getInterface(context, instanceId)
wf = iface.getWorkflow(workflowId)
if not wf:
return _err(name, f"Workflow {workflowId} not found")
graph = dict(wf.get("graph", {}) or {})
nodes = list(graph.get("nodes", []) or [])
connections = list(graph.get("connections", []) or [])
if not nodes:
return _ok(name, {"message": "No nodes to layout", "nodeCount": 0})
graph["nodes"] = _computeAutoLayout(nodes, connections)
iface.updateWorkflow(workflowId, {"graph": graph})
return _ok(name, {
"message": f"Auto-layout applied to {len(nodes)} nodes",
"nodeCount": len(nodes),
"layerCount": max((c.get("y", 0) for c in graph["nodes"]), default=_LAYOUT_START_Y) // (_NODE_HEIGHT + _LAYOUT_V_GAP) + 1,
})
except Exception as e:
logger.exception("autoLayoutWorkflow failed: %s", e)
return _err(name, str(e))
async def _validateGraph(params: Dict[str, Any], context: Any) -> ToolResult:
"""Validate a workflow graph for common issues."""
name = "validateGraph"
@ -507,11 +743,56 @@ def getWorkflowToolDefinitions() -> List[Dict[str, Any]]:
{
"name": "listAvailableNodeTypes",
"handler": _listAvailableNodeTypes,
"description": "List all available node types for the flow builder. Call this once to discover ids before using addNode.",
"description": (
"List all available node types (compact catalog: id, label, "
"description, paramCount, requiredParamCount, usesAi). Call this "
"once to discover ids; then call describeNodeType for each type "
"you intend to add to learn the parameter schema."
),
"parameters": {"type": "object", "properties": {}},
"readOnly": True,
"toolSet": TOOLBOX_ID,
},
{
"name": "describeNodeType",
"handler": _describeNodeType,
"description": (
"Return the FULL parameter schema for a single node type "
"(name, type, required, default, allowedValues, description) "
"plus input/output ports. ALWAYS call this before addNode for "
"any node type that has requiredParamCount > 0, and pass all "
"required parameters into addNode — otherwise the user sees an "
"empty configuration card. For parameters with "
"frontendType='userConnection' call listConnections to obtain "
"a connectionId."
),
"parameters": {
"type": "object",
"properties": {
"nodeType": {"type": "string", "description": "Node type id from listAvailableNodeTypes (e.g. 'email.checkEmail', 'ai.prompt')"},
},
"required": ["nodeType"],
},
"readOnly": True,
"toolSet": TOOLBOX_ID,
},
{
"name": "autoLayoutWorkflow",
"handler": _autoLayoutWorkflow,
"description": (
"Re-arrange ALL nodes into a clean top-down layered layout "
"(same algorithm as the editor's 'Arrange' button). Call this "
"AFTER you finished adding nodes and connections, otherwise the "
"user sees a pile of overlapping boxes. Idempotent — safe to "
"call multiple times."
),
"parameters": {
"type": "object",
"properties": {**_idFields},
"required": [],
},
"toolSet": TOOLBOX_ID,
},
{
"name": "validateGraph",
"handler": _validateGraph,