Raise error if inputs are not connected with output in functional model (#20705)

* Raise error if inputs are not connected with output in functional model

* Fix Failing test case for unconnected inputs/outputs

* fix formatting issue

Author: Surya2k1

keras/src/backend/tensorflow/saved_model_test.py

@@ -191,24 +191,37 @@ def call(self, inputs):
     def test_multi_input_model(self):
         input_1 = layers.Input(shape=(3,))
         input_2 = layers.Input(shape=(5,))
-        model = models.Model([input_1, input_2], [input_1, input_2])
-        path = os.path.join(self.get_temp_dir(), "my_keras_model")
 
-        tf.saved_model.save(model, path)
-        restored_model = tf.saved_model.load(path)
+        y1 = layers.Dense(1)(input_1)
+        y2 = layers.Dense(1)(input_2)
+        layer_2 = layers.Dense(1, activation="relu")
+        output_1 = layer_2(y1)
+        output_2 = layer_2(y2)
+        model = models.Model([input_1, input_2], [output_1, output_2])
+
         input_arr_1 = np.random.random((1, 3)).astype("float32")
         input_arr_2 = np.random.random((1, 5)).astype("float32")
 
-        outputs = restored_model.signatures["serving_default"](
+        model = models.Model([input_1, input_2], [output_1, output_2])
+        path = os.path.join(self.get_temp_dir(), "my_keras_model")
+        outputs_1 = model(
+            inputs=[
+                tf.convert_to_tensor(input_arr_1, dtype=tf.float32),
+                tf.convert_to_tensor(input_arr_2, dtype=tf.float32),
+            ],
+        )
+        tf.saved_model.save(model, path)
+        restored_model = tf.saved_model.load(path)
+
+        outputs_2 = restored_model.signatures["serving_default"](
             inputs=tf.convert_to_tensor(input_arr_1, dtype=tf.float32),
             inputs_1=tf.convert_to_tensor(input_arr_2, dtype=tf.float32),
         )
-
         self.assertAllClose(
-            input_arr_1, outputs["output_0"], rtol=1e-4, atol=1e-4
+            outputs_1[0], outputs_2["output_0"], rtol=1e-4, atol=1e-4
         )
         self.assertAllClose(
-            input_arr_2, outputs["output_1"], rtol=1e-4, atol=1e-4
+            outputs_1[1], outputs_2["output_1"], rtol=1e-4, atol=1e-4
         )
 
     def test_multi_input_custom_model_and_layer(self):

keras/src/ops/function.py

@@ -81,6 +81,12 @@ def __init__(self, inputs, outputs, name=None):
         self._nodes_by_depth = nodes_by_depth
         self._operations = operations
         self._operations_by_depth = operations_by_depth
+        for input in self._inputs:
+            if (
+                input._keras_history.operation
+                and not input._keras_history.operation._outbound_nodes
+            ):
+                raise ValueError("`inputs` not connected to `outputs`")
 
     @property
     def operations(self):

keras/src/ops/function_test.py

@@ -7,6 +7,7 @@
 from keras.src.layers import Dense
 from keras.src.layers import Input
 from keras.src.models import Model
+from keras.src.models import Sequential
 from keras.src.ops import function
 from keras.src.ops import numpy as knp
 
@@ -142,3 +143,26 @@ def test_function_with_empty_inputs(self):
             ValueError, "`inputs` argument cannot be empty"
         ):
             _ = function.Function(inputs=[], outputs=x)
+
+    def test_function_with_unconnected_inputs(self):
+        model_1 = Sequential(
+            [
+                Input(shape=(6,)),
+                Dense(3, activation="sigmoid"),
+            ]
+        )
+        model_2 = Sequential(
+            [
+                Input(shape=(3,)),
+                Dense(2, activation="sigmoid"),
+            ],
+        )
+        with self.assertRaisesRegex(
+            ValueError, "`inputs` not connected to `outputs`"
+        ):
+            _ = Model(Input(shape=(6,)), model_2(model_1(Input(shape=(6,)))))
+
+        with self.assertRaisesRegex(
+            ValueError, "`inputs` not connected to `outputs`"
+        ):
+            _ = Model(model_1(Input(shape=(6,))), model_2(Input(shape=(3,))))