tests/validate_runtime.py

"""Ручная интеграционная проверка настоящей модели на файле из базы (не unit-тест)."""
from __future__ import annotations

import argparse
import asyncio
import json
from pathlib import Path
import re
import sys
import time

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from core.agent_loop import AgentLoop
from core.formula_engine import FormulaEngine
from core.parser import DocumentParser
from core.qa_engine import QASystem


def main() -> None:
    cli = argparse.ArgumentParser()
    cli.add_argument("--document", type=Path, required=True)
    cli.add_argument("--model", default="llama3.1:8b")
    cli.add_argument("--no-llm", action="store_true")
    cli.add_argument("--output", type=Path, default=Path("data/processed/runtime_validation.json"))
    args = cli.parse_args()
    started = time.monotonic()
    document = DocumentParser().parse_file(args.document)
    assert document and document.get("chunks"), "Документ не прочитан"
    qa = QASystem(use_llm=not args.no_llm, use_embeddings=True, ollama_model=args.model, top_k=3)
    assert qa.embedding_model is not None, "Настоящая модель эмбеддингов не загружена"
    assert qa.build_index([document])
    assert qa.chunk_embeddings is not None, "Не построены настоящие эмбеддинги"
    question = "Как обеспечивается удаление загрязнённого воздуха из помещений?"
    results = qa.search(question, top_k=3)
    assert results and any(item.semantic_score > 0 for item in results)
    assert any("7.3.16" in item.text and "наиболее загрязнен" in item.text for item in results), "Не найдено прямое правило удаления воздуха"
    # Этот пример должен воздержаться от определения, даже когда LLM доступна.
    negative = qa.answer("Что такое рабочая зона по СП 60.13330? Найди определение.")
    assert negative["needs_clarification"] and not negative["used_llm"], negative["answer"]
    assert not negative["sources"], "Косвенные упоминания представлены как доказательство определения"
    definition = qa.answer("Что такое рециркуляция воздуха по СП 60.13330.2020?")
    assert definition["evidence_mode"] == "exact_excerpt", definition["answer"]
    assert "Смешение воздуха" in definition["answer"] and "[Источник 1]" in definition["answer"]
    norm = qa.answer("Какую температуру поверхности панелей радиационного обогрева рабочих мест допускает СП 60.13330.2020?")
    assert norm["evidence_mode"] == "exact_excerpt" and "60 °C" in norm["answer"], norm["answer"]
    # Вызов Ollama проверяется отдельно на обычном объяснении с прямой опорой.
    answer = qa.answer_from_results(question, results)
    assert answer["sources"] and not answer["needs_clarification"], answer["answer"]
    lower = answer["answer"].lower().replace("ё", "е")
    assert "зон" in lower and "загрязн" in lower and re.search(r"наиболее|наибольш|максималь", lower), answer["answer"]
    assert qa._has_valid_citations(answer["answer"], len(answer["sources"]))
    # Числа сами по себе не доказывают истинность, но новых числовых норм быть не должно.
    factual_answer = re.sub(r"\[Источник\s+\d+\]|(?m:^\s*\d+[.)]\s+)", "", answer["answer"])
    numbers = lambda value: {number.replace(",", ".") for number in re.findall(r"\d+(?:[.,]\d+)*", value)}
    assert numbers(factual_answer) <= numbers(answer["context"]), "Ответ добавил неподтверждённые числа"
    engine = FormulaEngine(qa)
    agent = AgentLoop(qa, engine)
    calculation = asyncio.run(agent.run("Рассчитай вентиляцию: L=100 м³/ч, t_в=20, t_н=-25"))
    assert not calculation.get("needs_clarification"), calculation["answer"]
    assert re.search(r"1\s?507[,.]5", calculation["answer"]), calculation["answer"]
    if not args.no_llm:
        assert answer["used_llm"], f"Модель не сформировала проверяемый ответ: {qa.last_llm_error}"
    report = {
        "document": args.document.name,
        "chunks": len(qa.chunks),
        "embedding_shape": list(qa.chunk_embeddings.shape),
        "retrieval": [{"source": item.doc_name, "score": item.score, "semantic_score": item.semantic_score, "excerpt": item.text[:350]} for item in results],
        "model": args.model,
        "used_llm": answer["used_llm"],
        "llm_error": qa.last_llm_error,
        "answer": answer["answer"],
        "sources": answer["sources"],
        "definition_rejection": negative,
        "supported_definition": definition,
        "supported_norm": norm,
        "content_checks_passed": True,
        "calculation": calculation["answer"],
        "elapsed_seconds": round(time.monotonic() - started, 2),
    }
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
    print(json.dumps(report, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()

# ИСПРАВЛЕНО: воспроизводимая проверка реального чтения, эмбеддингов, Ollama и async-калькулятора.