gendesign/backend/app/services/chat/tools.py

"""Read-only «секционные» tool'ы чата (#957, Step 2) — срезы УЖЕ-загруженного отчёта.

LLM в tool-loop'е (см. orchestrator.py) просит секции отчёта через function-calling.
Здесь — две вещи и НИЧЕГО больше:

  1. OpenAI tool-спеки (JSON-schema) пяти read-only секционных tool'ов
     (`get_exec_summary` / `get_product_recommendation` / `get_forecast` /
     `get_risks` / `get_scenarios`). Без параметров — каждый отдаёт фиксированную
     секцию(и) отчёта (модель не управляет вычислениями, только запрашивает данные).
  2. ЧИСТЫЕ executors: режут УЖЕ-ЗАГРУЖЕННЫЙ in-memory `report_dict`
     (`SiteFinderReport.as_dict()`, 8 секций). НИКАКОЙ БД, НИКАКОГО пере-расчёта,
     НИКАКОЙ движковой математики — только срез готового dict'а.

КРИТИЧНО (§16 grounding): числа берутся ВЕРБАТИМ из отчёта. Tool'ы не считают и не
выдумывают — они достают под-dict секции как есть. Модель потом только оборачивает
эти числа в RU-прозу (compose), не изобретая своих.

ROBUST: отчёт может быть тонким/частичным (каждая секция Optional). Executors НИКОГДА
не бросают KeyError — отсутствует секция → `{}` или маленький маркер «недоступно».
"""

from __future__ import annotations

from collections.abc import Callable
from typing import Any

# Маркер «секции нет в отчёте» (graceful, не KeyError). Модель по нему скажет, что
# данных нет, а не выдумает их (см. system-prompt в prompts.py).
_NOT_AVAILABLE: dict[str, Any] = {"available": False}


def _section(report: dict[str, Any], key: str) -> dict[str, Any]:
    """Достать секцию-dict по ключу или вернуть маркер «недоступно». PURE, без KeyError."""
    value = report.get(key)
    if isinstance(value, dict) and value:
        return value
    return dict(_NOT_AVAILABLE)


# ── Executors (PURE срезы готового report_dict; НЕТ БД/recompute/engine-math) ────


def get_exec_summary(report: dict[str, Any]) -> dict[str, Any]:
    """§13.1 exec_summary — вердикт-заголовок + ключевые числа + общая уверенность."""
    return _section(report, "exec_summary")


def get_product_recommendation(report: dict[str, Any]) -> dict[str, Any]:
    """§13.4 product_tz — рекомендация продукта (класс / квартирография / USP / §16)."""
    return _section(report, "product_tz")


def get_forecast(report: dict[str, Any]) -> dict[str, Any]:
    """§13.3 future_market — прогноз спроса/предложения по горизонтам + future-supply."""
    return _section(report, "future_market")


def get_risks(report: dict[str, Any]) -> dict[str, Any]:
    """Риски — §13.6 scoring (спец-индексы §25) + §13.7 confidence (уровень/факторы).

    Объединяет ДВЕ секции в один срез: риски в отчёте размазаны по scoring
    (special_indices, вкл. каннибализацию) и confidence (уровень + тянущие факторы).
    Каждая под-секция graceful: отсутствует → маркер «недоступно», без KeyError.
    """
    return {
        "scoring": _section(report, "scoring"),
        "confidence": _section(report, "confidence"),
    }


def get_scenarios(report: dict[str, Any]) -> dict[str, Any]:
    """§13.5 scenarios — сводка conservative/base/aggressive + summary."""
    return _section(report, "scenarios")


# ── Реестр имя→executor + имя→секции отчёта (для provenance grounded_in.sections) ─
# ЕДИНЫЙ источник истины: и спеки, и orchestrator берут имена/маппинг отсюда.

# Имя tool'а → (executor, какие секции отчёта он реально трогает). sections нужны для
# честного grounded_in.sections: что СПРАШИВАЛ LLM = на чём заземлён ответ.
_TOOLS: dict[str, tuple[Callable[[dict[str, Any]], dict[str, Any]], tuple[str, ...]]] = {
    "get_exec_summary": (get_exec_summary, ("exec_summary",)),
    "get_product_recommendation": (get_product_recommendation, ("product_tz",)),
    "get_forecast": (get_forecast, ("future_market",)),
    "get_risks": (get_risks, ("scoring", "confidence")),
    "get_scenarios": (get_scenarios, ("scenarios",)),
}

# RU-описания tool'ов для модели (что внутри секции — чтобы LLM выбирал верный tool).
_TOOL_DESCRIPTIONS: dict[str, str] = {
    "get_exec_summary": (
        "Краткое резюме отчёта: вердикт «что строить», ключевые числа, общая уверенность."
    ),
    "get_product_recommendation": (
        "Рекомендация продукта: класс объекта, квартирография (mix), коммерция, "
        "USP и §16-обоснования."
    ),
    "get_forecast": (
        "Прогноз будущего рынка: спрос/предложение по горизонтам, дефицит/затоварка, "
        "давление будущего предложения."
    ),
    "get_risks": (
        "Риски участка: специальные индексы §25 (включая каннибализацию портфеля) "
        "и уровень/факторы уверенности отчёта."
    ),
    "get_scenarios": (
        "Сценарии развития: разброс по консервативному / базовому / агрессивному."
    ),
}


def _spec(name: str) -> dict[str, Any]:
    """Собрать одну OpenAI function-tool спеку (без параметров — секция фиксирована)."""
    return {
        "type": "function",
        "function": {
            "name": name,
            "description": _TOOL_DESCRIPTIONS[name],
            # Без аргументов: tool отдаёт фиксированную секцию, модель не управляет
            # вычислениями. additionalProperties=False — строгая пустая схема.
            "parameters": {
                "type": "object",
                "properties": {},
                "additionalProperties": False,
            },
        },
    }


def tool_specs() -> list[dict[str, Any]]:
    """Все 5 секционных tool-спек (JSON-schema) для передачи в ``complete(tools=...)``."""
    return [_spec(name) for name in _TOOLS]


def is_known_tool(name: str) -> bool:
    """Известен ли tool по имени (защита от галлюцинации имени моделью)."""
    return name in _TOOLS


def execute_tool(name: str, report: dict[str, Any]) -> dict[str, Any]:
    """Выполнить секционный executor по имени против УЖЕ-загруженного report_dict.

    PURE-срез: НИКАКОЙ БД/recompute/engine-math. Неизвестное имя → маркер «недоступно»
    (модель не уронит loop, если выдумает имя tool'а). KeyError невозможен (executors
    graceful).
    """
    entry = _TOOLS.get(name)
    if entry is None:
        return dict(_NOT_AVAILABLE)
    executor, _ = entry
    return executor(report)


def sections_for_tool(name: str) -> tuple[str, ...]:
    """Какие секции отчёта трогает tool (для grounded_in.sections). Неизвестный → ()."""
    entry = _TOOLS.get(name)
    return entry[1] if entry is not None else ()


__all__ = [
    "execute_tool",
    "get_exec_summary",
    "get_forecast",
    "get_product_recommendation",
    "get_risks",
    "get_scenarios",
    "is_known_tool",
    "sections_for_tool",
    "tool_specs",
]