finalize release 1 changes

README.md

@@ -21,6 +21,8 @@ This repo contains:
 - **Modular**: PDF → images, embedding, Qdrant indexing, retrieval can be used independently.
 - **Multi-stage retrieval**: two-stage and three-stage retrieval modes built for Qdrant named vectors.
 - **Model-aware embedding**: ColSmol, ColPali, and ColQwen2/2.5 support behind a single `VisualEmbedder` interface.
+- **Configurable pooling**: adaptive mean-pooling cap for ColQwen2.5 (`--max-mean-pool-vectors`), and experimental pooling stored as Qdrant named vectors (`experimental_pooling` (ColQwen Gaussian alias), `experimental_pooling_gaussian`, `experimental_pooling_triangular`, `experimental_pooling_{k}` (ColPali), `experimental_pooling_2d` (ColSmol)).
+- **Single-stage ablations**: direct search modes over experimental pooled vectors (tokens-vs-doc and pooled-query-vs-doc) for fast storage-reduction experiments.
 - **Token hygiene**: query special-token filtering by default for more stable MaxSim behavior.
 - **Practical pipelines**: robust indexing, retries, optional Cloudinary image URLs, evaluation reporting.
 

benchmarks/vidore_beir_qdrant/run_qdrant_beir.py

@@ -132,7 +132,24 @@ def _default_output_filename(*, args, datasets: List[str]) -> str:
     parts = [model_tag, mode_tag]
     if str(args.mode) == "two_stage":
         parts.append(_safe_filename(str(args.stage1_mode)))
+        # Disambiguate which experimental named vector was used.
+        # Without this, `--experimental-pooling-k 3` overwrites the default `experimental_pooling` run.
+        if str(args.stage1_mode) in (
+            "pooled_query_vs_experimental_pooling",
+            "tokens_vs_experimental_pooling",
+            "pooled_query_vs_experimental",
+            "tokens_vs_experimental",
+        ):
+            if getattr(args, "experimental_pooling_technique", None):
+                parts.append(f"exptech{_safe_filename(str(args.experimental_pooling_technique))}")
+            if getattr(args, "experimental_pooling_k", None) is not None:
+                parts.append(f"expk{int(args.experimental_pooling_k)}")
         parts.append(f"pk{int(args.prefetch_k)}")
+    if str(args.mode) in ("single_experimental_tokens", "single_experimental_pooled"):
+        if getattr(args, "experimental_pooling_technique", None):
+            parts.append(f"exptech{_safe_filename(str(args.experimental_pooling_technique))}")
+        if getattr(args, "experimental_pooling_k", None) is not None:
+            parts.append(f"expk{int(args.experimental_pooling_k)}")
     if str(args.mode) == "three_stage":
         parts.append("tokens_vs_global")
         parts.append(f"s1k{int(args.stage1_k)}")
@@ -555,31 +572,47 @@ def _index_beir_corpus(
     is_colqwen25 = "colqwen2.5" in model_lower or "colqwen2_5" in model_lower
     is_colsmol = "colsmol" in model_lower
     kernel_arg = str(experimental_pooling_kernel or "auto").lower().strip()
-    if kernel_arg == "auto":
-        kernel = "gaussian" if is_colqwen25 else "legacy"
+    if is_colqwen25:
+        # ColQwen2.5: store technique variants (k is fixed at 3)
+        if pooling_windows:
+            raise ValueError(
+                "ColQwen2.5 does not support --pooling-windows; it stores gaussian+triangular variants."
+            )
+        if kernel_arg not in ("auto", "gaussian", "triangular"):
+            raise ValueError(
+                "ColQwen2.5 experimental pooling kernel is fixed to gaussian+triangular variants (k=3)."
+            )
+        kernel = "gaussian"
+        ks_norm = [3]
+        experimental_vector_names = [
+            "experimental_pooling_gaussian",
+            "experimental_pooling_triangular",
+        ]
     else:
-        kernel = kernel_arg
+        # ColPali-style experimental pooling supports different window sizes
+        if kernel_arg == "auto":
+            kernel = "legacy"
+        else:
+            kernel = kernel_arg
 
-    default_k = 5 if is_colqwen25 else 3
-    if kernel != "legacy":
         default_k = 3
-    ks = pooling_windows if pooling_windows else [default_k]
-    seen_ks = set()
-    ks_norm: List[int] = []
-    for k in ks:
-        try:
-            ki = int(k)
-        except Exception:
-            continue
-        if ki <= 0:
-            continue
-        if ki in seen_ks:
-            continue
-        seen_ks.add(ki)
-        ks_norm.append(ki)
-    if not ks_norm:
-        ks_norm = [default_k]
-    experimental_vector_names = [f"experimental_pooling_{int(k)}" for k in ks_norm]
+        ks = pooling_windows if pooling_windows else [default_k]
+        seen_ks = set()
+        ks_norm = []
+        for k in ks:
+            try:
+                ki = int(k)
+            except Exception:
+                continue
+            if ki <= 0:
+                continue
+            if ki in seen_ks:
+                continue
+            seen_ks.add(ki)
+            ks_norm.append(ki)
+        if not ks_norm:
+            ks_norm = [default_k]
+        experimental_vector_names = [f"experimental_pooling_{int(k)}" for k in ks_norm]
     if is_colsmol and bool(colsmol_experimental_2d):
         experimental_vector_names.append("experimental_pooling_2d")
 
@@ -898,23 +931,48 @@ def _index_beir_corpus(
                     )
 
                     experimental_pooled_by_name: Dict[str, Any] = {}
-                    canonical_k = int(ks_norm[0]) if ks_norm else int(default_k)
-                    for k in ks_norm:
-                        exp = embedder.experimental_pool_visual_embedding(
+                    if is_colqwen25:
+                        exp_gaussian = embedder.experimental_pool_visual_embedding(
+                            visual_embedding,
+                            token_info,
+                            target_vectors=tv,
+                            mean_pool=tile_pooled,
+                            window_size=3,
+                            kernel="gaussian",
+                        )
+                        exp_triangular = embedder.experimental_pool_visual_embedding(
                             visual_embedding,
                             token_info,
                             target_vectors=tv,
                             mean_pool=tile_pooled,
-                            window_size=int(k),
-                            kernel=kernel,
+                            window_size=3,
+                            kernel="triangular",
+                        )
+                        experimental_pooled_by_name["experimental_pooling"] = exp_gaussian
+                        experimental_pooled_by_name["experimental_pooling_gaussian"] = exp_gaussian
+                        experimental_pooled_by_name["experimental_pooling_triangular"] = (
+                            exp_triangular
                         )
-                        experimental_pooled_by_name[f"experimental_pooling_{int(k)}"] = exp
-                        if int(k) == int(canonical_k):
-                            experimental_pooled_by_name["experimental_pooling"] = exp
+                    else:
+                        canonical_k = int(ks_norm[0]) if ks_norm else 3
+                        for k in ks_norm:
+                            exp = embedder.experimental_pool_visual_embedding(
+                                visual_embedding,
+                                token_info,
+                                target_vectors=tv,
+                                mean_pool=tile_pooled,
+                                window_size=int(k),
+                                kernel=kernel,
+                            )
+                            experimental_pooled_by_name[f"experimental_pooling_{int(k)}"] = exp
+                            if int(k) == int(canonical_k):
+                                experimental_pooled_by_name["experimental_pooling"] = exp
 
                     if is_colsmol and bool(colsmol_experimental_2d):
                         try:
-                            from visual_rag.embedding.pooling import colsmol_tile_4n_pooling_from_tiles
+                            from visual_rag.embedding.pooling import (
+                                colsmol_tile_4n_pooling_from_tiles,
+                            )
 
                             n_rows = (token_info or {}).get("n_rows")
                             n_cols = (token_info or {}).get("n_cols")
@@ -1073,7 +1131,9 @@ def _index_beir_corpus(
                     "experimental_pooling_windows": ks_norm,
                     "experimental_pooling_default_window": int(ks_norm[0]) if ks_norm else None,
                     "experimental_pooling_kernel": str(kernel),
-                    "colsmol_experimental_2d": bool(colsmol_experimental_2d) if is_colsmol else None,
+                    "colsmol_experimental_2d": (
+                        bool(colsmol_experimental_2d) if is_colsmol else None
+                    ),
                     "max_mean_pool_vectors": (
                         int(max_mean_pool_vectors) if max_mean_pool_vectors is not None else None
                     ),
@@ -1217,7 +1277,7 @@ def main() -> None:
         nargs="+",
         default=None,
         help=(
-            "Experimental pooling window size(s). Provide one int to override the default window, "
+            "ColPali only: experimental pooling window size(s). Provide one int to override the default window, "
             "or multiple ints to index/store multiple experimental vectors. "
             "When multiple are provided, vectors are stored as 'experimental_pooling_{k}' and "
             "the canonical 'experimental_pooling' aliases the first provided k."
@@ -1232,7 +1292,8 @@ def main() -> None:
         help=(
             "Experimental pooling kernel. "
             "'legacy' uses the historical ColPali conv-style pooling (N->N+2r; default for ColPali). "
-            "'gaussian'/'triangular'/'uniform' use weighted same-length smoothing (N->N; default for ColQwen2.5)."
+            "'gaussian'/'triangular'/'uniform' use weighted same-length smoothing (N->N). "
+            "Ignored for ColQwen2.5 indexing (which stores gaussian+triangular variants with k=3)."
         ),
     )
     parser.add_argument(
@@ -1303,7 +1364,15 @@ def main() -> None:
         "--mode",
         type=str,
         default="single_full",
-        choices=["single_full", "single_tiles", "single_global", "two_stage", "three_stage"],
+        choices=[
+            "single_full",
+            "single_tiles",
+            "single_global",
+            "single_experimental_tokens",
+            "single_experimental_pooled",
+            "two_stage",
+            "three_stage",
+        ],
     )
     parser.add_argument(
         "--stage1-mode",
@@ -1335,10 +1404,22 @@ def main() -> None:
         type=int,
         default=None,
         help=(
-            "When using an experimental stage1-mode, select which indexed experimental vector to use "
+            "ColPali only: when using an experimental stage1-mode, select which indexed experimental vector to use "
             "(Qdrant named vector: 'experimental_pooling_{k}'). If omitted, uses 'experimental_pooling'."
         ),
     )
+    parser.add_argument(
+        "--experimental-pooling-technique",
+        "--experimental_pooling_technique",
+        type=str,
+        default=None,
+        choices=["gaussian", "triangular"],
+        help=(
+            "ColQwen only: select which experimental pooling technique to use for stage-1/single-stage experimental "
+            "modes. Maps to Qdrant named vectors: 'experimental_pooling_gaussian' or 'experimental_pooling_triangular'. "
+            "If omitted, uses 'experimental_pooling' (Gaussian alias)."
+        ),
+    )
     parser.add_argument(
         "--stage1-k", type=int, default=1000, help="Three-stage stage1 top_k (default: 1000)"
     )
@@ -1366,14 +1447,14 @@ def main() -> None:
         "--continue-on-error",
         dest="continue_on_error",
         action="store_true",
-        default=True,
-        help="Continue evaluating remaining datasets if one dataset fails (default: true).",
+        default=False,
+        help="Continue evaluating remaining datasets if one dataset fails (default: false).",
     )
     cont_group.add_argument(
         "--no-continue-on-error",
         dest="continue_on_error",
         action="store_false",
-        help="Stop the run immediately on the first dataset evaluation failure.",
+        help="Stop the run immediately on the first dataset evaluation failure (default).",
     )
     parser.add_argument("--output", type=str, default="auto")
     parser.add_argument(
@@ -1684,23 +1765,25 @@ def main() -> None:
             )
             # Verify by printing current on_disk flags (what the UI reads)
             try:
-                vectors = (
-                    (((info_after or {}).get("config") or {}).get("params") or {}).get("vectors") or {}
-                )
+                vectors = (((info_after or {}).get("config") or {}).get("params") or {}).get(
+                    "vectors"
+                ) or {}
                 if isinstance(vectors, dict):
                     vec_flags = {}
                     for name, cfg in vectors.items():
                         if not isinstance(cfg, dict):
                             continue
-                        vec_flags[str(name)] = bool(cfg.get("on_disk")) if cfg.get("on_disk") is not None else None
+                        vec_flags[str(name)] = (
+                            bool(cfg.get("on_disk")) if cfg.get("on_disk") is not None else None
+                        )
                 else:
                     vec_flags = {}
                 hnsw_on_disk = (
-                    (((info_after or {}).get("config") or {}).get("hnsw_config") or {}).get("on_disk")
-                )
+                    ((info_after or {}).get("config") or {}).get("hnsw_config") or {}
+                ).get("on_disk")
                 on_disk_payload = (
-                    (((info_after or {}).get("config") or {}).get("params") or {}).get("on_disk_payload")
-                )
+                    ((info_after or {}).get("config") or {}).get("params") or {}
+                ).get("on_disk_payload")
             except Exception:
                 vec_flags = {}
                 hnsw_on_disk = None
@@ -1719,21 +1802,45 @@ def main() -> None:
             print(f"⚠️ ensure-in-ram failed: {type(e).__name__}: {e}")
             sys.stdout.flush()
 
+    def _is_colqwen_model(model_name: str) -> bool:
+        return "colqwen" in str(model_name).lower()
+
     exp_vector_name = "experimental_pooling"
-    if (
-        args.experimental_pooling_k is not None
-        and str(args.stage1_mode) in ("pooled_query_vs_experimental_pooling", "tokens_vs_experimental_pooling")
+    uses_experimental_vector = (
+        str(args.stage1_mode)
+        in ("pooled_query_vs_experimental_pooling", "tokens_vs_experimental_pooling")
         and str(args.mode) in ("two_stage", "three_stage")
+    ) or str(args.mode) in ("single_experimental_tokens", "single_experimental_pooled")
+
+    if (
+        getattr(args, "experimental_pooling_technique", None)
+        and getattr(args, "experimental_pooling_k", None) is not None
     ):
+        raise ValueError(
+            "Use only one of --experimental-pooling-technique or --experimental-pooling-k (not both)."
+        )
+
+    if uses_experimental_vector and getattr(args, "experimental_pooling_technique", None):
+        if not _is_colqwen_model(args.model):
+            raise ValueError(
+                "--experimental-pooling-technique is only supported for ColQwen models."
+            )
+        exp_vector_name = (
+            f"experimental_pooling_{str(args.experimental_pooling_technique).strip().lower()}"
+        )
+
+    if uses_experimental_vector and getattr(args, "experimental_pooling_k", None) is not None:
+        if _is_colqwen_model(args.model):
+            raise ValueError(
+                "--experimental-pooling-k is intended for ColPali (experimental_pooling_{k}), not ColQwen."
+            )
         exp_vector_name = f"experimental_pooling_{int(args.experimental_pooling_k)}"
 
     retriever = MultiVectorRetriever(
         collection_name=args.collection,
         embedder=embedder,
         qdrant_url=os.getenv("QDRANT_URL"),
-        qdrant_api_key=(
-            os.getenv("QDRANT_API_KEY")
-        ),
+        qdrant_api_key=(os.getenv("QDRANT_API_KEY")),
         prefer_grpc=args.prefer_grpc,
         request_timeout=int(args.qdrant_timeout),
         max_retries=int(args.qdrant_retries),
@@ -1741,13 +1848,18 @@ def main() -> None:
         experimental_vector_name=exp_vector_name,
     )
 
-    if (
-        str(args.stage1_mode) in ("pooled_query_vs_experimental_pooling", "tokens_vs_experimental_pooling")
+    if str(args.mode) in ("single_experimental_tokens", "single_experimental_pooled") or (
+        str(args.stage1_mode)
+        in ("pooled_query_vs_experimental_pooling", "tokens_vs_experimental_pooling")
         and str(args.mode) in ("two_stage", "three_stage")
     ):
-        existing_vectors = _collection_vector_names(client=retriever.client, collection_name=str(args.collection))
+        existing_vectors = _collection_vector_names(
+            client=retriever.client, collection_name=str(args.collection)
+        )
         if exp_vector_name not in existing_vectors:
-            candidates = sorted([v for v in existing_vectors if str(v).startswith("experimental_pooling")])
+            candidates = sorted(
+                [v for v in existing_vectors if str(v).startswith("experimental_pooling")]
+            )
             raise ValueError(
                 f"Requested experimental vector '{exp_vector_name}' is not present in the collection. "
                 f"Available experimental vectors: {candidates or '[]'}. "
@@ -1798,11 +1910,12 @@ def main() -> None:
             "qdrant_retries": int(args.qdrant_retries),
             "qdrant_retry_sleep": float(args.qdrant_retry_sleep),
             "full_scan_threshold": int(args.full_scan_threshold),
-            "max_mean_pool_vectors": int(args.max_mean_pool_vectors)
-            if args.max_mean_pool_vectors is not None
-            else None,
+            "max_mean_pool_vectors": (
+                int(args.max_mean_pool_vectors) if args.max_mean_pool_vectors is not None else None
+            ),
             "pooling_windows": args.pooling_windows,
             "experimental_pooling_k": args.experimental_pooling_k,
+            "experimental_pooling_technique": getattr(args, "experimental_pooling_technique", None),
             "eval_wall_time_s": float(max(time.time() - eval_started_at, 0.0)),
             "metrics": single_metrics,
             "metrics_by_dataset": metrics_by_dataset,
@@ -1918,6 +2031,18 @@ def main() -> None:
             sys.stdout.flush()
         except Exception as e:
             dataset_errors[ds_name] = f"{type(e).__name__}: {e}"
+            print(f"❌ Evaluation failed for dataset={ds_name}: {dataset_errors[ds_name]}")
+            # Common cause for per_dataset: missing payload index for 'dataset'
+            msg = str(e)
+            if (
+                'Index required but not found for "dataset"' in msg
+                or "Index required but not found for 'dataset'" in msg
+            ):
+                print(
+                    "   Hint: Qdrant requires a payload index for key 'dataset' to use per-dataset filtering.\n"
+                    "   Fix: create payload index (keyword) on the collection, or rerun indexing with metadata indexes enabled."
+                )
+            sys.stdout.flush()
             if not bool(args.continue_on_error):
                 _write_json_atomic(out_path, _build_run_record())
                 raise

benchmarks/vidore_tatdqa_test/run_qdrant.py

@@ -742,10 +742,22 @@ def main() -> None:
         type=int,
         default=None,
         help=(
-            "When using an experimental stage1-mode, select which indexed experimental vector to use "
+            "ColPali only: when using an experimental stage1-mode, select which indexed experimental vector to use "
             "(Qdrant named vector: 'experimental_pooling_{k}'). If omitted, uses 'experimental_pooling'."
         ),
     )
+    parser.add_argument(
+        "--experimental-pooling-technique",
+        "--experimental_pooling_technique",
+        type=str,
+        default=None,
+        choices=["gaussian", "triangular"],
+        help=(
+            "ColQwen only: choose experimental pooling named vector for experimental stage-1. "
+            "Maps to: 'experimental_pooling_gaussian' or 'experimental_pooling_triangular'. "
+            "If omitted, uses 'experimental_pooling' (Gaussian alias)."
+        ),
+    )
     parser.add_argument("--output", type=str, default="results/qdrant_vidore_tatdqa_test.json")
 
     args = parser.parse_args()
@@ -816,13 +828,37 @@ def main() -> None:
                 full_scan_threshold=args.full_scan_threshold,
             )
 
+    def _is_colqwen_model(model_name: str) -> bool:
+        return "colqwen" in str(model_name).lower()
+
     exp_vector_name = "experimental_pooling"
+    uses_experimental = str(args.stage1_mode) in (
+        "pooled_query_vs_experimental_pooling",
+        "tokens_vs_experimental_pooling",
+    ) and str(args.mode) in ("two_stage", "three_stage")
+
     if (
-        args.experimental_pooling_k is not None
-        and str(args.stage1_mode)
-        in ("pooled_query_vs_experimental_pooling", "tokens_vs_experimental_pooling")
-        and str(args.mode) in ("two_stage", "three_stage")
+        getattr(args, "experimental_pooling_technique", None)
+        and getattr(args, "experimental_pooling_k", None) is not None
     ):
+        raise ValueError(
+            "Use only one of --experimental-pooling-technique or --experimental-pooling-k."
+        )
+
+    if uses_experimental and getattr(args, "experimental_pooling_technique", None):
+        if not _is_colqwen_model(args.model):
+            raise ValueError(
+                "--experimental-pooling-technique is only supported for ColQwen models."
+            )
+        exp_vector_name = (
+            f"experimental_pooling_{str(args.experimental_pooling_technique).strip().lower()}"
+        )
+
+    if uses_experimental and getattr(args, "experimental_pooling_k", None) is not None:
+        if _is_colqwen_model(args.model):
+            raise ValueError(
+                "--experimental-pooling-k is intended for ColPali (experimental_pooling_{k}), not ColQwen."
+            )
         exp_vector_name = f"experimental_pooling_{int(args.experimental_pooling_k)}"
 
     retriever = MultiVectorRetriever(

demo/qdrant_utils.py

@@ -12,14 +12,8 @@ def get_qdrant_credentials() -> Tuple[Optional[str], Optional[str]]:
 
     Priority: session_state > QDRANT_URL/QDRANT_API_KEY
     """
-    url = (
-        st.session_state.get("qdrant_url_input")
-        or os.getenv("QDRANT_URL")
-    )
-    api_key = (
-        st.session_state.get("qdrant_key_input")
-        or os.getenv("QDRANT_API_KEY")
-    )
+    url = st.session_state.get("qdrant_url_input") or os.getenv("QDRANT_URL")
+    api_key = st.session_state.get("qdrant_key_input") or os.getenv("QDRANT_API_KEY")
     return url, api_key
 
 

examples/COMMANDS.md

@@ -11,6 +11,21 @@ export QDRANT_API_KEY="..."  # optional
 
 Or create a `.env` file in `visual-rag-toolkit/` with the same variables.
 
+## Experimental pooling knobs (index-time)
+
+When indexing, you can control:
+
+- **Adaptive mean pooling cap** (ColQwen2.5): `--max-mean-pool-vectors 32` (default), or `<=0` for **no cap**
+- **ColPali experimental pooling window(s)**: `--pooling-windows 3` or `--pooling-windows 1 3 5`
+  - Multiple windows are stored as named vectors: `experimental_pooling_{k}`
+  - The canonical `experimental_pooling` uses the **first** provided window
+- **ColQwen experimental pooling variants (always written)**:
+  - `experimental_pooling` (Gaussian alias)
+  - `experimental_pooling_gaussian`
+  - `experimental_pooling_triangular`
+- **Experimental pooling kernel** (ColPali): `--experimental-pooling-kernel auto|legacy|uniform|triangular|gaussian`
+- **ColSmol 2D experimental pooling**: `--colsmol-experimental-2d` (stores `experimental_pooling_2d`)
+
 ## Index + evaluate (single run)
 
 This is the “all-in-one” script (indexes, then evaluates once):
@@ -123,7 +138,7 @@ python -m benchmarks.vidore_beir_qdrant.run_qdrant_beir \
   --prefer-grpc \
   --torch-dtype float32 \
   --qdrant-vector-dtype float32 \
-  --batch-size 1 \
+  --batch-size 6 \
   --upload-batch-size 4 \
   --upload-workers 0 \
   --no-cloudinary \
@@ -131,7 +146,7 @@ python -m benchmarks.vidore_beir_qdrant.run_qdrant_beir \
 ```
 
 Notes:
-- **`--batch-size 1`** is the safest starting point on Apple Silicon (MPS). Increase cautiously if stable.
+- On Apple Silicon (MPS), batched queries should be stable for ColQwen2.5; if you see NaNs, reduce `--batch-size` and/or set `VISUALRAG_SORT_QUERIES_BY_LENGTH=1`.
 - This does **not** enable cropping (we do **not** pass `--crop-empty`).
 
 ## Evaluate later (optional)
@@ -174,4 +189,52 @@ python -m benchmarks.vidore_beir_qdrant.run_qdrant_beir \
   --evaluation-scope per_dataset
 ```
 
+Single-stage experiments on **experimental pooling** (no rerank):
+
+- **Tokens vs experimental pooled vectors** (MaxSim query tokens vs `experimental_pooling`):
+
+```bash
+python -m benchmarks.vidore_beir_qdrant.run_qdrant_beir \
+  --datasets vidore/esg_reports_v2 \
+  --collection vidore_v2__colqwen25_fp32 \
+  --model vidore/colqwen2.5-v0.2 \
+  --prefer-grpc \
+  --torch-dtype float32 \
+  --qdrant-vector-dtype float32 \
+  --mode single_experimental_tokens \
+  --top-k 100
+```
+
+- **Pooled query vs experimental pooled vectors** (pooled query vs `experimental_pooling`):
+
+```bash
+python -m benchmarks.vidore_beir_qdrant.run_qdrant_beir \
+  --datasets vidore/esg_reports_v2 \
+  --collection vidore_v2__colqwen25_fp32 \
+  --model vidore/colqwen2.5-v0.2 \
+  --prefer-grpc \
+  --torch-dtype float32 \
+  --qdrant-vector-dtype float32 \
+  --mode single_experimental_pooled \
+  --top-k 100
+```
+
+If you indexed multiple windows (ColPali; e.g. `--pooling-windows 1 3 5`), select one via:
+
+```bash
+--experimental-pooling-k 3
+```
+
+If you’re using ColQwen and want the alternate variant, select via:
+
+```bash
+--experimental-pooling-technique triangular
+```
+
+Internal helper: update existing collections’ experimental vectors (no re-embedding)
+
+```bash
+python -m scripts.qdrant_update_experimental_poolings
+```
+
 

scripts/qdrant_rebuild_collection_no_index.py

@@ -232,6 +232,16 @@ def main() -> None:
             "('experimental_pooling_{k}') in the rebuilt collection schema."
         ),
     )
+    parser.add_argument(
+        "--add-colqwen-techniques",
+        dest="add_colqwen_techniques",
+        action="store_true",
+        default=False,
+        help=(
+            "Include ColQwen experimental technique named vectors in the rebuilt schema: "
+            "'experimental_pooling_gaussian' and 'experimental_pooling_triangular'."
+        ),
+    )
     parser.add_argument(
         "--keep-temp", action="store_true", help="Do not delete temp collection at the end"
     )
@@ -250,6 +260,10 @@ def main() -> None:
         seen.add(ki)
         ks_norm.append(ki)
     experimental_vector_names = [f"experimental_pooling_{k}" for k in ks_norm]
+    if bool(getattr(args, "add_colqwen_techniques", False)):
+        experimental_vector_names.extend(
+            ["experimental_pooling_gaussian", "experimental_pooling_triangular"]
+        )
 
     if DOTENV_AVAILABLE:
         load_dotenv()

scripts/qdrant_recompute_colqwen_pooling_from_initial.py

@@ -43,7 +43,7 @@ from qdrant_client.http import models as qm
 
 from visual_rag.embedding.pooling import (
     adaptive_row_mean_pooling_from_grid,
-    colpali_experimental_pooling_from_rows,
+    weighted_row_smoothing_same_length,
 )
 
 
@@ -159,8 +159,9 @@ def main() -> None:
         nargs="+",
         default=None,
         help=(
-            "Experimental pooling window size(s). If multiple are provided, vectors are updated as "
-            "'experimental_pooling_{k}' and the canonical 'experimental_pooling' aliases the first provided k."
+            "Deprecated (ColQwen now uses technique variants). Ignored. "
+            "This script always writes: experimental_pooling (Gaussian alias), "
+            "experimental_pooling_gaussian and experimental_pooling_triangular (both k=3)."
         ),
     )
     args = ap.parse_args()
@@ -180,22 +181,12 @@ def main() -> None:
         check_compatibility=False,
     )
 
-    # Requested experimental window sizes and required named vectors
-    ks = args.pooling_windows if args.pooling_windows else [5]
-    seen = set()
-    ks_norm = []
-    for k in ks:
-        try:
-            ki = int(k)
-        except Exception:
-            continue
-        if ki <= 0 or ki in seen:
-            continue
-        seen.add(ki)
-        ks_norm.append(ki)
-    if not ks_norm:
-        ks_norm = [5]
-    exp_names = ["experimental_pooling"] + [f"experimental_pooling_{k}" for k in ks_norm]
+    # Required named vectors for ColQwen experimental variants
+    exp_names = [
+        "experimental_pooling",
+        "experimental_pooling_gaussian",
+        "experimental_pooling_triangular",
+    ]
     try:
         info = client.get_collection(str(args.collection))
         vectors = info.config.params.vectors or {}
@@ -314,17 +305,12 @@ def main() -> None:
                 ),
                 output_dtype=np.float32,
             )
-            exp_by_name = {}
-            canonical_k = int(ks_norm[0])
-            for k in ks_norm:
-                exp = colpali_experimental_pooling_from_rows(
-                    mean_pool,
-                    window_size=int(k),
-                    output_dtype=np.float32,
-                )
-                exp_by_name[f"experimental_pooling_{int(k)}"] = exp
-                if int(k) == canonical_k:
-                    exp_by_name["experimental_pooling"] = exp
+            exp_gaussian = weighted_row_smoothing_same_length(
+                mean_pool, window_size=3, kernel="gaussian", output_dtype=np.float32
+            )
+            exp_triangular = weighted_row_smoothing_same_length(
+                mean_pool, window_size=3, kernel="triangular", output_dtype=np.float32
+            )
             glob = mean_pool.mean(axis=0).astype(np.float32)
 
             pv_batch.append(
@@ -333,7 +319,9 @@ def main() -> None:
                     vector={
                         "mean_pooling": mean_pool.tolist(),
                         "global_pooling": glob.tolist(),
-                        **{name: vec.tolist() for name, vec in exp_by_name.items()},
+                        "experimental_pooling": exp_gaussian.tolist(),
+                        "experimental_pooling_gaussian": exp_gaussian.tolist(),
+                        "experimental_pooling_triangular": exp_triangular.tolist(),
                     },
                 )
             )

visual_rag/cli/main.py

@@ -114,35 +114,46 @@ def cmd_process(args):
         processor_speed=str(getattr(args, "processor_speed", "fast")),
     )
 
-    # Experimental pooling windows/kernel (for additional Qdrant named vectors)
+    # Experimental pooling vectors (for additional Qdrant named vectors)
     model_lower = (model_name or "").lower()
     is_colqwen25 = "colqwen2.5" in model_lower or "colqwen2_5" in model_lower
     is_colsmol = "colsmol" in model_lower
-    kernel_arg = str(getattr(args, "experimental_pooling_kernel", "auto") or "auto").lower().strip()
-    if kernel_arg == "auto":
-        kernel = "gaussian" if is_colqwen25 else "legacy"
+    experimental_vector_names = []
+    if is_colqwen25:
+        # ColQwen2.5: always store both named vectors explicitly.
+        experimental_vector_names.extend(
+            ["experimental_pooling_gaussian", "experimental_pooling_triangular"]
+        )
+        if getattr(args, "pooling_windows", None):
+            logger.warning(
+                "⚠️ --pooling-windows is ignored for ColQwen2.5 (use technique variants instead)."
+            )
+        if str(
+            getattr(args, "experimental_pooling_kernel", "auto") or "auto"
+        ).lower().strip() not in ("auto", "gaussian", "triangular"):
+            logger.warning(
+                "⚠️ --experimental-pooling-kernel is ignored for ColQwen2.5 (fixed gaussian+triangular k=3)."
+            )
     else:
-        kernel = kernel_arg
-    default_k = 5 if is_colqwen25 else 3
-    if kernel != "legacy":
+        # ColPali-style: optional multiple ks stored as experimental_pooling_{k}
         default_k = 3
-    ks = args.pooling_windows if getattr(args, "pooling_windows", None) else [default_k]
-    seen_ks = set()
-    ks_norm = []
-    for k in ks:
-        try:
-            ki = int(k)
-        except Exception:
-            continue
-        if ki <= 0:
-            continue
-        if ki in seen_ks:
-            continue
-        seen_ks.add(ki)
-        ks_norm.append(ki)
-    if not ks_norm:
-        ks_norm = [default_k]
-    experimental_vector_names = [f"experimental_pooling_{int(k)}" for k in ks_norm]
+        ks = args.pooling_windows if getattr(args, "pooling_windows", None) else [default_k]
+        seen_ks = set()
+        ks_norm = []
+        for k in ks:
+            try:
+                ki = int(k)
+            except Exception:
+                continue
+            if ki <= 0:
+                continue
+            if ki in seen_ks:
+                continue
+            seen_ks.add(ki)
+            ks_norm.append(ki)
+        if not ks_norm:
+            ks_norm = [default_k]
+        experimental_vector_names = [f"experimental_pooling_{int(k)}" for k in ks_norm]
     if is_colsmol and bool(getattr(args, "colsmol_experimental_2d", False)):
         experimental_vector_names.append("experimental_pooling_2d")
 
@@ -297,9 +308,53 @@ def cmd_search(args):
         check_compatibility=False,
     )
 
+    def _is_colqwen_model(model_name: str) -> bool:
+        return "colqwen" in str(model_name).lower()
+
     exp_vector_name = "experimental_pooling"
+    uses_experimental_vector = str(args.strategy) in (
+        "single_experimental_tokens",
+        "single_experimental_pooled",
+    ) or (
+        str(args.strategy) == "two_stage"
+        and str(args.stage1_mode)
+        in ("pooled_query_vs_experimental_pooling", "tokens_vs_experimental_pooling")
+    )
+    if (
+        getattr(args, "experimental_pooling_technique", None)
+        and getattr(args, "experimental_pooling_k", None) is not None
+    ):
+        raise SystemExit(
+            "Use only one of --experimental-pooling-technique or --experimental-pooling-k."
+        )
+
+    if getattr(args, "experimental_pooling_technique", None):
+        if not uses_experimental_vector:
+            logger.warning(
+                "--experimental-pooling-technique was provided but this strategy does not use experimental vectors; ignoring."
+            )
+        else:
+            if not _is_colqwen_model(args.model):
+                raise SystemExit(
+                    "--experimental-pooling-technique is only supported for ColQwen models."
+                )
+            exp_vector_name = (
+                f"experimental_pooling_{str(args.experimental_pooling_technique).strip().lower()}"
+            )
+
     if getattr(args, "experimental_pooling_k", None) is not None:
-        if str(args.stage1_mode) in ("pooled_query_vs_experimental_pooling", "tokens_vs_experimental_pooling"):
+        if _is_colqwen_model(args.model):
+            raise SystemExit(
+                "--experimental-pooling-k is intended for ColPali (experimental_pooling_{k}), not ColQwen."
+            )
+        if not uses_experimental_vector:
+            logger.warning(
+                "--experimental-pooling-k was provided but this strategy does not use experimental vectors; ignoring."
+            )
+        elif str(args.stage1_mode) in (
+            "pooled_query_vs_experimental_pooling",
+            "tokens_vs_experimental_pooling",
+        ) or str(args.strategy) in ("single_experimental_tokens", "single_experimental_pooled"):
             exp_vector_name = f"experimental_pooling_{int(args.experimental_pooling_k)}"
         else:
             logger.warning(
@@ -309,9 +364,14 @@ def cmd_search(args):
     two_stage = TwoStageRetriever(
         client, args.collection, experimental_vector_name=str(exp_vector_name)
     )
-    single_stage = SingleStageRetriever(client, args.collection)
+    single_stage = SingleStageRetriever(
+        client, args.collection, experimental_vector_name=str(exp_vector_name)
+    )
 
-    if str(args.stage1_mode) in ("pooled_query_vs_experimental_pooling", "tokens_vs_experimental_pooling"):
+    if str(args.stage1_mode) in (
+        "pooled_query_vs_experimental_pooling",
+        "tokens_vs_experimental_pooling",
+    ):
         try:
             info = client.get_collection(str(args.collection))
             vectors = info.config.params.vectors or {}
@@ -323,7 +383,7 @@ def cmd_search(args):
             raise SystemExit(
                 f"Requested experimental vector '{exp_vector_name}' is not present in the collection. "
                 f"Available experimental vectors: {candidates or '[]'}. "
-                "Re-index with --pooling-windows (and --force-recreate) to add it."
+                "Re-index (and --force-recreate) to add it."
             )
 
     # Embed query
@@ -362,6 +422,20 @@ def cmd_search(args):
             strategy="pooled_global",
             filter_obj=filter_obj,
         )
+    elif args.strategy == "single_experimental_tokens":
+        results = single_stage.search(
+            query_embedding=query_np,
+            top_k=args.top_k,
+            strategy="experimental_maxsim",
+            filter_obj=filter_obj,
+        )
+    elif args.strategy == "single_experimental_pooled":
+        results = single_stage.search(
+            query_embedding=query_np,
+            top_k=args.top_k,
+            strategy="pooled_experimental",
+            filter_obj=filter_obj,
+        )
     else:
         results = two_stage.search(
             query_embedding=query_np,
@@ -564,9 +638,10 @@ Examples:
         nargs="+",
         default=None,
         help=(
-            "Experimental pooling window size(s). Provide one int to override the default window, "
+            "ColPali only: experimental pooling window size(s). Provide one int to override the default window, "
             "or multiple ints to index/store multiple experimental vectors as "
-            "'experimental_pooling_{k}' (and 'experimental_pooling' aliases the first provided k)."
+            "'experimental_pooling_{k}' (and 'experimental_pooling' aliases the first provided k). "
+            "Ignored for ColQwen2.5 (which stores gaussian+triangular variants)."
         ),
     )
     process_parser.add_argument(
@@ -578,7 +653,8 @@ Examples:
         help=(
             "Experimental pooling kernel. "
             "'legacy' uses the historical ColPali conv-style pooling (N->N+2r; default for ColPali). "
-            "'gaussian'/'triangular'/'uniform' use weighted same-length smoothing (N->N; default for ColQwen2.5)."
+            "'gaussian'/'triangular'/'uniform' use weighted same-length smoothing (N->N). "
+            "Ignored for ColQwen2.5 (which stores gaussian+triangular variants with k=3)."
         ),
     )
     process_parser.add_argument(
@@ -637,7 +713,14 @@ Examples:
         "--strategy",
         type=str,
         default="single_full",
-        choices=["single_full", "single_tiles", "single_global", "two_stage"],
+        choices=[
+            "single_full",
+            "single_tiles",
+            "single_global",
+            "single_experimental_tokens",
+            "single_experimental_pooled",
+            "two_stage",
+        ],
         help="Search strategy",
     )
     search_parser.add_argument(
@@ -667,10 +750,22 @@ Examples:
         type=int,
         default=None,
         help=(
-            "When using an experimental stage1-mode, select which indexed experimental vector to use "
+            "ColPali only: when using an experimental stage1-mode, select which indexed experimental vector to use "
             "(Qdrant named vector: 'experimental_pooling_{k}'). If omitted, uses 'experimental_pooling'."
         ),
     )
+    search_parser.add_argument(
+        "--experimental-pooling-technique",
+        "--experimental_pooling_technique",
+        type=str,
+        default=None,
+        choices=["gaussian", "triangular"],
+        help=(
+            "ColQwen only: choose experimental pooling named vector for experimental strategies/stage-1. "
+            "Maps to: 'experimental_pooling_gaussian' or 'experimental_pooling_triangular'. "
+            "If omitted, uses 'experimental_pooling' (Gaussian alias)."
+        ),
+    )
     search_parser.add_argument("--year", type=int, help="Filter by year")
     search_parser.add_argument("--source", type=str, help="Filter by source")
     search_parser.add_argument("--district", type=str, help="Filter by district")

visual_rag/embedding/visual_embedder.py

@@ -441,7 +441,8 @@ class VisualEmbedder:
             try:
                 if tok is not None:
                     lengths = [
-                        len(tok(q, add_special_tokens=True).get("input_ids", [])) for q in query_texts
+                        len(tok(q, add_special_tokens=True).get("input_ids", []))
+                        for q in query_texts
                     ]
                 else:
                     lengths = [len(str(q)) for q in query_texts]
@@ -804,7 +805,9 @@ class VisualEmbedder:
         if grid * grid == num_tokens:
             # For ColQwen2.5 with unset cap, keep all rows (grid) rather than defaulting to 32.
             effective_target_rows = (
-                int(grid) if (is_colqwen25 and target_vectors_cap is None) else int(target_vectors_cap)
+                int(grid)
+                if (is_colqwen25 and target_vectors_cap is None)
+                else int(target_vectors_cap)
             )
             if int(grid) == int(effective_target_rows):
                 return colpali_row_mean_pooling(
@@ -911,7 +914,11 @@ class VisualEmbedder:
         return weighted_row_smoothing_same_length(
             rows,
             window_size=window,
-            kernel=("gaussian" if k == "gaussian" else ("triangular" if k == "triangular" else "uniform")),
+            kernel=(
+                "gaussian"
+                if k == "gaussian"
+                else ("triangular" if k == "triangular" else "uniform")
+            ),
             output_dtype=self.output_dtype,
         )
 

visual_rag/indexing/pipeline.py

@@ -16,11 +16,10 @@ The metadata stored includes everything needed for saliency visualization:
 """
 
 import gc
-import time
 import hashlib
 import logging
 from pathlib import Path
-from typing import Dict, Any, List, Optional, Set, Tuple
+from typing import Any, Dict, List, Optional, Set
 
 import numpy as np
 import torch
@@ -31,7 +30,7 @@ logger = logging.getLogger(__name__)
 class ProcessingPipeline:
     """
     End-to-end pipeline for PDF processing and indexing.
-    
+
     This pipeline:
     1. Converts PDFs to images
     2. Resizes for ColPali processing
@@ -39,7 +38,7 @@ class ProcessingPipeline:
     4. Computes pooling (strategy-dependent)
     5. Uploads images to Cloudinary (optional)
     6. Stores in Qdrant with full saliency metadata
-    
+
     Args:
         embedder: VisualEmbedder instance
         indexer: QdrantIndexer instance (optional)
@@ -52,34 +51,34 @@ class ProcessingPipeline:
               This is our NOVEL contribution - preserves spatial structure while reducing size.
             - "standard": Push ALL tokens as-is (including special tokens, padding)
               This is the baseline approach for comparison.
-    
+
     Example:
         >>> from visual_rag import VisualEmbedder, QdrantIndexer, CloudinaryUploader
         >>> from visual_rag.indexing.pipeline import ProcessingPipeline
-        >>> 
+        >>>
         >>> # Our novel pooling strategy (default)
         >>> pipeline = ProcessingPipeline(
         ...     embedder=VisualEmbedder(),
         ...     indexer=QdrantIndexer(url, api_key, "my_collection"),
         ...     embedding_strategy="pooling",  # Visual tokens only + tile pooling
         ... )
-        >>> 
+        >>>
         >>> # Standard baseline (all tokens, no filtering)
         >>> pipeline_baseline = ProcessingPipeline(
         ...     embedder=VisualEmbedder(),
         ...     indexer=QdrantIndexer(url, api_key, "my_collection_baseline"),
         ...     embedding_strategy="standard",  # All tokens as-is
         ... )
-        >>> 
+        >>>
         >>> pipeline.process_pdf(Path("report.pdf"))
     """
-    
+
     # Valid embedding strategies
     # - "pooling": Visual tokens only + tile-level pooling (NOVEL)
     # - "standard": All tokens + global mean (BASELINE)
     # - "all": Embed once, push BOTH representations (efficient comparison)
     STRATEGIES = ["pooling", "standard", "all"]
-    
+
     def __init__(
         self,
         embedder=None,
@@ -104,7 +103,7 @@ class ProcessingPipeline:
         self.cloudinary_uploader = cloudinary_uploader
         self.metadata_mapping = metadata_mapping or {}
         self.config = config or {}
-        
+
         # Validate and set embedding strategy
         if embedding_strategy not in self.STRATEGIES:
             raise ValueError(
@@ -117,31 +116,34 @@ class ProcessingPipeline:
         self.crop_empty_percentage_to_remove = float(crop_empty_percentage_to_remove)
         self.crop_empty_remove_page_number = bool(crop_empty_remove_page_number)
         self.crop_empty_preserve_border_px = int(crop_empty_preserve_border_px)
-        self.crop_empty_uniform_rowcol_std_threshold = float(crop_empty_uniform_rowcol_std_threshold)
+        self.crop_empty_uniform_rowcol_std_threshold = float(
+            crop_empty_uniform_rowcol_std_threshold
+        )
 
         self.max_mean_pool_vectors = max_mean_pool_vectors
         self.pooling_windows = pooling_windows
         self.experimental_pooling_kernel = str(experimental_pooling_kernel or "auto")
         self.colsmol_experimental_2d = bool(colsmol_experimental_2d)
-        
+
         logger.info(f"📊 Embedding strategy: {embedding_strategy}")
         if embedding_strategy == "pooling":
             logger.info("   → Visual tokens only + tile-level mean pooling (NOVEL)")
         else:
             logger.info("   → All tokens as-is (BASELINE)")
-        
+
         # Create PDF processor if not provided
         if pdf_processor is None:
             from visual_rag.indexing.pdf_processor import PDFProcessor
+
             dpi = self.config.get("processing", {}).get("dpi", 140)
             pdf_processor = PDFProcessor(dpi=dpi)
         self.pdf_processor = pdf_processor
-        
+
         # Config defaults
         self.embedding_batch_size = self.config.get("batching", {}).get("embedding_batch_size", 8)
         self.upload_batch_size = self.config.get("batching", {}).get("upload_batch_size", 8)
         self.delay_between_uploads = self.config.get("delays", {}).get("between_uploads", 0.5)
-    
+
     def process_pdf(
         self,
         pdf_path: Path,
@@ -153,7 +155,7 @@ class ProcessingPipeline:
     ) -> Dict[str, Any]:
         """
         Process a single PDF end-to-end.
-        
+
         Args:
             pdf_path: Path to PDF file
             skip_existing: Skip pages that already exist in Qdrant
@@ -161,7 +163,7 @@ class ProcessingPipeline:
             upload_to_qdrant: Upload embeddings to Qdrant
             original_filename: Original filename (use this instead of pdf_path.name for temp files)
             progress_callback: Optional callback(stage, current, total, message) for progress updates
-        
+
         Returns:
             Dict with processing results:
             {
@@ -176,15 +178,15 @@ class ProcessingPipeline:
         pdf_path = Path(pdf_path)
         filename = original_filename or pdf_path.name
         logger.info(f"📚 Processing PDF: {filename}")
-        
+
         # Check existing pages
         existing_ids: Set[str] = set()
         if skip_existing and self.indexer:
             existing_ids = self.indexer.get_existing_ids(filename)
             if existing_ids:
                 logger.info(f"   Found {len(existing_ids)} existing pages")
-        
-        logger.info(f"🖼️ Converting PDF to images...")
+
+        logger.info("🖼️ Converting PDF to images...")
         if progress_callback:
             progress_callback("convert", 0, 0, "Converting PDF to images...")
         images, texts = self.pdf_processor.process_pdf(pdf_path)
@@ -192,48 +194,55 @@ class ProcessingPipeline:
         logger.info(f"   ✅ Converted {total_pages} pages")
         if progress_callback:
             progress_callback("convert", total_pages, total_pages, f"Converted {total_pages} pages")
-        
+
         extra_metadata = self._get_extra_metadata(filename)
         if extra_metadata:
             logger.info(f"   📋 Found extra metadata: {list(extra_metadata.keys())}")
-        
+
         # Process in batches
         uploaded = 0
         skipped = 0
         failed = 0
         all_pages = []
         upload_queue = []
-        
+
         for batch_start in range(0, total_pages, self.embedding_batch_size):
             batch_end = min(batch_start + self.embedding_batch_size, total_pages)
             batch_images = images[batch_start:batch_end]
             batch_texts = texts[batch_start:batch_end]
-            
+
             logger.info(f"📦 Processing pages {batch_start + 1}-{batch_end}/{total_pages}")
             if progress_callback:
-                progress_callback("embed", batch_start, total_pages, f"Embedding pages {batch_start + 1}-{batch_end}")
-            
+                progress_callback(
+                    "embed",
+                    batch_start,
+                    total_pages,
+                    f"Embedding pages {batch_start + 1}-{batch_end}",
+                )
+
             pages_to_process = []
             for i, (img, text) in enumerate(zip(batch_images, batch_texts)):
                 page_num = batch_start + i + 1
                 chunk_id = self.generate_chunk_id(filename, page_num)
-                
+
                 if skip_existing and chunk_id in existing_ids:
                     skipped += 1
                     continue
-                
-                pages_to_process.append({
-                    "index": i,
-                    "page_num": page_num,
-                    "chunk_id": chunk_id,
-                    "raw_image": img,
-                    "text": text,
-                })
-            
+
+                pages_to_process.append(
+                    {
+                        "index": i,
+                        "page_num": page_num,
+                        "chunk_id": chunk_id,
+                        "raw_image": img,
+                        "text": text,
+                    }
+                )
+
             if not pages_to_process:
                 logger.info("   All pages in batch exist, skipping...")
                 continue
-            
+
             # Generate embeddings with token info
             logger.info(f"🤖 Generating embeddings for {len(pages_to_process)} pages...")
             from visual_rag.preprocessing.crop_empty import CropEmptyConfig, crop_empty
@@ -248,7 +257,9 @@ class ProcessingPipeline:
                             percentage_to_remove=float(self.crop_empty_percentage_to_remove),
                             remove_page_number=bool(self.crop_empty_remove_page_number),
                             preserve_border_px=int(self.crop_empty_preserve_border_px),
-                            uniform_rowcol_std_threshold=float(self.crop_empty_uniform_rowcol_std_threshold),
+                            uniform_rowcol_std_threshold=float(
+                                self.crop_empty_uniform_rowcol_std_threshold
+                            ),
                         ),
                     )
                     p["embed_image"] = cropped_img
@@ -258,14 +269,14 @@ class ProcessingPipeline:
                     p["embed_image"] = raw_img
                     p["crop_meta"] = None
                     images_to_embed.append(raw_img)
-            
+
             embeddings, token_infos = self.embedder.embed_images(
                 images_to_embed,
                 batch_size=self.embedding_batch_size,
                 return_token_info=True,
                 show_progress=True,
             )
-            
+
             for idx, page_info in enumerate(pages_to_process):
                 raw_img = page_info["raw_image"]
                 embed_img = page_info["embed_image"]
@@ -275,10 +286,15 @@ class ProcessingPipeline:
                 text = page_info["text"]
                 embedding = embeddings[idx]
                 token_info = token_infos[idx]
-                
+
                 if progress_callback:
-                    progress_callback("process", page_num, total_pages, f"Processing page {page_num}/{total_pages}")
-                
+                    progress_callback(
+                        "process",
+                        page_num,
+                        total_pages,
+                        f"Processing page {page_num}/{total_pages}",
+                    )
+
                 try:
                     page_data = self._process_single_page(
                         filename=filename,
@@ -295,34 +311,36 @@ class ProcessingPipeline:
                         upload_to_cloudinary=upload_to_cloudinary,
                         crop_meta=crop_meta,
                     )
-                    
+
                     all_pages.append(page_data)
-                    
+
                     if upload_to_qdrant and self.indexer:
                         upload_queue.append(page_data)
-                        
+
                         # Upload in batches
                         if len(upload_queue) >= self.upload_batch_size:
                             count = self._upload_batch(upload_queue)
                             uploaded += count
                             upload_queue = []
-                    
+
                 except Exception as e:
                     logger.error(f"   ❌ Failed page {page_num}: {e}")
                     failed += 1
-            
+
             # Memory cleanup
             gc.collect()
             if torch.cuda.is_available():
                 torch.cuda.empty_cache()
-        
+
         # Upload remaining pages
         if upload_queue and upload_to_qdrant and self.indexer:
             count = self._upload_batch(upload_queue)
             uploaded += count
-        
-        logger.info(f"✅ Completed {filename}: {uploaded} uploaded, {skipped} skipped, {failed} failed")
-        
+
+        logger.info(
+            f"✅ Completed {filename}: {uploaded} uploaded, {skipped} skipped, {failed} failed"
+        )
+
         return {
             "filename": filename,
             "total_pages": total_pages,
@@ -331,7 +349,7 @@ class ProcessingPipeline:
             "failed": failed,
             "pages": all_pages,
         }
-    
+
     def _process_single_page(
         self,
         filename: str,
@@ -350,17 +368,17 @@ class ProcessingPipeline:
     ) -> Dict[str, Any]:
         """Process a single page with full metadata for saliency."""
         from visual_rag.embedding.pooling import global_mean_pooling
-        
+
         # Resize image for ColPali
         resized_img, tile_rows, tile_cols = self.pdf_processor.resize_for_colpali(embed_img)
-        
+
         # Use processor's tile info if available (more accurate)
         proc_n_rows = token_info.get("n_rows")
         proc_n_cols = token_info.get("n_cols")
         if proc_n_rows and proc_n_cols:
             tile_rows = proc_n_rows
             tile_cols = proc_n_cols
-        
+
         # Convert embedding to numpy
         if isinstance(embedding, torch.Tensor):
             if embedding.dtype == torch.bfloat16:
@@ -370,18 +388,18 @@ class ProcessingPipeline:
         else:
             full_embedding = np.array(embedding)
         full_embedding = full_embedding.astype(np.float32)
-        
+
         # Token info for metadata
         visual_indices = token_info["visual_token_indices"]
         num_visual_tokens = token_info["num_visual_tokens"]
-        
+
         # =========================================================================
         # STRATEGY: "pooling" (NOVEL) vs "standard" (BASELINE) vs "all" (BOTH)
         # =========================================================================
-        
+
         # Always compute visual-only embedding (needed for pooling and saliency)
         visual_embedding = full_embedding[visual_indices]
-        
+
         # Mean pooling cap: <=0 or None => no cap (ColQwen2.5 keeps all effective rows).
         tv = self.max_mean_pool_vectors
         if tv is not None:
@@ -397,46 +415,71 @@ class ProcessingPipeline:
         model_lower = (getattr(self.embedder, "model_name", "") or "").lower()
         is_colqwen25 = "colqwen2.5" in model_lower or "colqwen2_5" in model_lower
         is_colsmol = "colsmol" in model_lower
-        kernel_arg = str(getattr(self, "experimental_pooling_kernel", "auto") or "auto").lower().strip()
-        if kernel_arg == "auto":
-            kernel = "gaussian" if is_colqwen25 else "legacy"
-        else:
-            kernel = kernel_arg
-        default_k = 5 if is_colqwen25 else 3
-        if kernel != "legacy":
-            default_k = 3
-        ks = self.pooling_windows if self.pooling_windows else [default_k]
-        # Normalize + keep order, avoid duplicates.
-        seen_ks = set()
-        ks_norm: List[int] = []
-        for k in ks:
-            try:
-                ki = int(k)
-            except Exception:
-                continue
-            if ki <= 0:
-                continue
-            if ki in seen_ks:
-                continue
-            seen_ks.add(ki)
-            ks_norm.append(ki)
-        if not ks_norm:
-            ks_norm = [default_k]
-
         experimental_pooled_by_name: Dict[str, Any] = {}
-        canonical_k = int(ks_norm[0])
-        for k in ks_norm:
-            exp = self.embedder.experimental_pool_visual_embedding(
+        # ColQwen2.5: keep experimental_pooling as Gaussian alias, and always store both variants explicitly.
+        if is_colqwen25:
+            exp_gaussian = self.embedder.experimental_pool_visual_embedding(
                 visual_embedding,
                 token_info,
                 target_vectors=tv,
                 mean_pool=tile_pooled,
-                window_size=int(k),
-                kernel=str(kernel),
+                window_size=3,
+                kernel="gaussian",
             )
-            experimental_pooled_by_name[f"experimental_pooling_{int(k)}"] = exp
-            if int(k) == canonical_k:
-                experimental_pooled_by_name["experimental_pooling"] = exp
+            exp_triangular = self.embedder.experimental_pool_visual_embedding(
+                visual_embedding,
+                token_info,
+                target_vectors=tv,
+                mean_pool=tile_pooled,
+                window_size=3,
+                kernel="triangular",
+            )
+            experimental_pooled_by_name["experimental_pooling"] = exp_gaussian
+            experimental_pooled_by_name["experimental_pooling_gaussian"] = exp_gaussian
+            experimental_pooled_by_name["experimental_pooling_triangular"] = exp_triangular
+            ks_norm = [3]
+            canonical_k = 3
+            kernel = "gaussian"
+        else:
+            kernel_arg = (
+                str(getattr(self, "experimental_pooling_kernel", "auto") or "auto").lower().strip()
+            )
+            if kernel_arg == "auto":
+                kernel = "legacy"
+            else:
+                kernel = kernel_arg
+            default_k = 3
+            ks = self.pooling_windows if self.pooling_windows else [default_k]
+            # Normalize + keep order, avoid duplicates.
+            seen_ks = set()
+            ks_norm = []
+            for k in ks:
+                try:
+                    ki = int(k)
+                except Exception:
+                    continue
+                if ki <= 0:
+                    continue
+                if ki in seen_ks:
+                    continue
+                seen_ks.add(ki)
+                ks_norm.append(ki)
+            if not ks_norm:
+                ks_norm = [default_k]
+
+            canonical_k = int(ks_norm[0])
+            for k in ks_norm:
+                exp = self.embedder.experimental_pool_visual_embedding(
+                    visual_embedding,
+                    token_info,
+                    target_vectors=tv,
+                    mean_pool=tile_pooled,
+                    window_size=int(k),
+                    kernel=str(kernel),
+                )
+                experimental_pooled_by_name[f"experimental_pooling_{int(k)}"] = exp
+                if int(k) == canonical_k:
+                    experimental_pooled_by_name["experimental_pooling"] = exp
 
         if is_colsmol and bool(getattr(self, "colsmol_experimental_2d", False)):
             try:
@@ -457,7 +500,11 @@ class ProcessingPipeline:
             except Exception:
                 pass
         global_pooled = global_mean_pooling(full_embedding)
-        global_pooling = self.embedder.global_pool_from_mean_pool(tile_pooled) if tile_pooled.size else global_pooled
+        global_pooling = (
+            self.embedder.global_pool_from_mean_pool(tile_pooled)
+            if tile_pooled.size
+            else global_pooled
+        )
 
         num_tiles = int(tile_pooled.shape[0])
         patches_per_tile = int(visual_embedding.shape[0] // max(num_tiles, 1)) if num_tiles else 0
@@ -466,62 +513,70 @@ class ProcessingPipeline:
         else:
             tile_rows = token_info.get("n_rows") or None
             tile_cols = token_info.get("n_cols") or None
-        
+
         if self.embedding_strategy == "pooling":
             # NOVEL APPROACH: Visual tokens only + tile-level pooling
             embedding_for_initial = visual_embedding
             embedding_for_pooling = tile_pooled
-            global_pooling = self.embedder.global_pool_from_mean_pool(tile_pooled) if tile_pooled.size else global_pooled
-            
+            global_pooling = (
+                self.embedder.global_pool_from_mean_pool(tile_pooled)
+                if tile_pooled.size
+                else global_pooled
+            )
+
         elif self.embedding_strategy == "standard":
             # BASELINE: All tokens + global mean
             embedding_for_initial = full_embedding
             embedding_for_pooling = global_pooled.reshape(1, -1)
             global_pooling = global_pooled
-            
+
         else:  # "all" - Push BOTH representations (efficient for comparison)
             # Embed once, store multiple vector representations
             # This allows comparing both strategies without re-embedding
             embedding_for_initial = visual_embedding  # Use visual for search
-            embedding_for_pooling = tile_pooled       # Use tile-level for fast prefetch
-            global_pooling = self.embedder.global_pool_from_mean_pool(tile_pooled) if tile_pooled.size else global_pooled
-            
+            embedding_for_pooling = tile_pooled  # Use tile-level for fast prefetch
+            global_pooling = (
+                self.embedder.global_pool_from_mean_pool(tile_pooled)
+                if tile_pooled.size
+                else global_pooled
+            )
+
             # ALSO store standard representations as additional vectors
             # These will be added to metadata for optional use
             pass  # Extra vectors handled in return dict below
-        
+
         # Upload to Cloudinary
         original_url = None
         cropped_url = None
         resized_url = None
-        
+
         if upload_to_cloudinary and self.cloudinary_uploader:
             base_filename = f"{pdf_stem}_page_{page_num}"
             if self.crop_empty:
-                original_url, cropped_url, resized_url = self.cloudinary_uploader.upload_original_cropped_and_resized(
-                    raw_img, embed_img, resized_img, base_filename
+                original_url, cropped_url, resized_url = (
+                    self.cloudinary_uploader.upload_original_cropped_and_resized(
+                        raw_img, embed_img, resized_img, base_filename
+                    )
                 )
             else:
                 original_url, resized_url = self.cloudinary_uploader.upload_original_and_resized(
                     raw_img, resized_img, base_filename
                 )
-        
+
         # Sanitize text
         safe_text = self._sanitize_text(text[:10000]) if text else ""
-        
+
         metadata = {
             "filename": filename,
             "page_number": page_num,
             "total_pages": total_pages,
             "has_text": bool(text and text.strip()),
             "text": safe_text,
-            
             # Image URLs
             "page": resized_url or "",  # For display
             "original_url": original_url or "",
             "cropped_url": cropped_url or "",
             "resized_url": resized_url or "",
-            
             # Dimensions (needed for saliency overlay)
             "original_width": raw_img.width,
             "original_height": raw_img.height,
@@ -529,117 +584,113 @@ class ProcessingPipeline:
             "cropped_height": int(embed_img.height) if self.crop_empty else int(raw_img.height),
             "resized_width": resized_img.width,
             "resized_height": resized_img.height,
-            
             # Tile structure (needed for saliency)
             "num_tiles": num_tiles,
             "tile_rows": tile_rows,
             "tile_cols": tile_cols,
             "patches_per_tile": patches_per_tile,
-            
             # Token info (needed for saliency)
             "num_visual_tokens": num_visual_tokens,
             "visual_token_indices": visual_indices,
             "total_tokens": len(full_embedding),  # Total tokens in raw embedding
-            
             # Strategy used (important for paper comparison)
             "embedding_strategy": self.embedding_strategy,
-
             "model_name": getattr(self.embedder, "model_name", None),
             "experimental_pooling_windows": ks_norm,
             "experimental_pooling_default_window": canonical_k,
             "experimental_pooling_kernel": str(kernel),
-            "colsmol_experimental_2d": bool(getattr(self, "colsmol_experimental_2d", False))
-            if is_colsmol
-            else None,
+            "colsmol_experimental_2d": (
+                bool(getattr(self, "colsmol_experimental_2d", False)) if is_colsmol else None
+            ),
             "max_mean_pool_vectors": (
                 int(self.max_mean_pool_vectors) if self.max_mean_pool_vectors is not None else None
             ),
-
             "crop_empty_enabled": bool(self.crop_empty),
             "crop_empty_crop_box": (crop_meta or {}).get("crop_box"),
             "crop_empty_remove_page_number": bool(self.crop_empty_remove_page_number),
             "crop_empty_percentage_to_remove": float(self.crop_empty_percentage_to_remove),
             "crop_empty_preserve_border_px": int(self.crop_empty_preserve_border_px),
-            "crop_empty_uniform_rowcol_std_threshold": float(self.crop_empty_uniform_rowcol_std_threshold),
-            
+            "crop_empty_uniform_rowcol_std_threshold": float(
+                self.crop_empty_uniform_rowcol_std_threshold
+            ),
             # Extra metadata (year, district, etc.)
             **extra_metadata,
         }
-        
+
         result = {
             "id": chunk_id,
-            "visual_embedding": embedding_for_initial,    # "initial" vector in Qdrant
+            "visual_embedding": embedding_for_initial,  # "initial" vector in Qdrant
             "tile_pooled_embedding": embedding_for_pooling,  # "mean_pooling" vector in Qdrant
-            "experimental_pooled_embedding": experimental_pooled_by_name,  # multiple "experimental_pooling_{k}"
+            "experimental_pooled_embedding": experimental_pooled_by_name,  # ColQwen: gaussian/triangular; ColPali: "experimental_pooling_{k}"
             "global_pooled_embedding": global_pooling,  # "global_pooling" vector in Qdrant
             "metadata": metadata,
             "image": raw_img,
             "resized_image": resized_img,
         }
-        
+
         # For "all" strategy, include BOTH representations for comparison
         if self.embedding_strategy == "all":
             result["extra_vectors"] = {
                 # Standard baseline vectors (for comparison)
-                "full_embedding": full_embedding,           # All tokens [total, 128]
-                "global_pooled": global_pooled,             # Global mean [128]
+                "full_embedding": full_embedding,  # All tokens [total, 128]
+                "global_pooled": global_pooled,  # Global mean [128]
                 # Pooling vectors (already in main result)
-                "visual_embedding": visual_embedding,       # Visual only [visual, 128]  
-                "tile_pooled": tile_pooled,                 # Tile-level [tiles, 128]
+                "visual_embedding": visual_embedding,  # Visual only [visual, 128]
+                "tile_pooled": tile_pooled,  # Tile-level [tiles, 128]
             }
-        
+
         return result
-    
+
     def _upload_batch(self, upload_queue: List[Dict[str, Any]]) -> int:
         """Upload batch to Qdrant."""
         if not upload_queue or not self.indexer:
             return 0
-        
+
         logger.info(f"📤 Uploading batch of {len(upload_queue)} pages...")
-        
+
         count = self.indexer.upload_batch(
             upload_queue,
             delay_between_batches=self.delay_between_uploads,
         )
-        
+
         return count
-    
+
     def _get_extra_metadata(self, filename: str) -> Dict[str, Any]:
         """Get extra metadata for a filename."""
         if not self.metadata_mapping:
             return {}
-        
+
         # Normalize filename
         filename_clean = filename.replace(".pdf", "").replace(".PDF", "").strip().lower()
-        
+
         # Try exact match
         if filename_clean in self.metadata_mapping:
             return self.metadata_mapping[filename_clean].copy()
-        
+
         # Try fuzzy match
         from difflib import SequenceMatcher
-        
+
         best_match = None
         best_score = 0.0
-        
+
         for known_filename, metadata in self.metadata_mapping.items():
             score = SequenceMatcher(None, filename_clean, known_filename.lower()).ratio()
             if score > best_score and score > 0.75:
                 best_score = score
                 best_match = metadata
-        
+
         if best_match:
             logger.debug(f"Fuzzy matched '{filename}' with score {best_score:.2f}")
             return best_match.copy()
-        
+
         return {}
-    
+
     def _sanitize_text(self, text: str) -> str:
         """Remove invalid Unicode characters."""
         if not text:
             return ""
         return text.encode("utf-8", errors="surrogatepass").decode("utf-8", errors="ignore")
-    
+
     @staticmethod
     def generate_chunk_id(filename: str, page_number: int) -> str:
         """Generate deterministic chunk ID."""
@@ -647,12 +698,12 @@ class ProcessingPipeline:
         hash_obj = hashlib.sha256(content.encode())
         hex_str = hash_obj.hexdigest()[:32]
         return f"{hex_str[:8]}-{hex_str[8:12]}-{hex_str[12:16]}-{hex_str[16:20]}-{hex_str[20:32]}"
-    
+
     @staticmethod
     def load_metadata_mapping(json_path: Path) -> Dict[str, Dict[str, Any]]:
         """
         Load metadata mapping from JSON file.
-        
+
         Expected format:
         {
             "filenames": {
@@ -660,7 +711,7 @@ class ProcessingPipeline:
                 ...
             }
         }
-        
+
         Or simple format:
         {
             "Report Name 2023": {"year": 2023, "source": "Local Government", ...},
@@ -668,22 +719,21 @@ class ProcessingPipeline:
         }
         """
         import json
-        
+
         with open(json_path, "r") as f:
             data = json.load(f)
-        
+
         # Check if nested under "filenames"
         if "filenames" in data and isinstance(data["filenames"], dict):
             mapping = data["filenames"]
         else:
             mapping = data
-        
+
         # Normalize keys to lowercase
         normalized = {}
         for filename, metadata in mapping.items():
             key = filename.lower().strip().replace(".pdf", "")
             normalized[key] = metadata
-        
+
         logger.info(f"📖 Loaded metadata for {len(normalized)} files")
         return normalized
-

visual_rag/indexing/qdrant_indexer.py

@@ -144,7 +144,8 @@ class QdrantIndexer:
         - initial: Full multi-vector embeddings (num_patches × dim)
         - mean_pooling: Tile-level pooled vectors (num_tiles × dim)
         - experimental_pooling: Experimental multi-vector pooling (varies by model)
-        - experimental_pooling_{k}: Optional additional experimental poolings with different window sizes
+        - experimental_pooling_{k}: (ColPali) Optional additional experimental poolings with different window sizes
+        - experimental_pooling_gaussian / experimental_pooling_triangular: (ColQwen) Technique variants (k=3)
         - global_pooling: Single vector pooled representation (dim)
 
         Args:
@@ -437,9 +438,9 @@ class QdrantIndexer:
                             self._np_vector_dtype, copy=False
                         )
                 elif exp_val is not None:
-                    exp_vectors["experimental_pooling"] = np.array(exp_val, dtype=np.float32).astype(
-                        self._np_vector_dtype, copy=False
-                    )
+                    exp_vectors["experimental_pooling"] = np.array(
+                        exp_val, dtype=np.float32
+                    ).astype(self._np_vector_dtype, copy=False)
 
                 qdrant_points.append(
                     qdrant_models.PointStruct(

visual_rag/qdrant_admin.py

@@ -35,9 +35,7 @@ def _resolve_qdrant_connection(
     _maybe_load_dotenv()
     resolved_url = url or os.getenv("QDRANT_URL")
     if not resolved_url:
-        raise ValueError(
-            "Qdrant URL not set (pass url= or set QDRANT_URL)."
-        )
+        raise ValueError("Qdrant URL not set (pass url= or set QDRANT_URL).")
     resolved_key = api_key or os.getenv("QDRANT_API_KEY")
     return QdrantConnection(url=str(resolved_url), api_key=resolved_key)
 

visual_rag/retrieval/multi_vector.py

@@ -127,6 +127,7 @@ class MultiVectorRetriever:
         self._single_stage = SingleStageRetriever(
             qdrant_client=qdrant_client,
             collection_name=collection_name,
+            experimental_vector_name=str(experimental_vector_name),
             request_timeout=request_timeout,
             max_retries=max_retries,
             retry_sleep=retry_sleep,
@@ -195,12 +196,35 @@ class MultiVectorRetriever:
                 filter_obj=filter_obj,
                 strategy="multi_vector",
             )
-        elif mode == "single_pooled":
+        elif mode in ("single_tiles", "single_pooled"):
+            # single_tiles: MaxSim on pooled vectors (multi-vector)
+            # single_pooled: backward-compatible alias (pooled query vs pooled vectors)
+            return self._single_stage.search(
+                query_embedding=query_embedding,
+                top_k=top_k,
+                filter_obj=filter_obj,
+                strategy=("tiles_maxsim" if mode == "single_tiles" else "pooled_tile"),
+            )
+        elif mode == "single_global":
+            return self._single_stage.search(
+                query_embedding=query_embedding,
+                top_k=top_k,
+                filter_obj=filter_obj,
+                strategy="pooled_global",
+            )
+        elif mode == "single_experimental_tokens":
+            return self._single_stage.search(
+                query_embedding=query_embedding,
+                top_k=top_k,
+                filter_obj=filter_obj,
+                strategy="experimental_maxsim",
+            )
+        elif mode == "single_experimental_pooled":
             return self._single_stage.search(
                 query_embedding=query_embedding,
                 top_k=top_k,
                 filter_obj=filter_obj,
-                strategy="pooled_tile",
+                strategy="pooled_experimental",
             )
         elif mode == "two_stage":
             return self._two_stage.search_server_side(

visual_rag/retrieval/single_stage.py

@@ -26,6 +26,8 @@ class SingleStageRetriever:
     - tiles_maxsim: Native MaxSim between query tokens and tile vectors (using="mean_pooling")
     - pooled_tile: Pooled query vs tile vectors (using="mean_pooling")
     - pooled_global: Pooled query vs global pooled doc vector (using="global_pooling")
+    - experimental_maxsim: Native MaxSim between query tokens and experimental vectors (using="experimental_pooling[_k]")
+    - pooled_experimental: Pooled query vs experimental vectors (using="experimental_pooling[_k]")
 
     Args:
         qdrant_client: Connected Qdrant client
@@ -43,12 +45,14 @@ class SingleStageRetriever:
         self,
         qdrant_client,
         collection_name: str,
+        experimental_vector_name: str = "experimental_pooling",
         request_timeout: int = 120,
         max_retries: int = 3,
         retry_sleep: float = 1.0,
     ):
         self.client = qdrant_client
         self.collection_name = collection_name
+        self.experimental_vector_name = str(experimental_vector_name)
         self.request_timeout = int(request_timeout)
         self.max_retries = max_retries
         self.retry_sleep = retry_sleep
@@ -100,6 +104,19 @@ class SingleStageRetriever:
             query_vector = query_pooled.tolist()
             logger.debug(f"🔍 Global-pooled search on '{vector_name}'")
 
+        elif strategy == "experimental_maxsim":
+            # Native multi-vector MaxSim against experimental pooled vectors
+            vector_name = self.experimental_vector_name
+            query_vector = query_np.tolist()
+            logger.debug(f"🎯 Experimental MaxSim search on '{vector_name}'")
+
+        elif strategy == "pooled_experimental":
+            # Pooled query vs experimental pooled vectors
+            vector_name = self.experimental_vector_name
+            query_pooled = query_np.mean(axis=0)
+            query_vector = query_pooled.tolist()
+            logger.debug(f"🔍 Experimental pooled search on '{vector_name}'")
+
         else:
             raise ValueError(f"Unknown strategy: {strategy}")