From a3d644a5c94adfd0a1cc39ff41ee42fc5e081462 Mon Sep 17 00:00:00 2001 From: ClaudioDrews Date: Wed, 3 Jun 2026 10:54:53 -0300 Subject: [PATCH] fix: resolve 4 code-critical issues from Fase 2 audit MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit V2.07 (A01): Replace subprocess.run(['arq', ...]) with programmatic arq.worker.run_worker(WorkerSettings). Fixes signal handling, graceful shutdown, and logging unification. V2.10 (D11): Fix filename → file_path in backfill_decay_metadata.py resolve_timestamp() now reads payload.file_path (consistent with what bulk_wiki_ingest.py writes). Replaced VAULT_ROOT path reconstruction with direct Path(file_path). V2.11 (D05): Auto-create Qdrant collection in bulk_wiki_ingest.py If collection doesn't exist, creates it with hybrid schema (dense 4096d Cosine + sparse BM25) instead of sys.exit(1). V2.12 (D27): Add sparse vector generation to bulk_wiki_ingest.py Optional fastembed BM25 sparse vectors alongside dense embeddings. Falls back to dense-only if fastembed not installed. Vectors stored as Qdrant named vectors: {'dense': ..., 'sparse': ...} --- docker/worker/main.py | 10 ++--- scripts/backfill_decay_metadata.py | 12 +++--- scripts/bulk_wiki_ingest.py | 64 +++++++++++++++++++++++++++--- 3 files changed, 68 insertions(+), 18 deletions(-) diff --git a/docker/worker/main.py b/docker/worker/main.py index d02508e..ddcc041 100644 --- a/docker/worker/main.py +++ b/docker/worker/main.py @@ -103,11 +103,11 @@ if __name__ == "__main__": import sys if len(sys.argv) > 1 and sys.argv[1] == "--run-worker": - # Start ARQ worker (concurrency via max_jobs, not multi-process) - import subprocess - logger.info("Starting ARQ worker...") - subprocess.run(["arq", "main.WorkerSettings"]) + from arq.worker import run_worker + logger.info("Starting ARQ worker (programmatic)...") + run_worker(WorkerSettings) else: print("Usage: python main.py --run-worker") print("") - print("To enqueue jobs, use the enqueue_host.py script") + print("To enqueue jobs via Redis, use the enqueue functions") + print("or the Hermes cron scripts in the scripts/ directory.") diff --git a/scripts/backfill_decay_metadata.py b/scripts/backfill_decay_metadata.py index 73251ad..efed499 100644 --- a/scripts/backfill_decay_metadata.py +++ b/scripts/backfill_decay_metadata.py @@ -107,7 +107,7 @@ def resolve_timestamp(point: dict) -> str: """Return ISO 8601 string for created_at/last_accessed_at. Session points: use payload.timestamp (Unix epoch float). - Wiki points: locate file via payload.filename → file mtime. + Wiki points: locate file via payload.file_path → file mtime. Fallback: now(). """ pl = point.get("payload", {}) @@ -120,13 +120,11 @@ def resolve_timestamp(point: dict) -> str: except (ValueError, TypeError, OSError): pass - # Wiki point — resolve filename to filesystem path - filename = pl.get("filename", "") + # Wiki point — resolve file_path to filesystem + file_path = pl.get("file_path", "") source = pl.get("source", "") - if filename and source.startswith("wiki-"): - # Map source to subfolder: wiki-concepts → concepts, wiki-entities → entities, etc. - subfolder = source.replace("wiki-", "") - candidate = VAULT_ROOT / "wiki" / subfolder / f"{filename}.md" + if file_path and source.startswith("wiki-"): + candidate = Path(file_path) try: mtime = candidate.stat().st_mtime return datetime.fromtimestamp(mtime, tz=timezone.utc).isoformat() diff --git a/scripts/bulk_wiki_ingest.py b/scripts/bulk_wiki_ingest.py index 264dade..0b03a45 100644 --- a/scripts/bulk_wiki_ingest.py +++ b/scripts/bulk_wiki_ingest.py @@ -16,6 +16,15 @@ from collections import Counter import aiohttp import asyncio +# Sparse embedding (BM25) — optional, falls back to dense-only +try: + from fastembed import SparseTextEmbedding + _sparse_model = SparseTextEmbedding(model_name="Qdrant/bm25") + _has_sparse = True +except ImportError: + _sparse_model = None + _has_sparse = False + # ─── Config ──────────────────────────────────────────────────────────────── OPENROUTER_KEY = os.environ.get("OPENROUTER_API_KEY") QDRANT_URL = "http://localhost:6333" @@ -70,7 +79,22 @@ def get_tags_from_frontmatter(meta: dict) -> list[str]: return tags if isinstance(tags, list) else [] async def get_embedding(session: aiohttp.ClientSession, text: str) -> list[float] | None: - """Gera embedding via OpenRouter.""" + """Gera embedding denso via OpenRouter.""" + + +def get_sparse_vector(text: str) -> dict | None: + """Gera sparse vector BM25 via fastembed (se disponível).""" + if not _has_sparse: + return None + try: + sparse_result = list(_sparse_model.embed([text]))[0] + return { + "indices": sparse_result.indices.tolist(), + "values": sparse_result.values.tolist(), + } + except Exception as e: + print(f"⚠️ Sparse embedding error: {e}") + return None payload = { "model": EMBEDDING_MODEL, "input": text[:MAX_TEXT_LEN], @@ -123,11 +147,32 @@ async def main(): connector = aiohttp.TCPConnector(limit=20) async with aiohttp.ClientSession(connector=connector) as session: - # Verificar coleção + # Verificar coleção — criar se não existir async with session.get(f"{QDRANT_URL}/collections/{COLLECTION}") as r: if r.status != 200: - print(f"❌ Coleção {COLLECTION} não existe!") - sys.exit(1) + print(f"⚠️ Coleção {COLLECTION} não existe. Criando...") + collection_config = { + "vectors": { + "dense": { + "size": EMBEDDING_DIMS, + "distance": "Cosine", + } + }, + "sparse_vectors": { + "sparse": {}, + }, + } + async with session.put( + f"{QDRANT_URL}/collections/{COLLECTION}", + headers={"Content-Type": "application/json"}, + json=collection_config, + timeout=aiohttp.ClientTimeout(total=10), + ) as cr: + if cr.status not in (200, 201): + body = await cr.text() + print(f"❌ Falha ao criar coleção ({cr.status}): {body[:200]}") + sys.exit(1) + print(f"✅ Coleção {COLLECTION} criada (dense {EMBEDDING_DIMS}d + sparse BM25)") print("\n🚀 Iniciando ingestão em batches...\n") @@ -167,14 +212,21 @@ async def main(): embed_tasks = [get_embedding(session, b["embed_text"]) for b in batch] vectors = await asyncio.gather(*embed_tasks) + # Gerar sparse vectors (sync, fastembed é CPU-bound) + sparse_vecs = [get_sparse_vector(b["embed_text"]) for b in batch] + # Preparar pontos Qdrant points = [] - for b, vec in zip(batch, vectors): + for b, vec, sparse in zip(batch, vectors, sparse_vecs): if vec is None: stats["fail"] += 1 errors.append(f"Embedding failed: {b['path']}") continue + vector_payload = {"dense": vec} + if sparse is not None: + vector_payload["sparse"] = sparse + # Heurística de importance_score baseada no path/nome importance_score = 0.5 path_str_lower = b["path"].lower() @@ -189,7 +241,7 @@ async def main(): point = { "id": str(uuid.uuid4()), - "vector": {"dense": vec}, + "vector": vector_payload, "payload": { "text": b["embed_text"], "source": b["source"],