/
artyuhovs
/
AIFeedBackTrainingBot
Обзор
Документация
Войти
/
artyuhovs
/
AIFeedBackTrainingBot
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
master
app/models/framework_quality.py
328 строк
12 KB
Codex
release: prepare public GitVerse release 0.9.3
27 июл 2026, 08:52
27 июл 2026, 08:52
1195d6b
Код
Авторство
О чём код?
from __future__ import annotations import json import math from decimal import ROUND_HALF_UP, Decimal from typing import Any, Literal, cast from pydantic import BaseModel, ConfigDict, ValidationError, ValidationInfo, field_validator, model_validator FRAMEWORK_QUALITY_PROMPT_VERSION = "feedback_framework_quality_v7" FRAMEWORK_ELEMENT_WEIGHTS: dict[str, dict[str, int | float]] = { "COIN": { "context": 2, "observation_factual": 3, "impact": 2.5, "next_steps": 2.5, }, "BOFF": { "behavior_factual": 2.5, "outcome": 2, "feelings": 3, "future": 2.5, }, "Я-сообщения": { "fact_reason": 2, "own_feelings": 3, "fact_feeling_causal_link": 3, "request_or_boundary": 2, }, "Бутерброд": { "effective_fact": 2, "factual_praise": 2, "alternative_from_my_place": 2, "tact_no_intrusion": 2, "non_directive": 2, }, } SUPPORTED_FRAMEWORKS = tuple(FRAMEWORK_ELEMENT_WEIGHTS) FrameworkName = Literal["COIN", "BOFF", "Я-сообщения", "Бутерброд"] FrameworkElementId = Literal[ "context", "observation_factual", "impact", "next_steps", "behavior_factual", "outcome", "feelings", "future", "fact_reason", "own_feelings", "fact_feeling_causal_link", "request_or_boundary", "effective_fact", "factual_praise", "alternative_from_my_place", "tact_no_intrusion", "non_directive", ] ElementStatus = Literal["met", "partially_met", "not_met", "unknown", "not_applicable"] AnalysisStatus = Literal["evaluated", "no_feedback_detected"] FrameworkFit = Literal["does_not_fit", "weak_fit", "partial_fit", "good_fit", "strong_fit"] class FrameworkElement(BaseModel): model_config = ConfigDict(extra="forbid", strict=True) id: FrameworkElementId status: ElementStatus score: float | None evidence: str comment: str @field_validator("score", mode="before") @classmethod def validate_score_value(cls, value: object) -> object: if isinstance(value, bool) or (value is not None and value not in {0, 0.5, 1}): raise ValueError("score must be 0, 0.5, 1, or null") return value @model_validator(mode="after") def validate_status_score(self) -> FrameworkElement: expected_scores: dict[str, float | None] = { "met": 1.0, "partially_met": 0.5, "not_met": 0.0, "unknown": None, "not_applicable": None, } if self.score != expected_scores[self.status]: raise ValueError(f"status {self.status!r} requires score {expected_scores[self.status]!r}") return self class FrameworkSequenceCheck(BaseModel): model_config = ConfigDict(extra="forbid", strict=True) expected_logic: list[str] observed_logic: list[str] is_logic_ok: bool comment: str class FrameworkQualityModelResponse(BaseModel): model_config = ConfigDict(extra="forbid", strict=True) prompt_version: Literal["feedback_framework_quality_v7"] framework_name: FrameworkName status: AnalysisStatus verdict: str framework_elements: list[FrameworkElement] sequence_check: FrameworkSequenceCheck main_framework_gaps: list[str] what_is_missing_for_10: list[str] even_if_10_can_improve: list[str] recommendations: list[str] rewritten_by_framework: str repeat_practice_task: str missing_data: list[str] transcription_risks: list[str] @model_validator(mode="after") def validate_framework_and_elements(self, info: ValidationInfo) -> FrameworkQualityModelResponse: expected_framework = (info.context or {}).get("expected_framework_name") if expected_framework is not None and self.framework_name != expected_framework: raise ValueError(f"framework_name must be {expected_framework!r}") expected_ids = list(FRAMEWORK_ELEMENT_WEIGHTS[self.framework_name]) actual_ids = [element.id for element in self.framework_elements] if actual_ids != expected_ids: raise ValueError(f"framework_elements ids must be exactly {expected_ids!r} in this order") return self def validate_framework_quality_response( payload: dict[str, Any], framework_name: str, ) -> FrameworkQualityModelResponse: return FrameworkQualityModelResponse.model_validate( payload, context={"expected_framework_name": framework_name}, ) def framework_quality_validation_errors(error: ValidationError) -> list[str]: messages: list[str] = [] for item in error.errors(include_url=False): location = ".".join(str(part) for part in item["loc"]) or "$" messages.append(f"{location}: {item['msg']}") return messages def normalize_framework_quality_response( response: FrameworkQualityModelResponse, ) -> dict[str, Any]: return _normalize_framework_quality_payload(response.model_dump()) def salvage_framework_quality_response( payload: dict[str, Any], framework_name: str, ) -> dict[str, Any]: weights = FRAMEWORK_ELEMENT_WEIGHTS[framework_name] elements_by_id: dict[str, list[object]] = {} raw_elements = payload.get("framework_elements") if isinstance(raw_elements, list): for item in raw_elements: if isinstance(item, dict) and isinstance(item.get("id"), str): elements_by_id.setdefault(item["id"], []).append(item) elements = [_salvage_element(element_id, elements_by_id.get(element_id, [])) for element_id in weights] salvaged = { "prompt_version": FRAMEWORK_QUALITY_PROMPT_VERSION, "framework_name": framework_name, "status": ("no_feedback_detected" if payload.get("status") == "no_feedback_detected" else "evaluated"), "verdict": _string_or_default( payload.get("verdict"), "Часть ответа модели не прошла проверку; оценка рассчитана только по доступным элементам.", ), "framework_elements": elements, "sequence_check": _salvage_sequence_check(payload.get("sequence_check"), list(weights)), "main_framework_gaps": _string_list(payload.get("main_framework_gaps")), "what_is_missing_for_10": _string_list(payload.get("what_is_missing_for_10")), "even_if_10_can_improve": _string_list(payload.get("even_if_10_can_improve")), "recommendations": _string_list(payload.get("recommendations")), "rewritten_by_framework": _string_or_default(payload.get("rewritten_by_framework"), ""), "repeat_practice_task": _string_or_default(payload.get("repeat_practice_task"), ""), "missing_data": _string_list(payload.get("missing_data")), "transcription_risks": _string_list(payload.get("transcription_risks")), } return _normalize_framework_quality_payload(salvaged) def framework_elements_json(framework_name: str) -> str: elements = [ { "id": element_id, "status": "met", "score": 1, "evidence": "короткая цитата или точный пересказ", "comment": "почему выбран этот статус и score", } for element_id in FRAMEWORK_ELEMENT_WEIGHTS[framework_name] ] return json.dumps(elements, ensure_ascii=False, indent=2) def _salvage_element(element_id: str, candidates: list[object]) -> dict[str, Any]: if len(candidates) == 1: try: parsed = FrameworkElement.model_validate(candidates[0]) except ValidationError: pass else: return parsed.model_dump() return { "id": element_id, "status": "unknown", "score": None, "evidence": "Недостаточно надёжных данных модели для оценки этого элемента.", "comment": "Элемент исключён из расчёта итогового балла.", } def _salvage_sequence_check(value: object, expected_logic: list[str]) -> dict[str, Any]: try: parsed = FrameworkSequenceCheck.model_validate(value) except ValidationError: return { "expected_logic": expected_logic, "observed_logic": [], "is_logic_ok": False, "comment": "Модель не вернула надёжную проверку последовательности.", } return parsed.model_dump() def _string_list(value: object) -> list[str]: if not isinstance(value, list): return [] return [item for item in value if isinstance(item, str)] def _string_or_default(value: object, default: str) -> str: return value if isinstance(value, str) else default def _normalize_framework_quality_payload(payload: dict[str, Any]) -> dict[str, Any]: framework_name = str(payload["framework_name"]) weights = FRAMEWORK_ELEMENT_WEIGHTS[framework_name] elements: list[dict[str, Any]] = [] available_points = Decimal("0") available_weight = Decimal("0") unavailable_elements: list[str] = [] for raw_element in cast(list[dict[str, Any]], payload["framework_elements"]): element_id = str(raw_element["id"]) weight = Decimal(str(weights[element_id])) score = raw_element.get("score") points: float | int | None = None if isinstance(score, int | float) and not isinstance(score, bool): points_decimal = Decimal(str(score)) * weight points = _plain_number(points_decimal) available_points += points_decimal available_weight += weight else: unavailable_elements.append(element_id) elements.append( { **raw_element, "weight": _plain_number(weight), "points": points, } ) no_feedback = payload.get("status") == "no_feedback_detected" if no_feedback: block_score: float | int = 0 framework_score: float | int | None = 0 poll_answer: int | None = 0 elif available_weight == 0: block_score = 0 framework_score = None poll_answer = None else: block_score = _plain_number(available_points.quantize(Decimal("0.1"), rounding=ROUND_HALF_UP)) normalized = (available_points / available_weight * Decimal("10")).quantize( Decimal("0.1"), rounding=ROUND_HALF_UP, ) framework_score = _plain_number(normalized) poll_answer = min(10, max(1, math.ceil(float(normalized)))) normalized_payload = dict(payload) normalized_payload.update( { "prompt_version": FRAMEWORK_QUALITY_PROMPT_VERSION, "framework_score_10": framework_score, "framework_block_score": block_score, "framework_poll_answer": poll_answer, "framework_fit": _framework_fit(framework_score), "framework_elements": elements, "evaluated_elements_count": len(elements) - len(unavailable_elements), "total_elements_count": len(elements), "unavailable_elements": unavailable_elements, } ) return normalized_payload def _framework_fit(score: float | int | None) -> FrameworkFit | None: if score is None: return None if score < 4: return "does_not_fit" if score < 6: return "weak_fit" if score < 8: return "partial_fit" if score < 9: return "good_fit" return "strong_fit" def _plain_number(value: Decimal) -> float | int: integral = value.to_integral_value() if value == integral: return int(integral) return float(value)