From 2ffad8aad308bf904d46b0abecaa4cc82630f5c7 Mon Sep 17 00:00:00 2001 From: ChemaVX Date: Mon, 15 Jun 2026 15:00:47 +0000 Subject: [PATCH] perf: scoring de calidad en lote + fuentes YouTube/Reddit opcionales MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #1 batch scoring (processor): - _score_quality_batch puntúa hasta 25 chunks por llamada a Claude en vez de una por chunk (una fuente de 19 chunks pasaba de 19 llamadas a 1) - parser robusto (último número por línea, padding neutro si faltan) - fallback por-chunk con Ollama si el batch falla #2 fuentes opcionales (config + scraper): - ENABLE_YOUTUBE / ENABLE_REDDIT, default False: la IP del homelab está bloqueada por Reddit (403) y YouTube (transcripts vacíos), eran peso muerto - se saltan también las URLs de yt/reddit descubiertas dentro de webs, sin gastar petición de red Co-Authored-By: Claude Opus 4.8 --- src/config.py | 5 ++ src/processor/processor.py | 100 ++++++++++++++++++++++++++++++++++--- src/scraper/exhaustive.py | 17 +++++-- 3 files changed, 111 insertions(+), 11 deletions(-) diff --git a/src/config.py b/src/config.py index 821bed3..76cbe44 100644 --- a/src/config.py +++ b/src/config.py @@ -32,6 +32,11 @@ class Settings(BaseSettings): request_delay: float = Field(1.0, env="REQUEST_DELAY") # seconds between requests min_content_length: int = Field(200, env="MIN_CONTENT_LENGTH") # chars + # Fuentes opcionales — desactivadas por defecto: la IP del homelab está + # bloqueada por Reddit (403) y YouTube (transcripts vacíos), eran peso muerto. + enable_youtube: bool = Field(False, env="ENABLE_YOUTUBE") + enable_reddit: bool = Field(False, env="ENABLE_REDDIT") + # Processing chunk_size: int = Field(800, env="CHUNK_SIZE") # tokens per chunk chunk_overlap: int = Field(100, env="CHUNK_OVERLAP") diff --git a/src/processor/processor.py b/src/processor/processor.py index c7deef8..7375d87 100644 --- a/src/processor/processor.py +++ b/src/processor/processor.py @@ -243,23 +243,32 @@ class ContentProcessor: return 0 chunks = simple_chunk(content, settings.chunk_size, settings.chunk_overlap) + # Solo se puntúan/almacenan chunks con suficiente contenido + candidates = [(i, ch) for i, ch in enumerate(chunks) + if len(ch.split()) >= 30] logger.info("Processing source", source_id=source_id, content_len=len(content), num_chunks=len(chunks), + candidates=len(candidates), quality_threshold=settings.quality_threshold) + if not candidates: + return 0 + + # Scoring en lote: 1 (o pocas) llamadas a Claude por fuente en vez de + # una por chunk — antes una fuente de 19 chunks = 19 llamadas. + qualities = await self._score_quality_batch( + [ch for _, ch in candidates], topic, session_id + ) + stored = 0 filtered_quality = 0 - for i, chunk in enumerate(chunks): - words = len(chunk.split()) - if words < 30: - continue - - quality = await self._score_quality(chunk, topic, session_id) + for (i, chunk), quality in zip(candidates, qualities): if quality < settings.quality_threshold: filtered_quality += 1 logger.debug("Chunk filtered by quality", source_id=source_id, chunk_index=i, quality=round(quality, 2), - threshold=settings.quality_threshold, words=words) + threshold=settings.quality_threshold, + words=len(chunk.split())) continue # Embeber el chunk completo (ya acotado a ~chunk_size palabras). @@ -272,7 +281,7 @@ class ContentProcessor: source_id=source_id, content=chunk, chunk_index=i, - token_count=words, + token_count=len(chunk.split()), quality_score=quality, embedding=embedding ) @@ -297,6 +306,81 @@ class ContentProcessor: return await self._score_with_claude(chunk, topic, session_id) return await self._score_with_ollama(chunk, topic) + # ─── Batch scoring ────────────────────────────────────────────────────── + + _BATCH_SCORE_SIZE = 25 + + async def _score_quality_batch(self, chunk_texts: list[str], topic: str, + session_id: int | None = None) -> list[float]: + """Puntúa varios chunks a la vez. Devuelve scores 0-1 en el mismo orden.""" + if not chunk_texts: + return [] + if not settings.anthropic_api_key: + # Ollama es local; no merece la pena batchear, se mantiene por-chunk + return [await self._score_with_ollama(c, topic) for c in chunk_texts] + + results: list[float] = [] + for i in range(0, len(chunk_texts), self._BATCH_SCORE_SIZE): + sub = chunk_texts[i:i + self._BATCH_SCORE_SIZE] + scores = await self._score_with_claude_batch(sub, topic, session_id) + if scores is None: # fallo del batch → fallback por-chunk con Ollama + scores = [await self._score_with_ollama(c, topic) for c in sub] + results.extend(scores) + return results + + @staticmethod + def _parse_batch_scores(text: str, n: int) -> list[float]: + """Extrae n scores normalizados (0-1) de la respuesta del modelo. + + Toma el último número de cada línea (robusto ante numeración tipo + '1. 8'); rellena con 0.6 neutro si faltan líneas.""" + scores: list[float] = [] + for line in text.splitlines(): + nums = re.findall(r'\d+(?:\.\d+)?', line) + if nums: + scores.append(min(1.0, float(nums[-1]) / 10.0)) + if len(scores) >= n: + return scores[:n] + return scores + [0.6] * (n - len(scores)) + + async def _score_with_claude_batch(self, chunks: list[str], topic: str, + session_id: int | None = None): + """Puntúa hasta _BATCH_SCORE_SIZE chunks en una sola llamada. + Devuelve lista de scores 0-1, o None si la llamada falla.""" + from src.llm import get_anthropic_client + listing = "\n\n".join(f"[{i + 1}]\n{c[:400]}" for i, c in enumerate(chunks)) + prompt = ( + f'Rate each of the following {len(chunks)} texts 0-10 for relevance ' + f'to the topic "{topic}". Be generous — tangentially related = 4+, ' + f'only below 3 if completely unrelated.\n' + f'Reply with EXACTLY {len(chunks)} lines, one integer 0-10 per line, ' + f'in the same order as the texts. Just the number (e.g. 7), ' + f'no labels, no other text.\n\n{listing}' + ) + try: + client = get_anthropic_client() + msg = await client.messages.create( + model=settings.claude_model, + max_tokens=8 * len(chunks) + 50, + messages=[{"role": "user", "content": prompt}] + ) + if session_id is not None: + try: + await self.db.log_api_call( + session_id, "scoring", settings.claude_model, + msg.usage.input_tokens, msg.usage.output_tokens + ) + except Exception as log_err: + logger.warning("Failed to log API usage", error=str(log_err)) + scores = self._parse_batch_scores(msg.content[0].text.strip(), len(chunks)) + logger.debug("Claude batch scored", n=len(chunks), + avg=round(sum(scores) / len(scores), 2)) + return scores + except Exception as e: + logger.warning("Claude batch scoring failed, will fallback", + error=str(e), n=len(chunks)) + return None + async def _score_with_claude(self, chunk: str, topic: str, session_id: int | None = None) -> float: from src.llm import get_anthropic_client diff --git a/src/scraper/exhaustive.py b/src/scraper/exhaustive.py index 690d557..3cc186b 100644 --- a/src/scraper/exhaustive.py +++ b/src/scraper/exhaustive.py @@ -162,13 +162,16 @@ class ExhaustiveScraper: async def seed(self): """Initial broad search across multiple sources""" - logger.info("Seeding research", topic=self.topic) + logger.info("Seeding research", topic=self.topic, + youtube=settings.enable_youtube, reddit=settings.enable_reddit) tasks = [ self._seed_search(), self._seed_wikipedia(), - self._seed_reddit(), - self._seed_youtube(), ] + if settings.enable_reddit: + tasks.append(self._seed_reddit()) + if settings.enable_youtube: + tasks.append(self._seed_youtube()) await asyncio.gather(*tasks, return_exceptions=True) async def _generate_ddg_queries(self) -> list[str]: @@ -420,6 +423,14 @@ class ExhaustiveScraper: url = source["url"] source_id = source["id"] + # Saltar fuentes desactivadas (también las descubiertas dentro de + # páginas web, no solo las del seed) sin gastar una petición de red. + if ((source_type == "youtube" and not settings.enable_youtube) or + (source_type == "reddit" and not settings.enable_reddit)): + await self.db.update_source(source_id, status="skipped", + error=f"{source_type} disabled") + return 0 + try: try: cached = await self.db.get_cached_content(url)