Upload 5 files

image-classifier/build_code_embeddings.py

@@ -0,0 +1,245 @@
+"""
+build_code_embeddings.py — Embed USPTO Design Codes for RAG retrieval
+=======================================================================
+One-time script that reads uspto_design_codes.json (from the scraper) and
+embeds every code's description using Voyage AI. The output feeds the
+RAG layer of design_code_classifier.py.
+
+WHY THIS EXISTS:
+The classifier needs to constrain Claude to USPTO's actual code vocabulary
+(no hallucinating fake codes). Embedding every code's description once
+lets us do fast vector search at classification time: Claude describes
+the image, we retrieve the most semantically similar codes, Claude picks
+from that validated menu.
+
+SETUP:
+    pip install voyageai numpy python-dotenv
+
+ENV VARS:
+    VOYAGE_API_KEY — Voyage AI API key (free tier covers 200M tokens)
+
+USAGE:
+    python build_code_embeddings.py
+
+    # Re-embed with a different model:
+    python build_code_embeddings.py --model voyage-3-large
+
+OUTPUT:
+    uspto_code_embeddings.pkl — pickle dict with:
+      - codes:        list[str]     ordered list of XX.YY.ZZ codes
+      - descriptions: list[str]     parallel list of code descriptions
+      - categories:   list[str]     parent category for each code (XX)
+      - embeddings:   np.ndarray    shape (N, dim), float32
+      - metadata:     dict          model name, dim, timestamp
+
+COST:
+    ~1,300 codes × ~15 tokens each = ~20K tokens.
+    voyage-3.5 is ~$0.06 per 1M tokens → effectively free under the
+    200M-token free tier.
+"""
+
+import os
+import sys
+import json
+import pickle
+import logging
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import List
+
+import numpy as np
+from dotenv import load_dotenv
+
+try:
+    import voyageai
+except ImportError:
+    print("ERROR: Voyage AI SDK not installed. Run:")
+    print("  pip install voyageai")
+    sys.exit(1)
+
+
+# ============================================================================
+# CONFIG
+# ============================================================================
+
+env_path = Path(__file__).parent / ".env"
+load_dotenv(dotenv_path=env_path)
+
+VOYAGE_API_KEY = os.getenv("VOYAGE_API_KEY")
+
+DEFAULT_MODEL = "voyage-3.5"   # current general-purpose default; voyage-3-large is the premium upgrade
+DEFAULT_DIM = 1024              # default for voyage-3.5; do not change without re-embedding
+
+# Voyage allows up to 1,000 texts per batch; we have ~1,300 codes, so 2 batches
+BATCH_SIZE = 1000
+
+INPUT_PATH = Path(__file__).parent / "uspto_design_codes.json"
+OUTPUT_PATH = Path(__file__).parent / "uspto_code_embeddings.pkl"
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(message)s",
+)
+logger = logging.getLogger("embeddings")
+
+
+# ============================================================================
+# LOAD CODES FROM SCRAPED MANUAL
+# ============================================================================
+
+def load_codes() -> tuple[List[str], List[str], List[str]]:
+    """Read uspto_design_codes.json and flatten into parallel lists.
+
+    Returns:
+        (codes, descriptions, categories) — three lists of equal length,
+        where codes[i] corresponds to descriptions[i] from category categories[i].
+    """
+    if not INPUT_PATH.exists():
+        logger.error(f"❌ {INPUT_PATH} not found. Run scrape_uspto_design_codes.py first.")
+        sys.exit(1)
+
+    data = json.loads(INPUT_PATH.read_text())
+    codes_dict = data.get("categories", {})
+
+    codes: List[str] = []
+    descriptions: List[str] = []
+    categories: List[str] = []
+    seen = set()  # deduplicate (the scraper produced some cross-listed codes)
+
+    for category_id, category_data in codes_dict.items():
+        for division_id, division_data in category_data.get("divisions", {}).items():
+            for section_code, section_data in division_data.get("sections", {}).items():
+                if section_code in seen:
+                    continue
+                description = section_data.get("description", "").strip()
+                if not description:
+                    continue
+                codes.append(section_code)
+                descriptions.append(description)
+                # Use the section's *real* category prefix (XX), not the JSON parent
+                # — this self-corrects the scraper's cross-listing duplicates
+                categories.append(section_code.split(".")[0])
+                seen.add(section_code)
+
+    logger.info(f"📋 Loaded {len(codes)} unique codes from {INPUT_PATH.name}")
+    return codes, descriptions, categories
+
+
+# ============================================================================
+# EMBED VIA VOYAGE
+# ============================================================================
+
+def build_searchable_text(code: str, description: str) -> str:
+    """Construct the text that gets embedded for each code.
+
+    We include the code itself in the text — the digits give the embedding
+    a tiny extra signal of which category/division things belong to, which
+    helps when descriptions are very generic ("Other plants" appears in
+    multiple divisions and would otherwise be indistinguishable).
+    """
+    return f"USPTO Design Code {code}: {description}"
+
+
+def embed_descriptions(
+    descriptions_with_codes: List[str],
+    model: str,
+) -> np.ndarray:
+    """Call Voyage to embed all descriptions. Returns (N, dim) float32 array."""
+    if not VOYAGE_API_KEY:
+        logger.error("❌ VOYAGE_API_KEY not set in .env")
+        sys.exit(1)
+
+    client = voyageai.Client(api_key=VOYAGE_API_KEY)
+
+    all_embeddings: List[List[float]] = []
+
+    for batch_idx in range(0, len(descriptions_with_codes), BATCH_SIZE):
+        batch = descriptions_with_codes[batch_idx : batch_idx + BATCH_SIZE]
+        logger.info(
+            f"🔄 Embedding batch {batch_idx // BATCH_SIZE + 1} "
+            f"({len(batch)} texts, total tokens ~{sum(len(t.split()) for t in batch)})"
+        )
+
+        try:
+            result = client.embed(
+                texts=batch,
+                model=model,
+                input_type="document",  # corpus side of retrieval
+            )
+        except Exception as e:
+            logger.error(f"❌ Voyage API error: {e}")
+            sys.exit(1)
+
+        all_embeddings.extend(result.embeddings)
+        logger.info(f"  ✅ Batch returned {len(result.embeddings)} embeddings")
+
+    arr = np.array(all_embeddings, dtype=np.float32)
+    logger.info(f"📐 Final embeddings shape: {arr.shape}")
+    return arr
+
+
+# ============================================================================
+# SAVE
+# ============================================================================
+
+def save_embeddings(
+    codes: List[str],
+    descriptions: List[str],
+    categories: List[str],
+    embeddings: np.ndarray,
+    model: str,
+):
+    """Persist everything to a single pickle for easy loading by the classifier."""
+    payload = {
+        "codes": codes,
+        "descriptions": descriptions,
+        "categories": categories,
+        "embeddings": embeddings,
+        "metadata": {
+            "model": model,
+            "dimension": embeddings.shape[1],
+            "code_count": len(codes),
+            "created_at": datetime.now(timezone.utc).isoformat(),
+            "source_file": INPUT_PATH.name,
+        },
+    }
+    with OUTPUT_PATH.open("wb") as f:
+        pickle.dump(payload, f, protocol=pickle.HIGHEST_PROTOCOL)
+    logger.info(f"💾 Saved to {OUTPUT_PATH} ({OUTPUT_PATH.stat().st_size / 1024:.1f} KB)")
+
+
+# ============================================================================
+# CLI
+# ============================================================================
+
+def main():
+    import argparse
+    parser = argparse.ArgumentParser(description="Embed USPTO design codes via Voyage AI")
+    parser.add_argument(
+        "--model", default=DEFAULT_MODEL,
+        help=f"Voyage model to use (default: {DEFAULT_MODEL}). "
+             "Upgrade to voyage-3-large for marginal quality gains."
+    )
+    args = parser.parse_args()
+
+    codes, descriptions, categories = load_codes()
+    texts_to_embed = [
+        build_searchable_text(c, d) for c, d in zip(codes, descriptions)
+    ]
+    embeddings = embed_descriptions(texts_to_embed, model=args.model)
+
+    save_embeddings(codes, descriptions, categories, embeddings, args.model)
+
+    logger.info("\n" + "=" * 60)
+    logger.info("📊 EMBEDDING BUILD COMPLETE")
+    logger.info("=" * 60)
+    logger.info(f"  Model:      {args.model}")
+    logger.info(f"  Codes:      {len(codes):,}")
+    logger.info(f"  Dimensions: {embeddings.shape[1]}")
+    logger.info(f"  Output:     {OUTPUT_PATH}")
+    logger.info("=" * 60)
+    logger.info("\n💡 Next step: design_code_classifier.py will load this file at startup")
+
+
+if __name__ == "__main__":
+    main()

image-classifier/design_code_classifier.py

@@ -0,0 +1,565 @@
+"""
+design_code_classifier.py — USPTO Design Search Code classifier (RAG version)
+==============================================================================
+Uses a two-stage RAG flow to classify uploaded trademark images into USPTO
+Design Search Codes — with the guarantee that returned codes are real USPTO
+codes pulled from the official manual, not invented by the model.
+
+ARCHITECTURE
+------------
+Stage 1 (visual description):
+    Claude vision model looks at the image and produces a structured natural-
+    language description of what's in it.
+
+Vector retrieval:
+    The description is embedded via Voyage and matched against pre-embedded
+    USPTO code descriptions. Top-K candidates are pulled.
+
+Stage 2 (constrained selection):
+    Claude sees the image AGAIN alongside the candidate codes (with their
+    official descriptions) and selects which apply, with confidence scores
+    and rationales.
+
+This means Claude is *picking from a verified menu*, not free-styling. No
+hallucinated codes.
+
+DESIGN PHILOSOPHY
+-----------------
+This module is fully STANDALONE — does not import from or modify
+nst_comparison.py. The output is structured JSON; how it gets plugged into
+the matching algorithm is a deliberate decision left to the caller.
+
+REQUIREMENTS
+------------
+    pip install anthropic voyageai pillow numpy python-dotenv
+
+ENV VARS
+--------
+    ANTHROPIC_API_KEY — Anthropic API key
+    VOYAGE_API_KEY    — Voyage AI API key (for query embedding only;
+                        codes were pre-embedded by build_code_embeddings.py)
+    CLASSIFIER_MODEL  — Optional: override Claude model (default: claude-sonnet-4-6)
+
+PREREQUISITES
+-------------
+You must run scrape_uspto_design_codes.py and build_code_embeddings.py first,
+producing uspto_code_embeddings.pkl in the same directory as this module.
+
+QUICK CLI TEST
+--------------
+    python design_code_classifier.py path/to/logo.jpg
+
+INTEGRATION (the "subtle change" path for nst_comparison.py)
+------------------------------------------------------------
+    # ── At top of nst_comparison.py ──
+    from .design_code_classifier import classify_image
+
+    # ── Inside execute_model(), after np_img is created ──
+    classification = await classify_image(np_img)
+    high_conf_codes = classification.high_confidence_codes(threshold=0.7)
+
+    # ── Add design_search_codes to the SELECT, then filter results: ──
+    if high_conf_codes:
+        response.data = [
+            r for r in response.data
+            if r.get("design_search_codes") and any(
+                c in r["design_search_codes"] for c in high_conf_codes
+            )
+        ]
+"""
+
+import os
+import io
+import json
+import base64
+import pickle
+import logging
+import asyncio
+from pathlib import Path
+from typing import Optional, Any, List
+from dataclasses import dataclass, asdict, field
+
+import numpy as np
+from anthropic import AsyncAnthropic
+from PIL import Image
+from dotenv import load_dotenv
+
+try:
+    import voyageai
+except ImportError:
+    raise ImportError("Voyage AI SDK required. Run: pip install voyageai")
+
+# Optional FastAPI types — only loaded if FastAPI is present
+try:
+    from fastapi import UploadFile, HTTPException
+    HAS_FASTAPI = True
+except ImportError:
+    HAS_FASTAPI = False
+    UploadFile = None  # type: ignore
+    HTTPException = None  # type: ignore
+
+# ============================================================================
+# CONFIG
+# ============================================================================
+
+env_path = Path(__file__).parent / ".env"
+load_dotenv(dotenv_path=env_path)
+
+ANTHROPIC_API_KEY = os.getenv("ANTHROPIC_API_KEY")
+VOYAGE_API_KEY = os.getenv("VOYAGE_API_KEY")
+
+DEFAULT_MODEL = os.getenv("CLASSIFIER_MODEL", "claude-sonnet-4-6")
+MAX_TOKENS_STAGE1 = 1024
+MAX_TOKENS_STAGE2 = 2048
+
+TOP_K_CANDIDATES = 30
+MIN_CONFIDENCE = 0.5
+
+MAX_IMAGE_DIMENSION = 1568
+
+EMBEDDINGS_PATH = Path(__file__).parent / "uspto_code_embeddings.pkl"
+
+logger = logging.getLogger("design_classifier")
+
+
+# ============================================================================
+# CODE INDEX (loads embeddings once, supports vector search)
+# ============================================================================
+
+class CodeIndex:
+    """In-memory vector index over USPTO design codes.
+
+    Loaded once on first use. All requests share the same index — no per-
+    request file I/O.
+    """
+
+    _instance: Optional["CodeIndex"] = None
+
+    def __init__(self):
+        if not EMBEDDINGS_PATH.exists():
+            raise FileNotFoundError(
+                f"{EMBEDDINGS_PATH.name} not found. "
+                "Run scrape_uspto_design_codes.py + build_code_embeddings.py first."
+            )
+
+        with EMBEDDINGS_PATH.open("rb") as f:
+            payload = pickle.load(f)
+
+        self.codes: List[str] = payload["codes"]
+        self.descriptions: List[str] = payload["descriptions"]
+        self.categories: List[str] = payload["categories"]
+        self.embeddings: np.ndarray = payload["embeddings"]
+        self.metadata: dict = payload["metadata"]
+
+        # Pre-normalize for fast cosine similarity (just dot product after this)
+        norms = np.linalg.norm(self.embeddings, axis=1, keepdims=True)
+        norms[norms == 0] = 1.0
+        self.embeddings_normalized = self.embeddings / norms
+
+        self.code_to_description = dict(zip(self.codes, self.descriptions))
+
+        logger.info(
+            f"📚 Loaded code index: {len(self.codes)} codes, "
+            f"{self.metadata['dimension']}-dim {self.metadata['model']} embeddings"
+        )
+
+    @classmethod
+    def get(cls) -> "CodeIndex":
+        if cls._instance is None:
+            cls._instance = CodeIndex()
+        return cls._instance
+
+    def search(self, query_embedding: np.ndarray, top_k: int = TOP_K_CANDIDATES) -> List[dict]:
+        """Find top-K most similar codes to the query embedding.
+
+        Returns list of {code, description, category, similarity} dicts,
+        sorted descending by similarity.
+        """
+        q = query_embedding / (np.linalg.norm(query_embedding) + 1e-12)
+        sims = self.embeddings_normalized @ q
+        top_idx = np.argpartition(sims, -top_k)[-top_k:]
+        top_idx = top_idx[np.argsort(-sims[top_idx])]
+
+        return [
+            {
+                "code": self.codes[i],
+                "description": self.descriptions[i],
+                "category": self.categories[i],
+                "similarity": float(sims[i]),
+            }
+            for i in top_idx
+        ]
+
+
+# ============================================================================
+# IMAGE NORMALIZATION
+# ============================================================================
+
+def _normalize_image(image_input: Any) -> tuple[str, str]:
+    """Convert various image input types to (base64_data, media_type).
+
+    Accepts: bytes, PIL.Image, numpy.ndarray, file path.
+    """
+    if hasattr(image_input, "shape") and hasattr(image_input, "dtype"):
+        arr = image_input
+        if arr.dtype != np.uint8:
+            if arr.max() <= 1.0:
+                arr = (arr * 255).astype(np.uint8)
+            else:
+                arr = arr.astype(np.uint8)
+        img = Image.fromarray(arr)
+    elif isinstance(image_input, bytes):
+        img = Image.open(io.BytesIO(image_input))
+    elif isinstance(image_input, Image.Image):
+        img = image_input
+    elif isinstance(image_input, (str, Path)):
+        img = Image.open(image_input)
+    else:
+        raise TypeError(
+            f"Unsupported image input type: {type(image_input)}. "
+            "Use bytes, PIL.Image, numpy.ndarray, or a file path."
+        )
+
+    if img.mode != "RGB":
+        img = img.convert("RGB")
+
+    if max(img.size) > MAX_IMAGE_DIMENSION:
+        img.thumbnail((MAX_IMAGE_DIMENSION, MAX_IMAGE_DIMENSION), Image.Resampling.LANCZOS)
+        logger.debug(f"Resized image to {img.size}")
+
+    buf = io.BytesIO()
+    img.save(buf, format="JPEG", quality=90)
+    base64_data = base64.standard_b64encode(buf.getvalue()).decode("utf-8")
+    return base64_data, "image/jpeg"
+
+
+# ============================================================================
+# RESULT TYPES
+# ============================================================================
+
+@dataclass
+class DesignCode:
+    code: str
+    description: str
+    confidence: float
+    rationale: str
+
+
+@dataclass
+class ClassificationResult:
+    codes: List[DesignCode] = field(default_factory=list)
+    image_description: str = ""
+    primary_category: Optional[str] = None
+    candidate_codes_considered: List[str] = field(default_factory=list)
+
+    def high_confidence_codes(self, threshold: float = 0.7) -> List[str]:
+        """Return just the code strings above the confidence threshold."""
+        return [c.code for c in self.codes if c.confidence >= threshold]
+
+    def to_dict(self) -> dict:
+        return {
+            "codes": [asdict(c) for c in self.codes],
+            "image_description": self.image_description,
+            "primary_category": self.primary_category,
+            "candidate_codes_considered": self.candidate_codes_considered,
+        }
+
+
+# ============================================================================
+# STAGE 1 — IMAGE → NATURAL LANGUAGE DESCRIPTION
+# ============================================================================
+
+STAGE1_SYSTEM_PROMPT = """You are an expert trademark image analyst. Your job is to describe trademark logos in a way that will help retrieve relevant USPTO Design Search Codes.
+
+When you describe an image, focus on the visual elements a USPTO examiner would code:
+- Living things: humans, animals, plants (be specific — "lion's head" not just "animal")
+- Geometric shapes: circles, squares, triangles, lines (and their arrangement)
+- Objects: tools, vehicles, buildings, food, clothing (be specific)
+- Symbols: arrows, crosses, stars, hearts, mathematical symbols
+- Text/letters: note their presence and style (stylized, in a frame, etc.)
+- Colors: only if the color is a distinctive design element
+- Composition: what's framing what, what's stylized vs. realistic
+
+Output a single paragraph (3-6 sentences). Be precise and use concrete vocabulary that a search index would match. Don't editorialize about meaning or branding — just describe what's literally visible.
+
+Output the description as plain text. No JSON, no prose preamble."""
+
+
+async def stage1_describe_image(
+    base64_data: str,
+    media_type: str,
+    client: AsyncAnthropic,
+    model: str,
+) -> str:
+    """Have Claude describe the image in retrieval-optimized natural language."""
+    response = await client.messages.create(
+        model=model,
+        max_tokens=MAX_TOKENS_STAGE1,
+        system=STAGE1_SYSTEM_PROMPT,
+        messages=[{
+            "role": "user",
+            "content": [
+                {
+                    "type": "image",
+                    "source": {
+                        "type": "base64",
+                        "media_type": media_type,
+                        "data": base64_data,
+                    },
+                },
+                {"type": "text", "text": "Describe this trademark image."},
+            ],
+        }],
+    )
+    description = response.content[0].text.strip()
+    logger.info(f"Stage 1 description ({len(description)} chars): {description[:120]}...")
+    return description
+
+
+# ============================================================================
+# RETRIEVAL — VOYAGE EMBED + VECTOR SEARCH
+# ============================================================================
+
+def embed_query(text: str, voyage_client) -> np.ndarray:
+    """Embed a single query string with Voyage. Returns a 1D numpy array.
+
+    NOTE: model name MUST match what was used in build_code_embeddings.py.
+    Reading the metadata from the index ensures consistency.
+    """
+    index = CodeIndex.get()
+    embedding_model = index.metadata["model"]
+
+    result = voyage_client.embed(
+        texts=[text],
+        model=embedding_model,
+        input_type="query",
+    )
+    return np.array(result.embeddings[0], dtype=np.float32)
+
+
+# ============================================================================
+# STAGE 2 — IMAGE + CANDIDATES → STRUCTURED CODE SELECTION
+# ============================================================================
+
+def _stage2_system_prompt(candidates: List[dict]) -> str:
+    """Build the system prompt with the candidate codes inline."""
+    candidate_lines = "\n".join(
+        f"  {c['code']}  —  {c['description']}"
+        for c in candidates
+    )
+
+    return f"""You are an expert USPTO Trademark Design Search Code classifier.
+
+Below is a list of candidate USPTO Design Search Codes that may apply to the image. Your job is to look at the image and select ONLY the codes that genuinely apply — based on what is actually visible.
+
+CANDIDATE CODES:
+{candidate_lines}
+
+CRITICAL RULES:
+- ONLY return codes from the candidate list above. Do not invent codes that are not in the list.
+- Be precise: include a code only if the corresponding visual element is clearly present.
+- Most logos use 2–6 codes total. Don't pad — quality over quantity.
+- If the logo contains text or letters, ALWAYS look at category 27 codes in the candidates.
+- If the logo has a geometric frame around the elements, ALWAYS look at category 26 codes in the candidates.
+
+Return ONLY valid JSON in this exact structure:
+{{
+  "codes": [
+    {{
+      "code": "XX.YY.ZZ",
+      "confidence": 0.95,
+      "rationale": "Brief explanation of what in the image triggered this code"
+    }}
+  ],
+  "primary_category": "XX"
+}}
+
+CONFIDENCE SCORING:
+- 0.90–1.00: Element unmistakably present
+- 0.70–0.89: Element clearly present, minor interpretation involved
+- 0.50–0.69: Element likely present but ambiguous
+- Below 0.50: DO NOT include the code
+
+Output valid JSON only. No prose before or after."""
+
+
+async def stage2_select_codes(
+    base64_data: str,
+    media_type: str,
+    candidates: List[dict],
+    client: AsyncAnthropic,
+    model: str,
+) -> dict:
+    """Have Claude pick the applicable codes from the candidate list."""
+    response = await client.messages.create(
+        model=model,
+        max_tokens=MAX_TOKENS_STAGE2,
+        system=_stage2_system_prompt(candidates),
+        messages=[{
+            "role": "user",
+            "content": [
+                {
+                    "type": "image",
+                    "source": {
+                        "type": "base64",
+                        "media_type": media_type,
+                        "data": base64_data,
+                    },
+                },
+                {
+                    "type": "text",
+                    "text": "Select the applicable codes from the candidates. Return JSON only.",
+                },
+            ],
+        }],
+    )
+
+    raw_text = response.content[0].text.strip()
+
+    # Strip markdown fences if present
+    if raw_text.startswith("```"):
+        lines = raw_text.split("\n")
+        if lines[-1].startswith("```"):
+            lines = lines[1:-1]
+        else:
+            lines = lines[1:]
+        raw_text = "\n".join(lines).strip()
+
+    try:
+        return json.loads(raw_text)
+    except json.JSONDecodeError as e:
+        logger.error(f"Stage 2 JSON parse error: {e}")
+        logger.error(f"Raw response: {raw_text[:500]}")
+        return {"codes": [], "primary_category": None}
+
+
+# ============================================================================
+# PUBLIC API
+# ============================================================================
+
+async def classify_image(
+    image_input: Any,
+    model: str = DEFAULT_MODEL,
+    min_confidence: float = MIN_CONFIDENCE,
+    top_k: int = TOP_K_CANDIDATES,
+) -> ClassificationResult:
+    """Classify a trademark image into USPTO Design Search Codes via two-stage RAG.
+
+    Args:
+        image_input:    bytes, PIL.Image, file path, or numpy.ndarray
+        model:          Claude model (default: claude-sonnet-4-6)
+        min_confidence: Drop codes below this confidence (default: 0.5)
+        top_k:          How many candidates to surface to stage 2 (default: 30)
+
+    Returns:
+        ClassificationResult with verified codes only — Claude cannot return
+        codes that aren't in USPTO's actual vocabulary.
+    """
+    if not ANTHROPIC_API_KEY:
+        raise ValueError("ANTHROPIC_API_KEY environment variable not set")
+    if not VOYAGE_API_KEY:
+        raise ValueError("VOYAGE_API_KEY environment variable not set")
+
+    # Lazy-load the index (reused across all classifications)
+    index = CodeIndex.get()
+
+    # Normalize image once — used in both stages
+    base64_data, media_type = _normalize_image(image_input)
+
+    anthropic_client = AsyncAnthropic(api_key=ANTHROPIC_API_KEY)
+    voyage_client = voyageai.Client(api_key=VOYAGE_API_KEY)
+
+    # ── Stage 1: describe ──
+    description = await stage1_describe_image(
+        base64_data, media_type, anthropic_client, model
+    )
+
+    # ── Retrieval: embed + vector search ──
+    query_emb = embed_query(description, voyage_client)
+    candidates = index.search(query_emb, top_k=top_k)
+    logger.info(
+        f"Retrieved {len(candidates)} candidate codes "
+        f"(top similarity: {candidates[0]['similarity']:.3f})"
+    )
+
+    # ── Stage 2: select ──
+    selection = await stage2_select_codes(
+        base64_data, media_type, candidates, anthropic_client, model
+    )
+
+    # ── Build result, validating that returned codes are in the candidate set ──
+    candidate_codes = {c["code"] for c in candidates}
+    result = ClassificationResult(
+        image_description=description,
+        primary_category=selection.get("primary_category"),
+        candidate_codes_considered=[c["code"] for c in candidates],
+    )
+
+    for sel in selection.get("codes", []):
+        code = sel.get("code", "").strip()
+        confidence = float(sel.get("confidence", 0))
+
+        if confidence < min_confidence:
+            continue
+        if code not in candidate_codes:
+            # Defense in depth: even if Claude tries to invent a code, refuse it
+            logger.warning(f"Claude returned code {code} not in candidate set — dropping")
+            continue
+
+        result.codes.append(DesignCode(
+            code=code,
+            description=index.code_to_description.get(code, ""),
+            confidence=confidence,
+            rationale=sel.get("rationale", ""),
+        ))
+
+    logger.info(
+        f"✅ Classified into {len(result.codes)} verified code(s); "
+        f"primary category: {result.primary_category}"
+    )
+    return result
+
+
+# ============================================================================
+# OPTIONAL HELPERS
+# ============================================================================
+
+if HAS_FASTAPI:
+    async def classify_uploadfile(file: UploadFile, **kwargs) -> ClassificationResult:
+        """Convenience wrapper for FastAPI UploadFile inputs."""
+        if file.content_type not in ("image/jpeg", "image/png", "image/gif", "image/webp"):
+            raise HTTPException(
+                status_code=400,
+                detail=f"Unsupported image type: {file.content_type}",
+            )
+        image_bytes = await file.read()
+        return await classify_image(image_bytes, **kwargs)
+
+
+# ============================================================================
+# CLI
+# ============================================================================
+
+async def _cli():
+    import argparse
+    parser = argparse.ArgumentParser(description="Test the USPTO design code classifier")
+    parser.add_argument("image_path", help="Path to a trademark image")
+    parser.add_argument("--model", default=DEFAULT_MODEL, help="Claude model")
+    parser.add_argument("--threshold", type=float, default=MIN_CONFIDENCE)
+    parser.add_argument("--top-k", type=int, default=TOP_K_CANDIDATES)
+    args = parser.parse_args()
+
+    logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
+
+    result = await classify_image(
+        args.image_path,
+        model=args.model,
+        min_confidence=args.threshold,
+        top_k=args.top_k,
+    )
+
+    print(json.dumps(result.to_dict(), indent=2))
+
+
+if __name__ == "__main__":
+    asyncio.run(_cli())

image-classifier/uspto_code_embeddings.pkl

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:ddb2a0c81b1457c910226fe0e0c9a24e2723d0467f49a22a65fe289decbe42f3
+size 5814306

image-classifier/validate_classifier.py

@@ -0,0 +1,459 @@
+"""
+validate_classifier.py — Test the design code classifier against examiner ground truth
+========================================================================================
+Pulls a sample of image marks from your Supabase database that already have
+USPTO examiner-assigned design codes, runs each through the classifier, and
+compares the classifier's output against the examiner's codes.
+
+The output tells you concretely whether the classifier is good enough to
+ship — or whether we need to iterate before exposing it to attorney-tier
+customers.
+
+WHAT IT MEASURES
+----------------
+For each test image, four agreement levels:
+  - Section-exact:  XX.YY.ZZ matches exactly (strictest)
+  - Division-level: XX.YY matches (same code family)
+  - Category-level: XX matches (same broad category)
+  - Any overlap:    at least one code in common
+
+Plus retrieval-style metrics:
+  - Precision:  of codes the classifier returned, what fraction were correct?
+  - Recall:     of examiner codes, what fraction did the classifier find?
+  - F1:         harmonic mean
+
+WHAT TO LOOK FOR
+----------------
+For an MVP shipping to entrepreneurs at $49/mo:
+  - Category-level agreement >= 80% is reasonable
+  - Recall >= 60% means we catch most relevant matches
+
+For attorney-tier customers at $250+/mo:
+  - Section-exact agreement >= 70%
+  - Recall >= 80% (missing codes is a liability risk)
+
+If the numbers come in lower, look at the per-image CSV — usually the
+classifier is right but using a slightly different code than the examiner,
+or vice versa. That's data we can use to tune the prompt.
+
+REQUIREMENTS
+------------
+Same as design_code_classifier.py, plus:
+    pip install httpx supabase
+
+USAGE
+-----
+    # Quick test — 20 random samples
+    python validate_classifier.py --samples 20
+
+    # Full validation — 100 samples, slower but more reliable
+    python validate_classifier.py --samples 100
+
+    # Resume after a crash
+    python validate_classifier.py --resume
+
+OUTPUT
+------
+    validation_results.csv  — per-image: serial, examiner codes, classifier codes, scores
+    validation_summary.json — aggregate metrics
+"""
+
+import os
+import sys
+import csv
+import json
+import asyncio
+import logging
+import random
+from pathlib import Path
+from datetime import datetime, timezone
+from typing import List, Dict, Any
+
+import httpx
+from dotenv import load_dotenv
+from supabase import create_client, Client
+
+# Local import — must be in same directory
+from design_code_classifier import classify_image
+
+
+# ============================================================================
+# CONFIG
+# ============================================================================
+
+env_path = Path(__file__).parent / ".env"
+load_dotenv(dotenv_path=env_path)
+
+SUPABASE_URL = os.getenv("SUPABASE_URL")
+SUPABASE_KEY = os.getenv("SUPABASE_KEY")
+
+CONFIDENCE_THRESHOLD = 0.7   # only count classifier codes at or above this confidence
+
+OUTPUT_CSV = Path(__file__).parent / "validation_results.csv"
+OUTPUT_JSON = Path(__file__).parent / "validation_summary.json"
+CHECKPOINT_PATH = Path(__file__).parent / ".validation_checkpoint.json"
+
+# Politeness — don't hammer your DB or Claude API
+DELAY_BETWEEN_SAMPLES_S = 1.0
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(message)s",
+)
+logger = logging.getLogger("validate")
+
+
+# ============================================================================
+# CHECKPOINT (resume after crash)
+# ============================================================================
+
+def load_checkpoint() -> Dict[str, Any]:
+    if CHECKPOINT_PATH.exists():
+        try:
+            return json.loads(CHECKPOINT_PATH.read_text())
+        except Exception as e:
+            logger.warning(f"Checkpoint unreadable, starting fresh: {e}")
+    return {"completed_serials": [], "results": []}
+
+
+def save_checkpoint(state: Dict[str, Any]):
+    state["updated_at"] = datetime.now(timezone.utc).isoformat()
+    CHECKPOINT_PATH.write_text(json.dumps(state, indent=2))
+
+
+# ============================================================================
+# SAMPLE FETCHING
+# ============================================================================
+
+async def fetch_sample(supabase: Client, n: int, exclude_serials: set) -> List[Dict]:
+    """Pull N random image marks that have both an image and examiner codes.
+
+    We over-fetch and randomize client-side because Postgres ORDER BY RANDOM()
+    on millions of rows is brutally slow.
+    """
+    # Fetch a pool of candidates with codes + image (skip NOT_FOUND)
+    pool_size = min(n * 20, 5000)   # 20x oversample, capped
+    logger.info(f"📡 Fetching {pool_size} candidate image marks from Supabase...")
+
+    response = (
+        supabase.table("trademarks_images")
+        .select("serial_number,image_url,design_search_codes,mark_text")
+        .neq("image_url", "NOT_FOUND")
+        .not_.is_("image_url", "null")
+        .not_.is_("design_search_codes", "null")
+        .neq("design_search_codes", "")
+        .limit(pool_size)
+        .execute()
+    )
+
+    pool = [
+        r for r in response.data
+        if r["serial_number"] not in exclude_serials
+        and r.get("image_url")
+        and r.get("design_search_codes")
+    ]
+    logger.info(f"  Pool size after filtering: {len(pool)}")
+
+    # Random sample
+    random.shuffle(pool)
+    return pool[:n]
+
+
+# ============================================================================
+# IMAGE FETCHING
+# ============================================================================
+
+async def fetch_image_bytes(url: str, http_client: httpx.AsyncClient) -> bytes:
+    """Download a Supabase Storage image."""
+    resp = await http_client.get(url, timeout=20.0)
+    resp.raise_for_status()
+    return resp.content
+
+
+# ============================================================================
+# COMPARISON METRICS
+# ============================================================================
+
+def normalize_codes(codes_str: str) -> set:
+    """Parse a comma-separated codes string into a set of XX.YY.ZZ codes.
+
+    USPTO bulk XML stores codes as 6-digit strings without separators
+    (e.g., "260121"). The classifier returns dotted format (e.g., "26.01.21").
+    This function accepts either format and normalizes everything to dotted
+    XX.YY.ZZ so set comparison works correctly.
+    """
+    if not codes_str:
+        return set()
+    normalized = set()
+    for c in codes_str.split(","):
+        c = c.strip()
+        if not c:
+            continue
+        # Already-dotted format passes through unchanged
+        if "." in c:
+            normalized.add(c)
+            continue
+        # 6-digit USPTO format → insert dots: "260121" → "26.01.21"
+        if len(c) == 6 and c.isdigit():
+            normalized.add(f"{c[0:2]}.{c[2:4]}.{c[4:6]}")
+            continue
+        # Anything else: keep as-is (will simply not match, which is correct)
+        normalized.add(c)
+    return normalized
+
+
+def compare_codes(examiner: set, classifier: set) -> Dict[str, Any]:
+    """Compute multi-level agreement between two code sets."""
+    examiner_divisions = {".".join(c.split(".")[:2]) for c in examiner}
+    examiner_categories = {c.split(".")[0] for c in examiner}
+    classifier_divisions = {".".join(c.split(".")[:2]) for c in classifier}
+    classifier_categories = {c.split(".")[0] for c in classifier}
+
+    section_overlap = examiner & classifier
+    division_overlap = examiner_divisions & classifier_divisions
+    category_overlap = examiner_categories & classifier_categories
+
+    return {
+        "examiner_count": len(examiner),
+        "classifier_count": len(classifier),
+        "section_exact_matches": len(section_overlap),
+        "division_matches": len(division_overlap),
+        "category_matches": len(category_overlap),
+        "any_section_overlap": bool(section_overlap),
+        "any_division_overlap": bool(division_overlap),
+        "any_category_overlap": bool(category_overlap),
+        # Retrieval metrics (treat examiner codes as ground truth)
+        "precision": len(section_overlap) / max(len(classifier), 1),
+        "recall": len(section_overlap) / max(len(examiner), 1),
+    }
+
+
+def f1_score(precision: float, recall: float) -> float:
+    if precision + recall == 0:
+        return 0.0
+    return 2 * precision * recall / (precision + recall)
+
+
+# ============================================================================
+# PER-SAMPLE RUN
+# ============================================================================
+
+async def validate_one(
+    record: Dict,
+    http_client: httpx.AsyncClient,
+) -> Dict[str, Any]:
+    """Run the classifier on one image and compare to examiner codes."""
+    serial = record["serial_number"]
+    image_url = record["image_url"]
+    examiner_codes_raw = record["design_search_codes"]
+    examiner_codes = normalize_codes(examiner_codes_raw)
+
+    try:
+        image_bytes = await fetch_image_bytes(image_url, http_client)
+    except Exception as e:
+        return {
+            "serial_number": serial,
+            "status": "image_fetch_failed",
+            "error": str(e),
+        }
+
+    try:
+        result = await classify_image(image_bytes)
+    except Exception as e:
+        return {
+            "serial_number": serial,
+            "status": "classification_failed",
+            "error": str(e),
+        }
+
+    classifier_codes = set(result.high_confidence_codes(threshold=CONFIDENCE_THRESHOLD))
+    metrics = compare_codes(examiner_codes, classifier_codes)
+
+    return {
+        "serial_number": serial,
+        "mark_text": record.get("mark_text", ""),
+        "image_url": image_url,
+        "examiner_codes": sorted(examiner_codes),
+        "classifier_codes": sorted(classifier_codes),
+        "image_description": result.image_description,
+        "status": "ok",
+        **metrics,
+    }
+
+
+# ============================================================================
+# MAIN
+# ============================================================================
+
+async def run(n_samples: int, resume: bool):
+    if not all([SUPABASE_URL, SUPABASE_KEY]):
+        logger.error("❌ SUPABASE_URL / SUPABASE_KEY not set in .env")
+        sys.exit(1)
+
+    state = load_checkpoint() if resume else {"completed_serials": [], "results": []}
+    completed = set(state["completed_serials"])
+    results: List[Dict] = state["results"]
+
+    if resume and completed:
+        logger.info(f"📋 Resuming — {len(completed)} samples already done")
+        n_samples = max(0, n_samples - len(completed))
+        if n_samples == 0:
+            logger.info("✅ Sample target already met from checkpoint")
+
+    supabase = create_client(SUPABASE_URL, SUPABASE_KEY)
+
+    if n_samples > 0:
+        samples = await fetch_sample(supabase, n_samples, exclude_serials=completed)
+        logger.info(f"🎯 Will validate {len(samples)} new samples")
+    else:
+        samples = []
+
+    async with httpx.AsyncClient() as http_client:
+        for idx, record in enumerate(samples, 1):
+            serial = record["serial_number"]
+            logger.info(f"\n[{idx}/{len(samples)}] Validating {serial} ({record.get('mark_text', '')[:40]})")
+
+            try:
+                result = await validate_one(record, http_client)
+                results.append(result)
+                completed.add(serial)
+
+                if result["status"] == "ok":
+                    logger.info(
+                        f"  Examiner: {result['examiner_codes']}\n"
+                        f"  Classifier: {result['classifier_codes']}\n"
+                        f"  P={result['precision']:.2f} R={result['recall']:.2f} "
+                        f"category_match={result['any_category_overlap']}"
+                    )
+                else:
+                    logger.warning(f"  ⚠️ {result['status']}: {result.get('error', '')[:200]}")
+            except Exception as e:
+                logger.error(f"  ❌ Unexpected error: {e}")
+                results.append({
+                    "serial_number": serial,
+                    "status": "unexpected_error",
+                    "error": str(e),
+                })
+                completed.add(serial)
+
+            # Checkpoint after each sample
+            state["completed_serials"] = list(completed)
+            state["results"] = results
+            save_checkpoint(state)
+
+            await asyncio.sleep(DELAY_BETWEEN_SAMPLES_S)
+
+    # ── Aggregate ──
+    ok_results = [r for r in results if r.get("status") == "ok"]
+    if not ok_results:
+        logger.error("❌ No successful validations to aggregate")
+        return
+
+    avg_precision = sum(r["precision"] for r in ok_results) / len(ok_results)
+    avg_recall = sum(r["recall"] for r in ok_results) / len(ok_results)
+    avg_f1 = f1_score(avg_precision, avg_recall)
+
+    section_match_rate = sum(1 for r in ok_results if r["any_section_overlap"]) / len(ok_results)
+    division_match_rate = sum(1 for r in ok_results if r["any_division_overlap"]) / len(ok_results)
+    category_match_rate = sum(1 for r in ok_results if r["any_category_overlap"]) / len(ok_results)
+
+    summary = {
+        "generated_at": datetime.now(timezone.utc).isoformat(),
+        "total_samples": len(results),
+        "successful_samples": len(ok_results),
+        "failed_samples": len(results) - len(ok_results),
+        "confidence_threshold": CONFIDENCE_THRESHOLD,
+        "metrics": {
+            "section_exact_match_rate": section_match_rate,
+            "division_match_rate": division_match_rate,
+            "category_match_rate": category_match_rate,
+            "avg_precision": avg_precision,
+            "avg_recall": avg_recall,
+            "avg_f1": avg_f1,
+        },
+    }
+
+    OUTPUT_JSON.write_text(json.dumps(summary, indent=2))
+
+    # CSV — per-sample, easy to sort/filter in a spreadsheet
+    with OUTPUT_CSV.open("w", newline="", encoding="utf-8") as f:
+        writer = csv.writer(f)
+        writer.writerow([
+            "serial_number", "mark_text", "image_url",
+            "examiner_codes", "classifier_codes",
+            "section_overlap", "division_overlap", "category_overlap",
+            "precision", "recall", "image_description", "status",
+        ])
+        for r in results:
+            if r.get("status") != "ok":
+                writer.writerow([r.get("serial_number"), "", "", "", "", "", "", "", "", "", "", r.get("status")])
+                continue
+            writer.writerow([
+                r["serial_number"],
+                r["mark_text"],
+                r["image_url"],
+                ",".join(r["examiner_codes"]),
+                ",".join(r["classifier_codes"]),
+                r["any_section_overlap"],
+                r["any_division_overlap"],
+                r["any_category_overlap"],
+                f"{r['precision']:.3f}",
+                f"{r['recall']:.3f}",
+                r["image_description"],
+                "ok",
+            ])
+
+    # ── Print summary ──
+    logger.info("\n" + "=" * 70)
+    logger.info("📊 VALIDATION RESULTS")
+    logger.info("=" * 70)
+    logger.info(f"  Samples:                    {len(ok_results)} successful, {len(results) - len(ok_results)} failed")
+    logger.info(f"  Confidence threshold:       {CONFIDENCE_THRESHOLD}")
+    logger.info("")
+    logger.info(f"  Section-exact match rate:   {section_match_rate:.1%}  (any XX.YY.ZZ in common)")
+    logger.info(f"  Division match rate:        {division_match_rate:.1%}  (any XX.YY in common)")
+    logger.info(f"  Category match rate:        {category_match_rate:.1%}  (any XX in common)")
+    logger.info("")
+    logger.info(f"  Avg precision:              {avg_precision:.1%}  (classifier codes that were right)")
+    logger.info(f"  Avg recall:                 {avg_recall:.1%}  (examiner codes the classifier found)")
+    logger.info(f"  Avg F1:                     {avg_f1:.3f}")
+    logger.info("")
+    logger.info(f"  Per-sample CSV: {OUTPUT_CSV}")
+    logger.info(f"  Summary JSON:   {OUTPUT_JSON}")
+    logger.info("=" * 70)
+
+
+# ============================================================================
+# CLI
+# ============================================================================
+
+async def main():
+    import argparse
+    parser = argparse.ArgumentParser(description="Validate the design code classifier")
+    parser.add_argument(
+        "--samples", type=int, default=20,
+        help="How many random image marks to test against (default: 20)"
+    )
+    parser.add_argument(
+        "--resume", action="store_true",
+        help="Skip samples already in .validation_checkpoint.json"
+    )
+    parser.add_argument(
+        "--reset-checkpoint", action="store_true",
+        help="Delete checkpoint and start fresh"
+    )
+    args = parser.parse_args()
+
+    if args.reset_checkpoint and CHECKPOINT_PATH.exists():
+        CHECKPOINT_PATH.unlink()
+        logger.info("🗑️  Checkpoint cleared")
+
+    await run(n_samples=args.samples, resume=args.resume)
+
+
+if __name__ == "__main__":
+    try:
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        logger.info("\n⚠️ Interrupted — checkpoint saved, safe to re-run with --resume")
+        sys.exit(0)