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

backend/services/analyzer.py
329 строк12 KB
Ivan
Обновлен docker-compose.yml для монтирования локальной модели T-one и добавлен .dockerignore для исключения больших файлов и кэша. В api.py добавлена логика загрузки модели из локального пути или скачивания с HuggingFace. Модули whisper.py и analyzer.py помечены как устаревшие в связи с переходом на новые сервисы T-one и GigaChat.
07 фев 2026, 02:23
07 фев 2026, 02:2328713dc
"""
DEPRECATED: This module used gen-api.ru Gemini API.
Replaced by gigachat_client.py + script_analyzer.py as of Phase 4 integration.
Kept for backwards compatibility / fallback.

Original purpose: Analyzer service using gen-api.ru Gemini 3 Flash API.
"""

import httpx
import json
from typing import Optional, List, Dict, Any
from dataclasses import dataclass, field
from uuid import UUID

from config import settings


# System prompt for analysis (Russian)
SYSTEM_PROMPT = """Ты - AI-аналитик качества обслуживания клиентов в салонах красоты и продажах.

Твоя задача - проанализировать транскрипт телефонного разговора между сотрудником и клиентом.

КРИТЕРИИ ОЦЕНКИ КАЧЕСТВА (quality_score 0-100):
- Приветствие и представление (0-15 баллов)
- Выявление потребностей клиента (0-20 баллов)  
- Предложение услуг/товаров (0-20 баллов)
- Работа с возражениями (0-15 баллов)
- Закрытие / следующий шаг (0-15 баллов)
- Вежливость и профессионализм (0-15 баллов)

ТИПЫ РИСКОВ ДЛЯ ВЫЯВЛЕНИЯ:
- conflict: конфликт или явное недовольство клиента
- complaint_threat: угроза жалобы или негативного отзыва
- legal_threat: угроза обращения в суд или надзорные органы
- lost_client: потеря клиента (отказ от услуг, уход к конкуренту)

УРОВНИ СЕРЬЕЗНОСТИ РИСКА:
- low: незначительное недовольство
- medium: выраженное недовольство, требует внимания
- high: серьезный конфликт, возможна потеря клиента
- critical: угроза репутации, требует немедленной реакции

Верни результат СТРОГО в формате JSON."""


# User prompt template
USER_PROMPT_TEMPLATE = """Проанализируй следующий транскрипт разговора:

---
{transcript}
---

Верни JSON строго в следующем формате:
{{
  "quality_score": <число 0-100>,
  "sentiment_score": <число от -1.0 до 1.0>,
  "sentiment_class": "positive" | "neutral" | "negative",
  "summary": "<краткое резюме разговора в 2-3 предложениях>",
  "key_moments": [
    {{"timestamp": "MM:SS", "label": "<название момента>", "note": "<описание>"}}
  ],
  "risk_type": "<тип риска или null если нет>",
  "risk_severity": "low" | "medium" | "high" | "critical" | null,
  "topics": ["<тема1>", "<тема2>"],
  "recommendations": "<рекомендации по улучшению>"
}}"""


@dataclass
class AnalysisResult:
    """Result from LLM analysis."""
    quality_score: int  # 0-100
    sentiment_score: float  # -1.0 to 1.0
    sentiment_class: str  # positive, neutral, negative
    summary: str
    key_moments: List[Dict[str, Any]] = field(default_factory=list)
    risk_type: Optional[str] = None
    risk_severity: Optional[str] = None
    topics: List[str] = field(default_factory=list)
    recommendations: str = ""
    model_version: str = "gemini-3-flash"
    tokens_used: int = 0


class AnalyzerAPIError(Exception):
    """Error from Analyzer/LLM API."""
    
    def __init__(self, message: str, status_code: Optional[int] = None, response_body: Optional[str] = None):
        self.message = message
        self.status_code = status_code
        self.response_body = response_body
        super().__init__(self.message)


def _parse_analysis_response(response_text: str) -> Dict[str, Any]:
    """
    Parse LLM response text to extract JSON.
    Handles cases where response might be wrapped in markdown code blocks.
    """
    text = response_text.strip()
    
    # Remove markdown code blocks if present
    if text.startswith("```json"):
        text = text[7:]
    elif text.startswith("```"):
        text = text[3:]
    
    if text.endswith("```"):
        text = text[:-3]
    
    text = text.strip()
    
    try:
        return json.loads(text)
    except json.JSONDecodeError as e:
        raise AnalyzerAPIError(f"Failed to parse JSON response: {str(e)}", response_body=text[:500])


def _validate_analysis_data(data: Dict[str, Any]) -> AnalysisResult:
    """
    Validate and convert parsed JSON to AnalysisResult.
    Provides defaults for missing fields.
    """
    # Required fields with defaults
    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"):
        # Infer from score
        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")
    if risk_type and risk_type not in ("conflict", "complaint_threat", "legal_threat", "lost_client"):
        risk_type = None
    
    risk_severity = data.get("risk_severity")
    if risk_severity and risk_severity not in ("low", "medium", "high", "critical"):
        risk_severity = None
    
    topics = data.get("topics", [])
    if not isinstance(topics, list):
        topics = []
    topics = [str(t) for t in topics if t]
    
    recommendations = data.get("recommendations", "")
    if not isinstance(recommendations, str):
        recommendations = str(recommendations) if recommendations else ""
    
    return AnalysisResult(
        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,
        topics=topics,
        recommendations=recommendations,
        model_version="gemini-3-flash",
        tokens_used=0
    )


async def analyze_transcript(
    transcript_text: str,
    recording_id: UUID
) -> AnalysisResult:
    """
    Analyze conversation transcript using gen-api.ru Gemini 3 Flash API.
    
    Args:
        transcript_text: Full text of the transcript to analyze
        recording_id: Recording ID for logging
    
    Returns:
        AnalysisResult with quality score, sentiment, summary, and more
    
    Raises:
        AnalyzerAPIError: If API request fails or response is invalid
    """
    # Import logger here to avoid circular imports
    from services.logger import log_api_request, log_api_response, log_processing_error
    
    # Build endpoint - handle both base URL formats
    base_url = settings.GEN_API_URL.rstrip('/')
    if base_url.endswith('/networks'):
        endpoint = f"{base_url}/gemini-3-flash"
    else:
        endpoint = f"{base_url}/networks/gemini-3-flash"
    
    # Build user prompt with transcript
    user_prompt = USER_PROMPT_TEMPLATE.format(transcript=transcript_text)
    
    # Request payload (OpenAI-compatible format)
    payload = {
        "messages": [
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": user_prompt}
        ],
        "temperature": 0.3,  # Low temperature for consistent results
        "max_tokens": 2000,
        "response_format": {"type": "json_object"},
        "is_sync": True
    }
    
    headers = {
        "Authorization": f"Bearer {settings.GEN_API_KEY}",
        "Content-Type": "application/json"
    }
    
    log_api_request(str(recording_id), "Gemini API", endpoint)
    
    try:
        async with httpx.AsyncClient(timeout=settings.LLM_TIMEOUT) as client:
            response = await client.post(endpoint, json=payload, headers=headers)
            
            if response.status_code != 200:
                error_detail = response.text[:500] if response.text else "No response body"
                log_processing_error(
                    str(recording_id),
                    f"HTTP {response.status_code}: {error_detail}",
                    "AnalyzerAPIError"
                )
                raise AnalyzerAPIError(
                    message=f"Gemini API returned status {response.status_code}",
                    status_code=response.status_code,
                    response_body=error_detail
                )
            
            data = response.json()
            
            # Extract response content
            # Expected format: {"choices": [{"message": {"content": "..."}}]}
            # or direct response: {"content": "..."}
            response_text = ""
            tokens_used = 0
            
            if "choices" in data and data["choices"]:
                message = data["choices"][0].get("message", {})
                response_text = message.get("content", "")
            elif "output" in data:
                response_text = data["output"]
            elif "content" in data:
                response_text = data["content"]
            elif "text" in data:
                response_text = data["text"]
            else:
                # Try to use the whole response as JSON
                response_text = json.dumps(data)
            
            # Get token usage if available
            usage = data.get("usage", {})
            tokens_used = usage.get("total_tokens", 0)
            
            if not response_text:
                raise AnalyzerAPIError(
                    message="Empty response from Gemini API",
                    response_body=str(data)[:500]
                )
            
            # Parse and validate response
            parsed_data = _parse_analysis_response(response_text)
            result = _validate_analysis_data(parsed_data)
            result.tokens_used = tokens_used
            
            log_api_response(
                str(recording_id),
                "Gemini API",
                response.status_code,
                f"quality={result.quality_score}, sentiment={result.sentiment_class}, risk={result.risk_type}"
            )
            
            return result
            
    except httpx.TimeoutException:
        error_msg = f"Gemini API timeout after {settings.LLM_TIMEOUT}s"
        log_processing_error(str(recording_id), error_msg, "TimeoutError")
        raise AnalyzerAPIError(message=error_msg)
    
    except httpx.RequestError as e:
        error_msg = f"Network error: {str(e)}"
        log_processing_error(str(recording_id), error_msg, "RequestError")
        raise AnalyzerAPIError(message=error_msg)
    
    except Exception as e:
        if isinstance(e, AnalyzerAPIError):
            raise
        error_msg = f"Unexpected error: {type(e).__name__}: {str(e)}"
        log_processing_error(str(recording_id), error_msg, "UnexpectedError")
        raise AnalyzerAPIError(message=error_msg)


async def test_analyzer_connection() -> bool:
    """
    Test if Gemini API is accessible.
    
    Returns:
        True if API is reachable, False otherwise
    """
    try:
        async with httpx.AsyncClient(timeout=10) as client:
            response = await client.options(
                f"{settings.GEN_API_URL}/networks/gemini-3-flash",
                headers={"Authorization": f"Bearer {settings.GEN_API_KEY}"}
            )
            return response.status_code < 500
    except Exception:
        return False