Add flow visualization for tool builder
Implements node-based flow visualization using NodeGraphQt-QuiltiX-fork: New files: - src/cmdforge/gui/widgets/flow_graph.py: FlowGraphWidget with custom node types - InputNode: Shows tool inputs ($input + argument variables) - PromptNode: AI prompt steps with provider display - CodeNode: Python code steps - OutputNode: Final output - scripts/test_nodegraph.py: Standalone prototype for testing Modified files: - tool_builder_page.py: Added List/Flow view toggle with QStackedWidget - Lazy-loads flow widget on first use - Syncs between list and flow views - Double-click nodes to edit steps - styles.py: Added viewToggle button styling - widgets/__init__.py: Export FlowGraphWidget - pyproject.toml: Added 'flow' optional dependency group Features: - Toggle between List and Flow views in Steps panel - Nodes auto-connected based on step order - Auto-layout and fit-to-view on load - Double-click nodes opens step edit dialog - Graceful fallback if NodeGraphQt not installed Co-Authored-By: Claude Opus 4.5 <[email protected]>
This commit is contained in:
Claude Opus 4.5
@@ -0,0 +1,240 @@
|
||||
"""Flow visualization widget using NodeGraphQt."""
|
||||
|
||||
from typing import Optional, List
|
||||
|
||||
from PySide6.QtWidgets import QWidget, QVBoxLayout
|
||||
from PySide6.QtCore import Signal
|
||||
|
||||
from NodeGraphQt import NodeGraph, BaseNode
|
||||
|
||||
from ...tool import Tool, PromptStep, CodeStep
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Custom Node Types
|
||||
# =============================================================================
|
||||
|
||||
class CmdForgeBaseNode(BaseNode):
|
||||
"""Base class for CmdForge nodes with common styling."""
|
||||
|
||||
__identifier__ = 'cmdforge'
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
# Store reference to the step data
|
||||
self._step_data = None
|
||||
self._step_index = -1
|
||||
|
||||
|
||||
class InputNode(CmdForgeBaseNode):
|
||||
"""Node representing tool input (stdin and arguments)."""
|
||||
|
||||
NODE_NAME = 'Input'
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.set_color(90, 90, 160) # Indigo
|
||||
# Default output for stdin
|
||||
self.add_output('input', color=(180, 180, 250))
|
||||
|
||||
def set_arguments(self, arguments: list):
|
||||
"""Add output ports for each argument variable."""
|
||||
for arg in arguments:
|
||||
if arg.variable:
|
||||
# Check if port already exists
|
||||
existing = [p.name() for p in self.output_ports()]
|
||||
if arg.variable not in existing:
|
||||
self.add_output(arg.variable, color=(180, 180, 250))
|
||||
|
||||
|
||||
class PromptNode(CmdForgeBaseNode):
|
||||
"""Node representing a prompt step."""
|
||||
|
||||
NODE_NAME = 'Prompt'
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.set_color(102, 126, 234) # Indigo (matching CmdForge theme)
|
||||
self.add_input('in', color=(180, 180, 250), multi_input=True)
|
||||
self.add_output('out', color=(180, 250, 180))
|
||||
|
||||
# Add properties for display
|
||||
self.add_text_input('provider', 'Provider', text='claude')
|
||||
self.add_text_input('output_var', 'Output', text='response')
|
||||
|
||||
def set_step(self, step: PromptStep, index: int):
|
||||
"""Configure node from a PromptStep."""
|
||||
self._step_data = step
|
||||
self._step_index = index
|
||||
self.set_property('provider', step.provider or 'claude')
|
||||
self.set_property('output_var', step.output_var or 'response')
|
||||
# Update output port name
|
||||
if self.output_ports():
|
||||
self.output_ports()[0]._name = step.output_var or 'response'
|
||||
|
||||
|
||||
class CodeNode(CmdForgeBaseNode):
|
||||
"""Node representing a code step."""
|
||||
|
||||
NODE_NAME = 'Code'
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.set_color(72, 187, 120) # Green
|
||||
self.add_input('in', color=(180, 250, 180), multi_input=True)
|
||||
self.add_output('out', color=(250, 220, 180))
|
||||
|
||||
# Add properties
|
||||
self.add_text_input('output_var', 'Output', text='result')
|
||||
|
||||
def set_step(self, step: CodeStep, index: int):
|
||||
"""Configure node from a CodeStep."""
|
||||
self._step_data = step
|
||||
self._step_index = index
|
||||
self.set_property('output_var', step.output_var or 'result')
|
||||
# Update output port name
|
||||
if self.output_ports():
|
||||
self.output_ports()[0]._name = step.output_var or 'result'
|
||||
|
||||
|
||||
class OutputNode(CmdForgeBaseNode):
|
||||
"""Node representing tool output."""
|
||||
|
||||
NODE_NAME = 'Output'
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.set_color(237, 137, 54) # Orange
|
||||
self.add_input('in', color=(250, 220, 180), multi_input=True)
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Flow Graph Widget
|
||||
# =============================================================================
|
||||
|
||||
class FlowGraphWidget(QWidget):
|
||||
"""Widget for visualizing tool flow as a node graph."""
|
||||
|
||||
# Emitted when a node is double-clicked (step_index, step_type)
|
||||
node_double_clicked = Signal(int, str)
|
||||
|
||||
# Emitted when the flow structure changes
|
||||
flow_changed = Signal()
|
||||
|
||||
def __init__(self, parent=None):
|
||||
super().__init__(parent)
|
||||
self._tool: Optional[Tool] = None
|
||||
self._graph: Optional[NodeGraph] = None
|
||||
self._input_node = None
|
||||
self._output_node = None
|
||||
self._step_nodes: List[CmdForgeBaseNode] = []
|
||||
|
||||
self._setup_ui()
|
||||
|
||||
def _setup_ui(self):
|
||||
"""Set up the UI."""
|
||||
layout = QVBoxLayout(self)
|
||||
layout.setContentsMargins(0, 0, 0, 0)
|
||||
|
||||
# Create node graph
|
||||
self._graph = NodeGraph()
|
||||
|
||||
# Register custom node types
|
||||
self._graph.register_node(InputNode)
|
||||
self._graph.register_node(PromptNode)
|
||||
self._graph.register_node(CodeNode)
|
||||
self._graph.register_node(OutputNode)
|
||||
|
||||
# Connect signals
|
||||
self._graph.node_double_clicked.connect(self._on_node_double_clicked)
|
||||
|
||||
# Add graph widget
|
||||
layout.addWidget(self._graph.widget)
|
||||
|
||||
def set_tool(self, tool: Optional[Tool]):
|
||||
"""Set the tool to visualize."""
|
||||
self._tool = tool
|
||||
self._rebuild_graph()
|
||||
|
||||
def _rebuild_graph(self):
|
||||
"""Rebuild the graph from the tool data."""
|
||||
if not self._graph:
|
||||
return
|
||||
|
||||
# Clear existing nodes
|
||||
self._graph.clear_session()
|
||||
self._input_node = None
|
||||
self._output_node = None
|
||||
self._step_nodes = []
|
||||
|
||||
if not self._tool:
|
||||
return
|
||||
|
||||
# Create input node
|
||||
self._input_node = self._graph.create_node(
|
||||
'cmdforge.InputNode',
|
||||
name='Input',
|
||||
pos=[-300, 0]
|
||||
)
|
||||
if self._tool.arguments:
|
||||
self._input_node.set_arguments(self._tool.arguments)
|
||||
|
||||
# Create step nodes
|
||||
x_pos = 0
|
||||
prev_node = self._input_node
|
||||
|
||||
for i, step in enumerate(self._tool.steps or []):
|
||||
if isinstance(step, PromptStep):
|
||||
node = self._graph.create_node(
|
||||
'cmdforge.PromptNode',
|
||||
name=f'Prompt {i+1}',
|
||||
pos=[x_pos, 0]
|
||||
)
|
||||
node.set_step(step, i)
|
||||
elif isinstance(step, CodeStep):
|
||||
node = self._graph.create_node(
|
||||
'cmdforge.CodeNode',
|
||||
name=f'Code {i+1}',
|
||||
pos=[x_pos, 0]
|
||||
)
|
||||
node.set_step(step, i)
|
||||
else:
|
||||
continue
|
||||
|
||||
self._step_nodes.append(node)
|
||||
|
||||
# Connect to previous node
|
||||
if prev_node and prev_node.output_ports():
|
||||
prev_node.output_ports()[0].connect_to(node.input_ports()[0])
|
||||
|
||||
prev_node = node
|
||||
x_pos += 250
|
||||
|
||||
# Create output node
|
||||
self._output_node = self._graph.create_node(
|
||||
'cmdforge.OutputNode',
|
||||
name='Output',
|
||||
pos=[x_pos, 0]
|
||||
)
|
||||
|
||||
# Connect last step to output
|
||||
if prev_node and prev_node.output_ports():
|
||||
prev_node.output_ports()[0].connect_to(self._output_node.input_ports()[0])
|
||||
|
||||
# Auto-layout and fit view
|
||||
self._graph.auto_layout_nodes()
|
||||
self._graph.fit_to_selection()
|
||||
|
||||
def _on_node_double_clicked(self, node):
|
||||
"""Handle node double-click."""
|
||||
if hasattr(node, '_step_index') and node._step_index >= 0:
|
||||
step_type = 'prompt' if isinstance(node, PromptNode) else 'code'
|
||||
self.node_double_clicked.emit(node._step_index, step_type)
|
||||
|
||||
def refresh(self):
|
||||
"""Refresh the graph from current tool data."""
|
||||
self._rebuild_graph()
|
||||
|
||||
def get_graph(self) -> Optional[NodeGraph]:
|
||||
"""Get the underlying NodeGraph instance."""
|
||||
return self._graph
|
||||
Reference in New Issue
Block a user