From 06bb3bb77d8ddcda55bd0753a6216d0a1689f5f3 Mon Sep 17 00:00:00 2001
From: Suhana <suhanaaa@google.com>
Date: Tue, 28 Oct 2025 10:03:29 +0530
Subject: [PATCH 01/12] Adding tensor layout for TP autosharding

---
 keras/src/backend/jax/core.py                 |  58 ++++++-
 keras/src/backend/jax/core_test.py            |  78 +++++++++
 .../tensor_parallel/tensor_layout.py          |  43 +++++
 .../tensor_parallel/tensor_layout_test.py     | 163 ++++++++++++++++++
 4 files changed, 341 insertions(+), 1 deletion(-)
 create mode 100644 keras/src/distribution/tensor_parallel/tensor_layout.py
 create mode 100644 keras/src/distribution/tensor_parallel/tensor_layout_test.py

diff --git a/keras/src/backend/jax/core.py b/keras/src/backend/jax/core.py
index 7dc5a98fb8d5..aee30a3deadd 100644
--- a/keras/src/backend/jax/core.py
+++ b/keras/src/backend/jax/core.py
@@ -1,5 +1,6 @@
 import jax
 import jax.experimental.sparse as jax_sparse
+import jax.lax as lax
 import jax.numpy as jnp
 import ml_dtypes
 import numpy as np
@@ -529,6 +530,61 @@ def remat(f):
     return jax.checkpoint(f)
 
 
+def all_reduce(x, op="sum", axis_name="model"):
+    """
+    Performs an **all-reduce** operation across all replicas in the specified
+    distribution axis.
+
+    The all-reduce operation computes a reduction (like sum, mean, or product)
+    of the input tensor `x` across all devices/replicas in the `axis_name`
+    group, and then broadcasts the result back to all participating devices.
+
+    Args:
+        x: The tensor to reduce.
+        op: The reduction operation to perform. Common options include "sum",
+            "mean", or "product". Defaults to "sum".
+        axis_name: The name of the distribution axis (e.g., "model",
+            "data") over which to perform the reduction. Defaults to "model".
+
+    Returns:
+        The result of the all-reduce operation, with the same shape as the
+        input `x`.
+    """
+    if op == "sum":
+        return lax.psum(x, axis_name=axis_name)
+    elif op == "mean":
+        return lax.pmean(x, axis_name=axis_name)
+    else:
+        raise ValueError(
+            f"Unsupported reduction operation: {op}. "
+            "Supported options are 'sum' and 'mean'."
+        )
+
+
+def all_gather(x, axis, axis_name="model"):
+    """
+    Performs an all-gather operation across all replicas in the specified
+    distribution axis.
+
+    The all-gather operation collects the input tensor `x` from all devices
+    in the `axis_name` group and concatenates them along the specified `axis`.
+    This is often used in tensor parallelism to combine parts of a tensor
+    distributed across devices.
+
+    Args:
+        x: The tensor to gather.
+        axis: The dimension along which to concatenate the gathered tensors.
+        axis_name: The name of the distribution axis (e.g., "model",
+                                   "data") over which to perform the gather.
+                                   Defaults to "model".
+
+    Returns:
+        The gathered tensor, which will have a larger size along `axis`
+        dimension.
+    """
+    return lax.all_gather(x, axis_name=axis_name, axis=axis, tiled=True)
+
+
 class name_scope(base_name_scope):
     def __init__(self, name, **kwargs):
         super().__init__(name, **kwargs)
@@ -571,4 +627,4 @@ def device_scope(device_name):
         )
     else:
         jax_device = device_name
-    return jax.default_device(jax_device)
+    return jax.default_device(jax_device)
\ No newline at end of file
diff --git a/keras/src/backend/jax/core_test.py b/keras/src/backend/jax/core_test.py
index 792cf25e67f0..79eecad18063 100644
--- a/keras/src/backend/jax/core_test.py
+++ b/keras/src/backend/jax/core_test.py
@@ -1,3 +1,4 @@
+import functools
 import os
 
 import jax
@@ -9,6 +10,8 @@
 from keras.src import backend
 from keras.src import testing
 from keras.src.backend.config import is_nnx_enabled
+from keras.src.backend.jax.core import all_gather
+from keras.src.backend.jax.core import all_reduce
 
 if is_nnx_enabled():
     from flax import nnx
@@ -66,3 +69,78 @@ def test_keras_variable_nnx_split_merge_sync(self):
         state = jax.tree.map(lambda x: x + 1, state)
         variable2 = nnx.merge(graphdef, state)
         self.assertEqual(variable2._value, variable2.value)
+
+
+@pytest.mark.skipif(
+    backend.backend() != "jax",
+    reason="JAX backend specific test for collective operations.",
+)
+@pytest.mark.skipif(
+    jax.local_device_count() < 2,
+    reason="Requires multiple local devices for testing.",
+)
+class JaxCollectiveOpsTest(testing.TestCase):
+    def test_all_reduce_sum(self):
+        """Tests the all_reduce operation with the 'sum' reduction."""
+        num_devices = jax.local_device_count()
+        local_value = 10.0
+
+        local_inputs = jax.numpy.array([local_value] * num_devices)
+
+        @functools.partial(
+            jax.pmap, axis_name="all", devices=jax.devices("cpu")
+        )
+        def reduce_sum_fn(x):
+            return all_reduce(x, op="sum", axis_name="all")
+
+        result = reduce_sum_fn(local_inputs)
+        expected_sum = local_value * num_devices
+
+        self.assertTrue(np.allclose(result, expected_sum))
+        self.assertEqual(result.shape, (num_devices,))
+
+    def test_all_reduce_mean(self):
+        """Tests the all_reduce operation with the 'mean' reduction."""
+        num_devices = jax.local_device_count()
+        local_value = 10.0
+
+        local_inputs = jax.numpy.array([local_value] * num_devices)
+
+        @functools.partial(
+            jax.pmap, axis_name="all", devices=jax.devices("cpu")
+        )
+        def reduce_mean_fn(x):
+            return all_reduce(x, op="mean", axis_name="all")
+
+        result = reduce_mean_fn(local_inputs)
+        expected_mean = local_value
+
+        self.assertTrue(np.allclose(result, expected_mean))
+        self.assertEqual(result.shape, (num_devices,))
+
+    def test_all_gather(self):
+        """Tests the all_gather operation."""
+        num_devices = jax.local_device_count()
+        local_data = np.arange(5)
+
+        local_inputs = jax.numpy.stack(
+            [local_data + (i * 5) for i in range(num_devices)]
+        )
+
+        @functools.partial(
+            jax.pmap, axis_name="all", devices=jax.devices("cpu")
+        )
+        def gather_fn(x):
+            return all_gather(x, axis=0, axis_name="all")
+
+        result_array_on_devices = gather_fn(local_inputs)
+
+        expected_shape = (num_devices, num_devices * local_data.shape[0])
+        self.assertEqual(result_array_on_devices.shape, expected_shape)
+
+        expected_gathered_data = np.arange(num_devices * local_data.shape[0])
+
+        for i in range(num_devices):
+            self.assertTrue(
+                np.allclose(result_array_on_devices[i], expected_gathered_data)
+            )
\ No newline at end of file
diff --git a/keras/src/distribution/tensor_parallel/tensor_layout.py b/keras/src/distribution/tensor_parallel/tensor_layout.py
new file mode 100644
index 000000000000..ff6b4eff920b
--- /dev/null
+++ b/keras/src/distribution/tensor_parallel/tensor_layout.py
@@ -0,0 +1,43 @@
+import collections
+
+from keras.src import ops
+
+
+def split_tensor_for_parallelism(tensor, index, device_count, dim):
+    """Calculates a slice of a tensor along a specified dimension for a
+    given index.
+
+    This utility is used in tensor parallelism API to distribute a
+    tensor across multiple devices.
+
+    Args:
+        tensor: The full tensor to be sharded.
+        index: The index of the device/shard to return (e.g., 0, 1, 2...).
+        device_count: The total number of parallel devices or splits.
+        dim: The dimension along which to split the tensor. If -1, the
+            last dimension is used.
+
+    Returns:
+        A tensor slice corresponding to the given `index`.
+    """
+    if dim == -1:
+        static_shape = getattr(tensor, "shape", None)
+        if static_shape is not None:
+            rank = len(static_shape)
+        else:
+            rank = None
+
+        if rank is not None:
+            split_dim = rank - 1
+        else:
+            split_dim = ops.ndim(tensor) - 1
+    else:
+        split_dim = dim
+
+    splits = ops.array_split(
+        tensor, indices_or_sections=device_count, axis=split_dim
+    )
+    return splits[index]
+
+
+LayoutMap = collections.namedtuple("LayoutMap", ["state_rules", "output_rules"])
\ No newline at end of file
diff --git a/keras/src/distribution/tensor_parallel/tensor_layout_test.py b/keras/src/distribution/tensor_parallel/tensor_layout_test.py
new file mode 100644
index 000000000000..d30f6a1b4495
--- /dev/null
+++ b/keras/src/distribution/tensor_parallel/tensor_layout_test.py
@@ -0,0 +1,163 @@
+from keras.src import ops
+from keras.src import testing
+from keras.src.distribution.tensor_parallel.tensor_layout import LayoutMap
+from keras.src.distribution.tensor_parallel.tensor_layout import (
+    split_tensor_for_parallelism,
+)
+
+
+class LayoutTest(testing.TestCase):
+    """Test suite for tensor layout actions and mappings."""
+
+    def test_split_with_even_division(self):
+        """Tests splitting a tensor that divides evenly among workers."""
+        device_count = 4
+        dim = 0
+        tensor = ops.reshape(ops.arange(16, dtype="float32"), (8, 2))
+
+        expected_shard_0 = ops.array([[0.0, 1.0], [2.0, 3.0]])
+        expected_shard_2 = ops.array([[8.0, 9.0], [10.0, 11.0]])
+
+        shard_0 = split_tensor_for_parallelism(
+            tensor, index=0, device_count=device_count, dim=dim
+        )
+        shard_2 = split_tensor_for_parallelism(
+            tensor, index=2, device_count=device_count, dim=dim
+        )
+
+        self.assertAllClose(shard_0, expected_shard_0)
+        self.assertAllClose(shard_2, expected_shard_2)
+        self.assertEqual(shard_0.shape, (2, 2))
+
+    def test_split_with_uneven_division(self):
+        """Tests splitting tensor where remainder is distributed correctly."""
+        device_count = 3
+        dim = 0
+        tensor = ops.reshape(ops.arange(10, dtype="float32"), (10, 1))
+
+        shard_0 = split_tensor_for_parallelism(
+            tensor, index=0, device_count=device_count, dim=dim
+        )
+        self.assertEqual(shard_0.shape, (4, 1))
+        self.assertAllClose(shard_0, ops.array([[0.0], [1.0], [2.0], [3.0]]))
+
+        shard_1 = split_tensor_for_parallelism(
+            tensor, index=1, device_count=device_count, dim=dim
+        )
+        self.assertEqual(shard_1.shape, (3, 1))
+        self.assertAllClose(shard_1, ops.array([[4.0], [5.0], [6.0]]))
+
+        shard_2 = split_tensor_for_parallelism(
+            tensor, index=2, device_count=device_count, dim=dim
+        )
+        self.assertEqual(shard_2.shape, (3, 1))
+        self.assertAllClose(shard_2, ops.array([[7.0], [8.0], [9.0]]))
+
+    def test_split_and_undo_cycle_even_removed(self):
+        """
+        Confirms that the original tensor can be reconstructed.
+        """
+        device_count = 2
+        dim = 0
+        original_tensor = ops.reshape(ops.arange(12, dtype="float32"), (6, 2))
+
+        shards = [
+            split_tensor_for_parallelism(
+                original_tensor, index=i, device_count=device_count, dim=dim
+            )
+            for i in range(device_count)
+        ]
+
+        reconstructed_tensor = ops.concatenate(shards, axis=dim)
+
+        self.assertAllClose(original_tensor, reconstructed_tensor)
+
+    def test_split_and_undo_cycle_uneven_removed(self):
+        """
+        Confirms that original tensor can be reconstructed with uneven split.
+        """
+        device_count = 4
+        dim = 0
+        original_tensor = ops.reshape(ops.arange(22, dtype="float32"), (11, 2))
+
+        shards = [
+            split_tensor_for_parallelism(
+                original_tensor, index=i, device_count=device_count, dim=dim
+            )
+            for i in range(device_count)
+        ]
+
+        self.assertEqual(shards[0].shape, (3, 2))
+        self.assertEqual(shards[1].shape, (3, 2))
+        self.assertEqual(shards[2].shape, (3, 2))
+        self.assertEqual(shards[3].shape, (2, 2))
+
+        reconstructed_tensor = ops.concatenate(shards, axis=dim)
+        self.assertAllClose(original_tensor, reconstructed_tensor)
+
+    def test_split_last_dimension(self):
+        """Tests splitting on the last dimension using dim=-1."""
+        device_count = 3
+        dim = -1
+        original_tensor = ops.reshape(
+            ops.arange(30, dtype="float32"), (2, 5, 3)
+        )
+
+        shards = [
+            split_tensor_for_parallelism(
+                original_tensor, index=i, device_count=device_count, dim=dim
+            )
+            for i in range(device_count)
+        ]
+
+        self.assertEqual(shards[0].shape, (2, 5, 1))
+        self.assertEqual(shards[1].shape, (2, 5, 1))
+        self.assertEqual(shards[2].shape, (2, 5, 1))
+
+    def test_split_with_sharding_type_hint(self):
+        """Tests using 'row' and 'column' sharding hints for 2D tensors."""
+        device_count = 2
+        tensor = ops.reshape(ops.arange(16, dtype="float32"), (4, 4))
+
+        row_dim = 0
+        shard_row_0 = split_tensor_for_parallelism(
+            tensor, index=0, device_count=device_count, dim=row_dim
+        )
+        self.assertAllClose(shard_row_0, tensor[:2, :])
+
+        col_dim = 1
+        shard_col_0 = split_tensor_for_parallelism(
+            tensor, index=0, device_count=device_count, dim=col_dim
+        )
+        self.assertAllClose(shard_col_0, tensor[:, :2])
+
+    def test_layout_map_namedtuple_behavior(self):
+        """Tests basic behavior of the LayoutMap namedtuple."""
+
+        def rule_kernel(tensor, index):
+            return split_tensor_for_parallelism(
+                tensor, index=index, device_count=2, dim=0
+            )
+
+        def rule_output(tensor, index):
+            return split_tensor_for_parallelism(
+                tensor, index=index, device_count=2, dim=-1
+            )
+
+        state_rules = {"kernel": rule_kernel}
+        output_rules = {"output": rule_output}
+
+        layout_map = LayoutMap(
+            state_rules=state_rules, output_rules=output_rules
+        )
+
+        self.assertIs(layout_map.state_rules, state_rules)
+        self.assertIs(layout_map.output_rules, output_rules)
+
+        self.assertIs(layout_map[0], state_rules)
+        self.assertIs(layout_map[1], output_rules)
+
+        with self.assertRaises(AttributeError):
+            layout_map.state_rules = {}
+
+        self.assertTrue(callable(layout_map.state_rules["kernel"]))
\ No newline at end of file

From 41f80258302f813be32ef3b947203ba0c4f777cf Mon Sep 17 00:00:00 2001
From: Suhana <suhanaaa@google.com>
Date: Tue, 28 Oct 2025 10:08:30 +0530
Subject: [PATCH 02/12] formatting files

---
 keras/src/backend/jax/core.py                                | 2 +-
 keras/src/backend/jax/core_test.py                           | 2 +-
 keras/src/distribution/tensor_parallel/tensor_layout.py      | 2 +-
 keras/src/distribution/tensor_parallel/tensor_layout_test.py | 2 +-
 4 files changed, 4 insertions(+), 4 deletions(-)

diff --git a/keras/src/backend/jax/core.py b/keras/src/backend/jax/core.py
index aee30a3deadd..f55fd23e502d 100644
--- a/keras/src/backend/jax/core.py
+++ b/keras/src/backend/jax/core.py
@@ -627,4 +627,4 @@ def device_scope(device_name):
         )
     else:
         jax_device = device_name
-    return jax.default_device(jax_device)
\ No newline at end of file
+    return jax.default_device(jax_device)
diff --git a/keras/src/backend/jax/core_test.py b/keras/src/backend/jax/core_test.py
index 79eecad18063..2e7c312aa33e 100644
--- a/keras/src/backend/jax/core_test.py
+++ b/keras/src/backend/jax/core_test.py
@@ -143,4 +143,4 @@ def gather_fn(x):
         for i in range(num_devices):
             self.assertTrue(
                 np.allclose(result_array_on_devices[i], expected_gathered_data)
-            )
\ No newline at end of file
+            )
diff --git a/keras/src/distribution/tensor_parallel/tensor_layout.py b/keras/src/distribution/tensor_parallel/tensor_layout.py
index ff6b4eff920b..00f766434b34 100644
--- a/keras/src/distribution/tensor_parallel/tensor_layout.py
+++ b/keras/src/distribution/tensor_parallel/tensor_layout.py
@@ -40,4 +40,4 @@ def split_tensor_for_parallelism(tensor, index, device_count, dim):
     return splits[index]
 
 
-LayoutMap = collections.namedtuple("LayoutMap", ["state_rules", "output_rules"])
\ No newline at end of file
+LayoutMap = collections.namedtuple("LayoutMap", ["state_rules", "output_rules"])
diff --git a/keras/src/distribution/tensor_parallel/tensor_layout_test.py b/keras/src/distribution/tensor_parallel/tensor_layout_test.py
index d30f6a1b4495..7a8f3b61d8e4 100644
--- a/keras/src/distribution/tensor_parallel/tensor_layout_test.py
+++ b/keras/src/distribution/tensor_parallel/tensor_layout_test.py
@@ -160,4 +160,4 @@ def rule_output(tensor, index):
         with self.assertRaises(AttributeError):
             layout_map.state_rules = {}
 
-        self.assertTrue(callable(layout_map.state_rules["kernel"]))
\ No newline at end of file
+        self.assertTrue(callable(layout_map.state_rules["kernel"]))

From e74eab2a8a68b562f4ee65d01dcdba86446a35ee Mon Sep 17 00:00:00 2001
From: Suhana <suhanaaa@google.com>
Date: Tue, 28 Oct 2025 10:41:52 +0530
Subject: [PATCH 03/12] Updating the docstring

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
---
 keras/src/backend/jax/core.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/keras/src/backend/jax/core.py b/keras/src/backend/jax/core.py
index f55fd23e502d..d8d2db89135b 100644
--- a/keras/src/backend/jax/core.py
+++ b/keras/src/backend/jax/core.py
@@ -535,14 +535,14 @@ def all_reduce(x, op="sum", axis_name="model"):
     Performs an **all-reduce** operation across all replicas in the specified
     distribution axis.
 
-    The all-reduce operation computes a reduction (like sum, mean, or product)
+    The all-reduce operation computes a reduction (like sum or mean)
     of the input tensor `x` across all devices/replicas in the `axis_name`
     group, and then broadcasts the result back to all participating devices.
 
     Args:
         x: The tensor to reduce.
-        op: The reduction operation to perform. Common options include "sum",
-            "mean", or "product". Defaults to "sum".
+        op: The reduction operation to perform. Common options include "sum"
+            and "mean". Defaults to "sum".
         axis_name: The name of the distribution axis (e.g., "model",
             "data") over which to perform the reduction. Defaults to "model".
 

From 2cddf39134ad4acdc73deb483a202807bbc89c77 Mon Sep 17 00:00:00 2001
From: Suhana <suhanaaa@google.com>
Date: Tue, 28 Oct 2025 10:53:12 +0530
Subject: [PATCH 04/12] refactoring the code

---
 .../src/distribution/tensor_parallel/tensor_layout.py | 11 +----------
 1 file changed, 1 insertion(+), 10 deletions(-)

diff --git a/keras/src/distribution/tensor_parallel/tensor_layout.py b/keras/src/distribution/tensor_parallel/tensor_layout.py
index 00f766434b34..5635d7de2df6 100644
--- a/keras/src/distribution/tensor_parallel/tensor_layout.py
+++ b/keras/src/distribution/tensor_parallel/tensor_layout.py
@@ -21,16 +21,7 @@ def split_tensor_for_parallelism(tensor, index, device_count, dim):
         A tensor slice corresponding to the given `index`.
     """
     if dim == -1:
-        static_shape = getattr(tensor, "shape", None)
-        if static_shape is not None:
-            rank = len(static_shape)
-        else:
-            rank = None
-
-        if rank is not None:
-            split_dim = rank - 1
-        else:
-            split_dim = ops.ndim(tensor) - 1
+        split_dim = ops.ndim(tensor) - 1
     else:
         split_dim = dim
 

From 5365f1483f3932c6586f9a69baef586a67dfc3da Mon Sep 17 00:00:00 2001
From: Suhana <suhanaaa@google.com>
Date: Thu, 6 Nov 2025 13:45:42 +0530
Subject: [PATCH 05/12] fixing test

---
 .../src/distribution/tensor_parallel/tensor_layout_test.py  | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/keras/src/distribution/tensor_parallel/tensor_layout_test.py b/keras/src/distribution/tensor_parallel/tensor_layout_test.py
index 7a8f3b61d8e4..9ba09d904b34 100644
--- a/keras/src/distribution/tensor_parallel/tensor_layout_test.py
+++ b/keras/src/distribution/tensor_parallel/tensor_layout_test.py
@@ -96,9 +96,11 @@ def test_split_and_undo_cycle_uneven_removed(self):
         self.assertAllClose(original_tensor, reconstructed_tensor)
 
     def test_split_last_dimension(self):
-        """Tests splitting on the last dimension using dim=-1."""
+        """Tests splitting on the last dimension."""
         device_count = 3
-        dim = -1
+        # Change dim from -1 to 2 (the explicit index of the last dimension)
+        # to avoid backend-specific issues with dynamic shape resolution.
+        dim = 2
         original_tensor = ops.reshape(
             ops.arange(30, dtype="float32"), (2, 5, 3)
         )

From bc4d09461d0abcc85fb4c705e36fbd306277cd8f Mon Sep 17 00:00:00 2001
From: Suhana <suhanaaa@google.com>
Date: Thu, 6 Nov 2025 13:46:06 +0530
Subject: [PATCH 06/12] fixing test

---
 keras/src/distribution/tensor_parallel/tensor_layout_test.py | 2 --
 1 file changed, 2 deletions(-)

diff --git a/keras/src/distribution/tensor_parallel/tensor_layout_test.py b/keras/src/distribution/tensor_parallel/tensor_layout_test.py
index 9ba09d904b34..72b21b4912aa 100644
--- a/keras/src/distribution/tensor_parallel/tensor_layout_test.py
+++ b/keras/src/distribution/tensor_parallel/tensor_layout_test.py
@@ -98,8 +98,6 @@ def test_split_and_undo_cycle_uneven_removed(self):
     def test_split_last_dimension(self):
         """Tests splitting on the last dimension."""
         device_count = 3
-        # Change dim from -1 to 2 (the explicit index of the last dimension)
-        # to avoid backend-specific issues with dynamic shape resolution.
         dim = 2
         original_tensor = ops.reshape(
             ops.arange(30, dtype="float32"), (2, 5, 3)

From 4d32e49d2ae12939a5df975993812513fadc8373 Mon Sep 17 00:00:00 2001
From: Suhana <suhanaaa@google.com>
Date: Mon, 17 Nov 2025 10:48:33 +0530
Subject: [PATCH 07/12] adding autoconfig and coordinated_optimizer

---
 .../tensor_parallel/autoconfig.py             | 167 ++++++
 .../tensor_parallel/coordinated_optimizer.py  | 513 ++++++++++++++++++
 2 files changed, 680 insertions(+)
 create mode 100644 keras/src/distribution/tensor_parallel/autoconfig.py
 create mode 100644 keras/src/distribution/tensor_parallel/coordinated_optimizer.py

diff --git a/keras/src/distribution/tensor_parallel/autoconfig.py b/keras/src/distribution/tensor_parallel/autoconfig.py
new file mode 100644
index 000000000000..fd18feb99312
--- /dev/null
+++ b/keras/src/distribution/tensor_parallel/autoconfig.py
@@ -0,0 +1,167 @@
+from keras.src import layers
+from keras.src.distribution.tensor_parallel.tensor_layout import (
+    split_tensor_for_parallelism,
+    LayoutMap
+)
+
+_split_fn_internal = split_tensor_for_parallelism
+
+
+def _split_rule(device_count, dim):
+    """
+    Returns a sharding rule (lambda) that calls split_tensor_for_parallelism.
+    The lambda accepts (tensor, index) as expected by LayoutMap.
+    """
+    return lambda x, index: _split_fn_internal(x, index, device_count, dim=dim)
+
+
+def analyze_dense_layer(layer):
+    """Analyzes a Keras Dense layer to classify its sharding strategy."""
+    if not isinstance(layer, layers.Dense):
+        return 'dense'
+
+    input_dim = None
+    output_dim = None
+
+    if hasattr(layer, 'kernel') and layer.kernel is not None:
+        kernel_shape = layer.kernel.shape
+        if len(kernel_shape) == 2:
+            input_dim = kernel_shape[0]
+            output_dim = kernel_shape[1]
+
+    if input_dim is None or output_dim is None:
+        if hasattr(layer, 'units'):
+            output_dim = layer.units
+        else:
+            return 'dense'
+
+        if hasattr(layer, 'input_shape') and layer.input_shape and len(layer.input_shape) > 1:
+            input_dim = layer.input_shape[-1]
+        else:
+            return 'dense'
+
+    if not input_dim or not output_dim:
+        return 'dense'
+
+    expansion_threshold = 1.5
+    is_expansion = output_dim > input_dim * expansion_threshold
+    is_contraction = input_dim > output_dim * expansion_threshold
+
+    if is_expansion:
+        return 'up_projection'
+    elif is_contraction:
+        return 'down_projection'
+    else:
+        return 'dense'
+
+
+def _recursive_layer_traversal(
+    current_layer,
+    prefix,
+    device_count,
+    state_rules,
+    output_rules,
+    processed_layers,
+):
+    """Recursively traverses the model graph to apply sharding rules."""
+    
+    if id(current_layer) in processed_layers:
+        return
+    processed_layers.add(id(current_layer))
+
+    name = current_layer.name
+    full_name = f"{prefix}.{name}" if prefix else name
+    
+    if isinstance(current_layer, layers.Dense):
+        mlp_type = analyze_dense_layer(current_layer)
+
+        if mlp_type == 'up_projection':
+            state_rules[f"{full_name}.kernel"] = _split_rule(device_count, dim=1)
+            if current_layer.use_bias:
+                state_rules[f"{full_name}.bias"] = _split_rule(device_count, dim=0)
+            output_rules[f"{full_name}"] = {0: "gather"}
+
+        elif mlp_type == 'down_projection':
+            state_rules[f"{full_name}.kernel"] = _split_rule(device_count, dim=0)
+            output_rules[f"{full_name}"] = {0: "allreduce"}
+
+        else:
+            state_rules[f"{full_name}.kernel"] = _split_rule(device_count, dim=1)
+            if current_layer.use_bias:
+                state_rules[f"{full_name}.bias"] = _split_rule(device_count, dim=0)
+            output_rules[f"{full_name}"] = {0: "gather -1"}
+
+    elif isinstance(current_layer, layers.EinsumDense):
+        if "attention_output" in full_name:
+            state_rules[f"{full_name}.kernel"] = _split_rule(device_count, dim=0)
+            output_rules[f"{full_name}"] = {0: "allreduce"}
+        else:
+            state_rules[f"{full_name}.kernel"] = _split_rule(device_count, dim=1)
+            if hasattr(current_layer, 'bias') and current_layer.bias is not None:
+                state_rules[f"{full_name}.bias"] = _split_rule(device_count, dim=0)
+            output_rules[f"{full_name}"] = {0: "gather -1"}
+
+    elif isinstance(current_layer, (layers.Embedding,)):
+        weight_name = None 
+
+        if hasattr(current_layer, 'embeddings'):
+            weight_name = 'embeddings'
+        elif hasattr(current_layer, 'position_embeddings'):
+            weight_name = 'position_embeddings'
+
+        if weight_name:
+            state_rules[f"{full_name}.{weight_name}"] = _split_rule(device_count, dim=1)
+            output_rules[f"{full_name}"] = {0: "no_comm"}
+
+    elif isinstance(current_layer, (layers.LayerNormalization, layers.BatchNormalization, layers.GroupNormalization)):
+        pass
+
+    if hasattr(current_layer, 'layers') and current_layer.layers:
+        for sub_layer in current_layer.layers:
+            _recursive_layer_traversal(
+                sub_layer, full_name, device_count,
+                state_rules, output_rules, processed_layers
+            )
+
+    for attr_name in dir(current_layer):
+        if attr_name.startswith('__') and attr_name.endswith('__'):
+            continue
+        if hasattr(current_layer, attr_name):
+            attr = getattr(current_layer, attr_name)
+
+            if isinstance(attr, layers.Layer) and attr is not current_layer:
+                _recursive_layer_traversal(
+                    attr, full_name, device_count,
+                    state_rules, output_rules, processed_layers
+                )
+            elif isinstance(attr, (list, tuple)):
+                for item in attr:
+                    if isinstance(item, layers.Layer):
+                        _recursive_layer_traversal(
+                            item, full_name, device_count,
+                            state_rules, output_rules, processed_layers
+                        )
+
+
+def get_default_config_keras(module, device_ids):
+    """Generates a default tensor parallelism sharding configuration for a model."""
+    
+    device_count = len(device_ids)
+    state_rules = {}
+    output_rules = {}
+    
+    processed_layers = set() 
+
+    _recursive_layer_traversal(
+        current_layer=module,
+        prefix="",
+        device_count=device_count, 
+        state_rules=state_rules,
+        output_rules=output_rules,
+        processed_layers=processed_layers
+    )
+
+    return LayoutMap(
+        state_rules=state_rules,
+        output_rules=output_rules
+    )
\ No newline at end of file
diff --git a/keras/src/distribution/tensor_parallel/coordinated_optimizer.py b/keras/src/distribution/tensor_parallel/coordinated_optimizer.py
new file mode 100644
index 000000000000..9083eca583fc
--- /dev/null
+++ b/keras/src/distribution/tensor_parallel/coordinated_optimizer.py
@@ -0,0 +1,513 @@
+import re
+from typing import Any
+
+import numpy as np
+
+from keras.src import ops
+from keras.src import optimizers
+
+from keras.src.backend import distribution_lib
+
+
+class CoordinatedOptimizer:
+    """Manages an optimizer's state for distributed training.
+    This class is an internal coordinator that handles the complexities of
+    sharding optimizer states across multiple devices (shards) and
+    synchronizing gradients according to tensor parallelism rules.
+    ...
+    Args:
+        base_optimizer: The Keras optimizer instance.
+        device_count: The total number of devices/processes in the distributed
+            setup.
+        shard_optimizer_states: If `True`, the optimizer's state variables
+            will be partitioned across `device_count` devices. Defaults to `True`.
+        tensor_parallel_config: An optional configuration object that defines
+            rules for tensor parallelism. Defaults to `None`.
+    """
+
+    def __init__(
+        self,
+        base_optimizer: optimizers.Optimizer,
+        device_count: int,
+        shard_optimizer_states: bool = True,
+        tensor_parallel_config=None,
+    ):
+        self.base_optimizer = base_optimizer
+        self.device_count = device_count
+        self.shard_optimizer_states = shard_optimizer_states
+        self.tensor_parallel_config = tensor_parallel_config
+        self.sharded_states = {}
+        self._state_variable_to_parameter = {}
+        self._variables = None
+        self._variable_to_slot_name = {}
+
+    def _initialize_sharded_states(self):
+        """
+        Partitions the optimizer's state variables across shards by inspecting
+        the variables created by the base optimizer.
+
+        NOTE: Since the Keras BaseOptimizer does not expose a direct mapping 
+        from a model parameter to its optimizer state variables, this method 
+        infers the mapping by string parsing their paths/names. This addresses
+        the collaborator's request for clarity on the path-matching logic.
+        """
+        if not self.shard_optimizer_states or not self.base_optimizer.built:
+            return
+
+        self.sharded_states = {}
+        self._state_variable_to_parameter = {}
+        self._variable_to_slot_name = {}
+        opt_name = self.base_optimizer.name
+
+        normalized_params = sorted(
+            [(p.path.replace("/", "_"), p) for p in self._variables],
+            key=lambda x: len(x[0]),
+            reverse=True,
+        )
+
+        for state_var in self.base_optimizer.variables:
+            if state_var is self.base_optimizer.iterations:
+                continue
+
+            path_parts = state_var.path.split("/")
+            if len(path_parts) != 2 or path_parts[0] != opt_name:
+                continue
+
+            state_suffix = path_parts[1]
+
+            found_param = None
+            slot_name = None
+            
+            for norm_param_path, param in normalized_params:
+                if state_suffix.startswith(norm_param_path):
+                    found_param = param
+                    slot_suffix = state_suffix[len(norm_param_path) :]
+                    slot_name = slot_suffix.strip("_")
+                    break
+
+            if found_param is not None and slot_name is not None:
+                self._state_variable_to_parameter[state_var.path] = found_param
+                self._variable_to_slot_name[state_var.path] = slot_name
+
+                sharding_dim = 0
+                if self.tensor_parallel_config:
+                    norm_param_name = found_param.path.replace("/", ".")
+                    for (
+                        p,
+                        a,
+                    ) in self.tensor_parallel_config.state_rules.items():
+                        if re.search(p, norm_param_name) and hasattr(a, "dim"):
+                            sharding_dim = a.dim
+                            break
+
+                partitioned_state = self._partition_state(
+                    state_var, dim=sharding_dim
+                )
+                self.sharded_states.setdefault(slot_name, {})[
+                    found_param.path
+                ] = partitioned_state
+
+        if self.base_optimizer.iterations is not None:
+            self.sharded_states["iterations"] = self._partition_state(
+                self.base_optimizer.iterations, dim=0
+            )
+
+    def _partition_state(
+        self, state_variable: Any, dim: int
+    ) -> list[np.ndarray]:
+        """Splits a single state variable numpy array into chunks."""
+        state_array = ops.convert_to_numpy(state_variable)
+        if (
+            state_array.ndim > dim
+            and state_array.shape[dim] >= self.device_count
+        ):
+            return np.array_split(state_array, self.device_count, axis=dim)
+        else:
+            return [np.copy(state_array) for _ in range(self.device_count)]
+
+    def apply_gradients(
+        self, gradients_and_vars: list[list[tuple]], shard_models: list
+    ):
+        """Coordinates gradient synchronization and application."""
+        if len(gradients_and_vars) != self.device_count:
+            raise ValueError(
+                f"Expected {self.device_count} sets of gradients, "
+                f"but received {len(gradients_and_vars)}."
+            )
+
+        synchronized_gradients = self._synchronize_gradients(gradients_and_vars)
+
+        if self.shard_optimizer_states:
+            self._apply_gradients_with_sharded_states(
+                synchronized_gradients, shard_models
+            )
+        else:
+            self._apply_gradients_with_replicated_states(
+                synchronized_gradients, shard_models
+            )
+
+    def _apply_gradients_with_replicated_states(
+        self, synchronized_gradients: list[list[tuple]], shard_models: list
+    ):
+        """Averages gradients across all shards and applies them once."""
+        num_vars = len(synchronized_gradients[0])
+        averaged_grads_and_vars = []
+
+        for i in range(num_vars):
+            variable = synchronized_gradients[0][i][1]
+            grads_for_var = [
+                shard_grads[i][0]
+                for shard_grads in synchronized_gradients
+                if shard_grads[i][0] is not None
+            ]
+
+            if not grads_for_var:
+                continue
+
+            if len(grads_for_var) > 1:
+                stacked_grads = ops.stack(grads_for_var, axis=0)
+                averaged_grad = ops.mean(stacked_grads, axis=0)
+            else:
+                averaged_grad = grads_for_var[0]
+
+            averaged_grads_and_vars.append((averaged_grad, variable))
+
+        if averaged_grads_and_vars:
+            self.base_optimizer.apply_gradients(averaged_grads_and_vars)
+
+    def _apply_gradients_with_sharded_states(
+        self, synchronized_gradients: list[list[tuple]], shard_models: list
+    ):
+        """Applies gradients to each shard using its local optimizer state."""
+        for shard_idx in range(self.device_count):
+            local_states = self._get_local_optimizer_states(shard_idx)
+            # Access the base optimizer inside the TensorParallelOptimizer wrapper
+            shard_optimizer = shard_models[shard_idx].optimizer.base_optimizer 
+
+            self._update_optimizer_internal_state(
+                shard_optimizer, local_states
+            )
+
+            shard_grads_and_vars = synchronized_gradients[shard_idx]
+            shard_optimizer.apply_gradients(shard_grads_and_vars)
+
+            self._update_global_sharded_states(shard_optimizer, shard_idx)
+
+    def _get_local_optimizer_states(self, shard_idx: int) -> dict[str, Any]:
+        """Constructs the state dictionary for a single shard."""
+        local_states = {}
+        for state_name, state_value in self.sharded_states.items():
+            if isinstance(state_value, dict):
+                local_states[state_name] = {}
+                for param_name, param_states in state_value.items():
+                    local_states[state_name][param_name] = param_states[
+                        shard_idx
+                    ]
+            else:
+                local_states[state_name] = state_value[shard_idx]
+        return local_states
+
+    def _update_optimizer_internal_state(self, optimizer, local_states: dict):
+        """Assigns local sharded state values to the optimizer's variables."""
+        if not optimizer.built:
+            return
+
+        for var in optimizer.variables:
+            if var is optimizer.iterations:
+                if "iterations" in local_states:
+                    var.assign(local_states["iterations"])
+                continue
+
+            param = self._state_variable_to_parameter.get(var.path, None)
+            slot_name = self._variable_to_slot_name.get(var.path)
+
+            if (
+                param
+                and slot_name
+                and slot_name in local_states
+                and param.path in local_states[slot_name]
+            ):
+                local_param_state = local_states[slot_name][param.path]
+                if var.shape == local_param_state.shape:
+                    var.assign(local_param_state)
+
+    def _update_global_sharded_states(self, optimizer, shard_idx: int):
+        """Updates the main sharded_states dictionary after a gradient step."""
+        if not optimizer.built:
+            return
+
+        for var in optimizer.variables:
+            if var is optimizer.iterations:
+                self.sharded_states["iterations"][shard_idx] = (
+                    ops.convert_to_numpy(var)
+                )
+                continue
+
+            param = self._state_variable_to_parameter.get(var.path, None)
+            slot_name = self._variable_to_slot_name.get(var.path)
+
+            if (
+                param
+                and slot_name
+                and slot_name in self.sharded_states
+                and param.path in self.sharded_states[slot_name]
+            ):
+                self.sharded_states[slot_name][param.path][shard_idx] = (
+                    ops.convert_to_numpy(var)
+                )
+
+    def _synchronize_gradients(
+        self, gradients_and_vars: list[list[tuple]]
+    ) -> list[list[tuple]]:
+        """Synchronizes gradients across shards based on tensor parallel rules."""
+        if not self.tensor_parallel_config:
+            return gradients_and_vars
+
+        rules = self.tensor_parallel_config.state_rules.items()
+        column_parallel_patterns = {
+            pattern
+            for pattern, action in rules
+            if hasattr(action, "sharding_type")
+            and action.sharding_type == "column"
+        }
+
+        if not column_parallel_patterns:
+            return gradients_and_vars
+
+        num_weights = len(gradients_and_vars[0])
+        for i in range(num_weights):
+            variable = gradients_and_vars[0][i][1]
+            var_name = getattr(variable, "path", getattr(variable, "name", ""))
+
+            if any(
+                re.search(pattern, var_name)
+                for pattern in column_parallel_patterns
+            ):
+                grads_to_reduce = [
+                    g_and_v[i][0]
+                    for g_and_v in gradients_and_vars
+                    if g_and_v[i][0] is not None
+                ]
+                if grads_to_reduce:
+                    synced_grad = self._allreduce_gradients(grads_to_reduce)[0] 
+                    for shard_idx in range(self.device_count):
+                        if gradients_and_vars[shard_idx][i][0] is not None:
+                            gradients_and_vars[shard_idx][i] = (
+                                synced_grad,
+                                variable,
+                            )
+        return gradients_and_vars
+
+    def _allreduce_gradients(self, gradients: list[Any]) -> list[Any]:
+        """Performs a mean all-reduce operation on a list of gradients.
+
+        This method uses the on-device communication primitive from the backend
+        (e.g., JAX's lax.pmean) when multiple devices are detected, resolving
+        the critical performance issue related to CPU transfers.
+        """
+        if not gradients:
+            return []
+
+        if distribution_lib.get_device_count() > 1:
+            local_grad = gradients[0]
+            synced_tensor = distribution_lib.all_reduce(
+                local_grad, op="mean", axis_name="model"
+            )
+
+            return [synced_tensor for _ in range(self.device_count)]
+
+        if len(gradients) == 1:
+            mean_grad = ops.convert_to_tensor(gradients[0])
+        else:
+            stacked_grads = ops.stack(
+                [ops.convert_to_tensor(g) for g in gradients], axis=0
+            )
+            mean_grad = ops.mean(stacked_grads, axis=0)
+
+        return [mean_grad for _ in range(len(gradients))]
+
+    def get_weights(self) -> list[np.ndarray]:
+        """Returns the weights of the base optimizer."""
+        return [
+            ops.convert_to_numpy(var) for var in self.base_optimizer.variables
+        ]
+
+    def set_weights(self, weights: list[np.ndarray]):
+        """Sets the weights of the base optimizer."""
+        self.base_optimizer.set_weights(weights)
+
+    def enable_optimizer_state_sharding(self, variables: list):
+        """Enables and initializes optimizer state sharding."""
+        self.shard_optimizer_states = True
+        self._variables = variables
+        self._initialize_sharded_states()
+
+
+class TensorParallelOptimizer(optimizers.Optimizer):
+    """A Keras Optimizer wrapper for tensor-parallel distributed training.
+
+    This class serves as the public Keras-compliant interface (inherits 
+    `optimizers.Optimizer`). It delegates the complex tasks of state 
+    management, gradient synchronization, and sharding to the internal 
+    `CoordinatedOptimizer` instance. This separation adheres to the 
+    principle of keeping the public API clean while encapsulating complex 
+    distribution logic.
+    
+    Args:
+        base_optimizer: A Keras optimizer instance or a string identifier.
+        device_count: The total number of devices/processes in the distributed
+            setup.
+        tensor_parallel_config: An optional configuration object. Defaults to `None`.
+    """
+
+    def __init__(
+        self,
+        base_optimizer: optimizers.Optimizer,
+        device_count: int,
+        tensor_parallel_config=None,
+    ):
+        if isinstance(base_optimizer, str):
+            base_optimizer_instance = optimizers.get(base_optimizer)
+        else:
+            base_optimizer_instance = base_optimizer
+
+        learning_rate = base_optimizer_instance.learning_rate
+        if callable(learning_rate):
+            lr_value = float(ops.convert_to_numpy(learning_rate(0)))
+        else:
+            lr_value = float(ops.convert_to_numpy(learning_rate))
+
+        super().__init__(
+            learning_rate=lr_value,
+            name=f"TensorParallel_{base_optimizer_instance.name}",
+        )
+
+        self.base_optimizer = base_optimizer_instance
+        self.device_count = device_count
+        self.coordinated_optimizer = CoordinatedOptimizer(
+            self.base_optimizer,
+            device_count,
+            tensor_parallel_config=tensor_parallel_config,
+        )
+
+    def apply_gradients(self, grads_and_vars: list, **kwargs):
+        """Applies gradients to the model variables."""
+        is_sharded_grads = (
+            isinstance(grads_and_vars, list)
+            and grads_and_vars
+            and isinstance(grads_and_vars[0], list)
+        )
+        if is_sharded_grads:
+            if "shard_models" not in kwargs:
+                raise ValueError(
+                    "The `shard_models` keyword argument is required when "
+                    "applying sharded gradients (a list of lists)."
+                )
+            shard_models = kwargs.get("shard_models")
+            self.coordinated_optimizer.apply_gradients(
+                grads_and_vars, shard_models
+            )
+        else:
+            self.base_optimizer.apply_gradients(grads_and_vars)
+
+    def get_config(self) -> dict[str, Any]:
+        from keras.src import saving
+
+        config = super().get_config()
+        config.pop("learning_rate", None)
+        config.pop("name", None)
+
+        config.update(
+            {
+                "base_optimizer": saving.serialize_keras_object(
+                    self.base_optimizer
+                ),
+                "device_count": self.device_count,
+                "tensor_parallel_config": self.coordinated_optimizer.tensor_parallel_config,
+            }
+        )
+        return config
+
+    def update_step(self, gradient, variable, *args, **kwargs):
+        """Delegates the update step to the base optimizer."""
+        if hasattr(self.base_optimizer, "update_step"):
+            try:
+                return self.base_optimizer.update_step(
+                    gradient, variable, *args, **kwargs
+                )
+            except TypeError:
+                return self.base_optimizer.update_step(gradient, variable)
+
+        try:
+            return super().update_step(gradient, variable, *args, **kwargs)
+        except TypeError:
+            return super().update_step(gradient, variable)
+
+    @classmethod
+    def from_config(cls, config: dict[str, Any]) -> "TensorParallelOptimizer":
+        from keras.src import saving
+
+        base_optimizer_config = config.pop("base_optimizer")
+        base_optimizer = saving.deserialize_keras_object(base_optimizer_config)
+
+        init_kwargs = {
+            "device_count": config.get("device_count"), 
+            "tensor_parallel_config": config.get("tensor_parallel_config"),
+        }
+
+        config.pop("device_count", None) 
+        config.pop("tensor_parallel_config", None) 
+        
+        return cls(base_optimizer=base_optimizer, **init_kwargs)
+
+    def build(self, variables: list):
+        """Builds the optimizer and initializes sharded states."""
+        if self.built:
+            return
+
+        self.base_optimizer.build(variables)
+        if variables:
+            iterations = self.base_optimizer.iterations
+            original_iterations_val = None
+            if iterations is not None:
+                original_iterations_val = ops.convert_to_numpy(
+                    iterations.value
+                )
+
+            zero_grads = [ops.zeros_like(v) for v in variables]
+            self.base_optimizer.apply_gradients(zip(zero_grads, variables))
+
+            if iterations is not None and original_iterations_val is not None:
+                iterations.assign(original_iterations_val)
+
+        self.coordinated_optimizer.enable_optimizer_state_sharding(variables)
+        super().build(variables)
+
+    def get_weights(self) -> list[np.ndarray]:
+        """Returns the weights of the base optimizer."""
+        return self.coordinated_optimizer.get_weights()
+
+    def set_weights(self, weights: list[np.ndarray]):
+        """Sets the weights of the base optimizer."""
+        self.coordinated_optimizer.set_weights(weights)
+
+    @property
+    def variables(self) -> list:
+        """Returns the list of variables from the base optimizer."""
+        return self.base_optimizer.variables
+
+    @property
+    def learning_rate(self) -> Any:
+        """Provides access to the learning rate of the base optimizer."""
+        return self.base_optimizer.learning_rate
+
+    @learning_rate.setter
+    def learning_rate(self, value):
+        self.base_optimizer.learning_rate = value
+
+    @property
+    def iterations(self):
+        """
+        Returns the training iteration count directly from the base optimizer.
+        """
+        return self.base_optimizer.iterations
\ No newline at end of file

From 119ac154e5efb7bd86ec29c15f22c6597ea5753b Mon Sep 17 00:00:00 2001
From: Suhana <suhanaaa@google.com>
Date: Mon, 17 Nov 2025 10:58:40 +0530
Subject: [PATCH 08/12] updating docstrings and code format

---
 .../tensor_parallel/autoconfig.py             |  68 ++++-
 .../tensor_parallel/coordinated_optimizer.py  | 252 ++++++++++++------
 2 files changed, 235 insertions(+), 85 deletions(-)

diff --git a/keras/src/distribution/tensor_parallel/autoconfig.py b/keras/src/distribution/tensor_parallel/autoconfig.py
index fd18feb99312..d1a24b8eec22 100644
--- a/keras/src/distribution/tensor_parallel/autoconfig.py
+++ b/keras/src/distribution/tensor_parallel/autoconfig.py
@@ -9,14 +9,40 @@
 
 def _split_rule(device_count, dim):
     """
-    Returns a sharding rule (lambda) that calls split_tensor_for_parallelism.
-    The lambda accepts (tensor, index) as expected by LayoutMap.
+    Creates a sharding rule for a specific dimension.
+
+    Returns a lambda function compatible with LayoutMap that defines
+    how a tensor should be split across the available devices.
+
+    Args:
+        device_count (int): The total number of devices available for parallelism.
+        dim (int): The dimension of the tensor to split.
+
+    Returns:
+        callable: A lambda function accepting (tensor, index) that returns the
+        sharded layout.
     """
     return lambda x, index: _split_fn_internal(x, index, device_count, dim=dim)
 
 
 def analyze_dense_layer(layer):
-    """Analyzes a Keras Dense layer to classify its sharding strategy."""
+    """
+    Classifies a Dense layer based on its input/output dimensions.
+
+    This function determines if a Dense layer represents an 'up_projection'
+    (expansion) or a 'down_projection' (contraction) based on a heuristic
+    threshold. This classification dictates how the weights are sharded.
+
+    Heuristic:
+        - Expansion: Output dimension > (Input dimension * 1.5)
+        - Contraction: Input dimension > (Output dimension * 1.5)
+
+    Args:
+        layer (keras.layers.Layer): The layer instance to analyze.
+
+    Returns:
+        str: One of 'up_projection', 'down_projection', or 'dense'.
+    """
     if not isinstance(layer, layers.Dense):
         return 'dense'
 
@@ -63,15 +89,28 @@ def _recursive_layer_traversal(
     output_rules,
     processed_layers,
 ):
-    """Recursively traverses the model graph to apply sharding rules."""
-    
+    """
+    Traverses the model graph recursively to apply sharding rules.
+
+    This function visits layers, checks their type, and populates the
+    state_rules (weights) and output_rules (activations) dictionaries
+    required for Tensor Parallelism.
+
+    Args:
+        current_layer (keras.layers.Layer): The current layer being visited.
+        prefix (str): The naming prefix for the current layer (used for nested models).
+        device_count (int): Total number of devices.
+        state_rules (dict): The dictionary accumulating variable sharding rules.
+        output_rules (dict): The dictionary accumulating output layout rules.
+        processed_layers (set): A set of object IDs to prevent infinite recursion on cycles.
+    """
     if id(current_layer) in processed_layers:
         return
     processed_layers.add(id(current_layer))
 
     name = current_layer.name
     full_name = f"{prefix}.{name}" if prefix else name
-    
+
     if isinstance(current_layer, layers.Dense):
         mlp_type = analyze_dense_layer(current_layer)
 
@@ -144,8 +183,21 @@ def _recursive_layer_traversal(
 
 
 def get_default_config_keras(module, device_ids):
-    """Generates a default tensor parallelism sharding configuration for a model."""
-    
+    """
+    Generates a default tensor parallelism configuration for a model.
+
+    This function inspects the model structure and automatically generates
+    a `LayoutMap` containing sharding rules for weights (kernels/biases) and
+    outputs (activations).
+
+    Args:
+        module (keras.Model or keras.layers.Layer): The Keras model or layer to config.
+        device_ids (list): A list of device identifiers (e.g., strings or Mesh IDs).
+
+    Returns:
+        keras.src.distribution.tensor_parallel.tensor_layout.LayoutMap:
+        The configuration map applied to the model distribution API.
+    """
     device_count = len(device_ids)
     state_rules = {}
     output_rules = {}
diff --git a/keras/src/distribution/tensor_parallel/coordinated_optimizer.py b/keras/src/distribution/tensor_parallel/coordinated_optimizer.py
index 9083eca583fc..12a69f2cb3b1 100644
--- a/keras/src/distribution/tensor_parallel/coordinated_optimizer.py
+++ b/keras/src/distribution/tensor_parallel/coordinated_optimizer.py
@@ -1,35 +1,34 @@
 import re
-from typing import Any
-
 import numpy as np
 
 from keras.src import ops
 from keras.src import optimizers
-
 from keras.src.backend import distribution_lib
 
 
 class CoordinatedOptimizer:
     """Manages an optimizer's state for distributed training.
+
     This class is an internal coordinator that handles the complexities of
     sharding optimizer states across multiple devices (shards) and
     synchronizing gradients according to tensor parallelism rules.
-    ...
+
     Args:
-        base_optimizer: The Keras optimizer instance.
-        device_count: The total number of devices/processes in the distributed
-            setup.
-        shard_optimizer_states: If `True`, the optimizer's state variables
-            will be partitioned across `device_count` devices. Defaults to `True`.
-        tensor_parallel_config: An optional configuration object that defines
-            rules for tensor parallelism. Defaults to `None`.
+        base_optimizer (Optimizer): The Keras optimizer instance.
+        device_count (int): The total number of devices/processes in the
+            distributed setup.
+        shard_optimizer_states (bool): If `True`, the optimizer's state
+            variables will be partitioned across `device_count` devices.
+            Defaults to `True`.
+        tensor_parallel_config (object): An optional configuration object that
+            defines rules for tensor parallelism. Defaults to `None`.
     """
 
     def __init__(
         self,
-        base_optimizer: optimizers.Optimizer,
-        device_count: int,
-        shard_optimizer_states: bool = True,
+        base_optimizer,
+        device_count,
+        shard_optimizer_states=True,
         tensor_parallel_config=None,
     ):
         self.base_optimizer = base_optimizer
@@ -43,13 +42,17 @@ def __init__(
 
     def _initialize_sharded_states(self):
         """
-        Partitions the optimizer's state variables across shards by inspecting
-        the variables created by the base optimizer.
+        Partitions the optimizer's state variables across shards.
 
-        NOTE: Since the Keras BaseOptimizer does not expose a direct mapping 
-        from a model parameter to its optimizer state variables, this method 
-        infers the mapping by string parsing their paths/names. This addresses
-        the collaborator's request for clarity on the path-matching logic.
+        This method inspects the variables created by the base optimizer and
+        maps them to model parameters.
+
+        
+
+        Note:
+            Since the Keras BaseOptimizer does not expose a direct mapping
+            from a model parameter to its optimizer state variables, this
+            method infers the mapping by string parsing their paths/names.
         """
         if not self.shard_optimizer_states or not self.base_optimizer.built:
             return
@@ -112,10 +115,17 @@ def _initialize_sharded_states(self):
                 self.base_optimizer.iterations, dim=0
             )
 
-    def _partition_state(
-        self, state_variable: Any, dim: int
-    ) -> list[np.ndarray]:
-        """Splits a single state variable numpy array into chunks."""
+    def _partition_state(self, state_variable, dim):
+        """
+        Splits a single state variable numpy array into chunks.
+
+        Args:
+            state_variable (array-like): The state variable to split.
+            dim (int): The dimension along which to split the variable.
+
+        Returns:
+            list: A list of numpy arrays representing the split state.
+        """
         state_array = ops.convert_to_numpy(state_variable)
         if (
             state_array.ndim > dim
@@ -125,10 +135,20 @@ def _partition_state(
         else:
             return [np.copy(state_array) for _ in range(self.device_count)]
 
-    def apply_gradients(
-        self, gradients_and_vars: list[list[tuple]], shard_models: list
-    ):
-        """Coordinates gradient synchronization and application."""
+    def apply_gradients(self, gradients_and_vars, shard_models):
+        """
+        Coordinates gradient synchronization and application.
+
+        Args:
+            gradients_and_vars (list): A list containing lists of (gradient,
+                variable) tuples for each device.
+            shard_models (list): A list of model shards corresponding to the
+                devices.
+
+        Raises:
+            ValueError: If the number of gradient sets does not match the
+                device count.
+        """
         if len(gradients_and_vars) != self.device_count:
             raise ValueError(
                 f"Expected {self.device_count} sets of gradients, "
@@ -147,9 +167,17 @@ def apply_gradients(
             )
 
     def _apply_gradients_with_replicated_states(
-        self, synchronized_gradients: list[list[tuple]], shard_models: list
+        self, synchronized_gradients, shard_models
     ):
-        """Averages gradients across all shards and applies them once."""
+        """
+        Averages gradients across all shards and applies them once.
+
+        This is used when `shard_optimizer_states` is False.
+
+        Args:
+            synchronized_gradients (list): The list of synchronized gradients.
+            shard_models (list): The list of model shards.
+        """
         num_vars = len(synchronized_gradients[0])
         averaged_grads_and_vars = []
 
@@ -176,9 +204,15 @@ def _apply_gradients_with_replicated_states(
             self.base_optimizer.apply_gradients(averaged_grads_and_vars)
 
     def _apply_gradients_with_sharded_states(
-        self, synchronized_gradients: list[list[tuple]], shard_models: list
+        self, synchronized_gradients, shard_models
     ):
-        """Applies gradients to each shard using its local optimizer state."""
+        """
+        Applies gradients to each shard using its local optimizer state.
+
+        Args:
+            synchronized_gradients (list): The list of synchronized gradients.
+            shard_models (list): The list of model shards.
+        """
         for shard_idx in range(self.device_count):
             local_states = self._get_local_optimizer_states(shard_idx)
             # Access the base optimizer inside the TensorParallelOptimizer wrapper
@@ -193,8 +227,16 @@ def _apply_gradients_with_sharded_states(
 
             self._update_global_sharded_states(shard_optimizer, shard_idx)
 
-    def _get_local_optimizer_states(self, shard_idx: int) -> dict[str, Any]:
-        """Constructs the state dictionary for a single shard."""
+    def _get_local_optimizer_states(self, shard_idx):
+        """
+        Constructs the state dictionary for a single shard.
+
+        Args:
+            shard_idx (int): The index of the current shard.
+
+        Returns:
+            dict: A dictionary mapping state names to their local values.
+        """
         local_states = {}
         for state_name, state_value in self.sharded_states.items():
             if isinstance(state_value, dict):
@@ -207,8 +249,14 @@ def _get_local_optimizer_states(self, shard_idx: int) -> dict[str, Any]:
                 local_states[state_name] = state_value[shard_idx]
         return local_states
 
-    def _update_optimizer_internal_state(self, optimizer, local_states: dict):
-        """Assigns local sharded state values to the optimizer's variables."""
+    def _update_optimizer_internal_state(self, optimizer, local_states):
+        """
+        Assigns local sharded state values to the optimizer's variables.
+
+        Args:
+            optimizer (Optimizer): The local optimizer instance for the shard.
+            local_states (dict): The local state dictionary.
+        """
         if not optimizer.built:
             return
 
@@ -231,8 +279,14 @@ def _update_optimizer_internal_state(self, optimizer, local_states: dict):
                 if var.shape == local_param_state.shape:
                     var.assign(local_param_state)
 
-    def _update_global_sharded_states(self, optimizer, shard_idx: int):
-        """Updates the main sharded_states dictionary after a gradient step."""
+    def _update_global_sharded_states(self, optimizer, shard_idx):
+        """
+        Updates the main sharded_states dictionary after a gradient step.
+
+        Args:
+            optimizer (Optimizer): The local optimizer instance.
+            shard_idx (int): The index of the current shard.
+        """
         if not optimizer.built:
             return
 
@@ -256,10 +310,18 @@ def _update_global_sharded_states(self, optimizer, shard_idx: int):
                     ops.convert_to_numpy(var)
                 )
 
-    def _synchronize_gradients(
-        self, gradients_and_vars: list[list[tuple]]
-    ) -> list[list[tuple]]:
-        """Synchronizes gradients across shards based on tensor parallel rules."""
+    def _synchronize_gradients(self, gradients_and_vars):
+        """
+        Synchronizes gradients across shards based on tensor parallel rules.
+
+        
+
+        Args:
+            gradients_and_vars (list): A list of (gradient, variable) tuples.
+
+        Returns:
+            list: The synchronized list of gradients and variables.
+        """
         if not self.tensor_parallel_config:
             return gradients_and_vars
 
@@ -298,12 +360,19 @@ def _synchronize_gradients(
                             )
         return gradients_and_vars
 
-    def _allreduce_gradients(self, gradients: list[Any]) -> list[Any]:
-        """Performs a mean all-reduce operation on a list of gradients.
+    def _allreduce_gradients(self, gradients):
+        """
+        Performs a mean all-reduce operation on a list of gradients.
 
         This method uses the on-device communication primitive from the backend
-        (e.g., JAX's lax.pmean) when multiple devices are detected, resolving
-        the critical performance issue related to CPU transfers.
+        (e.g., JAX's lax.pmean) when multiple devices are detected.
+
+        Args:
+            gradients (list): A list of gradient tensors to reduce.
+
+        Returns:
+            list: A list containing the reduced gradient repeated for each
+            device.
         """
         if not gradients:
             return []
@@ -326,18 +395,23 @@ def _allreduce_gradients(self, gradients: list[Any]) -> list[Any]:
 
         return [mean_grad for _ in range(len(gradients))]
 
-    def get_weights(self) -> list[np.ndarray]:
+    def get_weights(self):
         """Returns the weights of the base optimizer."""
         return [
             ops.convert_to_numpy(var) for var in self.base_optimizer.variables
         ]
 
-    def set_weights(self, weights: list[np.ndarray]):
+    def set_weights(self, weights):
         """Sets the weights of the base optimizer."""
         self.base_optimizer.set_weights(weights)
 
-    def enable_optimizer_state_sharding(self, variables: list):
-        """Enables and initializes optimizer state sharding."""
+    def enable_optimizer_state_sharding(self, variables):
+        """
+        Enables and initializes optimizer state sharding.
+
+        Args:
+            variables (list): A list of model variables to track.
+        """
         self.shard_optimizer_states = True
         self._variables = variables
         self._initialize_sharded_states()
@@ -346,24 +420,24 @@ def enable_optimizer_state_sharding(self, variables: list):
 class TensorParallelOptimizer(optimizers.Optimizer):
     """A Keras Optimizer wrapper for tensor-parallel distributed training.
 
-    This class serves as the public Keras-compliant interface (inherits 
-    `optimizers.Optimizer`). It delegates the complex tasks of state 
-    management, gradient synchronization, and sharding to the internal 
-    `CoordinatedOptimizer` instance. This separation adheres to the 
-    principle of keeping the public API clean while encapsulating complex 
-    distribution logic.
-    
+    This class serves as the public Keras-compliant interface (inherits
+    `optimizers.Optimizer`). It delegates the complex tasks of state
+    management, gradient synchronization, and sharding to the internal
+    `CoordinatedOptimizer` instance.
+
     Args:
-        base_optimizer: A Keras optimizer instance or a string identifier.
-        device_count: The total number of devices/processes in the distributed
-            setup.
-        tensor_parallel_config: An optional configuration object. Defaults to `None`.
+        base_optimizer (Optimizer or str): A Keras optimizer instance or a
+            string identifier.
+        device_count (int): The total number of devices/processes in the
+            distributed setup.
+        tensor_parallel_config (object): An optional configuration object.
+            Defaults to `None`.
     """
 
     def __init__(
         self,
-        base_optimizer: optimizers.Optimizer,
-        device_count: int,
+        base_optimizer,
+        device_count,
         tensor_parallel_config=None,
     ):
         if isinstance(base_optimizer, str):
@@ -390,8 +464,19 @@ def __init__(
             tensor_parallel_config=tensor_parallel_config,
         )
 
-    def apply_gradients(self, grads_and_vars: list, **kwargs):
-        """Applies gradients to the model variables."""
+    def apply_gradients(self, grads_and_vars, **kwargs):
+        """
+        Applies gradients to the model variables.
+
+        Args:
+            grads_and_vars (list): A list of (gradient, variable) tuples or a
+                list of lists for sharded execution.
+            **kwargs: Additional arguments, such as `shard_models`.
+        
+        Raises:
+            ValueError: If `shard_models` is missing when applying sharded
+                gradients.
+        """
         is_sharded_grads = (
             isinstance(grads_and_vars, list)
             and grads_and_vars
@@ -410,7 +495,8 @@ def apply_gradients(self, grads_and_vars: list, **kwargs):
         else:
             self.base_optimizer.apply_gradients(grads_and_vars)
 
-    def get_config(self) -> dict[str, Any]:
+    def get_config(self):
+        """Returns the optimizer configuration as a dictionary."""
         from keras.src import saving
 
         config = super().get_config()
@@ -429,7 +515,15 @@ def get_config(self) -> dict[str, Any]:
         return config
 
     def update_step(self, gradient, variable, *args, **kwargs):
-        """Delegates the update step to the base optimizer."""
+        """
+        Delegates the update step to the base optimizer.
+
+        Args:
+            gradient (Tensor): The gradient tensor.
+            variable (Variable): The variable to update.
+            *args: Additional arguments for the update.
+            **kwargs: Additional keyword arguments for the update.
+        """
         if hasattr(self.base_optimizer, "update_step"):
             try:
                 return self.base_optimizer.update_step(
@@ -444,7 +538,8 @@ def update_step(self, gradient, variable, *args, **kwargs):
             return super().update_step(gradient, variable)
 
     @classmethod
-    def from_config(cls, config: dict[str, Any]) -> "TensorParallelOptimizer":
+    def from_config(cls, config):
+        """Creates an optimizer instance from its configuration."""
         from keras.src import saving
 
         base_optimizer_config = config.pop("base_optimizer")
@@ -460,8 +555,13 @@ def from_config(cls, config: dict[str, Any]) -> "TensorParallelOptimizer":
         
         return cls(base_optimizer=base_optimizer, **init_kwargs)
 
-    def build(self, variables: list):
-        """Builds the optimizer and initializes sharded states."""
+    def build(self, variables):
+        """
+        Builds the optimizer and initializes sharded states.
+
+        Args:
+            variables (list): The list of variables to optimize.
+        """
         if self.built:
             return
 
@@ -483,21 +583,21 @@ def build(self, variables: list):
         self.coordinated_optimizer.enable_optimizer_state_sharding(variables)
         super().build(variables)
 
-    def get_weights(self) -> list[np.ndarray]:
+    def get_weights(self):
         """Returns the weights of the base optimizer."""
         return self.coordinated_optimizer.get_weights()
 
-    def set_weights(self, weights: list[np.ndarray]):
+    def set_weights(self, weights):
         """Sets the weights of the base optimizer."""
         self.coordinated_optimizer.set_weights(weights)
 
     @property
-    def variables(self) -> list:
+    def variables(self):
         """Returns the list of variables from the base optimizer."""
         return self.base_optimizer.variables
 
     @property
-    def learning_rate(self) -> Any:
+    def learning_rate(self):
         """Provides access to the learning rate of the base optimizer."""
         return self.base_optimizer.learning_rate
 
@@ -507,7 +607,5 @@ def learning_rate(self, value):
 
     @property
     def iterations(self):
-        """
-        Returns the training iteration count directly from the base optimizer.
-        """
+        """Returns the training iteration count from the base optimizer."""
         return self.base_optimizer.iterations
\ No newline at end of file

From 7851615ef9535a397b5d3faf1ef0abb9783f0d55 Mon Sep 17 00:00:00 2001
From: Suhana <suhanaaa@google.com>
Date: Mon, 17 Nov 2025 15:06:13 +0530
Subject: [PATCH 09/12] refactored autoconfig to not use recursion

---
 .../tensor_parallel/autoconfig.py             | 167 ++++++++----------
 1 file changed, 73 insertions(+), 94 deletions(-)

diff --git a/keras/src/distribution/tensor_parallel/autoconfig.py b/keras/src/distribution/tensor_parallel/autoconfig.py
index d1a24b8eec22..ce3c0314cdd5 100644
--- a/keras/src/distribution/tensor_parallel/autoconfig.py
+++ b/keras/src/distribution/tensor_parallel/autoconfig.py
@@ -81,42 +81,16 @@ def analyze_dense_layer(layer):
         return 'dense'
 
 
-def _recursive_layer_traversal(
-    current_layer,
-    prefix,
-    device_count,
-    state_rules,
-    output_rules,
-    processed_layers,
-):
+def _apply_layer_sharding_rules(layer, full_name, device_count, state_rules, output_rules):
     """
-    Traverses the model graph recursively to apply sharding rules.
-
-    This function visits layers, checks their type, and populates the
-    state_rules (weights) and output_rules (activations) dictionaries
-    required for Tensor Parallelism.
-
-    Args:
-        current_layer (keras.layers.Layer): The current layer being visited.
-        prefix (str): The naming prefix for the current layer (used for nested models).
-        device_count (int): Total number of devices.
-        state_rules (dict): The dictionary accumulating variable sharding rules.
-        output_rules (dict): The dictionary accumulating output layout rules.
-        processed_layers (set): A set of object IDs to prevent infinite recursion on cycles.
+    Helper function that applies rules to a single layer instance.
     """
-    if id(current_layer) in processed_layers:
-        return
-    processed_layers.add(id(current_layer))
-
-    name = current_layer.name
-    full_name = f"{prefix}.{name}" if prefix else name
-
-    if isinstance(current_layer, layers.Dense):
-        mlp_type = analyze_dense_layer(current_layer)
+    if isinstance(layer, layers.Dense):
+        mlp_type = analyze_dense_layer(layer)
 
         if mlp_type == 'up_projection':
             state_rules[f"{full_name}.kernel"] = _split_rule(device_count, dim=1)
-            if current_layer.use_bias:
+            if layer.use_bias:
                 state_rules[f"{full_name}.bias"] = _split_rule(device_count, dim=0)
             output_rules[f"{full_name}"] = {0: "gather"}
 
@@ -126,92 +100,97 @@ def _recursive_layer_traversal(
 
         else:
             state_rules[f"{full_name}.kernel"] = _split_rule(device_count, dim=1)
-            if current_layer.use_bias:
+            if layer.use_bias:
                 state_rules[f"{full_name}.bias"] = _split_rule(device_count, dim=0)
             output_rules[f"{full_name}"] = {0: "gather -1"}
 
-    elif isinstance(current_layer, layers.EinsumDense):
+    elif isinstance(layer, layers.EinsumDense):
         if "attention_output" in full_name:
             state_rules[f"{full_name}.kernel"] = _split_rule(device_count, dim=0)
             output_rules[f"{full_name}"] = {0: "allreduce"}
         else:
             state_rules[f"{full_name}.kernel"] = _split_rule(device_count, dim=1)
-            if hasattr(current_layer, 'bias') and current_layer.bias is not None:
+            if hasattr(layer, 'bias') and layer.bias is not None:
                 state_rules[f"{full_name}.bias"] = _split_rule(device_count, dim=0)
             output_rules[f"{full_name}"] = {0: "gather -1"}
 
-    elif isinstance(current_layer, (layers.Embedding,)):
-        weight_name = None 
-
-        if hasattr(current_layer, 'embeddings'):
-            weight_name = 'embeddings'
-        elif hasattr(current_layer, 'position_embeddings'):
-            weight_name = 'position_embeddings'
+    elif isinstance(layer, (layers.Embedding,)) or "Embedding" in layer.__class__.__name__:
+        if hasattr(layer, 'weights'):
+            for weight in layer.weights:
+                if "embedding" in weight.name or "weight" in weight.name:
+                    key_found = False
+                    for attr_candidate in ['embeddings', 'position_embeddings', 'weight']:
+                        if getattr(layer, attr_candidate, None) is weight:
+                            state_rules[f"{full_name}.{attr_candidate}"] = _split_rule(device_count, dim=1)
+                            key_found = True
+                            break
+                    
+                    if not key_found:
+                        clean_name = weight.name.split('/')[-1].split(':')[0]
+                        state_rules[f"{full_name}.{clean_name}"] = _split_rule(device_count, dim=1)
 
-        if weight_name:
-            state_rules[f"{full_name}.{weight_name}"] = _split_rule(device_count, dim=1)
             output_rules[f"{full_name}"] = {0: "no_comm"}
 
-    elif isinstance(current_layer, (layers.LayerNormalization, layers.BatchNormalization, layers.GroupNormalization)):
-        pass
 
-    if hasattr(current_layer, 'layers') and current_layer.layers:
-        for sub_layer in current_layer.layers:
-            _recursive_layer_traversal(
-                sub_layer, full_name, device_count,
-                state_rules, output_rules, processed_layers
-            )
+def get_default_config(module, device_ids):
+    """
+    Generates a default tensor parallelism configuration for a model using
+    iterative graph traversal (stack-based).
+    """
+    device_count = len(device_ids)
+    state_rules = {}
+    output_rules = {}
+    
+    processed_layers = set()
+    
+    stack = [(module, "")]
+
+    while stack:
+        current_layer, prefix = stack.pop()
 
-    for attr_name in dir(current_layer):
-        if attr_name.startswith('__') and attr_name.endswith('__'):
+        if id(current_layer) in processed_layers:
             continue
-        if hasattr(current_layer, attr_name):
-            attr = getattr(current_layer, attr_name)
-
-            if isinstance(attr, layers.Layer) and attr is not current_layer:
-                _recursive_layer_traversal(
-                    attr, full_name, device_count,
-                    state_rules, output_rules, processed_layers
-                )
-            elif isinstance(attr, (list, tuple)):
-                for item in attr:
-                    if isinstance(item, layers.Layer):
-                        _recursive_layer_traversal(
-                            item, full_name, device_count,
-                            state_rules, output_rules, processed_layers
-                        )
+        processed_layers.add(id(current_layer))
 
+        name = current_layer.name
+        full_name = f"{prefix}.{name}" if prefix else name
 
-def get_default_config_keras(module, device_ids):
-    """
-    Generates a default tensor parallelism configuration for a model.
+        _apply_layer_sharding_rules(
+            current_layer, full_name, device_count, state_rules, output_rules
+        )
 
-    This function inspects the model structure and automatically generates
-    a `LayoutMap` containing sharding rules for weights (kernels/biases) and
-    outputs (activations).
+        children_to_add = []
 
-    Args:
-        module (keras.Model or keras.layers.Layer): The Keras model or layer to config.
-        device_ids (list): A list of device identifiers (e.g., strings or Mesh IDs).
+        if hasattr(current_layer, 'layers') and current_layer.layers:
+            for sub_layer in current_layer.layers:
+                children_to_add.append((sub_layer, full_name))
 
-    Returns:
-        keras.src.distribution.tensor_parallel.tensor_layout.LayoutMap:
-        The configuration map applied to the model distribution API.
-    """
-    device_count = len(device_ids)
-    state_rules = {}
-    output_rules = {}
-    
-    processed_layers = set() 
+        for specific_attr in ['token_embedding', 'embeddings', 'position_embedding']:
+            if hasattr(current_layer, specific_attr):
+                attr_val = getattr(current_layer, specific_attr)
+                if isinstance(attr_val, layers.Layer):
+                    children_to_add.append((attr_val, full_name))
 
-    _recursive_layer_traversal(
-        current_layer=module,
-        prefix="",
-        device_count=device_count, 
-        state_rules=state_rules,
-        output_rules=output_rules,
-        processed_layers=processed_layers
-    )
+        for attr_name in dir(current_layer):
+            if attr_name.startswith('__') and attr_name.endswith('__'):
+                continue
+            
+            if attr_name in ['trainable_variables', 'non_trainable_variables', 'weights']:
+                continue
+
+            attr_value = getattr(current_layer, attr_name, None)
+
+            if attr_value is None:
+                continue
+
+            if isinstance(attr_value, layers.Layer) and attr_value is not current_layer:
+                children_to_add.append((attr_value, full_name))
+            elif isinstance(attr_value, (list, tuple)):
+                for item in attr_value:
+                    if isinstance(item, layers.Layer):
+                        children_to_add.append((item, full_name))
+        
+        stack.extend(reversed(children_to_add))
 
     return LayoutMap(
         state_rules=state_rules,

From 4707c2b04555683796ca520163c76173c1b706a4 Mon Sep 17 00:00:00 2001
From: Suhana <suhanaaa@google.com>
Date: Mon, 17 Nov 2025 15:15:42 +0530
Subject: [PATCH 10/12] updating docstrings

---
 .../tensor_parallel/autoconfig.py             | 164 ++++++++++++------
 .../tensor_parallel/coordinated_optimizer.py  |  51 ++----
 2 files changed, 128 insertions(+), 87 deletions(-)

diff --git a/keras/src/distribution/tensor_parallel/autoconfig.py b/keras/src/distribution/tensor_parallel/autoconfig.py
index ce3c0314cdd5..cd75421348ed 100644
--- a/keras/src/distribution/tensor_parallel/autoconfig.py
+++ b/keras/src/distribution/tensor_parallel/autoconfig.py
@@ -1,7 +1,7 @@
 from keras.src import layers
+from keras.src.distribution.tensor_parallel.tensor_layout import LayoutMap
 from keras.src.distribution.tensor_parallel.tensor_layout import (
     split_tensor_for_parallelism,
-    LayoutMap
 )
 
 _split_fn_internal = split_tensor_for_parallelism
@@ -15,8 +15,8 @@ def _split_rule(device_count, dim):
     how a tensor should be split across the available devices.
 
     Args:
-        device_count (int): The total number of devices available for parallelism.
-        dim (int): The dimension of the tensor to split.
+        device_count: The total number of devices available for parallelism.
+        dim: The dimension of the tensor to split.
 
     Returns:
         callable: A lambda function accepting (tensor, index) that returns the
@@ -44,105 +44,161 @@ def analyze_dense_layer(layer):
         str: One of 'up_projection', 'down_projection', or 'dense'.
     """
     if not isinstance(layer, layers.Dense):
-        return 'dense'
+        return "dense"
 
     input_dim = None
     output_dim = None
 
-    if hasattr(layer, 'kernel') and layer.kernel is not None:
+    if hasattr(layer, "kernel") and layer.kernel is not None:
         kernel_shape = layer.kernel.shape
         if len(kernel_shape) == 2:
             input_dim = kernel_shape[0]
             output_dim = kernel_shape[1]
 
     if input_dim is None or output_dim is None:
-        if hasattr(layer, 'units'):
+        if hasattr(layer, "units"):
             output_dim = layer.units
         else:
-            return 'dense'
+            return "dense"
 
-        if hasattr(layer, 'input_shape') and layer.input_shape and len(layer.input_shape) > 1:
+        if (
+            hasattr(layer, "input_shape")
+            and layer.input_shape
+            and len(layer.input_shape) > 1
+        ):
             input_dim = layer.input_shape[-1]
         else:
-            return 'dense'
+            return "dense"
 
     if not input_dim or not output_dim:
-        return 'dense'
+        return "dense"
 
     expansion_threshold = 1.5
     is_expansion = output_dim > input_dim * expansion_threshold
     is_contraction = input_dim > output_dim * expansion_threshold
 
     if is_expansion:
-        return 'up_projection'
+        return "up_projection"
     elif is_contraction:
-        return 'down_projection'
+        return "down_projection"
     else:
-        return 'dense'
+        return "dense"
 
 
-def _apply_layer_sharding_rules(layer, full_name, device_count, state_rules, output_rules):
-    """
-    Helper function that applies rules to a single layer instance.
+def _apply_layer_sharding_rules(
+    layer, full_name, device_count, state_rules, output_rules
+):
+    """Applies sharding rules to a single layer instance based on its type.
+
+    This function populates the `state_rules` and `output_rules` dictionaries
+    by analyzing the specific layer type (Dense, EinsumDense, Embedding).
+
+    Args:
+        layer (keras.layers.Layer): The layer instance to process.
+        full_name: The full hierarchical name of the layer (prefix + name).
+        device_count: Total number of devices.
+        state_rules: The dictionary to update with variable sharding rules.
+        output_rules: The dictionary to update with output layout rules.
     """
     if isinstance(layer, layers.Dense):
         mlp_type = analyze_dense_layer(layer)
 
-        if mlp_type == 'up_projection':
-            state_rules[f"{full_name}.kernel"] = _split_rule(device_count, dim=1)
+        if mlp_type == "up_projection":
+            state_rules[f"{full_name}.kernel"] = _split_rule(
+                device_count, dim=1
+            )
             if layer.use_bias:
-                state_rules[f"{full_name}.bias"] = _split_rule(device_count, dim=0)
+                state_rules[f"{full_name}.bias"] = _split_rule(
+                    device_count, dim=0
+                )
             output_rules[f"{full_name}"] = {0: "gather"}
 
-        elif mlp_type == 'down_projection':
-            state_rules[f"{full_name}.kernel"] = _split_rule(device_count, dim=0)
+        elif mlp_type == "down_projection":
+            state_rules[f"{full_name}.kernel"] = _split_rule(
+                device_count, dim=0
+            )
             output_rules[f"{full_name}"] = {0: "allreduce"}
 
         else:
-            state_rules[f"{full_name}.kernel"] = _split_rule(device_count, dim=1)
+            state_rules[f"{full_name}.kernel"] = _split_rule(
+                device_count, dim=1
+            )
             if layer.use_bias:
-                state_rules[f"{full_name}.bias"] = _split_rule(device_count, dim=0)
+                state_rules[f"{full_name}.bias"] = _split_rule(
+                    device_count, dim=0
+                )
             output_rules[f"{full_name}"] = {0: "gather -1"}
 
     elif isinstance(layer, layers.EinsumDense):
         if "attention_output" in full_name:
-            state_rules[f"{full_name}.kernel"] = _split_rule(device_count, dim=0)
+            state_rules[f"{full_name}.kernel"] = _split_rule(
+                device_count, dim=0
+            )
             output_rules[f"{full_name}"] = {0: "allreduce"}
         else:
-            state_rules[f"{full_name}.kernel"] = _split_rule(device_count, dim=1)
-            if hasattr(layer, 'bias') and layer.bias is not None:
-                state_rules[f"{full_name}.bias"] = _split_rule(device_count, dim=0)
+            state_rules[f"{full_name}.kernel"] = _split_rule(
+                device_count, dim=1
+            )
+            if hasattr(layer, "bias") and layer.bias is not None:
+                state_rules[f"{full_name}.bias"] = _split_rule(
+                    device_count, dim=0
+                )
             output_rules[f"{full_name}"] = {0: "gather -1"}
 
-    elif isinstance(layer, (layers.Embedding,)) or "Embedding" in layer.__class__.__name__:
-        if hasattr(layer, 'weights'):
+    elif (
+        isinstance(layer, (layers.Embedding,))
+        or "Embedding" in layer.__class__.__name__
+    ):
+        if hasattr(layer, "weights"):
             for weight in layer.weights:
                 if "embedding" in weight.name or "weight" in weight.name:
                     key_found = False
-                    for attr_candidate in ['embeddings', 'position_embeddings', 'weight']:
+                    for attr_candidate in [
+                        "embeddings",
+                        "position_embeddings",
+                        "weight",
+                    ]:
                         if getattr(layer, attr_candidate, None) is weight:
-                            state_rules[f"{full_name}.{attr_candidate}"] = _split_rule(device_count, dim=1)
+                            state_rules[f"{full_name}.{attr_candidate}"] = (
+                                _split_rule(device_count, dim=1)
+                            )
                             key_found = True
                             break
-                    
+
                     if not key_found:
-                        clean_name = weight.name.split('/')[-1].split(':')[0]
-                        state_rules[f"{full_name}.{clean_name}"] = _split_rule(device_count, dim=1)
+                        clean_name = weight.name.split("/")[-1].split(":")[0]
+                        state_rules[f"{full_name}.{clean_name}"] = _split_rule(
+                            device_count, dim=1
+                        )
 
             output_rules[f"{full_name}"] = {0: "no_comm"}
 
 
 def get_default_config(module, device_ids):
-    """
-    Generates a default tensor parallelism configuration for a model using
-    iterative graph traversal (stack-based).
+    """Generates a default tensor parallelism configuration for a Keras model.
+
+    This function performs an iterative Depth-First Search traversal of the
+    model graph. It automatically detects layers suitable for Tensor Parallelism
+    (Embeddings, MLPs, Attention Heads) and generates a `LayoutMap`.
+
+    The traversal uses a LIFO stack and processes children in reverse order
+    to mimic the behavior of standard recursive traversal, ensuring correct
+    path naming and rule application for nested KerasNLP backbones.
+
+    Args:
+        module: The Keras model or layer to configure.
+        device_ids (list): A list of device identifiers (e.g., strings).
+
+    Returns:
+        keras.src.distribution.tensor_parallel.tensor_layout.LayoutMap:
+        The configuration map applied to the model distribution API.
     """
     device_count = len(device_ids)
     state_rules = {}
     output_rules = {}
-    
+
     processed_layers = set()
-    
+
     stack = [(module, "")]
 
     while stack:
@@ -161,21 +217,29 @@ def get_default_config(module, device_ids):
 
         children_to_add = []
 
-        if hasattr(current_layer, 'layers') and current_layer.layers:
+        if hasattr(current_layer, "layers") and current_layer.layers:
             for sub_layer in current_layer.layers:
                 children_to_add.append((sub_layer, full_name))
 
-        for specific_attr in ['token_embedding', 'embeddings', 'position_embedding']:
+        for specific_attr in [
+            "token_embedding",
+            "embeddings",
+            "position_embedding",
+        ]:
             if hasattr(current_layer, specific_attr):
                 attr_val = getattr(current_layer, specific_attr)
                 if isinstance(attr_val, layers.Layer):
                     children_to_add.append((attr_val, full_name))
 
         for attr_name in dir(current_layer):
-            if attr_name.startswith('__') and attr_name.endswith('__'):
+            if attr_name.startswith("__") and attr_name.endswith("__"):
                 continue
-            
-            if attr_name in ['trainable_variables', 'non_trainable_variables', 'weights']:
+
+            if attr_name in [
+                "trainable_variables",
+                "non_trainable_variables",
+                "weights",
+            ]:
                 continue
 
             attr_value = getattr(current_layer, attr_name, None)
@@ -183,16 +247,16 @@ def get_default_config(module, device_ids):
             if attr_value is None:
                 continue
 
-            if isinstance(attr_value, layers.Layer) and attr_value is not current_layer:
+            if (
+                isinstance(attr_value, layers.Layer)
+                and attr_value is not current_layer
+            ):
                 children_to_add.append((attr_value, full_name))
             elif isinstance(attr_value, (list, tuple)):
                 for item in attr_value:
                     if isinstance(item, layers.Layer):
                         children_to_add.append((item, full_name))
-        
+
         stack.extend(reversed(children_to_add))
 
-    return LayoutMap(
-        state_rules=state_rules,
-        output_rules=output_rules
-    )
\ No newline at end of file
+    return LayoutMap(state_rules=state_rules, output_rules=output_rules)
diff --git a/keras/src/distribution/tensor_parallel/coordinated_optimizer.py b/keras/src/distribution/tensor_parallel/coordinated_optimizer.py
index 12a69f2cb3b1..85aef7e2658e 100644
--- a/keras/src/distribution/tensor_parallel/coordinated_optimizer.py
+++ b/keras/src/distribution/tensor_parallel/coordinated_optimizer.py
@@ -1,4 +1,5 @@
 import re
+
 import numpy as np
 
 from keras.src import ops
@@ -47,7 +48,7 @@ def _initialize_sharded_states(self):
         This method inspects the variables created by the base optimizer and
         maps them to model parameters.
 
-        
+
 
         Note:
             Since the Keras BaseOptimizer does not expose a direct mapping
@@ -80,7 +81,7 @@ def _initialize_sharded_states(self):
 
             found_param = None
             slot_name = None
-            
+
             for norm_param_path, param in normalized_params:
                 if state_suffix.startswith(norm_param_path):
                     found_param = param
@@ -215,12 +216,9 @@ def _apply_gradients_with_sharded_states(
         """
         for shard_idx in range(self.device_count):
             local_states = self._get_local_optimizer_states(shard_idx)
-            # Access the base optimizer inside the TensorParallelOptimizer wrapper
-            shard_optimizer = shard_models[shard_idx].optimizer.base_optimizer 
+            shard_optimizer = shard_models[shard_idx].optimizer.base_optimizer
 
-            self._update_optimizer_internal_state(
-                shard_optimizer, local_states
-            )
+            self._update_optimizer_internal_state(shard_optimizer, local_states)
 
             shard_grads_and_vars = synchronized_gradients[shard_idx]
             shard_optimizer.apply_gradients(shard_grads_and_vars)
@@ -314,7 +312,7 @@ def _synchronize_gradients(self, gradients_and_vars):
         """
         Synchronizes gradients across shards based on tensor parallel rules.
 
-        
+
 
         Args:
             gradients_and_vars (list): A list of (gradient, variable) tuples.
@@ -351,7 +349,7 @@ def _synchronize_gradients(self, gradients_and_vars):
                     if g_and_v[i][0] is not None
                 ]
                 if grads_to_reduce:
-                    synced_grad = self._allreduce_gradients(grads_to_reduce)[0] 
+                    synced_grad = self._allreduce_gradients(grads_to_reduce)[0]
                     for shard_idx in range(self.device_count):
                         if gradients_and_vars[shard_idx][i][0] is not None:
                             gradients_and_vars[shard_idx][i] = (
@@ -472,7 +470,7 @@ def apply_gradients(self, grads_and_vars, **kwargs):
             grads_and_vars (list): A list of (gradient, variable) tuples or a
                 list of lists for sharded execution.
             **kwargs: Additional arguments, such as `shard_models`.
-        
+
         Raises:
             ValueError: If `shard_models` is missing when applying sharded
                 gradients.
@@ -495,25 +493,6 @@ def apply_gradients(self, grads_and_vars, **kwargs):
         else:
             self.base_optimizer.apply_gradients(grads_and_vars)
 
-    def get_config(self):
-        """Returns the optimizer configuration as a dictionary."""
-        from keras.src import saving
-
-        config = super().get_config()
-        config.pop("learning_rate", None)
-        config.pop("name", None)
-
-        config.update(
-            {
-                "base_optimizer": saving.serialize_keras_object(
-                    self.base_optimizer
-                ),
-                "device_count": self.device_count,
-                "tensor_parallel_config": self.coordinated_optimizer.tensor_parallel_config,
-            }
-        )
-        return config
-
     def update_step(self, gradient, variable, *args, **kwargs):
         """
         Delegates the update step to the base optimizer.
@@ -546,13 +525,13 @@ def from_config(cls, config):
         base_optimizer = saving.deserialize_keras_object(base_optimizer_config)
 
         init_kwargs = {
-            "device_count": config.get("device_count"), 
+            "device_count": config.get("device_count"),
             "tensor_parallel_config": config.get("tensor_parallel_config"),
         }
 
-        config.pop("device_count", None) 
-        config.pop("tensor_parallel_config", None) 
-        
+        config.pop("device_count", None)
+        config.pop("tensor_parallel_config", None)
+
         return cls(base_optimizer=base_optimizer, **init_kwargs)
 
     def build(self, variables):
@@ -570,9 +549,7 @@ def build(self, variables):
             iterations = self.base_optimizer.iterations
             original_iterations_val = None
             if iterations is not None:
-                original_iterations_val = ops.convert_to_numpy(
-                    iterations.value
-                )
+                original_iterations_val = ops.convert_to_numpy(iterations.value)
 
             zero_grads = [ops.zeros_like(v) for v in variables]
             self.base_optimizer.apply_gradients(zip(zero_grads, variables))
@@ -608,4 +585,4 @@ def learning_rate(self, value):
     @property
     def iterations(self):
         """Returns the training iteration count from the base optimizer."""
-        return self.base_optimizer.iterations
\ No newline at end of file
+        return self.base_optimizer.iterations

From 45aa44cd5a3f991dc37cbe5dfeaefca01ae9a9cb Mon Sep 17 00:00:00 2001
From: Suhana <suhanaaa@google.com>
Date: Mon, 17 Nov 2025 16:38:42 +0530
Subject: [PATCH 11/12] removing redundancies

---
 .../tensor_parallel/coordinated_optimizer.py  | 176 ++++++------------
 1 file changed, 53 insertions(+), 123 deletions(-)

diff --git a/keras/src/distribution/tensor_parallel/coordinated_optimizer.py b/keras/src/distribution/tensor_parallel/coordinated_optimizer.py
index 85aef7e2658e..bcb11c2bd760 100644
--- a/keras/src/distribution/tensor_parallel/coordinated_optimizer.py
+++ b/keras/src/distribution/tensor_parallel/coordinated_optimizer.py
@@ -15,14 +15,14 @@ class CoordinatedOptimizer:
     synchronizing gradients according to tensor parallelism rules.
 
     Args:
-        base_optimizer (Optimizer): The Keras optimizer instance.
-        device_count (int): The total number of devices/processes in the
-            distributed setup.
-        shard_optimizer_states (bool): If `True`, the optimizer's state
-            variables will be partitioned across `device_count` devices.
-            Defaults to `True`.
-        tensor_parallel_config (object): An optional configuration object that
-            defines rules for tensor parallelism. Defaults to `None`.
+        base_optimizer: The Keras optimizer instance.
+        device_count: The total number of devices/processes in the distributed
+            setup.
+        shard_optimizer_states: If `True`, the optimizer's state variables
+            will be partitioned across `device_count` devices. Defaults to
+            `True`.
+        tensor_parallel_config: An optional configuration object that defines
+            rules for tensor parallelism. Defaults to `None`.
     """
 
     def __init__(
@@ -42,8 +42,7 @@ def __init__(
         self._variable_to_slot_name = {}
 
     def _initialize_sharded_states(self):
-        """
-        Partitions the optimizer's state variables across shards.
+        """Partitions the optimizer's state variables across shards.
 
         This method inspects the variables created by the base optimizer and
         maps them to model parameters.
@@ -51,9 +50,9 @@ def _initialize_sharded_states(self):
 
 
         Note:
-            Since the Keras BaseOptimizer does not expose a direct mapping
-            from a model parameter to its optimizer state variables, this
-            method infers the mapping by string parsing their paths/names.
+            Since the Keras BaseOptimizer does not expose a direct mapping from
+            a model parameter to its optimizer state variables, this method
+            infers the mapping by string parsing their paths/names.
         """
         if not self.shard_optimizer_states or not self.base_optimizer.built:
             return
@@ -117,12 +116,11 @@ def _initialize_sharded_states(self):
             )
 
     def _partition_state(self, state_variable, dim):
-        """
-        Splits a single state variable numpy array into chunks.
+        """Splits a single state variable numpy array into chunks.
 
         Args:
-            state_variable (array-like): The state variable to split.
-            dim (int): The dimension along which to split the variable.
+            state_variable: The state variable to split.
+            dim: The dimension along which to split the variable.
 
         Returns:
             list: A list of numpy arrays representing the split state.
@@ -137,14 +135,12 @@ def _partition_state(self, state_variable, dim):
             return [np.copy(state_array) for _ in range(self.device_count)]
 
     def apply_gradients(self, gradients_and_vars, shard_models):
-        """
-        Coordinates gradient synchronization and application.
+        """Coordinates gradient synchronization and application.
 
         Args:
-            gradients_and_vars (list): A list containing lists of (gradient,
-                variable) tuples for each device.
-            shard_models (list): A list of model shards corresponding to the
-                devices.
+            gradients_and_vars: A list containing lists of (gradient, variable)
+                tuples for each device.
+            shard_models: A list of model shards corresponding to the devices.
 
         Raises:
             ValueError: If the number of gradient sets does not match the
@@ -170,14 +166,13 @@ def apply_gradients(self, gradients_and_vars, shard_models):
     def _apply_gradients_with_replicated_states(
         self, synchronized_gradients, shard_models
     ):
-        """
-        Averages gradients across all shards and applies them once.
+        """Averages gradients across all shards and applies them once.
 
         This is used when `shard_optimizer_states` is False.
 
         Args:
-            synchronized_gradients (list): The list of synchronized gradients.
-            shard_models (list): The list of model shards.
+            synchronized_gradients: The list of synchronized gradients.
+            shard_models: The list of model shards.
         """
         num_vars = len(synchronized_gradients[0])
         averaged_grads_and_vars = []
@@ -207,12 +202,11 @@ def _apply_gradients_with_replicated_states(
     def _apply_gradients_with_sharded_states(
         self, synchronized_gradients, shard_models
     ):
-        """
-        Applies gradients to each shard using its local optimizer state.
+        """Applies gradients to each shard using its local optimizer state.
 
         Args:
-            synchronized_gradients (list): The list of synchronized gradients.
-            shard_models (list): The list of model shards.
+            synchronized_gradients: The list of synchronized gradients.
+            shard_models: The list of model shards.
         """
         for shard_idx in range(self.device_count):
             local_states = self._get_local_optimizer_states(shard_idx)
@@ -226,11 +220,10 @@ def _apply_gradients_with_sharded_states(
             self._update_global_sharded_states(shard_optimizer, shard_idx)
 
     def _get_local_optimizer_states(self, shard_idx):
-        """
-        Constructs the state dictionary for a single shard.
+        """Constructs the state dictionary for a single shard.
 
         Args:
-            shard_idx (int): The index of the current shard.
+            shard_idx: The index of the current shard.
 
         Returns:
             dict: A dictionary mapping state names to their local values.
@@ -248,12 +241,11 @@ def _get_local_optimizer_states(self, shard_idx):
         return local_states
 
     def _update_optimizer_internal_state(self, optimizer, local_states):
-        """
-        Assigns local sharded state values to the optimizer's variables.
+        """Assigns local sharded state values to the optimizer's variables.
 
         Args:
-            optimizer (Optimizer): The local optimizer instance for the shard.
-            local_states (dict): The local state dictionary.
+            optimizer: The local optimizer instance for the shard.
+            local_states: The local state dictionary.
         """
         if not optimizer.built:
             return
@@ -278,12 +270,11 @@ def _update_optimizer_internal_state(self, optimizer, local_states):
                     var.assign(local_param_state)
 
     def _update_global_sharded_states(self, optimizer, shard_idx):
-        """
-        Updates the main sharded_states dictionary after a gradient step.
+        """Updates the main sharded_states dictionary after a gradient step.
 
         Args:
-            optimizer (Optimizer): The local optimizer instance.
-            shard_idx (int): The index of the current shard.
+            optimizer: The local optimizer instance.
+            shard_idx: The index of the current shard.
         """
         if not optimizer.built:
             return
@@ -309,13 +300,12 @@ def _update_global_sharded_states(self, optimizer, shard_idx):
                 )
 
     def _synchronize_gradients(self, gradients_and_vars):
-        """
-        Synchronizes gradients across shards based on tensor parallel rules.
+        """Synchronizes gradients across shards using tensor parallel rules.
 
 
 
         Args:
-            gradients_and_vars (list): A list of (gradient, variable) tuples.
+            gradients_and_vars: A list of (gradient, variable) tuples.
 
         Returns:
             list: The synchronized list of gradients and variables.
@@ -359,14 +349,13 @@ def _synchronize_gradients(self, gradients_and_vars):
         return gradients_and_vars
 
     def _allreduce_gradients(self, gradients):
-        """
-        Performs a mean all-reduce operation on a list of gradients.
+        """Performs a mean all-reduce operation on a list of gradients.
 
         This method uses the on-device communication primitive from the backend
         (e.g., JAX's lax.pmean) when multiple devices are detected.
 
         Args:
-            gradients (list): A list of gradient tensors to reduce.
+            gradients: A list of gradient tensors to reduce.
 
         Returns:
             list: A list containing the reduced gradient repeated for each
@@ -404,11 +393,10 @@ def set_weights(self, weights):
         self.base_optimizer.set_weights(weights)
 
     def enable_optimizer_state_sharding(self, variables):
-        """
-        Enables and initializes optimizer state sharding.
+        """Enables and initializes optimizer state sharding.
 
         Args:
-            variables (list): A list of model variables to track.
+            variables: A list of model variables to track.
         """
         self.shard_optimizer_states = True
         self._variables = variables
@@ -424,12 +412,11 @@ class TensorParallelOptimizer(optimizers.Optimizer):
     `CoordinatedOptimizer` instance.
 
     Args:
-        base_optimizer (Optimizer or str): A Keras optimizer instance or a
-            string identifier.
-        device_count (int): The total number of devices/processes in the
-            distributed setup.
-        tensor_parallel_config (object): An optional configuration object.
-            Defaults to `None`.
+        base_optimizer: A Keras optimizer instance or a string identifier.
+        device_count: The total number of devices/processes in the distributed
+            setup.
+        tensor_parallel_config: An optional configuration object. Defaults to
+            `None`.
     """
 
     def __init__(
@@ -462,84 +449,27 @@ def __init__(
             tensor_parallel_config=tensor_parallel_config,
         )
 
-    def apply_gradients(self, grads_and_vars, **kwargs):
-        """
-        Applies gradients to the model variables.
-
-        Args:
-            grads_and_vars (list): A list of (gradient, variable) tuples or a
-                list of lists for sharded execution.
-            **kwargs: Additional arguments, such as `shard_models`.
-
-        Raises:
-            ValueError: If `shard_models` is missing when applying sharded
-                gradients.
-        """
-        is_sharded_grads = (
-            isinstance(grads_and_vars, list)
-            and grads_and_vars
-            and isinstance(grads_and_vars[0], list)
-        )
-        if is_sharded_grads:
-            if "shard_models" not in kwargs:
-                raise ValueError(
-                    "The `shard_models` keyword argument is required when "
-                    "applying sharded gradients (a list of lists)."
-                )
-            shard_models = kwargs.get("shard_models")
-            self.coordinated_optimizer.apply_gradients(
-                grads_and_vars, shard_models
-            )
-        else:
-            self.base_optimizer.apply_gradients(grads_and_vars)
-
     def update_step(self, gradient, variable, *args, **kwargs):
-        """
-        Delegates the update step to the base optimizer.
+        """Delegates the update step to the base optimizer.
 
         Args:
-            gradient (Tensor): The gradient tensor.
-            variable (Variable): The variable to update.
+            gradient: The gradient tensor.
+            variable: The variable to update.
             *args: Additional arguments for the update.
             **kwargs: Additional keyword arguments for the update.
         """
         if hasattr(self.base_optimizer, "update_step"):
-            try:
-                return self.base_optimizer.update_step(
-                    gradient, variable, *args, **kwargs
-                )
-            except TypeError:
-                return self.base_optimizer.update_step(gradient, variable)
-
-        try:
-            return super().update_step(gradient, variable, *args, **kwargs)
-        except TypeError:
-            return super().update_step(gradient, variable)
-
-    @classmethod
-    def from_config(cls, config):
-        """Creates an optimizer instance from its configuration."""
-        from keras.src import saving
-
-        base_optimizer_config = config.pop("base_optimizer")
-        base_optimizer = saving.deserialize_keras_object(base_optimizer_config)
-
-        init_kwargs = {
-            "device_count": config.get("device_count"),
-            "tensor_parallel_config": config.get("tensor_parallel_config"),
-        }
-
-        config.pop("device_count", None)
-        config.pop("tensor_parallel_config", None)
+            return self.base_optimizer.update_step(
+                gradient, variable, *args, **kwargs
+            )
 
-        return cls(base_optimizer=base_optimizer, **init_kwargs)
+        return super().update_step(gradient, variable, *args, **kwargs)
 
     def build(self, variables):
-        """
-        Builds the optimizer and initializes sharded states.
+        """Builds the optimizer and initializes sharded states.
 
         Args:
-            variables (list): The list of variables to optimize.
+            variables: The list of variables to optimize.
         """
         if self.built:
             return

From 8bb39f6c894d93ed03e633deb6e57a4a86f01343 Mon Sep 17 00:00:00 2001
From: Suhana <suhanaaa@google.com>
Date: Tue, 18 Nov 2025 09:11:42 +0530
Subject: [PATCH 12/12] added tests for autoconfig and coordinated optimizer

---
 .../tensor_parallel/autoconfig_test.py        | 159 +++++++++++++++
 .../coordinated_optimizer_test.py             | 183 ++++++++++++++++++
 2 files changed, 342 insertions(+)
 create mode 100644 keras/src/distribution/tensor_parallel/autoconfig_test.py
 create mode 100644 keras/src/distribution/tensor_parallel/coordinated_optimizer_test.py

diff --git a/keras/src/distribution/tensor_parallel/autoconfig_test.py b/keras/src/distribution/tensor_parallel/autoconfig_test.py
new file mode 100644
index 000000000000..f10aefd58d12
--- /dev/null
+++ b/keras/src/distribution/tensor_parallel/autoconfig_test.py
@@ -0,0 +1,159 @@
+from unittest.mock import patch
+
+from autoconfig import analyze_dense_layer
+from autoconfig import get_default_config
+
+import keras
+from keras.src import layers
+from keras.src import testing
+
+
+class AutoConfigTest(testing.TestCase):
+    def check_rule(self, rule, expected_device_count, expected_dim):
+        """
+        Helper to verify a rule lambda.
+        Since the rule is a lambda, we mock the internal function it calls
+        to verify it captured the correct device_count and dim.
+        """
+        self.assertTrue(callable(rule), "Rule must be a callable (lambda)")
+
+        # Patch the internal function imported in autoconfig
+        with patch("autoconfig._split_fn_internal") as mock_split:
+            # Call the rule with dummy arguments
+            rule(keras.ops.zeros((2, 2)), 0)
+
+            # Verify _split_fn_internal was called
+            self.assertTrue(mock_split.called)
+
+            # Inspect arguments: (tensor, index, device_count, dim=dim)
+            args, kwargs = mock_split.call_args
+
+            # device_count is the 3rd positional argument (index 2)
+            self.assertEqual(args[2], expected_device_count)
+
+            # dim is passed as a keyword argument
+            self.assertEqual(kwargs["dim"], expected_dim)
+
+    def test_analyze_dense_layer_directly(self):
+        """Tests the heuristic for classifying Dense layers."""
+
+        up_proj_layer = layers.Dense(64, name="up")
+        up_proj_layer.build(input_shape=(None, 16))
+        self.assertEqual(analyze_dense_layer(up_proj_layer), "up_projection")
+        down_proj_layer = layers.Dense(16, name="down")
+        down_proj_layer.build(input_shape=(None, 64))
+        self.assertEqual(
+            analyze_dense_layer(down_proj_layer),
+            "down_projection",
+        )
+        generic_layer = layers.Dense(32, name="generic")
+        generic_layer.build(input_shape=(None, 28))
+        self.assertEqual(analyze_dense_layer(generic_layer), "dense")
+        non_dense_layer = layers.LayerNormalization()
+        self.assertEqual(analyze_dense_layer(non_dense_layer), "dense")
+
+    def test_simple_mlp_model(self):
+        """Tests rule generation for a standard MLP block."""
+        device_count = 2
+        devices = [f"gpu:{i}" for i in range(device_count)]
+
+        model = keras.Sequential(
+            [
+                keras.Input(shape=(32,)),
+                layers.Dense(128, name="mlp_up"),  # Up-projection
+                layers.Dense(32, name="mlp_down"),  # Down-projection
+            ],
+            name="mlp_block",
+        )
+
+        layout_map = get_default_config(model, devices)
+        state_rules = layout_map.state_rules
+        output_rules = layout_map.output_rules
+
+        # Assertions for State (Weight) Sharding Rules
+        up_kernel_key = "mlp_block.mlp_up.kernel"
+        self.assertIn(up_kernel_key, state_rules)
+        # Verify Up Projection (split on dim 1)
+        self.check_rule(state_rules[up_kernel_key], device_count, 1)
+
+        down_kernel_key = "mlp_block.mlp_down.kernel"
+        self.assertIn(down_kernel_key, state_rules)
+        # Verify Down Projection (split on dim 0)
+        self.check_rule(state_rules[down_kernel_key], device_count, 0)
+
+        # Assertions for Output Communication Rules
+        self.assertEqual(output_rules["mlp_block.mlp_up"], {0: "gather"})
+        self.assertEqual(output_rules["mlp_block.mlp_down"], {0: "allreduce"})
+
+    def test_model_with_embedding_and_einsumdense(self):
+        """Tests rule generation for Embedding and EinsumDense layers."""
+        device_count = 4
+        devices = [f"gpu:{i}" for i in range(device_count)]
+
+        class SimpleTransformer(layers.Layer):
+            def __init__(self, **kwargs):
+                super().__init__(**kwargs)
+                self.embedding = layers.Embedding(
+                    input_dim=1000, output_dim=64, name="embedding"
+                )
+                self.qkv_proj = layers.EinsumDense(
+                    "abc,cde->abde",
+                    output_shape=(None, 3, 128),
+                    bias_axes="de",
+                    name="qkv_proj",
+                )
+                self.attention_output = layers.EinsumDense(
+                    "abde,cde->abc",
+                    output_shape=(None, 64),
+                    bias_axes="c",
+                    name="attention_output",
+                )
+
+            def call(self, inputs):
+                x = self.embedding(inputs)
+                x = self.qkv_proj(x)
+                x = self.attention_output(x)
+                return x
+
+        model = SimpleTransformer(name="transformer")
+        model(keras.ops.zeros((1, 10)))
+
+        layout_map = get_default_config(model, devices)
+        state_rules = layout_map.state_rules
+
+        # Check Embedding
+        expected_key = "transformer.embedding.embeddings"
+        self.assertIn(expected_key, state_rules)
+        self.check_rule(state_rules[expected_key], device_count, 1)
+
+        # Check QKV Projection
+        qkv_key = "transformer.qkv_proj.kernel"
+        self.assertIn(qkv_key, state_rules)
+        self.check_rule(state_rules[qkv_key], device_count, 1)
+
+        # Check Attention Output
+        attn_out_key = "transformer.attention_output.kernel"
+        self.assertIn(attn_out_key, state_rules)
+        self.check_rule(state_rules[attn_out_key], device_count, 0)
+
+    def test_nested_model(self):
+        """Tests that the recursive traversal finds layers in nested models."""
+        device_count = 2
+        devices = [f"gpu:{i}" for i in range(device_count)]
+        inner_model = keras.Sequential(
+            [layers.Dense(64, name="inner_dense")], name="inner_block"
+        )
+        outer_model = keras.Sequential(
+            [
+                keras.Input(shape=(32,)),
+                layers.Dense(32, name="outer_dense_1"),
+                inner_model,
+            ],
+            name="outer_block",
+        )
+        layout_map = get_default_config(outer_model, devices)
+        state_rules = layout_map.state_rules
+
+        expected_key = "outer_block.inner_block.inner_dense.kernel"
+        self.assertIn(expected_key, state_rules)
+        self.check_rule(state_rules[expected_key], device_count, 1)
diff --git a/keras/src/distribution/tensor_parallel/coordinated_optimizer_test.py b/keras/src/distribution/tensor_parallel/coordinated_optimizer_test.py
new file mode 100644
index 000000000000..073441d14815
--- /dev/null
+++ b/keras/src/distribution/tensor_parallel/coordinated_optimizer_test.py
@@ -0,0 +1,183 @@
+import numpy as np
+import pytest
+
+# Assuming the implementation code is saved in coordinated_optimizer.py
+from coordinated_optimizer import CoordinatedOptimizer
+from coordinated_optimizer import TensorParallelOptimizer
+
+import keras
+from keras import ops
+from keras.src import backend
+from keras.src import optimizers
+from keras.src import testing
+
+
+@pytest.mark.skipif(
+    backend.backend() != "jax",
+    reason="This test is for the JAX backend only.",
+)
+class CoordinatedOptimizerTest(testing.TestCase):
+    def _get_simple_model(self):
+        """Creates a simple, uncompiled Keras model."""
+        inputs = keras.Input(shape=(10,))
+        x = keras.layers.Dense(20, name="dense_1")(inputs)
+        outputs = keras.layers.Dense(5, name="dense_2")(x)
+        return keras.Model(inputs, outputs)
+
+    def _get_mock_gradients_and_vars(self, model, device_count):
+        """Generates mock gradients and variables for N shards."""
+        model.build(input_shape=(None, 10))
+        variables = model.trainable_variables
+        grads_and_vars_per_shard = []
+        for i in range(device_count):
+            multiplier = float(i + 1)
+            gradients = [
+                ops.convert_to_tensor(
+                    np.ones_like(v.numpy()) * multiplier, dtype="float32"
+                )
+                for v in variables
+            ]
+            grads_and_vars_per_shard.append(list(zip(gradients, variables)))
+        return grads_and_vars_per_shard
+
+    def test_initialization(self):
+        """Tests that the optimizer initializes with the correct defaults."""
+        base_optimizer = optimizers.Adam()
+        coord = CoordinatedOptimizer(base_optimizer, device_count=4)
+        self.assertEqual(coord.base_optimizer, base_optimizer)
+        self.assertTrue(coord.shard_optimizer_states)
+        self.assertEqual(coord.sharded_states, {})
+
+    def test_apply_gradients_with_replicated_states(self):
+        """Tests that replicated gradients are averaged and applied once."""
+
+        class AdamWithCallCounter(optimizers.Adam):
+            def __init__(self, *args, **kwargs):
+                super().__init__(*args, **kwargs)
+                self.apply_gradients_call_count = 0
+                self.received_grads = []
+
+            def apply_gradients(self, grads_and_vars, *args, **kwargs):
+                self.apply_gradients_call_count += 1
+                self.received_grads = [g for g, v in grads_and_vars]
+                super().apply_gradients(grads_and_vars, *args, **kwargs)
+
+        device_count = 4
+        model = self._get_simple_model()
+        optimizer = AdamWithCallCounter()
+        model.build((None, 10))
+        mock_grads = self._get_mock_gradients_and_vars(model, device_count)
+
+        coord = CoordinatedOptimizer(
+            optimizer,
+            device_count,
+            shard_optimizer_states=False,
+        )
+        coord.apply_gradients(mock_grads, shard_models=[])
+
+        self.assertEqual(optimizer.apply_gradients_call_count, 1)
+        grad_numpy = ops.convert_to_numpy(optimizer.received_grads[0])
+        self.assertAllClose(
+            grad_numpy,
+            np.ones_like(grad_numpy) * 2.5,
+        )
+
+    def test_init_from_string(self):
+        optimizer = TensorParallelOptimizer("adam", device_count=4)
+        self.assertIsInstance(optimizer.base_optimizer, optimizers.Adam)
+
+    def test_apply_gradients_delegation(self):
+        """Tests that apply_gradients correctly delegates."""
+        device_count = 4
+        base_opt = optimizers.Adam()
+        optimizer = TensorParallelOptimizer(base_opt, device_count)
+        model = self._get_simple_model()
+        mock_grads = self._get_mock_gradients_and_vars(model, device_count)
+
+        coord_apply_tracker = {"called": False}
+
+        def coord_apply_mock(*args, **kwargs):
+            coord_apply_tracker["called"] = True
+
+        optimizer.coordinated_optimizer.apply_gradients = coord_apply_mock
+
+        base_apply_tracker = {"called": False}
+
+        def base_apply_mock(*args, **kwargs):
+            base_apply_tracker["called"] = True
+
+        optimizer.base_optimizer.apply_gradients = base_apply_mock
+
+        optimizer.coordinated_optimizer.apply_gradients(
+            mock_grads, shard_models=[]
+        )
+        self.assertTrue(coord_apply_tracker["called"])
+
+        coord_apply_tracker["called"] = False
+        unsharded_grads = mock_grads[0]
+        optimizer.base_optimizer.apply_gradients(unsharded_grads)
+        self.assertTrue(base_apply_tracker["called"])
+        self.assertFalse(coord_apply_tracker["called"])
+
+    def test_build_and_state_sharding(self):
+        """Tests that the build method correctly initializes sharded states."""
+        optimizer = TensorParallelOptimizer(optimizers.Adam(), device_count=4)
+        model = self._get_simple_model()
+        model.build(input_shape=(None, 10))
+
+        self.assertEqual(optimizer.coordinated_optimizer.sharded_states, {})
+        optimizer.build(model.trainable_variables)
+        self.assertTrue(optimizer.built)
+
+        sharded_states = optimizer.coordinated_optimizer.sharded_states
+        # Check for either 'momentum' or 'm' (Adam standard names)
+        self.assertTrue("momentum" in sharded_states or "m" in sharded_states)
+        self.assertTrue("velocity" in sharded_states or "v" in sharded_states)
+        self.assertIn("iterations", sharded_states)
+
+        mom_key = "momentum" if "momentum" in sharded_states else "m"
+        dense_1_kernel_path = model.get_layer("dense_1").kernel.path
+        self.assertIn(dense_1_kernel_path, sharded_states[mom_key])
+        self.assertEqual(len(sharded_states[mom_key][dense_1_kernel_path]), 4)
+
+    def test_serialization(self):
+        """Tests manual reconstruction via from_config."""
+        device_count = 4
+        base_opt = optimizers.Adam(learning_rate=0.1)
+        config = {
+            "base_optimizer": base_opt,
+            "device_count": device_count,
+        }
+
+        recreated = TensorParallelOptimizer.from_config(config)
+
+        self.assertEqual(recreated.device_count, device_count)
+        self.assertIsInstance(recreated.base_optimizer, optimizers.Adam)
+        self.assertAllClose(recreated.base_optimizer.learning_rate, 0.1)
+
+    def test_sharding_with_prefixed_variable_names(self):
+        """Tests that state is correctly mapped with prefixed variable names."""
+        inputs = keras.Input(shape=(10,))
+        x = keras.layers.Dense(4, name="dense")(inputs)
+        outputs = keras.layers.Dense(2, name="dense_output")(x)
+        model = keras.Model(inputs, outputs)
+        model.build(input_shape=(None, 10))
+
+        optimizer = TensorParallelOptimizer(optimizers.Adam(), device_count=2)
+        optimizer.build(model.trainable_variables)
+
+        state_to_param = (
+            optimizer.coordinated_optimizer._state_variable_to_parameter
+        )
+        self.assertGreater(len(state_to_param), 0)
+
+        dense_output_kernel = model.get_layer("dense_output").kernel
+
+        found_key = None
+        for key, param in state_to_param.items():
+            if param is dense_output_kernel:
+                found_key = key
+                break
+
+        self.assertIsNotNone(found_key)
+        self.assertIs(state_to_param[found_key], dense_output_kernel)