Spaces:
Running
Running
| """Ручная интеграционная проверка настоящей модели на файле из базы (не 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-калькулятора. | |