feat(ccip): incremental character-prototype builder (#1317, m138 step 2)
refresh_character_prototypes (sync, celery ml worker): - Cheap GLOBAL gate (a few COUNTs) → no-op when nothing that affects references changed since the last refresh (the operator's "only recompute if something was tagged" trigger). - Else a per-character fingerprint diff (one GROUP BY: ref count + max region id) rebuilds ONLY the characters whose references moved — each capped to MLSettings.ccip_prototype_cap — and drops characters that lost all refs. Cost scales with WHAT changed, not library size. Reuses ccip's reference predicate (single-character, non-hygiene, figure CCIP) so prototypes match the legacy matcher exactly. The async matcher (next step) will READ the table. Tests: gate no-op when idle, only-changed-character rebuild, capping, single-character exclusion, lost-reference cleanup. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CDgx8bQS5YrGRK76v8HUnM
This commit is contained in:
@@ -0,0 +1,175 @@
|
||||
"""Precomputed CCIP character prototypes — incremental builder (#1317, m138).
|
||||
|
||||
Turns the per-character CCIP reference set into a precomputed artifact so the
|
||||
matcher never rebuilds it on the request path. Sync (runs in the celery ml
|
||||
worker via tasks.ml.refresh_character_prototypes); the async matcher only READS
|
||||
the character_prototype table.
|
||||
|
||||
`refresh_character_prototypes`:
|
||||
1. Cheap GLOBAL gate — a few COUNTs (`_global_signature`). Unchanged + not
|
||||
`full` → no-op (nothing that affects references changed since last refresh).
|
||||
2. Per-character fingerprint diff (one GROUP BY) → rebuild ONLY the characters
|
||||
whose references changed (or are new); drop characters that lost all refs.
|
||||
Each rebuild loads just ONE character's reference vectors, caps them to
|
||||
MLSettings.ccip_prototype_cap, and replaces that character's prototype rows — so
|
||||
cost scales with WHAT changed, not with the library size.
|
||||
|
||||
The reference PREDICATE (single-character, non-hygiene, figure CCIP) is imported
|
||||
from ccip so the prototypes match exactly what the legacy matcher selected.
|
||||
"""
|
||||
|
||||
import random
|
||||
from datetime import UTC, datetime
|
||||
|
||||
from sqlalchemy import delete, func, select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from ...models import (
|
||||
CcipPrototypeState,
|
||||
CharacterPrototype,
|
||||
ImageRegion,
|
||||
MLSettings,
|
||||
Tag,
|
||||
TagKind,
|
||||
)
|
||||
from ...models.tag import image_tag
|
||||
from .ccip import _FIGURE_KINDS, _hygiene_tagged_images, _single_character_images
|
||||
|
||||
# Deterministic per-tag capping so a rebuild of an UNCHANGED reference set
|
||||
# resamples identically (stable prototypes, no churn between refreshes).
|
||||
_SAMPLE_SEED = 1317
|
||||
|
||||
|
||||
def _global_signature(session: Session) -> str:
|
||||
"""Cheap 'could any references have changed' gate: character-tag count,
|
||||
figure-CCIP region count + max id, hygiene-tag count. A few COUNTs — the same
|
||||
quantities the legacy per-request signature used, now computed once per
|
||||
refresh instead of on every /suggestions call."""
|
||||
n_tags = session.execute(
|
||||
select(func.count())
|
||||
.select_from(image_tag)
|
||||
.join(Tag, Tag.id == image_tag.c.tag_id)
|
||||
.where(Tag.kind == TagKind.character)
|
||||
).scalar_one()
|
||||
n_regs, max_id = session.execute(
|
||||
select(func.count(), func.max(ImageRegion.id)).where(
|
||||
ImageRegion.kind.in_(_FIGURE_KINDS),
|
||||
ImageRegion.ccip_embedding.is_not(None),
|
||||
)
|
||||
).one()
|
||||
n_hygiene = session.execute(
|
||||
select(func.count())
|
||||
.select_from(image_tag)
|
||||
.join(Tag, Tag.id == image_tag.c.tag_id)
|
||||
.where(Tag.is_system.is_(True))
|
||||
).scalar_one()
|
||||
return f"{n_tags}:{n_regs}:{max_id or 0}:{n_hygiene}"
|
||||
|
||||
|
||||
def _current_fingerprints(session: Session) -> dict[int, str]:
|
||||
"""Per-character (reference count, max reference region id) over the SAME
|
||||
predicate the matcher's references use. One GROUP BY → the change detector:
|
||||
a character whose fingerprint moved (gained/lost a reference) needs a
|
||||
rebuild; everyone else is left untouched."""
|
||||
rows = session.execute(
|
||||
select(
|
||||
image_tag.c.tag_id,
|
||||
func.count(ImageRegion.id),
|
||||
func.max(ImageRegion.id),
|
||||
)
|
||||
.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))
|
||||
.where(ImageRegion.image_record_id.in_(_single_character_images()))
|
||||
.where(ImageRegion.image_record_id.not_in(_hygiene_tagged_images()))
|
||||
.group_by(image_tag.c.tag_id)
|
||||
).all()
|
||||
return {tag_id: f"{cnt}:{mx}" for tag_id, cnt, mx in rows}
|
||||
|
||||
|
||||
def _rebuild_one(session: Session, tag_id: int, cap: int) -> int:
|
||||
"""Replace ONE character's prototype rows from its current references, capped
|
||||
to `cap`. Loads only this character's vectors (bounded by its popularity)."""
|
||||
rows = session.execute(
|
||||
select(ImageRegion.id, ImageRegion.ccip_embedding)
|
||||
.select_from(ImageRegion)
|
||||
.join(
|
||||
image_tag,
|
||||
image_tag.c.image_record_id == ImageRegion.image_record_id,
|
||||
)
|
||||
.where(image_tag.c.tag_id == tag_id)
|
||||
.where(ImageRegion.kind.in_(_FIGURE_KINDS))
|
||||
.where(ImageRegion.ccip_embedding.is_not(None))
|
||||
.where(ImageRegion.image_record_id.in_(_single_character_images()))
|
||||
.where(ImageRegion.image_record_id.not_in(_hygiene_tagged_images()))
|
||||
).all()
|
||||
# Cap for bounded MATCH cost. A random sample (not most-recent) keeps the
|
||||
# prototypes representative of the whole reference set; a fixed per-tag seed
|
||||
# makes an unchanged set resample identically.
|
||||
if cap > 0 and len(rows) > cap:
|
||||
rows = random.Random(f"{_SAMPLE_SEED}:{tag_id}").sample(rows, cap)
|
||||
session.execute(
|
||||
delete(CharacterPrototype).where(CharacterPrototype.tag_id == tag_id)
|
||||
)
|
||||
for region_id, vec in rows:
|
||||
session.add(
|
||||
CharacterPrototype(
|
||||
tag_id=tag_id, region_id=region_id, ccip_embedding=vec
|
||||
)
|
||||
)
|
||||
return len(rows)
|
||||
|
||||
|
||||
def refresh_character_prototypes(
|
||||
session: Session, *, full: bool = False
|
||||
) -> dict[str, int | bool]:
|
||||
"""Incrementally refresh the prototype store. `full=True` rebuilds every
|
||||
character regardless of the gate/fingerprints (nightly reconcile). Returns
|
||||
{skipped, rebuilt, removed}; commits."""
|
||||
settings = session.execute(
|
||||
select(MLSettings).where(MLSettings.id == 1)
|
||||
).scalar_one()
|
||||
sig = _global_signature(session)
|
||||
if not full and settings.ccip_ref_signature == sig:
|
||||
return {"skipped": True, "rebuilt": 0, "removed": 0}
|
||||
|
||||
cap = settings.ccip_prototype_cap
|
||||
current = _current_fingerprints(session)
|
||||
stored = dict(
|
||||
session.execute(
|
||||
select(CcipPrototypeState.tag_id, CcipPrototypeState.fingerprint)
|
||||
).all()
|
||||
)
|
||||
now = datetime.now(UTC)
|
||||
rebuilt = 0
|
||||
for tag_id, fp in current.items():
|
||||
if full or stored.get(tag_id) != fp:
|
||||
_rebuild_one(session, tag_id, cap)
|
||||
state = session.get(CcipPrototypeState, tag_id)
|
||||
if state is None:
|
||||
state = CcipPrototypeState(tag_id=tag_id)
|
||||
session.add(state)
|
||||
state.fingerprint = fp
|
||||
state.updated_at = now
|
||||
rebuilt += 1
|
||||
# Characters that lost every reference (refs removed / re-kinded / image now
|
||||
# multi-character) → drop their prototypes + state so they stop matching.
|
||||
removed = 0
|
||||
for tag_id in set(stored) - set(current):
|
||||
session.execute(
|
||||
delete(CharacterPrototype).where(CharacterPrototype.tag_id == tag_id)
|
||||
)
|
||||
session.execute(
|
||||
delete(CcipPrototypeState).where(CcipPrototypeState.tag_id == tag_id)
|
||||
)
|
||||
removed += 1
|
||||
|
||||
settings.ccip_ref_signature = sig
|
||||
session.commit()
|
||||
return {"skipped": False, "rebuilt": rebuilt, "removed": removed}
|
||||
@@ -0,0 +1,163 @@
|
||||
"""Incremental CCIP prototype builder (#1317, m138). Sync (celery ml worker),
|
||||
so tested directly via db_sync — the global gate, per-character fingerprint diff,
|
||||
capping, single-character exclusion, and lost-reference cleanup."""
|
||||
import pytest
|
||||
from sqlalchemy import func, select
|
||||
|
||||
from backend.app.models import (
|
||||
CcipPrototypeState,
|
||||
CharacterPrototype,
|
||||
ImageRecord,
|
||||
ImageRegion,
|
||||
MLSettings,
|
||||
Tag,
|
||||
TagKind,
|
||||
)
|
||||
from backend.app.models.tag import image_tag
|
||||
from backend.app.services.ml.character_prototypes import (
|
||||
refresh_character_prototypes,
|
||||
)
|
||||
|
||||
pytestmark = pytest.mark.integration
|
||||
|
||||
|
||||
def _ccip(slot: int = 0) -> list[float]:
|
||||
v = [0.0] * 768
|
||||
v[slot] = 1.0
|
||||
return v
|
||||
|
||||
|
||||
def _img(db, sha: str) -> 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)
|
||||
db.flush()
|
||||
return img
|
||||
|
||||
|
||||
def _figure(db, image_id: int, ccip=None) -> None:
|
||||
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 if ccip is not None else _ccip(),
|
||||
embedding_version="ccip-test",
|
||||
))
|
||||
db.flush()
|
||||
|
||||
|
||||
def _char(db, name: str) -> Tag:
|
||||
t = Tag(name=name, kind=TagKind.character)
|
||||
db.add(t)
|
||||
db.flush()
|
||||
return t
|
||||
|
||||
|
||||
def _tag(db, image_id: int, tag_id: int) -> None:
|
||||
db.execute(image_tag.insert().values(
|
||||
image_record_id=image_id, tag_id=tag_id, source="manual",
|
||||
))
|
||||
|
||||
|
||||
def _proto_count(db, tag_id: int) -> int:
|
||||
return db.execute(
|
||||
select(func.count()).select_from(CharacterPrototype)
|
||||
.where(CharacterPrototype.tag_id == tag_id)
|
||||
).scalar_one()
|
||||
|
||||
|
||||
def _set_cap(db, cap: int) -> None:
|
||||
s = db.execute(select(MLSettings).where(MLSettings.id == 1)).scalar_one()
|
||||
s.ccip_prototype_cap = cap
|
||||
db.flush()
|
||||
|
||||
|
||||
def test_refresh_builds_then_gate_no_ops(db_sync):
|
||||
raven = _char(db_sync, "Raven")
|
||||
img = _img(db_sync, "a" * 64)
|
||||
_figure(db_sync, img.id)
|
||||
_tag(db_sync, img.id, raven.id)
|
||||
db_sync.commit()
|
||||
|
||||
r1 = refresh_character_prototypes(db_sync)
|
||||
assert r1["skipped"] is False and r1["rebuilt"] == 1
|
||||
assert _proto_count(db_sync, raven.id) == 1
|
||||
assert db_sync.get(CcipPrototypeState, raven.id) is not None
|
||||
|
||||
# Nothing changed → the global gate short-circuits (no per-character work).
|
||||
r2 = refresh_character_prototypes(db_sync)
|
||||
assert r2["skipped"] is True and r2["rebuilt"] == 0
|
||||
|
||||
|
||||
def test_only_changed_character_rebuilt(db_sync):
|
||||
raven = _char(db_sync, "Raven")
|
||||
daphne = _char(db_sync, "Daphne")
|
||||
ri = _img(db_sync, "b" * 64)
|
||||
_figure(db_sync, ri.id)
|
||||
_tag(db_sync, ri.id, raven.id)
|
||||
di = _img(db_sync, "c" * 64)
|
||||
_figure(db_sync, di.id)
|
||||
_tag(db_sync, di.id, daphne.id)
|
||||
db_sync.commit()
|
||||
assert refresh_character_prototypes(db_sync)["rebuilt"] == 2
|
||||
|
||||
# A new figure on Raven's image changes only Raven's fingerprint.
|
||||
_figure(db_sync, ri.id, _ccip(1))
|
||||
db_sync.commit()
|
||||
r = refresh_character_prototypes(db_sync)
|
||||
assert r["rebuilt"] == 1 # Daphne untouched
|
||||
assert _proto_count(db_sync, raven.id) == 2
|
||||
assert _proto_count(db_sync, daphne.id) == 1
|
||||
|
||||
|
||||
def test_cap_bounds_prototypes(db_sync):
|
||||
_set_cap(db_sync, 2)
|
||||
raven = _char(db_sync, "Raven")
|
||||
for i in range(3): # 3 single-character images
|
||||
img = _img(db_sync, f"{i}" * 64)
|
||||
_figure(db_sync, img.id)
|
||||
_tag(db_sync, img.id, raven.id)
|
||||
db_sync.commit()
|
||||
|
||||
refresh_character_prototypes(db_sync)
|
||||
assert _proto_count(db_sync, raven.id) == 2 # capped from 3 → 2
|
||||
|
||||
|
||||
def test_multi_character_image_not_referenced(db_sync):
|
||||
# A figure on a 2-character image is ambiguous (tag is image-level), so it
|
||||
# must NOT become a reference for either — parity with the legacy matcher.
|
||||
raven = _char(db_sync, "Raven")
|
||||
daphne = _char(db_sync, "Daphne")
|
||||
img = _img(db_sync, "d" * 64)
|
||||
_figure(db_sync, img.id)
|
||||
_tag(db_sync, img.id, raven.id)
|
||||
_tag(db_sync, img.id, daphne.id)
|
||||
db_sync.commit()
|
||||
|
||||
refresh_character_prototypes(db_sync)
|
||||
assert _proto_count(db_sync, raven.id) == 0
|
||||
assert _proto_count(db_sync, daphne.id) == 0
|
||||
|
||||
|
||||
def test_lost_references_are_removed(db_sync):
|
||||
raven = _char(db_sync, "Raven")
|
||||
img = _img(db_sync, "e" * 64)
|
||||
_figure(db_sync, img.id)
|
||||
_tag(db_sync, img.id, raven.id)
|
||||
db_sync.commit()
|
||||
refresh_character_prototypes(db_sync)
|
||||
assert _proto_count(db_sync, raven.id) == 1
|
||||
|
||||
# Untag Raven → the image is no longer a reference → prototypes + state drop.
|
||||
db_sync.execute(
|
||||
image_tag.delete()
|
||||
.where(image_tag.c.image_record_id == img.id)
|
||||
.where(image_tag.c.tag_id == raven.id)
|
||||
)
|
||||
db_sync.commit()
|
||||
r = refresh_character_prototypes(db_sync)
|
||||
assert r["removed"] == 1
|
||||
assert _proto_count(db_sync, raven.id) == 0
|
||||
assert db_sync.get(CcipPrototypeState, raven.id) is None
|
||||
Reference in New Issue
Block a user