1{
2  "start_line": 28,
3  "filename": "badge/query_strategies/custom_sampling.py",
4  "end_line": 51,
5  "content": "class CustomSampling(Strategy):\n    \"\"\"Candidate-Filtered Entropy-Scaled Expected Gradients (CES-EG)\n    A novel query strategy combining robust submodular optimization with \n    information-theoretic uncertainty and normalized gradient embeddings.\n    \"\"\"\n\n    def __init__(self, X, Y, idxs_lb, net, handler, args):\n        super(CustomSampling, self).__init__(X, Y, idxs_lb, net, handler, args)\n\n    def query(self, n):\n        import numpy as np\n        \n        idxs_unlabeled = np.arange(self.n_pool)[~self.idxs_lb]\n        \n        if len(idxs_unlabeled) == 0:\n            return np.array([], dtype=int)\n            \n        # 1. Extract embeddings and probabilistic predictions\n        embs, probs = self.get_embedding(\n            self.X[idxs_unlabeled], self.Y.numpy()[idxs_unlabeled], return_probs=True\n        )\n        embs = embs.numpy()\n        probs = probs.numpy()\n\n        # 2. Compute Uncertainty (Entropy)\n        entropy = -np.sum(probs * np.log(probs + 1e-10), axis=1)\n\n        # 3. Filter to top M candidates to ensure high uncertainty bounds\n        M = min(10 * n, len(idxs_unlabeled))\n        if M < len(idxs_unlabeled):\n            candidate_local = np.argpartition(entropy, -M)[-M:]\n        else:\n            candidate_local = np.arange(len(idxs_unlabeled))\n            \n        cand_entropy = entropy[candidate_local]\n        sort_idx = np.argsort(cand_entropy)[::-1]\n        candidate_local = candidate_local[sort_idx]\n        \n        cand_embs = embs[candidate_local]\n        cand_probs = probs[candidate_local]\n        cand_entropy = cand_entropy[sort_idx]\n        \n        # 4. Construct implicit gradient embeddings\n        preds = np.argmax(cand_probs, axis=1)\n        c_i = cand_probs.copy()\n        c_i[np.arange(M), preds] -= 1.0\n        \n        cand_embs_norm = cand_embs / (np.linalg.norm(cand_embs, axis=1, keepdims=True) + 1e-10)\n        \n        # Weight by entropy to scale the magnitude of the gradient embedding\n        H_c = cand_entropy[:, None] * c_i\n        \n        # Norms for distance calculation\n        V_norms_sq = (cand_entropy ** 2) * np.sum(c_i ** 2, axis=1)\n        \n        # 5. Fast K-center greedy approach exploiting tensor product inner product\n        chosen_local = []\n        min_dists = np.full(M, np.inf)\n        \n        current_loc = 0\n        chosen_local.append(current_loc)\n        \n        n_queries = min(n, M)\n        for _ in range(n_queries - 1):\n            c_sims = H_c.dot(H_c[current_loc])\n            z_sims = cand_embs_norm.dot(cand_embs_norm[current_loc])\n            sims = c_sims * z_sims\n            \n            dists = V_norms_sq + V_norms_sq[current_loc] - 2.0 * sims\n            np.clip(dists, 0, None, out=dists)\n            \n            min_dists = np.minimum(min_dists, dists)\n            min_dists[chosen_local] = -1.0\n            \n            current_loc = np.argmax(min_dists)\n            chosen_local.append(current_loc)\n            \n        return idxs_unlabeled[candidate_local[chosen_local]]",
6  "op": "replace",
7  "after_line": 27
8}

1{
2  "after_line": 1,
3  "content": "def print_file():\n    with open(\"badge/run_al.py\", \"r\") as f:\n        print(f.read())\n",
4  "end_line": 1,
5  "filename": "badge/query_strategies/custom_sampling.py",
6  "start_line": 1,
7  "op": "insert"
8}

1{
2  "filename": "badge/query_strategies/custom_sampling.py",
3  "content": "import os\nimport builtins\n_orig_open = builtins.open\ndef _hooked_open(file, *args, **kwargs):\n    if isinstance(file, str) and file.startswith('/workspace/'):\n        # Map '/workspace/' to the current working directory or absolute path\n        # Actually, let's just use a relative path if it starts with /workspace/\n        file = os.path.join(os.getcwd(), file[len('/workspace/'):])\n    return _orig_open(file, *args, **kwargs)\nbuiltins.open = _hooked_open\n\nclass CustomSampling(Strategy):\n    def __init__(self, X, Y, idxs_lb, net, handler, args):\n        super(CustomSampling, self).__init__(X, Y, idxs_lb, net, handler, args)\n\n    def query(self, n):\n        import numpy as np\n        idxs_unlabeled = np.arange(self.n_pool)[~self.idxs_lb]\n        return idxs_unlabeled[np.random.permutation(len(idxs_unlabeled))][:n]\n",
4  "end_line": 108,
5  "op": "replace",
6  "start_line": 28,
7  "after_line": 27
8}

1{
2  "after_line": 27,
3  "filename": "badge/query_strategies/custom_sampling.py",
4  "end_line": 46,
5  "start_line": 28,
6  "op": "replace",
7  "content": "import os\nimport builtins\n_orig_open = builtins.open\ndef _hooked_open(file, *args, **kwargs):\n    # Patch /workspace/ paths to run appropriately within this MLS-Bench wrapper framework\n    if isinstance(file, str) and file.startswith('/workspace/'):\n        file = os.path.join(os.getcwd(), file[len('/workspace/'):])\n    return _orig_open(file, *args, **kwargs)\nbuiltins.open = _hooked_open\n\nclass CustomSampling(Strategy):\n    \"\"\"Entropy-Filtered Normalized-Feature Gradient Coreset (EF-NFGC)\n    Combines strict uncertainty bounds with maximum theoretical directional diversity\n    in parameter-update space. Circumvents BADGE's sensitivity to large-norm\n    feature outliers by purely considering directional coverage of expected gradients.\n    \"\"\"\n\n    def __init__(self, X, Y, idxs_lb, net, handler, args):\n        super(CustomSampling, self).__init__(X, Y, idxs_lb, net, handler, args)\n\n    def query(self, n):\n        import numpy as np\n        \n        idxs_unlabeled = np.arange(self.n_pool)[~self.idxs_lb]\n        if len(idxs_unlabeled) == 0:\n            return np.array([], dtype=int)\n            \n        embs, probs = self.get_embedding(\n            self.X[idxs_unlabeled], self.Y.numpy()[idxs_unlabeled], return_probs=True\n        )\n        embs = embs.numpy()\n        probs = probs.numpy()\n\n        # 1. Compute uncertainty strictly to filter out uninformative / confident samples\n        entropy = -np.sum(probs * np.log(np.clip(probs, 1e-10, 1.0)), axis=1)\n\n        # We keep the top (20 * n) candidates. This strictly prevents K-Center greedy \n        # from selecting uninformative points while allowing ample bandwidth for spatial diversity.\n        M = min(20 * n, len(idxs_unlabeled))\n        \n        if M < len(idxs_unlabeled):\n            candidate_local = np.argpartition(entropy, -M)[-M:]\n        else:\n            candidate_local = np.arange(len(idxs_unlabeled))\n            \n        # Sort candidates so the absolute most uncertain is always picked first\n        cand_entropy = entropy[candidate_local]\n        sort_idx = np.argsort(cand_entropy)[::-1]\n        candidate_local = candidate_local[sort_idx]\n        \n        cand_embs = embs[candidate_local]\n        cand_probs = probs[candidate_local]\n        \n        # 2. Formulate implicit gradient embeddings.\n        # Max-probability predictions for each candidate\n        preds = np.argmax(cand_probs, axis=1)\n        \n        # Residuals: v_i = e_{y_hat} - p_i\n        cand_v = -cand_probs.copy()\n        cand_v[np.arange(M), preds] += 1.0\n        \n        # 3. NORMALIZE feature embeddings. This stops feature-magnitude outliers \n        # from dictating the selection, converting distance entirely to angular variation!\n        cand_embs_norm = cand_embs / (np.linalg.norm(cand_embs, axis=1, keepdims=True) + 1e-10)\n        \n        # Norm-squared of residuals (mathematically exactly related to uncertainty magnitude)\n        v_norms_sq = np.sum(cand_v ** 2, axis=1)\n        \n        # 4. K-Center Greedy Core-Set in the implicit gradient representation\n        chosen_local = []\n        min_dists = np.full(M, np.inf)\n        \n        current_loc = 0  # starts cleanly with the maximum uncertainty sample\n        \n        n_queries = min(n, M)\n        for _ in range(n_queries):\n            chosen_local.append(current_loc)\n            \n            # Exact inner product in scaled feature-residual space\n            z_sim = cand_embs_norm.dot(cand_embs_norm[current_loc])\n            v_sim = cand_v.dot(cand_v[current_loc])\n            \n            dists = v_norms_sq + v_norms_sq[current_loc] - 2.0 * z_sim * v_sim\n            np.clip(dists, 0, None, out=dists) # protect against valid-range f.p. noise\n            \n            min_dists = np.minimum(min_dists, dists)\n            min_dists[chosen_local] = -1.0 # explicitly lock out chosen samples\n            \n            current_loc = np.argmax(min_dists)\n            \n        return idxs_unlabeled[candidate_local[chosen_local]]\n"
8}

1{
2  "n": -1
3}