fckbot/rag/functions/soma_evaluate.py

# rag/functions/soma_evaluate.py
"""
SOMA-анализ (LLM-as-judge) для оценки качества ответа.
Оценивает ответ по критериям: релевантность, полнота, отсутствие галлюцинаций,
стиль, полезность. Возвращает оценки и вердикт.

Добавлено кэширование результатов (in-memory с TTL) для экономии токенов.
Ключ кэша — хеш от (query, context, answer).
TTL по умолчанию 300 секунд (5 минут), настраивается через config.soma_cache_ttl.
"""

import json
import logging
import hashlib
import time
from typing import Dict, Any

from rag.services.giga_client import GigaClient
from rag.config_models import AppConfig

logger = logging.getLogger(__name__)

# In-memory кэш для SOMA-оценок
# Структура: {cache_key: (timestamp, result_dict)}
_soma_cache: Dict[str, tuple] = {}
# TTL по умолчанию (секунды)
_DEFAULT_TTL = 300


def _get_cache_key(query: str, context: str, answer: str) -> str:
    """
    Формирует ключ кэша на основе хеша от (query, context, answer).
    """
    # Ограничиваем длину контекста для хеширования, чтобы не перегружать память
    # Берём первые 2000 символов контекста, чтобы избежать слишком длинных строк
    context_hash = hashlib.md5(context[:2000].encode('utf-8')).hexdigest()
    answer_hash = hashlib.md5(answer[:2000].encode('utf-8')).hexdigest()
    query_hash = hashlib.md5(query.encode('utf-8')).hexdigest()
    return f"soma:{query_hash}:{context_hash}:{answer_hash}"


async def soma_evaluate(
    query: str,
    context: str,
    answer: str,
    config: AppConfig,
    giga: GigaClient,
    prompts: Dict[str, str],
) -> Dict[str, Any]:
    """
    Оценивает ответ LLM с помощью другого вызова GigaChat.

    Аргументы:
        query: исходный запрос пользователя.
        context: контекст из базы знаний (может быть пустым).
        answer: сгенерированный ответ.
        config: объект конфигурации.
        giga: клиент GigaChat для вызова.
        prompts: словарь с содержимым промптов (ключ 'soma_evaluate').

    Возвращает:
        dict с полями:
            scores: dict с оценками по критериям (числа от 1 до 5).
            overall_score: средняя оценка (float).
            verdict: 'pass' или 'fail' (в зависимости от порога).
            feedback: текстовое обоснование.
            error: строка ошибки (если есть).
    """
    default_result = {
        "scores": {
            "relevance": 3.0,
            "completeness": 3.0,
            "no_hallucination": 3.0,
            "style": 3.0,
            "usefulness": 3.0,
        },
        "overall_score": 3.0,
        "verdict": "fail",
        "feedback": "Оценка не удалась, возвращены значения по умолчанию.",
        "error": None,
    }

    # ---- Проверка кэша ----
    ttl = getattr(config, 'soma_cache_ttl', _DEFAULT_TTL)
    cache_key = _get_cache_key(query, context, answer)

    if cache_key in _soma_cache:
        timestamp, cached_result = _soma_cache[cache_key]
        if time.time() - timestamp < ttl:
            logger.debug(f"SOMA-оценка взята из кэша (TTL {ttl} сек)")
            return cached_result
        else:
            # Истекло — удаляем
            del _soma_cache[cache_key]
            logger.debug("SOMA-кэш истёк")

    # ---- Проверка наличия промпта ----
    prompt_template = prompts.get("soma_evaluate")
    if not prompt_template:
        logger.error("Промпт для SOMA-оценки не найден")
        default_result["error"] = "Отсутствует промпт soma_evaluate"
        return default_result

    # ---- Формирование промпта ----
    full_prompt = prompt_template.format(
        query=query,
        context=context[:3000] if context else "Нет контекста",
        answer=answer,
    )

    temperature = getattr(config, 'soma_temperature', 0.1)
    timeout = getattr(config, 'soma_timeout', 30)

    try:
        response = await giga.chat(
            history=[],
            query=full_prompt,
            system_prompt=None,
            file_id=None,
            temperature=temperature,
            timeout=timeout,
        )
        data = json.loads(response.strip())
        scores = data.get("scores", {})
        feedback = data.get("feedback", "")

        for key in scores:
            try:
                scores[key] = float(scores[key])
            except (ValueError, TypeError):
                scores[key] = 3.0
        overall = sum(scores.values()) / len(scores) if scores else 3.0

        threshold = getattr(config, 'soma_threshold', 3.5)
        verdict = "pass" if overall >= threshold else "fail"

        result = {
            "scores": scores,
            "overall_score": overall,
            "verdict": verdict,
            "feedback": feedback,
            "error": None,
        }

        # ---- Сохранение в кэш ----
        _soma_cache[cache_key] = (time.time(), result)
        logger.debug(f"SOMA-оценка сохранена в кэш (всего {len(_soma_cache)} записей)")

        # Ограничим размер кэша, чтобы не переполнить память
        if len(_soma_cache) > 1000:
            # Удаляем самые старые записи (по времени)
            sorted_keys = sorted(_soma_cache.keys(), key=lambda k: _soma_cache[k][0])
            for old_key in sorted_keys[:200]:  # удаляем 200 самых старых
                del _soma_cache[old_key]
            logger.debug("Кэш SOMA очищен (старые записи удалены)")

        return result

    except json.JSONDecodeError as e:
        logger.error(f"Ошибка парсинга JSON от GigaChat: {e}, ответ: {response[:200]}")
        default_result["error"] = f"Ошибка парсинга JSON: {e}"
        return default_result
    except Exception as e:
        logger.exception(f"Ошибка при вызове SOMA-оценки: {e}")
        default_result["error"] = str(e)
        return default_result