From d57ca847e7070c1d9e700c7f7ce4d4f7c2d22c9c Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Mon, 29 Jun 2026 11:57:39 -0400 Subject: [PATCH] feat(ccip): few-shot character matcher (#114 slice 5) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The server-side brain that turns stored CCIP vectors into character suggestions — no GPU. character_references() gathers each character tag's prototype vectors (figure/face-region CCIP embeddings on images carrying that tag); match_image() cosine-matches an image's figure vectors against every character (multi- prototype: best over a character's examples), surfacing those above a tunable threshold as {tag_id, name, category:'character', score, source:'ccip'}, excluding already-applied characters. v1 = cosine on raw CCIP vectors; the exact CCIP metric/threshold gets validated against the model in the hands-on eval. Tests (synthetic vectors): same-character match across images, no-match for an orthogonal figure, already-applied exclusion, no-figure-vectors empty. NEXT: merge CCIP character suggestions into the rail; the agent container that actually produces the vectors (hands-on, GPU — not CI-verifiable). Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Ttrj5P7upUTueSfoJcxEqa --- backend/app/services/ml/ccip.py | 120 ++++++++++++++++++++++++++++++++ tests/test_ccip.py | 88 +++++++++++++++++++++++ 2 files changed, 208 insertions(+) create mode 100644 backend/app/services/ml/ccip.py create mode 100644 tests/test_ccip.py diff --git a/backend/app/services/ml/ccip.py b/backend/app/services/ml/ccip.py new file mode 100644 index 0000000..406fbed --- /dev/null +++ b/backend/app/services/ml/ccip.py @@ -0,0 +1,120 @@ +"""CCIP few-shot character matcher (#114) — server-side, numpy on stored vectors. + +CCIP is a FROZEN identity embedding; we don't train it. Instead the operator's +tagged characters become reference prototypes: a character tag's references are +the CCIP vectors of figure/face regions on images carrying that tag. To suggest +characters for a new image, we compare its figure-region CCIP vectors to every +character's references (multi-prototype: best match over a character's examples) +and surface the ones that clear a similarity threshold. No GPU here — the agent +already produced the vectors; this is cosine matching on what's stored. + +v1 uses cosine similarity on the raw CCIP vectors with a tunable threshold; the +exact CCIP difference metric/threshold gets validated against the model during +the hands-on eval. numpy is imported lazily (API worker has it via pgvector). +""" + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from ...models import ImageRegion, Tag, TagKind +from ...models.tag import image_tag + +# Cosine-similarity floor to call a figure the same character. Conservative +# default; tune from real matches (CCIP same-char clusters tightly). +DEFAULT_SIM_THRESHOLD = 0.75 +_FIGURE_KINDS = ("face", "figure") + + +def _l2norm(mat, np): + n = np.linalg.norm(mat, axis=1, keepdims=True) + n[n == 0] = 1.0 + return mat / n + + +async def character_references(session: AsyncSession) -> dict[int, list]: + """Per character-tag CCIP reference vectors: figure/face-region CCIP + embeddings on images that carry that character tag (the operator's examples). + Multi-prototype — several vectors per character.""" + rows = ( + await session.execute( + select(image_tag.c.tag_id, ImageRegion.ccip_embedding) + .select_from(ImageRegion) + .join( + image_tag, + image_tag.c.image_record_id == ImageRegion.image_record_id, + ) + .join(Tag, Tag.id == image_tag.c.tag_id) + .where(Tag.kind == TagKind.character) + .where(ImageRegion.kind.in_(_FIGURE_KINDS)) + .where(ImageRegion.ccip_embedding.is_not(None)) + ) + ).all() + refs: dict[int, list] = {} + for tag_id, vec in rows: + refs.setdefault(tag_id, []).append(vec) + return refs + + +async def _tag_names(session: AsyncSession, tag_ids: list[int]) -> dict[int, str]: + if not tag_ids: + return {} + return dict( + ( + await session.execute( + select(Tag.id, Tag.name).where(Tag.id.in_(tag_ids)) + ) + ).all() + ) + + +async def match_image( + session: AsyncSession, image_id: int, threshold: float = DEFAULT_SIM_THRESHOLD +) -> list[dict]: + """Character suggestions for one image from its figure-region CCIP vectors: + [{tag_id, name, category:'character', score, source:'ccip'}], ranked. + Already-applied character tags are excluded. Empty if the image has no figure + CCIP vectors or no character references exist yet.""" + import numpy as np + + qvecs = ( + await session.execute( + select(ImageRegion.ccip_embedding).where( + ImageRegion.image_record_id == image_id, + ImageRegion.kind.in_(_FIGURE_KINDS), + ImageRegion.ccip_embedding.is_not(None), + ) + ) + ).scalars().all() + if not qvecs: + return [] + refs = await character_references(session) + if not refs: + return [] + applied = set( + ( + await session.execute( + select(image_tag.c.tag_id).where( + image_tag.c.image_record_id == image_id + ) + ) + ).scalars() + ) + names = await _tag_names(session, [t for t in refs if t not in applied]) + + Q = _l2norm(np.vstack([np.asarray(v, dtype=np.float32) for v in qvecs]), np) + out = [] + for tag_id, vecs in refs.items(): + if tag_id in applied: + continue + R = _l2norm(np.vstack([np.asarray(v, dtype=np.float32) for v in vecs]), np) + best = float((Q @ R.T).max()) # best (query figure, reference) cosine + if best >= threshold: + out.append({ + "tag_id": tag_id, + "name": names.get(tag_id, str(tag_id)), + "category": "character", + "score": round(best, 4), + "source": "ccip", + }) + out.sort(key=lambda d: d["score"], reverse=True) + return out diff --git a/tests/test_ccip.py b/tests/test_ccip.py new file mode 100644 index 0000000..5d81c72 --- /dev/null +++ b/tests/test_ccip.py @@ -0,0 +1,88 @@ +"""CCIP few-shot character matcher (#114). numpy cosine on stored vectors — no +model needed, so it runs in CI with synthetic CCIP vectors.""" +import pytest + +from backend.app.models import ImageRecord, ImageRegion, TagKind +from backend.app.models.tag import image_tag +from backend.app.services.ml.ccip import match_image +from backend.app.services.tag_service import TagService + +pytestmark = pytest.mark.integration + + +def _ccip(slot: int) -> list[float]: + v = [0.0] * 768 + v[slot] = 1.0 + return v + + +async def _img(db, sha) -> ImageRecord: + img = ImageRecord( + path=f"/images/{sha}.jpg", sha256=sha, size_bytes=1, mime="image/jpeg", + width=1, height=1, origin="imported_filesystem", integrity_status="unknown", + ) + db.add(img) + await db.flush() + return img + + +async def _figure(db, image_id, ccip): + db.add(ImageRegion( + image_record_id=image_id, kind="figure", + rx=0.0, ry=0.0, rw=1.0, rh=1.0, + ccip_embedding=ccip, embedding_version="ccip-test", + )) + + +async def _tag_image(db, image_id, tag_id): + await db.execute(image_tag.insert().values( + image_record_id=image_id, tag_id=tag_id, source="manual", + )) + + +@pytest.mark.asyncio +async def test_matches_same_character_across_images(db): + raven = await TagService(db).find_or_create("Raven", TagKind.character) + ref = await _img(db, "a" * 64) # a tagged example = a prototype + await _figure(db, ref.id, _ccip(0)) + await _tag_image(db, ref.id, raven.id) + query = await _img(db, "b" * 64) # untagged, near-identical figure + await _figure(db, query.id, _ccip(0)) + await db.commit() + + matches = await match_image(db, query.id) + m = next(x for x in matches if x["tag_id"] == raven.id) + assert m["source"] == "ccip" and m["category"] == "character" + assert m["score"] > 0.9 + + +@pytest.mark.asyncio +async def test_no_match_for_different_character(db): + raven = await TagService(db).find_or_create("Raven", TagKind.character) + ref = await _img(db, "c" * 64) + await _figure(db, ref.id, _ccip(0)) + await _tag_image(db, ref.id, raven.id) + query = await _img(db, "d" * 64) + await _figure(db, query.id, _ccip(5)) # orthogonal → not Raven + await db.commit() + assert await match_image(db, query.id) == [] + + +@pytest.mark.asyncio +async def test_excludes_already_applied_character(db): + raven = await TagService(db).find_or_create("Raven", TagKind.character) + ref = await _img(db, "e" * 64) + await _figure(db, ref.id, _ccip(0)) + await _tag_image(db, ref.id, raven.id) + query = await _img(db, "f" * 64) + await _figure(db, query.id, _ccip(0)) + await _tag_image(db, query.id, raven.id) # already tagged → no re-suggest + await db.commit() + assert all(m["tag_id"] != raven.id for m in await match_image(db, query.id)) + + +@pytest.mark.asyncio +async def test_no_figure_vectors_means_no_match(db): + query = await _img(db, "g" * 64) + await db.commit() + assert await match_image(db, query.id) == []