bvandeusen
d57ca847e7
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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Ttrj5P7upUTueSfoJcxEqa
89 lines
3.0 KiB
Python
89 lines
3.0 KiB
Python
"""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) == []
|