| """Patch backend/personalization/worker.py of the UniHive `main` repo (python patch_worker.py <repo root>). |
| |
| Two problems of the original enrich_once, both made worse by seeding the interest catalogue (migration 012): |
| 1. The content hash includes the whole `domains` table, so any catalogue change re-processes and RE-EMBEDS every post (32 per cycle, 10 s sleep per cycle). |
| 2. A pass without the model (embeddings off, or the worker's fallback after an error) overwrote stored vectors with NULL. |
| New behaviour (no schema change; the text hash lives in the existing `features` jsonb, which nothing reads): |
| * a stored vector is never replaced by NULL (pass without a model skips rows that have one; the upsert also keeps an existing vector as a guard); |
| * when only the catalogue / tags / domain / community changed and the embedding text is the same, only the cheap features are refreshed (up to `refresh_limit` per call), no embedding is computed; |
| * rows written by the old worker have no text hash: a POST whose vector was written after its last edit (`edited_at`) is trusted, anything else is re-embedded as before; |
| * everything else is unchanged: new rows, edited text, and rows whose vector comes from another model are embedded, `limit` per call. The return value is the number of rows written.""" |
| import sys |
|
|
| path = sys.argv[1] + "/backend/personalization/worker.py" |
| s = open(path, encoding="utf8", newline="").read(); crlf = "\r\n" in s; s = s.replace("\r\n", "\n") |
| start, end = s.index("async def enrich_once(model=None,limit=32):"), s.index("async def main():") |
| NEW = '''def _edited_after(entity,when): |
| """True if the entity's `edited_at` is later than `when` (unparsable values count as edited, so the row is re-embedded).""" |
| from datetime import datetime,timezone |
| value=entity.get('edited_at') |
| if not value:return False |
| try: |
| edited=datetime.fromisoformat(str(value).replace('Z','+00:00')) |
| return (edited if edited.tzinfo else edited.replace(tzinfo=timezone.utc))>when |
| except (ValueError,TypeError):return True |
| |
| async def enrich_once(model=None,limit=32,refresh_limit=1000): |
| async with use_conn() as c: |
| domains=await c.fetch('SELECT id,name,slug FROM domains') |
| rows=await c.fetch(\'\'\'SELECT 'POST' AS entity_type,id,to_jsonb(p) AS entity FROM posts p WHERE status='PUBLISHED' |
| UNION ALL SELECT 'PROJECT',id,to_jsonb(p) FROM projects p WHERE status<>'DRAFT' \'\'\') |
| cached=await c.fetch("SELECT entity_type,entity_id,content_hash,embedding_model,(embedding IS NOT NULL) AS has_vector,features->>'_text_sha' AS text_sha,updated_at FROM personalization_content") |
| # Taxonomy changes also invalidate content features. |
| taxonomy=json.dumps([dict(d) for d in domains],default=str,sort_keys=True) |
| cache={(r['entity_type'],r['entity_id']):r for r in cached} |
| batch=[];refresh=[] # batch: needs a vector (or a first row); refresh: only the stored features / hash change, the vector stays valid |
| for row in rows: |
| entity=row['entity'];text='\\n'.join(str(entity.get(k) or '') for k in ('title','body','summary','description')) |
| signature=hashlib.sha256((text+json.dumps(entity.get('tags'))+str(entity.get('domain_id'))+str(entity.get('community_id'))+taxonomy).encode()).hexdigest() |
| previous=cache.get((row['entity_type'],row['id'])) |
| if previous and previous['content_hash']==signature and not (model and previous['embedding_model']!=EMBEDDING_MODEL):continue # up to date |
| text_sha=hashlib.sha256(text[:12000].encode()).hexdigest() |
| vector_ok=bool(previous and previous['has_vector'] and previous['embedding_model']==EMBEDDING_MODEL and (previous['text_sha']==text_sha or |
| (previous['text_sha'] is None and row['entity_type']=='POST' and not _edited_after(entity,previous['updated_at'])))) |
| if vector_ok: |
| if len(refresh)<refresh_limit:refresh.append((row,signature,text_sha)) |
| elif previous and previous['has_vector'] and not model:continue # no model now: never overwrite a stored vector with NULL; the next pass with the model refreshes it |
| elif len(batch)<limit:batch.append((row,signature,text,text_sha)) |
| if len(batch)>=limit and len(refresh)>=refresh_limit:break |
| vectors=await asyncio.to_thread(lambda:list(model.embed([text[:12000] for _,_,text,_ in batch]))) if model and batch else [None]*len(batch) |
| async with use_conn() as c: |
| for (row,signature,text,text_sha),vector in zip(batch,vectors): |
| value=None |
| if vector is not None: |
| import numpy as np |
| vector=np.asarray(vector,dtype=float);norm=np.linalg.norm(vector) |
| if vector.shape!=(384,) or not np.isfinite(vector).all() or not norm:raise ValueError('Invalid model embedding') |
| value=json.dumps((vector/norm).tolist()) |
| await c.execute(\'\'\'INSERT INTO personalization_content(entity_type,entity_id,content_hash,features,embedding,embedding_model) |
| VALUES($1,$2,$3,$4,$5::vector,$6) ON CONFLICT(entity_type,entity_id) DO UPDATE SET |
| content_hash=excluded.content_hash,features=excluded.features, |
| embedding=coalesce(excluded.embedding,personalization_content.embedding), |
| embedding_model=CASE WHEN excluded.embedding IS NULL THEN personalization_content.embedding_model ELSE excluded.embedding_model END,updated_at=now()\'\'\', |
| row['entity_type'],row['id'],signature,{**features(row['entity'],domains),'_text_sha':text_sha},value,EMBEDDING_MODEL if value else None) |
| for row,signature,text_sha in refresh: |
| await c.execute('UPDATE personalization_content SET content_hash=$3,features=$4,updated_at=now() WHERE entity_type=$1 AND entity_id=$2', |
| row['entity_type'],row['id'],signature,{**features(row['entity'],domains),'_text_sha':text_sha}) |
| return len(batch)+len(refresh) |
| |
| ''' |
| s = s[:start] + NEW + s[end:] |
| open(path, "w", encoding="utf8", newline="").write(s.replace("\n", "\r\n") if crlf else s) |
| print("patched", path) |
|
|
