feat(ccip): few-shot character matcher (#114 slice 5)
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
This commit is contained in:
@@ -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) == []
|
||||
Reference in New Issue
Block a user