File size: 6,114 Bytes
6b1bc4b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 | """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)
|