perf: scoring de calidad en lote + fuentes YouTube/Reddit opcionales
Build & Deploy ResearchOwl / build-and-push (push) Successful in 6s

#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 <noreply@anthropic.com>
This commit is contained in:
ChemaVX
2026-06-15 15:00:47 +00:00
co-authored by Claude Opus 4.8
parent 972bd2f883
commit 2ffad8aad3
3 changed files with 111 additions and 11 deletions
+5
View File
@@ -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")
+92 -8
View File
@@ -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
+14 -3
View File
@@ -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)