research-agent-plugin/scripts/research_validator.py

#!/usr/bin/env python3
"""Research Validator — проверка качества исследовательских отчётов.

Три проверки, которые ловят 80% брака от локальных LLM:
1. Цитаты — все [N] существуют, нет «мёртвых» ссылок
2. URL — все валидные (http, не битые relative)
3. Полнота — минимум источников, достаточно контента

Использование:
    from research_validator import validate_report

    ok, errors = validate_report(report_text, sources_list)
    if not ok:
        print("Проблемы:", errors)
        # → отправить errors обратно в промпт LLM на исправление

Также используется для предварительной фильтрации URL из SearXNG:
    clean_urls = filter_valid_urls(raw_results)
"""

import re
from typing import Any
from urllib.parse import urlparse


def validate_report(report: str, sources: list[dict]) -> tuple[bool, list[str]]:
    """Проверить отчёт на соответствие стандартам качества.

    Args:
        report: текст отчёта (Markdown)
        sources: список словарей с ключами ref (int), title (str), url (str)

    Returns:
        (True, []) — всё хорошо
        (False, ["ошибка 1", "ошибка 2"]) — нужно исправлять
    """
    errors = []

    # 1. Проверка цитат: все [N] в тексте должны существовать в sources
    cited_nums = set()
    for m in re.finditer(r'\[(\d+)\]', report):
        cited_nums.add(int(m.group(1)))

    source_nums = {s["ref"] for s in sources}

    missing = cited_nums - source_nums
    if missing:
        errors.append(
            f"Цитаты [{', '.join(str(n) for n in sorted(missing))}] "
            f"не существуют — нет таких источников"
        )

    unused = source_nums - cited_nums
    if unused:
        errors.append(
            f"Источники [{', '.join(str(n) for n in sorted(unused))}] "
            f"не процитированы в тексте"
        )

    # 2. Проверка URL: все валидные
    for s in sources:
        url = s.get("url", "")
        if not url:
            errors.append(f"[{s['ref']}] пустой URL")
        elif not url.startswith("http"):
            errors.append(f"[{s['ref']}] невалидный URL: {url[:80]}")
        else:
            parsed = urlparse(url)
            if not parsed.netloc:
                errors.append(f"[{s['ref']}] URL без домена: {url[:80]}")

    # 3. Проверка полноты
    if len(sources) < 3:
        errors.append(f"Слишком мало источников: {len(sources)} (минимум 3)")

    total_text = sum(len(s.get("text", "")) for s in sources)
    if total_text < 1000:
        errors.append(f"Слишком мало контента: {total_text} символов (минимум 1000)")

    # 4. Проверка структуры — есть ли ключевые секции
    has_overview = bool(re.search(r'###?\s*(Обзор|Overview)', report))
    has_sources = bool(re.search(r'###?\s*(Источники|Sources)', report))
    has_citations = bool(re.search(r'\[\d+\]', report))

    if not has_overview:
        errors.append("Нет секции «Обзор»")
    if not has_citations:
        errors.append("Нет inline-цитат [N] в тексте")
    if not has_sources and sources:
        errors.append("Нет секции «Источники» в конце отчёта")

    return len(errors) == 0, errors


def filter_valid_urls(results: list[dict]) -> list[dict]:
    """Отфильтровать результаты SearXNG: только с валидными URL.

    Args:
        results: список словарей от SearXNG (url, title, content, engine)

    Returns:
        Отфильтрованный список (только http/https URL)
    """
    valid = []
    for r in results:
        url = r.get("url", "")
        if not url or not url.startswith("http"):
            continue
        parsed = urlparse(url)
        if not parsed.netloc:
            continue
        # Отбрасываем явно битые
        if any(x in url.lower() for x in ["example.com", "localhost", "0.0.0.0"]):
            continue
        valid.append(r)
    return valid


def score_source(result: dict) -> float:
    """Оценить качество источника от 0 до 1.

    Учитывает: поисковый движок, длину сниппета, наличие даты.
    """
    score = 0.5  # базовый

    engine = result.get("engine", "").lower()
    engine_scores = {
        "wikipedia": 0.4,  # +0.4 за достоверность
        "arxiv": 0.3,
        "google": 0.2,
        "bing": 0.1,
        "duckduckgo": 0.1,
        "brave": 0.1,
        "reddit": -0.1,  # штраф за субъективность
    }
    score += engine_scores.get(engine, 0)

    # Бонус за длинный сниппет (больше информации)
    snippet = result.get("content", "") or result.get("snippet", "")
    if len(snippet) > 200:
        score += 0.1
    if len(snippet) > 500:
        score += 0.1

    # Штраф за отсутствие заголовка
    if not result.get("title"):
        score -= 0.3

    return max(0.0, min(1.0, score))


def build_fix_prompt(report: str, errors: list[str], sources: list[dict]) -> str:
    """Собрать промпт для LLM с конкретными ошибками для исправления."""
    sources_text = "\n".join(
        f"[{s['ref']}] {s['title']} — {s['url']}" for s in sources
    )
    errors_text = "\n".join(f"- {e}" for e in errors)

    return f"""Отчёт содержит ошибки. Исправь их и верни ПОЛНЫЙ исправленный отчёт.

ОШИБКИ:
{errors_text}

ДОСТУПНЫЕ ИСТОЧНИКИ:
{sources_text}

ТЕКУЩИЙ ОТЧЁТ:
{report[:4000]}

ТРЕБОВАНИЯ:
- Каждый факт с цитатой [N], где N — номер из списка источников
- Все указанные источники должны быть процитированы
- В конце — секция «### Источники» с полными URL
- Секция «### Обзор» обязательна
- Не добавляй источники, которых нет в списке"""


# ═══════════════════════════════════════════
# CLI
# ═══════════════════════════════════════════

if __name__ == "__main__":
    import sys, json

    if len(sys.argv) < 2:
        print("Usage: research_validator.py <report.md> [sources.json]")
        print("  sources.json: [{\"ref\": 1, \"title\": \"...\", \"url\": \"...\"}, ...]")
        sys.exit(1)

    report_text = open(sys.argv[1]).read()

    if len(sys.argv) > 2:
        sources = json.loads(open(sys.argv[2]).read())
    else:
        sources = []

    ok, errors = validate_report(report_text, sources)

    if ok:
        print("✓ Отчёт валиден")
    else:
        print(f"✗ Найдено {len(errors)} проблем:")
        for e in errors:
            print(f"  - {e}")
        sys.exit(1)