backend/services/script_analyzer.py - master - NoWay1co/AudioAnalytics

backend/services/script_analyzer.py
622 строки25 KB
Ivan
Добавлен новый роутер для управления скриптами, включая CRUD операции и генерацию LLM промптов. Обновлены основные маршруты приложения для интеграции нового функционала. Также добавлен сервис для анализа скриптов, адаптированный для работы с базой данных и асинхронным клиентом GigaChat.
07 фев 2026, 01:42
07 фев 2026, 01:425253246
"""
Script Analyzer Service for conversation quality assessment.

Adapted from ai_pipeline/script_analyzer.py for the backend.
Uses async GigaChatClient and loads scripts from PostgreSQL
(Script + ScriptStep models) instead of hardcoded defaults.

All methods are async. Results are compatible with the Analysis model.
"""

import json
import logging
from datetime import datetime
from typing import Any, Dict, List, Optional
from uuid import UUID

from sqlalchemy.orm import Session

from models import Script, ScriptStep
from services.gigachat_client import GigaChatClient, GigaChatAPIError

logger = logging.getLogger(__name__)


class ScriptAnalyzerError(Exception):
    """Raised when script analysis fails."""

    def __init__(self, message: str, cause: Optional[Exception] = None) -> None:
        self.message = message
        self.cause = cause
        super().__init__(self.message)


class ScriptAnalyzerService:
    """Async analyzer for checking transcriptions against manager scripts.

    Loads scripts from the database (Script + ScriptStep models) and uses
    GigaChatClient for LLM-based analysis.  All I/O is non-blocking.

    Args:
        gigachat_client: Async GigaChat client instance.
    """

    def __init__(self, gigachat_client: GigaChatClient) -> None:
        self.client = gigachat_client
        self._prompt_cache: Dict[UUID, str] = {}

        logger.info("ScriptAnalyzerService initialized")

    # ------------------------------------------------------------------
    # Prompt generation
    # ------------------------------------------------------------------

    async def generate_system_prompt(
        self,
        script: Script,
        db: Session,
        custom_instructions: str = "",
    ) -> str:
        """Generate an LLM system prompt from a script and its steps.

        The prompt is saved to ``script.generated_prompt`` in the database
        and cached in memory.

        Args:
            script: Script ORM instance with loaded ``steps`` relationship.
            db: Active SQLAlchemy session (used for commit).
            custom_instructions: Additional free-text instructions.

        Returns:
            Generated system prompt string.
        """
        logger.info("Generating system prompt for script %s ...", script.id)

        steps = sorted(script.steps, key=lambda s: s.order_index or 0)
        scripts_text = "\n".join(
            [
                (
                    f"{i + 1}. **{s.label}** (ID: {s.id})\n"
                    f"   - Описание: {s.description or 'Н/Д'}\n"
                    f"   - Обязательный: {'Да' if s.is_required else 'Нет'}\n"
                    f"   - Вес: {s.weight or 10}\n"
                    f"   - Ключевые слова: {', '.join(s.keywords or [])}"
                )
                for i, s in enumerate(steps)
            ]
        )

        requirements_section = ""
        if script.requirements_text:
            requirements_section = (
                f"\nДополнительные требования менеджера:\n{script.requirements_text}\n"
            )

        custom_section = ""
        if custom_instructions:
            custom_section = f"\nДополнительные инструкции: {custom_instructions}\n"

        meta_prompt = (
            "Ты - эксперт по созданию системных промптов для анализа качества "
            "обслуживания клиентов.\n\n"
            "Твоя задача: создать оптимальный системный промпт для LLM, которая будет "
            "анализировать транскрибации телефонных разговоров менеджеров салона красоты "
            "с клиентами.\n\n"
            "LLM должна проверять соблюдение следующих скриптов (правил общения):\n\n"
            f"{scripts_text}\n"
            f"{requirements_section}"
            f"{custom_section}\n"
            "Требования к системному промпту:\n"
            "1. Промпт должен быть на русском языке\n"
            "2. Должен четко объяснять задачу анализа\n"
            "3. Должен описывать формат выходных данных (JSON)\n"
            "4. Должен учитывать, что в одной транскрибации может быть несколько разговоров\n"
            "5. Должен объяснять критерии оценки каждого скрипта\n"
            "6. Должен быть структурированным и понятным\n\n"
            "Создай системный промпт. Выведи ТОЛЬКО текст системного промпта, "
            "без дополнительных комментариев."
        )

        try:
            raw = await self.client.chat(
                [{"role": "user", "content": meta_prompt}],
                temperature=0.3,
            )
            generated = raw.strip()
        except GigaChatAPIError as exc:
            raise ScriptAnalyzerError(
                message=f"Failed to generate system prompt: {exc.message}",
                cause=exc,
            ) from exc

        # Persist to DB
        script.generated_prompt = generated
        script.prompt_generated_at = datetime.utcnow()
        db.commit()
        db.refresh(script)

        # Cache
        self._prompt_cache[script.id] = generated

        logger.info(
            "System prompt generated for script %s (%d chars)",
            script.id,
            len(generated),
        )
        return generated

    def _get_cached_prompt(self, script: Script) -> str:
        """Return cached or persisted prompt; fall back to a basic template."""
        if script.id in self._prompt_cache:
            return self._prompt_cache[script.id]
        if script.generated_prompt:
            self._prompt_cache[script.id] = script.generated_prompt
            return script.generated_prompt
        return self._build_fallback_prompt(script)

    @staticmethod
    def _build_fallback_prompt(script: Script) -> str:
        """Build a minimal analysis prompt when no generated prompt exists."""
        steps = sorted(script.steps, key=lambda s: s.order_index or 0)
        step_lines = "\n".join(
            f"- {s.label}: {s.description or ''}" for s in steps
        )
        return (
            "Ты - AI-аналитик качества обслуживания клиентов в салонах красоты.\n"
            "Проанализируй транскрипт телефонного разговора между сотрудником и клиентом.\n"
            f"Скрипты для проверки:\n{step_lines}\n"
            "Верни результат СТРОГО в формате JSON."
        )

    # ------------------------------------------------------------------
    # Conversation splitting
    # ------------------------------------------------------------------

    async def split_conversations(
        self,
        transcription: str,
    ) -> List[Dict[str, Any]]:
        """Split a transcription into separate conversations using GigaChat.

        Args:
            transcription: Full transcription text.

        Returns:
            List of dicts, each with ``id``, ``text``, ``start_time``, ``end_time``.
        """
        logger.info("Splitting transcription into conversations ...")

        split_prompt = (
            "Проанализируй транскрибацию и раздели её на отдельные разговоры с клиентами.\n\n"
            "Признаки нового разговора:\n"
            "- Новое приветствие после прощания\n"
            "- Значительная пауза (если указаны временные метки)\n"
            "- Смена темы разговора на начальную (приветствие, выявление потребности)\n"
            "- Явное завершение предыдущего разговора (прощание)\n\n"
            "Верни JSON в формате:\n"
            "{\n"
            '    "conversations": [\n'
            "        {\n"
            '            "id": 1,\n'
            '            "start_marker": "текст начала разговора (первые 50 символов)",\n'
            '            "end_marker": "текст конца разговора (последние 50 символов)",\n'
            '            "start_time": null или число (секунды),\n'
            '            "end_time": null или число\n'
            "        }\n"
            "    ],\n"
            '    "total_conversations": число\n'
            "}\n\n"
            "Если в транскрибации только один разговор, верни массив с одним элементом.\n\n"
            "Транскрибация:\n"
        )

        try:
            response = await self.client.generate_json(
                [{"role": "user", "content": split_prompt + transcription}],
            )
        except GigaChatAPIError as exc:
            logger.warning("Conversation splitting failed, treating as single: %s", exc)
            return [{"id": 1, "text": transcription, "start_time": None, "end_time": None}]

        conversations: List[Dict[str, Any]] = []
        for conv in response.get("conversations", [{"id": 1}]):
            conv_id = conv.get("id", len(conversations) + 1)
            start_marker: str = conv.get("start_marker", "")
            end_marker: str = conv.get("end_marker", "")

            # Extract conversation text based on markers
            if start_marker and end_marker:
                start_idx = transcription.find(start_marker[:30])
                end_idx = transcription.rfind(end_marker[-30:])
                if start_idx >= 0 and end_idx >= 0:
                    conv_text = transcription[start_idx : end_idx + len(end_marker[-30:])]
                else:
                    conv_text = transcription
            else:
                conv_text = transcription

            conversations.append(
                {
                    "id": conv_id,
                    "text": conv_text,
                    "start_time": conv.get("start_time"),
                    "end_time": conv.get("end_time"),
                }
            )

        if not conversations:
            conversations = [
                {"id": 1, "text": transcription, "start_time": None, "end_time": None}
            ]

        logger.info("Found %d conversation(s)", len(conversations))
        return conversations

    # ------------------------------------------------------------------
    # Single-conversation analysis
    # ------------------------------------------------------------------

    async def analyze_conversation(
        self,
        conversation_text: str,
        script: Script,
        conversation_id: int = 1,
    ) -> Dict[str, Any]:
        """Analyze a single conversation against script steps.

        Returns a dict with all metrics required for the Analysis model:
        checklist, compliance_score, quality_score, sentiment_score,
        sentiment_class, risk_type, risk_severity, summary,
        recommendations, key_moments, topics.

        Args:
            conversation_text: Transcription text of one conversation.
            script: Script ORM instance with steps.
            conversation_id: Ordinal number of the conversation.

        Returns:
            Analysis result dict.
        """
        logger.info("Analyzing conversation %d against script %s ...", conversation_id, script.id)

        system_prompt = self._get_cached_prompt(script)

        steps = sorted(script.steps, key=lambda s: s.order_index or 0)
        steps_json = json.dumps(
            [
                {
                    "id": str(s.id),
                    "label": s.label,
                    "description": s.description or "",
                    "required": s.is_required,
                    "weight": s.weight or 10,
                    "keywords": s.keywords or [],
                }
                for s in steps
            ],
            ensure_ascii=False,
            indent=2,
        )

        analysis_prompt = (
            "Проанализируй разговор менеджера с клиентом.\n"
            "Оцени соблюдение скриптов И качество обслуживания в ОДНОМ ответе.\n\n"
            f"Скрипты:\n{steps_json}\n\n"
            "ВАЖНО: Верни ТОЛЬКО JSON без пояснений. Формат:\n"
            "{\n"
            '  "checklist": [\n'
            "    {\n"
            '      "script_step_id": "uuid",\n'
            '      "label": "название шага",\n'
            '      "status": "completed|partial|not_completed|not_applicable",\n'
            '      "confidence": 0.0-1.0,\n'
            '      "evidence": "цитата из разговора",\n'
            '      "notes": ""\n'
            "    }\n"
            "  ],\n"
            '  "compliance_score": 0-100,\n'
            '  "quality_score": 0-100,\n'
            '  "sentiment_score": от -1.0 до 1.0,\n'
            '  "sentiment_class": "positive|neutral|negative",\n'
            '  "summary": "краткое резюме разговора в 2-3 предложениях",\n'
            '  "key_moments": [\n'
            '    {"timestamp": "MM:SS", "label": "название момента", "note": "описание"}\n'
            "  ],\n"
            '  "risk_type": "conflict|complaint_threat|legal_threat|lost_client" или null,\n'
            '  "risk_severity": "low|medium|high|critical" или null,\n'
            '  "recommendations": "текст рекомендаций по улучшению",\n'
            '  "topics": ["тема1", "тема2"]\n'
            "}\n\n"
            "Статусы: completed=выполнен, partial=частично, "
            "not_completed=не выполнен, not_applicable=неприменим\n\n"
            "Разговор:\n"
            f"{conversation_text}\n\n"
            "Ответ (только JSON):"
        )

        try:
            response = await self.client.generate_json(
                [
                    {"role": "system", "content": system_prompt},
                    {"role": "user", "content": analysis_prompt},
                ],
                temperature=0.1,
            )
        except GigaChatAPIError as exc:
            raise ScriptAnalyzerError(
                message=(
                    f"LLM analysis failed for conversation {conversation_id}: {exc.message}"
                ),
                cause=exc,
            ) from exc

        result = self._validate_analysis_response(response, steps)
        result["conversation_id"] = conversation_id

        logger.info(
            "Conversation %d analyzed: compliance=%s, quality=%s",
            conversation_id,
            result.get("compliance_score"),
            result.get("quality_score"),
        )
        return result

    # ------------------------------------------------------------------
    # Full transcription analysis
    # ------------------------------------------------------------------

    async def analyze_transcription(
        self,
        transcription: str,
        script: Script,
        auto_split: bool = True,
    ) -> Dict[str, Any]:
        """Analyze a full transcription, optionally splitting conversations.

        Returns a merged result dict suitable for saving to the Analysis model.

        Args:
            transcription: Full transcription text.
            script: Script ORM instance.
            auto_split: Whether to split into separate conversations.

        Returns:
            Merged analysis result dict.
        """
        logger.info("Starting transcription analysis (auto_split=%s) ...", auto_split)

        if auto_split:
            conversations = await self.split_conversations(transcription)
        else:
            conversations = [
                {"id": 1, "text": transcription, "start_time": None, "end_time": None}
            ]

        results: List[Dict[str, Any]] = []
        for conv in conversations:
            try:
                analysis = await self.analyze_conversation(
                    conversation_text=conv["text"],
                    script=script,
                    conversation_id=conv["id"],
                )
                results.append(analysis)
            except ScriptAnalyzerError as exc:
                logger.error("Failed to analyze conversation %s: %s", conv["id"], exc.message)
                results.append(
                    {
                        "conversation_id": conv["id"],
                        "checklist": [],
                        "compliance_score": 0,
                        "quality_score": 0,
                        "sentiment_score": 0.0,
                        "sentiment_class": "neutral",
                        "summary": f"Ошибка анализа: {exc.message}",
                        "key_moments": [],
                        "risk_type": None,
                        "risk_severity": None,
                        "recommendations": "Повторить анализ вручную.",
                        "topics": [],
                    }
                )

        merged = self._merge_results(results)
        merged["conversation_count"] = len(conversations)

        logger.info(
            "Analysis complete: %d conversation(s), merged quality=%s",
            len(conversations),
            merged.get("quality_score"),
        )
        return merged

    # ------------------------------------------------------------------
    # Helpers
    # ------------------------------------------------------------------

    @staticmethod
    def _validate_analysis_response(
        data: Dict[str, Any],
        steps: List[ScriptStep],
    ) -> Dict[str, Any]:
        """Validate and normalise the LLM response into a consistent dict."""

        # -- checklist --
        raw_checklist = data.get("checklist", data.get("script_results", []))
        checklist: List[Dict[str, Any]] = []
        for item in raw_checklist if isinstance(raw_checklist, list) else []:
            checklist.append(
                {
                    "script_step_id": item.get("script_step_id", item.get("script_id", "")),
                    "label": item.get("label", item.get("script_name", "")),
                    "status": item.get("status", "not_completed"),
                    "confidence": float(item.get("confidence", 0.5)),
                    "evidence": item.get("evidence", ""),
                    "notes": item.get("notes", ""),
                }
            )

        # -- scores --
        compliance_score = data.get("compliance_score", data.get("overall_score", 0))
        if not isinstance(compliance_score, (int, float)):
            compliance_score = 0
        compliance_score = max(0, min(100, int(compliance_score)))

        quality_score = data.get("quality_score", 50)
        if not isinstance(quality_score, (int, float)):
            quality_score = 50
        quality_score = max(0, min(100, int(quality_score)))

        sentiment_score = data.get("sentiment_score", 0.0)
        if not isinstance(sentiment_score, (int, float)):
            sentiment_score = 0.0
        sentiment_score = max(-1.0, min(1.0, float(sentiment_score)))

        sentiment_class = data.get("sentiment_class", "neutral")
        if sentiment_class not in ("positive", "neutral", "negative"):
            if sentiment_score > 0.3:
                sentiment_class = "positive"
            elif sentiment_score < -0.3:
                sentiment_class = "negative"
            else:
                sentiment_class = "neutral"

        summary = data.get("summary", "")
        if not isinstance(summary, str):
            summary = str(summary)

        key_moments = data.get("key_moments", [])
        if not isinstance(key_moments, list):
            key_moments = []

        risk_type = data.get("risk_type")
        valid_risk_types = ("conflict", "complaint_threat", "legal_threat", "lost_client")
        if risk_type and risk_type not in valid_risk_types:
            risk_type = None

        risk_severity = data.get("risk_severity")
        valid_severities = ("low", "medium", "high", "critical")
        if risk_severity and risk_severity not in valid_severities:
            risk_severity = None

        recommendations = data.get("recommendations", "")
        if isinstance(recommendations, list):
            recommendations = "; ".join(str(r) for r in recommendations)
        elif not isinstance(recommendations, str):
            recommendations = str(recommendations) if recommendations else ""

        topics = data.get("topics", [])
        if not isinstance(topics, list):
            topics = []
        topics = [str(t) for t in topics if t]

        return {
            "checklist": checklist,
            "compliance_score": compliance_score,
            "quality_score": quality_score,
            "sentiment_score": sentiment_score,
            "sentiment_class": sentiment_class,
            "summary": summary,
            "key_moments": key_moments,
            "risk_type": risk_type,
            "risk_severity": risk_severity,
            "recommendations": recommendations,
            "topics": topics,
        }

    @staticmethod
    def _merge_results(results: List[Dict[str, Any]]) -> Dict[str, Any]:
        """Merge multiple conversation results into a single Analysis-compatible dict."""
        if not results:
            return {
                "checklist": [],
                "compliance_score": 0,
                "quality_score": 0,
                "sentiment_score": 0.0,
                "sentiment_class": "neutral",
                "summary": "",
                "key_moments": [],
                "risk_type": None,
                "risk_severity": None,
                "recommendations": "",
                "topics": [],
            }

        if len(results) == 1:
            return results[0]

        # Average numeric scores
        n = len(results)
        avg_compliance = sum(r.get("compliance_score", 0) for r in results) // n
        avg_quality = sum(r.get("quality_score", 0) for r in results) // n
        avg_sentiment = sum(r.get("sentiment_score", 0.0) for r in results) / n

        # Determine sentiment class from average
        if avg_sentiment > 0.3:
            sentiment_class = "positive"
        elif avg_sentiment < -0.3:
            sentiment_class = "negative"
        else:
            sentiment_class = "neutral"

        # Merge lists
        all_checklist: List[Dict[str, Any]] = []
        all_key_moments: List[Dict[str, Any]] = []
        all_topics: List[str] = []
        all_recommendations: List[str] = []
        summaries: List[str] = []

        # Pick worst risk
        severity_order = {"critical": 4, "high": 3, "medium": 2, "low": 1}
        worst_risk_type: Optional[str] = None
        worst_risk_severity: Optional[str] = None
        worst_severity_val = 0

        for r in results:
            conv_id = r.get("conversation_id", "?")
            for item in r.get("checklist", []):
                tagged = dict(item)
                tagged["conversation_id"] = conv_id
                all_checklist.append(tagged)

            all_key_moments.extend(r.get("key_moments", []))
            all_topics.extend(r.get("topics", []))

            rec = r.get("recommendations", "")
            if rec:
                all_recommendations.append(f"[Razgovor {conv_id}] {rec}")

            s = r.get("summary", "")
            if s:
                summaries.append(f"[Razgovor {conv_id}] {s}")

            rt = r.get("risk_type")
            rs = r.get("risk_severity")
            if rt and rs:
                sv = severity_order.get(rs, 0)
                if sv > worst_severity_val:
                    worst_severity_val = sv
                    worst_risk_type = rt
                    worst_risk_severity = rs

        # Deduplicate topics
        seen: set[str] = set()
        unique_topics: List[str] = []
        for t in all_topics:
            if t not in seen:
                seen.add(t)
                unique_topics.append(t)

        return {
            "checklist": all_checklist,
            "compliance_score": avg_compliance,
            "quality_score": avg_quality,
            "sentiment_score": round(avg_sentiment, 2),
            "sentiment_class": sentiment_class,
            "summary": "\n".join(summaries),
            "key_moments": all_key_moments,
            "risk_type": worst_risk_type,
            "risk_severity": worst_risk_severity,
            "recommendations": "\n".join(all_recommendations),
            "topics": unique_topics,
        }