streamlit app

.streamlit/config.toml

@@ -0,0 +1,9 @@
+[theme]
+base = "light"
+primaryColor = "#002BFF"
+backgroundColor = "#FAFAFA"
+secondaryBackgroundColor = "#F0F0F0"
+textColor = "#0A0A0A"
+
+[server]
+headless = true

app.py

@@ -0,0 +1,511 @@
+"""Comic panel search UI."""
+
+import base64
+import os
+import sys
+from io import BytesIO
+from pathlib import Path
+
+import streamlit as st
+from dotenv import load_dotenv
+from PIL import Image
+
+ROOT = Path(__file__).parent
+sys.path.insert(0, str(ROOT))
+load_dotenv(ROOT / ".env")
+
+st.set_page_config(
+    page_title="Comic Panel Search",
+    page_icon="💥",
+    layout="wide",
+    initial_sidebar_state="collapsed",
+)
+
+st.markdown("""
+<style>
+  #MainMenu, footer, [data-testid="collapsedControl"] { visibility: hidden; }
+  .block-container { padding-top: 1.5rem !important; max-width: 1400px !important; }
+
+  /* result card */
+  .rc {
+    background: #fff;
+    border: 1px solid #e5e5e5;
+    border-radius: 8px;
+    padding: 18px 20px;
+    display: grid;
+    grid-template-columns: 62px 1fr;
+    gap: 18px;
+    margin-bottom: 10px;
+  }
+  .rc:hover { box-shadow: 0 2px 10px rgba(0,0,0,0.07); }
+  .rc-left {
+    display: flex;
+    flex-direction: column;
+    align-items: center;
+    gap: 8px;
+    padding-top: 2px;
+  }
+  .rc-rank {
+    width: 40px; height: 40px;
+    background: rgba(0,43,255,0.05);
+    border-radius: 6px;
+    display: flex; align-items: center; justify-content: center;
+    font-weight: 700; font-size: 15px; color: #0a0a0a;
+    flex-shrink: 0;
+  }
+  .rc-score-lbl {
+    font-size: 10px; font-weight: 600; text-transform: uppercase;
+    letter-spacing: 0.06em; color: #737373; text-align: center;
+  }
+  .rc-score-val {
+    font-family: monospace; font-size: 12px;
+    color: #0a0a0a; text-align: center;
+  }
+  .rc-tag {
+    font-size: 10px; background: rgba(0,43,255,0.06);
+    color: #002BFF; border-radius: 4px;
+    padding: 2px 5px; font-weight: 600;
+    display: inline-block; margin: 1px;
+  }
+  .rc-right {
+    display: flex; flex-direction: column; gap: 8px; min-width: 0;
+  }
+  .rc-img {
+    max-width: 100%; max-height: 200px;
+    width: auto; height: auto;
+    object-fit: contain; border-radius: 4px;
+    display: block; background: #f5f5f5;
+  }
+  .rc-id {
+    font-family: monospace; font-size: 13px;
+    font-weight: 600; color: #0a0a0a;
+    overflow: hidden; text-overflow: ellipsis; white-space: nowrap;
+  }
+  .rc-field {
+    display: grid; grid-template-columns: 95px 1fr;
+    gap: 6px; font-size: 12px; line-height: 1.5;
+  }
+  .rc-key {
+    color: #737373; font-family: monospace;
+    font-size: 11px; overflow: hidden;
+    text-overflow: ellipsis; white-space: nowrap;
+  }
+  .rc-val {
+    color: #0a0a0a; word-break: break-word;
+  }
+  .rh {
+    font-size: 13px; color: #737373;
+    padding-bottom: 12px;
+    border-bottom: 1px solid #e5e5e5;
+    margin-bottom: 14px;
+  }
+  .rh strong { color: #0a0a0a; }
+  .ph {
+    text-align: center; padding: 72px 0;
+    color: #737373; font-size: 14px;
+  }
+  .section-label {
+    font-size: 11px; font-weight: 600;
+    text-transform: uppercase; letter-spacing: 0.08em;
+    color: #737373; margin: 0;
+  }
+</style>
+""", unsafe_allow_html=True)
+
+# ── constants & resources ─────────────────────────────────────────────────────
+
+NAMESPACE   = os.environ.get("PINECONE_NAMESPACE", "comics-v1")
+SHOW_FIELDS = ["comic_id", "book_id", "page_num", "panel_num", "source", "ocr_text"]
+
+OPS = ["==", "!=", ">", ">=", "<", "<=", "In", "Not In", "Match phrase", "Match all", "Match any"]
+_MATCH_OPS = {"Match phrase", "Match all", "Match any"}
+_OP_MAP = {
+    "==": "$eq",  "!=": "$ne",
+    ">":  "$gt",  ">=": "$gte",
+    "<":  "$lt",  "<=": "$lte",
+    "In": "$in",  "Not In": "$nin",
+    "Match phrase": "$match_phrase",
+    "Match all":    "$match_all",
+    "Match any":    "$match_any",
+}
+ALL_FIELDS = ["comic_id", "book_id", "page_num", "panel_num", "source", "ocr_text", "image_path", "is_ad_page"]
+
+@st.cache_resource(show_spinner="Connecting to Pinecone…")
+def _index():
+    from pinecone import Pinecone
+    pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"])
+    return pc.preview.index(name=os.environ.get("PINECONE_INDEX_NAME", "comic-panels"))
+
+@st.cache_resource(show_spinner="Loading CLIP model…")
+def _load_clip():
+    from src.embeddings.embed_images import _init_model
+    _init_model()
+
+# ── helpers ───────────────────────────────────────────────────────────────────
+
+def _b64(path: str) -> str | None:
+    try:
+        p = Path(path)
+        if not p.is_absolute():
+            p = ROOT / p
+        img = Image.open(p).convert("RGB")
+        scale = min(1.0, 500 / img.width)
+        img = img.resize((int(img.width * scale), int(img.height * scale)), Image.LANCZOS)
+        buf = BytesIO()
+        img.save(buf, "JPEG", quality=82)
+        return base64.b64encode(buf.getvalue()).decode()
+    except Exception:
+        return None
+
+def _esc(s: str) -> str:
+    return s.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
+
+def _coerce(s: str):
+    try: return int(s)
+    except ValueError: pass
+    try: return float(s)
+    except ValueError: pass
+    if s.lower() in ("true", "yes"): return True
+    if s.lower() in ("false", "no"): return False
+    return s
+
+def _add_filter():
+    fid = st.session_state.filter_next_id
+    st.session_state.filter_rows = st.session_state.filter_rows + [fid]
+    st.session_state.filter_next_id += 1
+
+def _remove_filter(fid: int):
+    st.session_state.filter_rows = [r for r in st.session_state.filter_rows if r != fid]
+
+def _build_filters() -> dict:
+    row_conds = []
+    for fid in st.session_state.filter_rows:
+        k       = st.session_state.get(f"fk_{fid}", "").strip()
+        op      = st.session_state.get(f"fo_{fid}", "==")
+        val_str = st.session_state.get(f"fv_{fid}", "").strip()
+        if not k or not val_str:
+            continue
+        if op in ("In", "Not In"):
+            vals = [_coerce(v.strip()) for v in val_str.split(",")]
+            row_conds.append({k: {"$in" if op == "In" else "$nin": vals}})
+        elif op in _MATCH_OPS:
+            row_conds.append({k: {_OP_MAP[op]: val_str}})
+        else:
+            row_conds.append({k: {_OP_MAP[op]: _coerce(val_str)}})
+
+    combinator = st.session_state.get("filter_combinator", "And")
+    if len(row_conds) == 0:
+        user_filter = None
+    elif len(row_conds) == 1:
+        user_filter = row_conds[0]
+    elif combinator == "Or":
+        user_filter = {"$or": row_conds}
+    else:
+        user_filter = {"$and": row_conds}
+
+    exclude = st.session_state.get("exclude_ads", True)
+    if exclude and user_filter:
+        return {"$and": [{"is_ad_page": {"$eq": False}}, user_filter]}
+    elif exclude:
+        return {"is_ad_page": {"$eq": False}}
+    elif user_filter:
+        return user_filter
+    return {}
+
+def _card(hit: dict, rank: int, show_fields: list):
+    score   = hit.get("_score") or hit.get("rrf_score") or 0.0
+    sources = hit.get("sources") or []
+    hit_id  = _esc(str(hit.get("_id", "—")))
+
+    tags = "".join(f'<span class="rc-tag">{s}</span>' for s in sources)
+
+    img_html = ""
+    if b64 := _b64(hit.get("image_path", "")):
+        img_html = f'<img class="rc-img" src="data:image/jpeg;base64,{b64}"/>'
+
+    fields = ""
+    for k in show_fields:
+        v = hit.get(k)
+        if v is None or str(v).strip() == "":
+            continue
+        s = str(v)
+        if len(s) > 500:
+            s = s[:500] + "…"
+        fields += f'<div class="rc-field"><span class="rc-key">{k}</span><span class="rc-val">{_esc(s)}</span></div>'
+
+    st.markdown(f"""
+<div class="rc">
+  <div class="rc-left">
+    <div class="rc-rank">{rank}</div>
+    <div class="rc-score-lbl">Score</div>
+    <div class="rc-score-val">{score:.4f}</div>
+    {tags}
+  </div>
+  <div class="rc-right">
+    {img_html}
+    <div class="rc-id">_id: {hit_id}</div>
+    {fields}
+  </div>
+</div>""", unsafe_allow_html=True)
+
+# ── example queries ───────────────────────────────────────────────────────────
+
+EXAMPLES = {
+    "🎨 Visual (Dense)": [
+        ("Hero flying",             "hero in cape flying through sky"),
+        ("Fistfight",               "two men punching fighting brawl"),
+        ("Explosion",               "explosion fire destruction chaos"),
+        ("Scientist in lab",        "scientist laboratory chemicals"),
+        ("Rocket in space",         "spaceship rocket outer space stars"),
+        ("Villain menacing",        "villain sinister evil grin"),
+        ("Romance",                 "couple romance embrace kissing"),
+        ("Detective",               "detective investigating crime scene"),
+        ("Car chase",               "car chase street pursuit"),
+        ("Monster horror",          "monster creature attacking horror"),
+        ("Woman captured",          "woman captured tied up danger"),
+        ("Man with gun",            "man pointing gun threatening"),
+    ],
+    "🔤 Keyword (Sparse)": [
+        ("Secret formula",          "secret formula"),
+        ("Help / danger",           "help me danger"),
+        ("Villain escape",          "villain escape capture"),
+        ("Mysterious stranger",     "mysterious stranger"),
+        ("Great power",             "great power"),
+        ("Bank robbery",            "robbery bank heist money"),
+        ("Crime mystery",           "crime mystery clue evidence"),
+        ("Space adventure",         "space adventure rocket planet"),
+    ],
+    "💥 Sound effects (FTS)": [
+        ("POW / BANG / ZAP",        "BANG OR POW OR ZAP"),
+        ("KAPOW / WHAM / CRASH",    "KAPOW OR WHAM OR CRASH"),
+        ("BOOM / BLAST",            "BOOM OR BLAST OR KA-BOOM"),
+        ("THUD / SMASH",            "THUD OR CRUNCH OR SMASH"),
+    ],
+    "💬 Phrases (FTS)": [
+        ('"secret formula"',        '"secret formula"'),
+        ('"great danger"',          '"great danger"'),
+        ("detective AND murder",    "detective AND murder"),
+        ("villain AND escape",      "villain AND escape"),
+        ("danger AND rescue",       "danger AND rescue"),
+    ],
+}
+
+# ── session state ─────────────────────────────────────────────────────────────
+
+for k, v in {
+    "fts_on": True, "dense_on": False, "sparse_on": False,
+    "fts_q": "*", "dense_q": "", "sparse_q": "",
+    "fts_type": "Query string",
+    "top_k": 20, "exclude_ads": True,
+    "filter_rows": [], "filter_next_id": 0, "filter_combinator": "And",
+    "show_fields": ["ocr_text"],
+    "results": None, "result_meta": {},
+    "run": False,
+}.items():
+    st.session_state.setdefault(k, v)
+
+
+def _use_example(query: str, signal: str):
+    """on_click callback — runs before rerun so widget keys can be set safely."""
+    if signal == "dense":
+        st.session_state.dense_on = True
+        st.session_state.dense_q  = query
+    elif signal == "sparse":
+        st.session_state.sparse_on = True
+        st.session_state.sparse_q  = query
+    else:
+        st.session_state.fts_on = True
+        st.session_state.fts_q  = query
+    st.session_state.run = True
+
+# ── header ────────────────────────────────────────────────────────────────────
+
+st.markdown("## Browse your index")
+st.caption(f"comic-panels · 1,229,664 panels · namespace: {NAMESPACE}")
+
+# ── two-column layout ─────────────────────────────────────────────────────────
+
+left, right = st.columns([1, 2.8], gap="large")
+
+with left:
+    # ── Search signals ────────────────────────────────────────────────────
+    st.markdown('<p class="section-label">Search signals</p>', unsafe_allow_html=True)
+
+    # Full-text
+    fts_on = st.toggle("Full-text search", key="fts_on")
+    if fts_on:
+        st.selectbox("Type", ["Query string", "BM25 (text)"], key="fts_type",
+                     label_visibility="collapsed")
+        st.text_input(
+            "fts_query", key="fts_q", label_visibility="collapsed",
+            placeholder='"phrase" OR keyword  ·  BANG OR POW  ·  field AND value',
+        )
+
+    # Dense
+    dense_on = st.toggle("Dense — visual similarity", key="dense_on")
+    if dense_on:
+        st.text_input(
+            "dense_query", key="dense_q", label_visibility="collapsed",
+            placeholder="describe what you see: hero flying through sky",
+            help="Embedded with OpenCLIP ViT-B/16",
+        )
+
+    # Sparse
+    sparse_on = st.toggle("Sparse — keyword expansion", key="sparse_on")
+    if sparse_on:
+        st.text_input(
+            "sparse_query", key="sparse_q", label_visibility="collapsed",
+            placeholder="keywords: secret formula villain escape",
+            help="Embedded with pinecone-sparse-english-v0",
+        )
+
+    st.divider()
+
+    # ── Filters ───────────────────────────────────────────────────────────
+    fhdr_l, fhdr_r = st.columns([1, 2])
+    with fhdr_l:
+        st.markdown('<p class="section-label">Filters</p>', unsafe_allow_html=True)
+    if st.session_state.filter_rows:
+        with fhdr_r:
+            st.selectbox("Combine", ["And", "Or"], key="filter_combinator",
+                         label_visibility="collapsed",
+                         help="How to combine multiple filter conditions")
+
+    for fid in st.session_state.filter_rows:
+        op  = st.session_state.get(f"fo_{fid}", "==")
+        op_w = max(2.0, min(3.2, len(op) * 0.25))
+        fv_w = (6.4 - op_w - 0.4) / 2
+        c1, c2, c3, c4 = st.columns([fv_w, op_w, fv_w, 0.4])
+        with c1:
+            st.text_input("Key", key=f"fk_{fid}",
+                          label_visibility="collapsed", placeholder="field")
+        with c2:
+            st.selectbox("Op", OPS, key=f"fo_{fid}", label_visibility="collapsed")
+        with c3:
+            st.text_input("Value", key=f"fv_{fid}",
+                          label_visibility="collapsed", placeholder="value")
+        with c4:
+            st.button("✕", key=f"fr_{fid}",
+                      on_click=_remove_filter, args=(fid,),
+                      use_container_width=True)
+
+    st.button("+ Add filter", key="add_filter",
+              on_click=_add_filter, use_container_width=True)
+    st.checkbox("Exclude ad pages", key="exclude_ads")
+
+    st.divider()
+
+    # ── Results ───────────────────────────────────────────────────────────
+    st.markdown('<p class="section-label">Results</p>', unsafe_allow_html=True)
+    st.number_input("Top K", min_value=1, max_value=200, step=5, key="top_k")
+    st.multiselect("Fields", ALL_FIELDS, key="show_fields",
+                   label_visibility="collapsed",
+                   placeholder="Fields to display in results")
+
+    st.write("")
+    if st.button("Run query", type="primary", use_container_width=True):
+        st.session_state.run = True
+
+    st.divider()
+
+    # ── Examples ──────────────────────────────────────────────────────────
+    with st.expander("💡 Example queries"):
+        for category, items in EXAMPLES.items():
+            st.markdown(f'<p class="section-label" style="margin:8px 0 4px">{category}</p>',
+                        unsafe_allow_html=True)
+            for label, query in items:
+                if "Visual" in category:
+                    sig = "dense"
+                elif "Keyword" in category:
+                    sig = "sparse"
+                else:
+                    sig = "fts"
+                st.button(label, key=f"ex·{label}", use_container_width=True,
+                          on_click=_use_example, args=(query, sig))
+
+# ── run search ────────────────────────────────────────────────────────────────
+
+if st.session_state.run:
+    st.session_state.run = False
+
+    fts_on    = st.session_state.fts_on
+    dense_on  = st.session_state.dense_on
+    sparse_on = st.session_state.sparse_on
+
+    if not any([fts_on, dense_on, sparse_on]):
+        st.session_state.result_meta = {"error": "Enable at least one search signal."}
+        st.session_state.results = []
+    else:
+        _load_clip()
+        idx     = _index()
+        filters = _build_filters()
+        top_k   = int(st.session_state.top_k)
+
+        from src.search.search_dense  import search_dense
+        from src.search.search_sparse import search_sparse
+        from src.search.search_fts    import search_fts
+        from src.search.fusion        import rrf_merge
+
+        groups, error = [], None
+        try:
+            if dense_on and st.session_state.dense_q.strip():
+                from src.embeddings.embed_images import embed_text_query
+                vec = embed_text_query(st.session_state.dense_q.strip())
+                r = search_dense(idx, NAMESPACE, vec, top_k=top_k, filters=filters)
+                groups.append(("dense", r.get("result", {}).get("hits", [])))
+
+            if sparse_on and st.session_state.sparse_q.strip():
+                from src.embeddings.embed_sparse_text import embed_sparse_query
+                svec = embed_sparse_query(st.session_state.sparse_q.strip())
+                if svec:
+                    r = search_sparse(idx, NAMESPACE, svec, top_k=top_k, filters=filters)
+                    groups.append(("sparse", r.get("result", {}).get("hits", [])))
+
+            if fts_on and st.session_state.fts_q.strip():
+                r = search_fts(idx, NAMESPACE, st.session_state.fts_q.strip(),
+                               top_k=top_k, filters=filters)
+                groups.append(("fts", r.get("result", {}).get("hits", [])))
+
+        except Exception as exc:
+            error = str(exc)
+
+        if error:
+            st.session_state.results     = []
+            st.session_state.result_meta = {"error": error}
+        elif not groups:
+            st.session_state.results     = []
+            st.session_state.result_meta = {"info": "No query text provided for the enabled signals."}
+        else:
+            merged = rrf_merge(groups) if len(groups) > 1 else groups[0][1]
+            src    = "hybrid · RRF" if len(groups) > 1 else groups[0][0]
+            st.session_state.results     = merged[:top_k]
+            st.session_state.result_meta = {"source": src, "top_k": top_k}
+
+# ── results panel ─────────────────────────────────────────────────────────────
+
+with right:
+    results = st.session_state.results
+    meta    = st.session_state.result_meta
+
+    if meta.get("error"):
+        st.error(meta["error"])
+    elif meta.get("info"):
+        st.warning(meta["info"])
+    elif results is None:
+        st.markdown('<div class="ph">Configure your query and click <strong>Run query</strong></div>',
+                    unsafe_allow_html=True)
+    elif len(results) == 0:
+        st.markdown('<div class="ph">No results — try adjusting your query or enabling more signals.</div>',
+                    unsafe_allow_html=True)
+    else:
+        src = meta.get("source", "")
+        tk  = meta.get("top_k", "?")
+        st.markdown(
+            f'<div class="rh">Search: <strong>{len(results)} results</strong>'
+            f'&nbsp;&nbsp;·&nbsp;&nbsp;top_k={tk}'
+            f'&nbsp;&nbsp;·&nbsp;&nbsp;{src}</div>',
+            unsafe_allow_html=True,
+        )
+        show_fields = st.session_state.show_fields or SHOW_FIELDS
+        for i, hit in enumerate(results, 1):
+            _card(hit, i, show_fields)

config.yaml

@@ -3,7 +3,7 @@ pinecone:
   namespace: comics-v1
 
 dense_embedding:
-  model_name: open_clip_ViT-B-32
+  model_name: open_clip_ViT-B-16
   field_name: image_dense
   dimension: 512
   metric: cosine

run_pipeline.sh

@@ -0,0 +1,41 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+PROJECT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+cd "$PROJECT_DIR"
+
+set -a && source .env && set +a
+
+PYTHON=".venv/bin/python"
+LOG_DIR="logs"
+mkdir -p "$LOG_DIR"
+
+log() { echo "[$(date '+%H:%M:%S')] $*" | tee -a "$LOG_DIR/pipeline.log"; }
+
+log "=== Pipeline start ==="
+
+log "Step 1: build_manifest"
+$PYTHON -m src.ingest.build_manifest 2>&1 | tee -a "$LOG_DIR/build_manifest.log"
+log "Step 1 done"
+
+log "Step 2: embed_images"
+$PYTHON -m src.embeddings.embed_images 2>&1 | tee -a "$LOG_DIR/embed_images.log"
+log "Step 2 done"
+
+log "Step 3: embed_sparse_text"
+$PYTHON -m src.embeddings.embed_sparse_text 2>&1 | tee -a "$LOG_DIR/embed_sparse_text.log"
+log "Step 3 done"
+
+log "Step 4: create_index (skips if exists)"
+$PYTHON -m src.pinecone.create_index 2>&1 | tee -a "$LOG_DIR/create_index.log"
+log "Step 4 done"
+
+log "Step 5: build_documents"
+$PYTHON -m src.pinecone.build_documents 2>&1 | tee -a "$LOG_DIR/build_documents.log"
+log "Step 5 done"
+
+log "Step 6: upsert_panels"
+$PYTHON -m src.pinecone.upsert_panels 2>&1 | tee -a "$LOG_DIR/upsert_panels.log"
+log "Step 6 done"
+
+log "=== Pipeline complete ==="

src/embeddings/embed_images.py

@@ -1,4 +1,9 @@
-"""Generate dense image embeddings for each panel using OpenCLIP."""
+"""Generate dense image embeddings for each panel using OpenCLIP.
+
+Processes panels in chunks of CHUNK_SIZE, writing one parquet file per chunk.
+Resumable: skips chunks whose output file already exists.
+Run merge_chunks() at the end or pass --merge to produce the final parquet.
+"""
 
 from pathlib import Path
 import os
@@ -6,14 +11,17 @@ import pandas as pd
 from PIL import Image
 from tqdm import tqdm
 import torch
+from torch.utils.data import Dataset, DataLoader
 import open_clip
 
 
 MANIFEST_PATH = Path("data/comics/processed/panels_manifest.parquet")
 OUTPUT_PATH = Path("data/comics/processed/image_dense_vectors.parquet")
+CHUNKS_DIR = Path("data/comics/processed/image_dense_chunks")
 FAILURES_PATH = Path("data/comics/processed/image_embedding_failures.csv")
+CHUNK_SIZE = 10_000
 
-_MODEL_NAME = os.environ.get("IMAGE_EMBED_MODEL", "open_clip_ViT-B-32")
+_MODEL_NAME = os.environ.get("IMAGE_EMBED_MODEL", "open_clip_ViT-B-16")
 _ARCH = _MODEL_NAME.replace("open_clip_", "")
 _DEVICE = (
     "mps" if torch.backends.mps.is_available()
@@ -21,22 +29,51 @@ _DEVICE = (
     else "cpu"
 )
 
-_model, _, _preprocess = open_clip.create_model_and_transforms(_ARCH, pretrained="openai")
-_model.eval()
-_model = _model.to(_DEVICE)
-_tokenizer = open_clip.get_tokenizer(_ARCH)
+# Lazy init so DataLoader workers don't each load the model on spawn
+_model = None
+_preprocess = None
+_tokenizer = None
+
+
+def _init_model():
+    global _model, _preprocess, _tokenizer
+    if _model is not None:
+        return
+    model, _, preprocess = open_clip.create_model_and_transforms(_ARCH, pretrained="openai")
+    model.eval()
+    model = model.to(_DEVICE)
+    _model = model
+    _preprocess = preprocess
+    _tokenizer = open_clip.get_tokenizer(_ARCH)
+
+
+class PanelDataset(Dataset):
+    def __init__(self, image_paths: list[str], preprocess):
+        self.image_paths = image_paths
+        self.preprocess = preprocess
 
+    def __len__(self):
+        return len(self.image_paths)
 
-def embed_image(image: Image.Image) -> list[float]:
-    tensor = _preprocess(image).unsqueeze(0).to(_DEVICE)
+    def __getitem__(self, idx):
+        try:
+            img = Image.open(self.image_paths[idx]).convert("RGB")
+            return self.preprocess(img), True
+        except Exception:
+            return torch.zeros(3, 224, 224), False
+
+
+def embed_images_batch(images: list[Image.Image]) -> list[list[float]]:
+    _init_model()
+    tensors = torch.stack([_preprocess(img) for img in images]).to(_DEVICE)
     with torch.no_grad():
-        features = _model.encode_image(tensor)
+        features = _model.encode_image(tensors)
         features = features / features.norm(dim=-1, keepdim=True)
-    return features[0].cpu().tolist()
+    return features.cpu().tolist()
 
 
 def embed_text_query(text: str) -> list[float]:
-    """Embed a text query into the same vector space as panel images."""
+    _init_model()
     tokens = _tokenizer([text]).to(_DEVICE)
     with torch.no_grad():
         features = _model.encode_text(tokens)
@@ -44,39 +81,113 @@ def embed_text_query(text: str) -> list[float]:
     return features[0].cpu().tolist()
 
 
+def merge_chunks() -> None:
+    chunk_files = sorted(CHUNKS_DIR.glob("chunk_*.parquet"))
+    if not chunk_files:
+        print("No chunk files found to merge")
+        return
+    print(f"Merging {len(chunk_files)} chunk files...")
+    df = pd.concat([pd.read_parquet(f) for f in chunk_files], ignore_index=True)
+    df.to_parquet(OUTPUT_PATH, index=False)
+    print(f"Merged {len(df):,} vectors → {OUTPUT_PATH}")
+
+
+def done_chunks() -> set[int]:
+    if not CHUNKS_DIR.exists():
+        return set()
+    return {int(f.stem.split("_")[1]) for f in CHUNKS_DIR.glob("chunk_*.parquet")}
+
+
 def main():
     import argparse
     parser = argparse.ArgumentParser()
-    parser.add_argument("--limit", type=int, default=None, help="Cap number of panels to embed (for dry runs)")
+    parser.add_argument("--limit", type=int, default=None)
+    parser.add_argument("--batch-size", type=int, default=96)
+    parser.add_argument("--num-workers", type=int, default=4)
+    parser.add_argument("--merge", action="store_true", help="Merge existing chunks and exit")
     args = parser.parse_args()
 
+    if args.merge:
+        merge_chunks()
+        return
+
+    _init_model()
+    CHUNKS_DIR.mkdir(parents=True, exist_ok=True)
+
     manifest = pd.read_parquet(MANIFEST_PATH)
     if args.limit:
         manifest = manifest.head(args.limit)
-    print(f"Embedding {len(manifest)} panels with {_ARCH} on {_DEVICE}")
 
-    rows = []
-    failures = []
-
-    for row in tqdm(manifest.itertuples(index=False), total=len(manifest)):
-        try:
-            image = Image.open(row.image_path).convert("RGB")
-            vector = embed_image(image)
-            rows.append({"panel_id": row.panel_id, "image_dense": vector})
-        except Exception as exc:
-            failures.append({
-                "panel_id": row.panel_id,
-                "image_path": row.image_path,
-                "error": str(exc),
-            })
-
-    OUTPUT_PATH.parent.mkdir(parents=True, exist_ok=True)
-    pd.DataFrame(rows).to_parquet(OUTPUT_PATH, index=False)
-    print(f"Saved {len(rows)} vectors → {OUTPUT_PATH}")
-
-    if failures:
-        pd.DataFrame(failures).to_csv(FAILURES_PATH, index=False)
-        print(f"Failed: {len(failures)} panels → {FAILURES_PATH}")
+    already_done = done_chunks()
+    total_chunks = (len(manifest) + CHUNK_SIZE - 1) // CHUNK_SIZE
+    remaining = [i for i in range(total_chunks) if i not in already_done]
+
+    print(f"Total panels:  {len(manifest):,}")
+    print(f"Total chunks:  {total_chunks:,} (chunk_size={CHUNK_SIZE})")
+    print(f"Done chunks:   {len(already_done):,}")
+    print(f"Remaining:     {len(remaining):,}")
+    print(f"Model: {_ARCH} on {_DEVICE}, batch_size={args.batch_size}, num_workers={args.num_workers}")
+
+    all_failures = []
+
+    for chunk_idx in remaining:
+        start = chunk_idx * CHUNK_SIZE
+        end = min(start + CHUNK_SIZE, len(manifest))
+        chunk_df = manifest.iloc[start:end]
+
+        panel_ids = chunk_df["panel_id"].tolist()
+        image_paths = chunk_df["image_path"].tolist()
+
+        dataset = PanelDataset(image_paths, _preprocess)
+        loader = DataLoader(
+            dataset,
+            batch_size=args.batch_size,
+            num_workers=args.num_workers,
+            pin_memory=False,
+            prefetch_factor=2 if args.num_workers > 0 else None,
+        )
+
+        rows = []
+        failures = []
+        item_offset = 0
+
+        pbar = tqdm(loader, desc=f"chunk {chunk_idx+1}/{total_chunks}", leave=False)
+        for tensors, valid_flags in pbar:
+            batch_len = len(tensors)
+            batch_ids = panel_ids[item_offset: item_offset + batch_len]
+            batch_paths = image_paths[item_offset: item_offset + batch_len]
+            item_offset += batch_len
+
+            valid_mask = valid_flags.bool()
+            valid_indices = valid_mask.nonzero(as_tuple=True)[0]
+            invalid_indices = (~valid_mask).nonzero(as_tuple=True)[0].tolist()
+
+            for i in invalid_indices:
+                failures.append({"panel_id": batch_ids[i], "image_path": batch_paths[i], "error": "image load failed"})
+
+            if valid_indices.numel() > 0:
+                valid_tensors = tensors[valid_indices].to(_DEVICE)
+                with torch.no_grad():
+                    features = _model.encode_image(valid_tensors)
+                    features = features / features.norm(dim=-1, keepdim=True)
+                for i, vec in enumerate(features.cpu().tolist()):
+                    rows.append({"panel_id": batch_ids[valid_indices[i].item()], "image_dense": vec})
+
+        chunk_path = CHUNKS_DIR / f"chunk_{chunk_idx:06d}.parquet"
+        pd.DataFrame(rows).to_parquet(chunk_path, index=False)
+        all_failures.extend(failures)
+
+        del loader  # release worker processes and file descriptors before next chunk
+
+        done_count = len(already_done) + (remaining.index(chunk_idx) + 1)
+        print(f"  chunk {chunk_idx:06d}: {len(rows)} vectors saved ({done_count}/{total_chunks} chunks done)")
+
+    if all_failures:
+        pd.DataFrame(all_failures).to_csv(FAILURES_PATH, index=False)
+        print(f"Failures: {len(all_failures)} → {FAILURES_PATH}")
+
+    print("All chunks complete. Merging...")
+    merge_chunks()
 
 
 if __name__ == "__main__":

src/embeddings/embed_sparse_text.py

@@ -1,7 +1,12 @@
-"""Generate sparse text embeddings for each panel using Pinecone inference."""
+"""Generate sparse text embeddings for each panel using Pinecone inference.
+
+Processes panels in chunks of CHUNK_SIZE, writing one parquet file per chunk.
+Resumable: skips chunks whose output file already exists.
+"""
 
 from pathlib import Path
 import os
+import time
 import pandas as pd
 from tqdm import tqdm
 from pinecone import Pinecone
@@ -9,65 +14,112 @@ from pinecone import Pinecone
 
 MANIFEST_PATH = Path("data/comics/processed/panels_manifest.parquet")
 OUTPUT_PATH = Path("data/comics/processed/text_sparse_vectors.parquet")
+CHUNKS_DIR = Path("data/comics/processed/sparse_chunks")
+CHUNK_SIZE = 10_000
+BATCH_SIZE = 96
 
 SPARSE_MODEL = os.environ.get("SPARSE_EMBED_MODEL", "pinecone-sparse-english-v0")
-BATCH_SIZE = 96  # Stay within Pinecone inference batch limits
-
 _pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"])
 
 
-def embed_sparse_batch(texts: list[str], input_type: str = "passage") -> list[dict | None]:
-    """Embed a batch of texts using pinecone-sparse-english-v0."""
+def embed_sparse_batch(texts: list[str], input_type: str = "passage", retries: int = 5) -> list[dict | None]:
     non_empty = [(i, t) for i, t in enumerate(texts) if t.strip()]
     results: list[dict | None] = [None] * len(texts)
-
     if not non_empty:
         return results
-
     indices, batch_texts = zip(*non_empty)
-    response = _pc.inference.embed(
-        model=SPARSE_MODEL,
-        inputs=list(batch_texts),
-        parameters={"input_type": input_type},
-    )
-
-    for idx, embedding in zip(indices, response):
-        if embedding.sparse_indices and embedding.sparse_values:
-            results[idx] = {
-                "indices": embedding.sparse_indices,
-                "values": embedding.sparse_values,
-            }
-
+    for attempt in range(retries):
+        try:
+            response = _pc.inference.embed(
+                model=SPARSE_MODEL,
+                inputs=list(batch_texts),
+                parameters={"input_type": input_type},
+            )
+            for idx, embedding in zip(indices, response):
+                if embedding.sparse_indices and embedding.sparse_values:
+                    results[idx] = {
+                        "indices": embedding.sparse_indices,
+                        "values": embedding.sparse_values,
+                    }
+            return results
+        except Exception as exc:
+            if attempt == retries - 1:
+                raise
+            wait = 2 ** attempt
+            print(f"  Retry {attempt + 1}: {exc}; waiting {wait}s")
+            time.sleep(wait)
     return results
 
 
 def embed_sparse_query(text: str) -> dict | None:
-    """Embed a single query string for search time."""
     if not text.strip():
         return None
-    results = embed_sparse_batch([text], input_type="query")
-    return results[0]
+    return embed_sparse_batch([text], input_type="query")[0]
 
 
-def main():
-    manifest = pd.read_parquet(MANIFEST_PATH)
-    print(f"Sparse-embedding {len(manifest)} panels with {SPARSE_MODEL}")
+def merge_chunks() -> None:
+    chunk_files = sorted(CHUNKS_DIR.glob("chunk_*.parquet"))
+    if not chunk_files:
+        print("No chunk files to merge")
+        return
+    print(f"Merging {len(chunk_files)} chunk files...")
+    df = pd.concat([pd.read_parquet(f) for f in chunk_files], ignore_index=True)
+    df.to_parquet(OUTPUT_PATH, index=False)
+    print(f"Merged {len(df):,} rows → {OUTPUT_PATH}")
+
 
-    texts = manifest["search_text"].fillna("").tolist()
-    panel_ids = manifest["panel_id"].tolist()
+def done_chunks() -> set[int]:
+    if not CHUNKS_DIR.exists():
+        return set()
+    return {int(f.stem.split("_")[1]) for f in CHUNKS_DIR.glob("chunk_*.parquet")}
 
-    all_sparse: list[dict | None] = []
-    for start in tqdm(range(0, len(texts), BATCH_SIZE)):
-        batch = texts[start : start + BATCH_SIZE]
-        all_sparse.extend(embed_sparse_batch(batch))
 
-    rows = [{"panel_id": pid, "text_sparse": sparse} for pid, sparse in zip(panel_ids, all_sparse)]
+def main():
+    import argparse
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--merge", action="store_true")
+    args = parser.parse_args()
 
-    OUTPUT_PATH.parent.mkdir(parents=True, exist_ok=True)
-    pd.DataFrame(rows).to_parquet(OUTPUT_PATH, index=False)
+    if args.merge:
+        merge_chunks()
+        return
 
-    filled = sum(1 for r in rows if r["text_sparse"] is not None)
-    print(f"Saved {len(rows)} rows ({filled} with sparse vectors) → {OUTPUT_PATH}")
+    CHUNKS_DIR.mkdir(parents=True, exist_ok=True)
+
+    manifest = pd.read_parquet(MANIFEST_PATH)
+    already_done = done_chunks()
+    total_chunks = (len(manifest) + CHUNK_SIZE - 1) // CHUNK_SIZE
+    remaining = [i for i in range(total_chunks) if i not in already_done]
+
+    print(f"Total panels: {len(manifest):,} | chunks done: {len(already_done)}/{total_chunks} | remaining: {len(remaining)}")
+    print(f"Model: {SPARSE_MODEL}")
+
+    for chunk_idx in remaining:
+        start = chunk_idx * CHUNK_SIZE
+        end = min(start + CHUNK_SIZE, len(manifest))
+        chunk_df = manifest.iloc[start:end]
+
+        texts = chunk_df["search_text"].fillna("").tolist()
+        panel_ids = chunk_df["panel_id"].tolist()
+        rows = []
+
+        pbar = tqdm(range(0, len(texts), BATCH_SIZE), desc=f"chunk {chunk_idx+1}/{total_chunks}", leave=False)
+        for start_b in pbar:
+            batch = texts[start_b: start_b + BATCH_SIZE]
+            batch_ids = panel_ids[start_b: start_b + BATCH_SIZE]
+            sparse_results = embed_sparse_batch(batch)
+            for pid, sparse in zip(batch_ids, sparse_results):
+                rows.append({"panel_id": pid, "text_sparse": sparse})
+
+        chunk_path = CHUNKS_DIR / f"chunk_{chunk_idx:06d}.parquet"
+        pd.DataFrame(rows).to_parquet(chunk_path, index=False)
+
+        done_count = len(already_done) + (remaining.index(chunk_idx) + 1)
+        filled = sum(1 for r in rows if r["text_sparse"] is not None)
+        print(f"  chunk {chunk_idx:06d}: {len(rows)} rows ({filled} with vectors) [{done_count}/{total_chunks} done]")
+
+    print("All chunks complete. Merging...")
+    merge_chunks()
 
 
 if __name__ == "__main__":

src/ingest/build_manifest.py

@@ -104,6 +104,10 @@ def build_ocr_lookup(ocr_df: pd.DataFrame) -> dict[str, str]:
 
     print(f"  Using columns: book={book_col}, page={page_col}, panel={panel_col}, text={text_col}")
 
+    # Aggregate all textbox texts per panel (CSV has one row per textbox)
+    from collections import defaultdict
+    panel_texts: dict[str, list[str]] = defaultdict(list)
+
     for row in ocr_df.itertuples(index=False):
         book = str(getattr(row, book_col)).strip()
         try:
@@ -111,15 +115,23 @@ def build_ocr_lookup(ocr_df: pd.DataFrame) -> dict[str, str]:
             panel = int(getattr(row, panel_col))
         except (ValueError, TypeError):
             continue
-        text = str(getattr(row, text_col)) if getattr(row, text_col, None) is not None else ""
-        key = f"{book}:{page}:{panel}"
-        lookup[key] = text
+        val = getattr(row, text_col, None)
+        if val is not None and str(val).strip() and str(val).lower() != "nan":
+            panel_texts[f"{book}:{page}:{panel}"].append(str(val).strip())
+
+    for key, texts in panel_texts.items():
+        lookup[key] = " ".join(texts)
 
     print(f"  OCR lookup entries: {len(lookup)}")
     return lookup
 
 
 def main():
+    import argparse
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--limit", type=int, default=None, help="Cap number of panels (for dry runs)")
+    args = parser.parse_args()
+
     panel_dir = RAW_DIR / "raw_panel_images"
     ocr_path = RAW_DIR / "COMICS_ocr_file.csv"
     ad_pages_path = RAW_DIR / "predadpages.txt"
@@ -138,6 +150,9 @@ def main():
         if p.is_file() and p.suffix.lower() in {".jpg", ".jpeg", ".png"}
     )
     print(f"Found {len(panel_paths)} panel images")
+    if args.limit:
+        panel_paths = panel_paths[:args.limit]
+        print(f"Limiting to {len(panel_paths)} panels (--limit {args.limit})")
 
     ocr_lookup: dict[str, str] = {}
     if ocr_path.exists():

src/pinecone/build_documents.py

@@ -12,10 +12,20 @@ SPARSE_PATH = PROCESSED_DIR / "text_sparse_vectors.parquet"
 OUTPUT_PATH = PROCESSED_DIR / "pinecone_documents.jsonl"
 
 
+def to_list(val) -> list:
+    """Convert numpy array or other sequence to a plain Python list."""
+    import numpy as np
+    if isinstance(val, np.ndarray):
+        return val.tolist()
+    return list(val)
+
+
 def valid_sparse(sparse) -> bool:
     if not isinstance(sparse, dict):
         return False
-    return bool(sparse.get("indices")) and bool(sparse.get("values"))
+    indices = sparse.get("indices")
+    values = sparse.get("values")
+    return indices is not None and len(indices) > 0 and values is not None and len(values) > 0
 
 
 def main():
@@ -37,7 +47,7 @@ def main():
         for row in df.itertuples(index=False):
             doc: dict = {
                 "_id": row.panel_id,
-                "image_dense": row.image_dense,
+                "image_dense": to_list(row.image_dense),
                 "ocr_text": row.ocr_text_clean or "",
                 "search_text": row.search_text or "",
                 "comic_id": row.comic_id,
@@ -52,7 +62,10 @@ def main():
 
             sparse_vec = getattr(row, "text_sparse", None)
             if valid_sparse(sparse_vec):
-                doc["text_sparse"] = sparse_vec
+                doc["text_sparse"] = {
+                    "indices": to_list(sparse_vec["indices"]),
+                    "values": to_list(sparse_vec["values"]),
+                }
 
             f.write(json.dumps(doc) + "\n")
             count += 1

src/pinecone/upsert_panels.py

@@ -51,10 +51,10 @@ def validate_search(index) -> None:
         score_by=[{"type": "dense_vector", "field": "image_dense", "values": dummy_vector}],
         include_fields=["panel_id", "comic_id", "page_num"],
     )
-    hits = results.get("result", {}).get("hits", [])
+    hits = results.matches
     print(f"Validation search returned {len(hits)} hits")
     for h in hits:
-        print(f"  {h.get('_id')} score={h.get('_score'):.4f}")
+        print(f"  {h.id} score={h.score:.4f}")
 
 
 def main():

src/search/search_dense.py

@@ -11,13 +11,14 @@ _INCLUDE_FIELDS = [
 
 
 def search_dense(index, namespace: str, query_vector: list[float], top_k: int = 20, filters: dict | None = None) -> dict:
-    return index.documents.search(
+    response = index.documents.search(
         namespace=namespace,
         top_k=top_k,
         score_by=[{"type": "dense_vector", "field": "image_dense", "values": query_vector}],
         filter=filters or {},
         include_fields=_INCLUDE_FIELDS,
     )
+    return {"result": {"hits": [h.to_dict() for h in response.matches]}}
 
 
 def search_dense_with_phrase_filter(
@@ -30,13 +31,14 @@ def search_dense_with_phrase_filter(
 ) -> dict:
     merged_filter = dict(filters) if filters else {}
     merged_filter["search_text"] = {"$match_phrase": phrase}
-    return index.documents.search(
+    response = index.documents.search(
         namespace=namespace,
         top_k=top_k,
         score_by=[{"type": "dense_vector", "field": "image_dense", "values": query_vector}],
         filter=merged_filter,
         include_fields=_INCLUDE_FIELDS,
     )
+    return {"result": {"hits": [h.to_dict() for h in response.matches]}}
 
 
 def main():

src/search/search_fts.py

@@ -11,13 +11,14 @@ _INCLUDE_FIELDS = [
 
 
 def search_fts(index, namespace: str, query: str, top_k: int = 20, filters: dict | None = None) -> dict:
-    return index.documents.search(
+    response = index.documents.search(
         namespace=namespace,
         top_k=top_k,
         score_by=[{"type": "text", "field": "search_text", "query": query}],
         filter=filters or {},
         include_fields=_INCLUDE_FIELDS,
     )
+    return {"result": {"hits": [h.to_dict() for h in response.matches]}}
 
 
 def main():

src/search/search_sparse.py

@@ -11,13 +11,14 @@ _INCLUDE_FIELDS = [
 
 
 def search_sparse(index, namespace: str, query_sparse: dict, top_k: int = 20, filters: dict | None = None) -> dict:
-    return index.documents.search(
+    response = index.documents.search(
         namespace=namespace,
         top_k=top_k,
         score_by=[{"type": "sparse_vector", "field": "text_sparse", "sparse_values": query_sparse}],
         filter=filters or {},
         include_fields=_INCLUDE_FIELDS,
     )
+    return {"result": {"hits": [h.to_dict() for h in response.matches]}}
 
 
 def main():