Skip to content

KerasCortex

centimators.model_estimators.keras_cortex

Keras Cortex: A self-improving Keras estimator wrapper using DSPy to self-reflect and improve its architecture.

This module provides KerasCortex, a scikit-learn compatible meta-estimator.

KerasCodeRefinements

Bases: Signature

Suggest modifications to build_model code to improve performance.

Source code in src/centimators/model_estimators/keras_cortex.py 
43
44
45
46
47
48
49
50
51
52
53
54
class KerasCodeRefinements(Signature):
    """Suggest modifications to build_model code to improve performance."""

    current_keras_code = InputField(desc="Source code of build_model method.")
    performance_log = InputField(desc="History of (code, metric) pairs.")
    optimization_goal = InputField(desc="Objective, e.g., 'improve validation scores'.")
    suggested_keras_code_modification = OutputField(
        desc=(
            "Modified build_model method body as code. No code fences. "
            "You must start with only 'def build_model(self):'"
        )
    )

Think

Bases: Module

DSPy Module for suggesting Keras model code modifications.

Source code in src/centimators/model_estimators/keras_cortex.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
class Think(Module):
    """DSPy Module for suggesting Keras model code modifications."""

    def __init__(self, verbose=False):
        super().__init__()
        self.suggest_code = ChainOfThought(KerasCodeRefinements)
        self.verbose = verbose

    def forward(self, current_keras_code, performance_log, optimization_goal):
        prediction = self.suggest_code(
            current_keras_code=current_keras_code,
            performance_log=performance_log,
            optimization_goal=optimization_goal,
        )
        if self.verbose:
            print(f"Reasoning: \n{prediction.reasoning}")
            print(f"Suggested code: \n{prediction.suggested_keras_code_modification}")
        return prediction.suggested_keras_code_modification

KerasCortex

Bases: RegressorMixin, BaseEstimator

A scikit-learn meta-estimator that iteratively refines a Keras model.

Source code in src/centimators/model_estimators/keras_cortex.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
class KerasCortex(RegressorMixin, BaseEstimator):
    """A scikit-learn meta-estimator that iteratively refines a Keras model."""

    def __init__(
        self,
        base_estimator=None,
        n_iterations=5,
        lm="openai/gpt-4o-mini",
        verbose=False,
    ):
        if base_estimator is None:
            base_estimator = MLPRegressor()
        self.base_estimator = base_estimator
        self.n_iterations = n_iterations
        self.lm = dspy.LM(lm)
        dspy.configure(lm=self.lm)
        self.verbose = verbose

    def think_loop(
        self, base_estimator, X, y, validation_data, n_iterations=5, **kwargs
    ) -> tuple[BaseEstimator, list[tuple[str, float]]]:
        baseline_model = clone(base_estimator)
        baseline_model.fit(X, y, **kwargs)

        X_val, y_val = validation_data
        best_metric = baseline_model.score(X_val, y_val)
        current_code = inspect.getsource(type(baseline_model).build_model)
        performance_log = [(current_code, best_metric)]

        best_model = baseline_model
        suggestion = current_code

        think = Think(verbose=self.verbose)
        for i in range(n_iterations):
            print(f"\n--- Iteration {i + 1} ---")
            try:
                suggestion = think.forward(
                    current_keras_code=suggestion,
                    performance_log=performance_log,
                    optimization_goal="improve validation metrics (R2)",
                )
                namespace = {}
                exec(suggestion, globals(), namespace)
                build_model_fn = namespace["build_model"]

                new_model = clone(base_estimator)
                new_model.build_model = types.MethodType(build_model_fn, new_model)
                new_model.fit(X, y, **kwargs)
                metric = new_model.score(X_val, y_val)

                performance_log.append((suggestion, metric))
                if metric > best_metric:
                    print(
                        f"Improvement! New validation score: {metric:.4f} > {best_metric:.4f}"
                    )
                    best_metric = metric
                    best_model = new_model
                else:
                    print(
                        f"No improvement ({metric:.4f} <= {best_metric:.4f}), keeping best code."
                    )
            except Exception as e:
                print("Error during optimization iteration:", e)
                break

        return best_model, performance_log

    def fit(self, X, y, validation_data=None, **kwargs):
        self.best_model_, self.performance_log_ = self.think_loop(
            base_estimator=self.base_estimator,
            X=X,
            y=y,
            validation_data=validation_data,
            n_iterations=self.n_iterations,
            **kwargs,
        )
        return self

    def predict(self, X):
        if not hasattr(self, "best_model_"):
            raise ValueError("Estimator not fitted. Call 'fit' first.")
        return self.best_model_.predict(X)