admin/proxy.py - main - amne/dca-agent

admin/proxy.py
379 строк14 KB
amne
feat: Phase 2 — pipeline_steps tracking + API
11 авг 2026, 14:52
11 авг 2026, 14:522b38c25
"""
Proxy module: calls Nexara STT and LLM on behalf of the agent.
All external API calls happen here — agent never touches them directly.
"""

import asyncio
import hashlib
import json
import logging
import tempfile
from typing import Optional

import aiohttp

import db

logger = logging.getLogger("dca-admin.proxy")

# Timeout: Nexara can take minutes for long audio
NEXARA_TIMEOUT = aiohttp.ClientTimeout(total=600)  # 10 min
LLM_TIMEOUT = aiohttp.ClientTimeout(total=300)  # 5 min

MAX_RETRIES = 2
RETRY_BACKOFF = 2  # seconds


async def _retry(fn, *args, max_retries=MAX_RETRIES, **kwargs):
    """Retry transient failures (5xx, network) with exponential backoff."""
    last_err = None
    for attempt in range(max_retries + 1):
        try:
            return await fn(*args, **kwargs)
        except Exception as e:
            last_err = e
            # Only retry on transient errors (network, 5xx), not 4xx
            err_str = str(e).lower()
            is_transient = any(s in err_str for s in [
                "timeout", "connection", "500", "502", "503", "504",
                "broken pipe", "reset"
            ])
            if not is_transient or attempt == max_retries:
                raise
            wait = RETRY_BACKOFF * (2 ** attempt)
            logger.warning(f"[proxy] retry {attempt+1}/{max_retries} after {wait}s: {e}")
            await asyncio.sleep(wait)
    raise last_err


async def call_nexara(session: aiohttp.ClientSession, nexara_url: str, nexara_key: str,
                      audio_data: bytes, filename: str) -> dict:
    """
    Send audio to Nexara for STT + diarization.
    Returns {"text": ..., "segments": [...], "duration": ..., "language": ...}
    """
    async def _do():
        data = aiohttp.FormData()
        data.add_field("file", audio_data, filename=filename, content_type="audio/mpeg")
        data.add_field("task", "diarize")
        data.add_field("diarization_setting", "telephonic")
        data.add_field("num_speakers", "2")
        data.add_field("response_format", "json")

        headers = {"Authorization": f"Bearer {nexara_key}"}

        async with session.post(nexara_url, data=data, headers=headers, timeout=NEXARA_TIMEOUT) as resp:
            body = await resp.text()
            if resp.status != 200:
                logger.error(f"[proxy] nexara error {resp.status}: {body[:300]}")
                raise RuntimeError(f"nexara error {resp.status}: {body[:200]}")
            return json.loads(body)

    result = await _retry(_do)
    logger.info(f"[proxy] nexara done: {len(result.get('segments', []))} segments, "
                f"{result.get('duration', 0):.1f}s")
    return result


async def call_llm(session: aiohttp.ClientSession, llm_url: str, llm_key: str, model: str,
                   temperature: float, system_prompt: str, user_message: str,
                   max_tokens: int = 20000) -> tuple:
    """
    Call LLM (OpenRouter/DeepSeek) with a chat completion request.
    Returns (content, prompt_tokens, completion_tokens).
    """
    async def _do():
        # Ensure URL ends with /chat/completions
        url = llm_url
        if not url.endswith("/chat/completions"):
            url = url.rstrip("/") + "/chat/completions"

        payload = {
            "model": model,
            "messages": [
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_message},
            ],
            "temperature": temperature,
            "max_tokens": max_tokens,
        }
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {llm_key}",
            "HTTP-Referer": "https://dca-admin.local",
            "X-Title": "DCA Admin Proxy",
        }

        async with session.post(url, json=payload, headers=headers, timeout=LLM_TIMEOUT) as resp:
            body = await resp.text()
            if resp.status != 200:
                logger.error(f"[proxy] LLM error {resp.status}: {body[:300]}")
                raise RuntimeError(f"LLM error {resp.status}: {body[:200]}")
            return json.loads(body)

    # Retry on empty content (DeepSeek sometimes returns null content under load)
    EMPTY_RETRIES = 3
    EMPTY_BACKOFFS = [3, 5, 8]

    data = None
    content = ""
    for empty_attempt in range(EMPTY_RETRIES):
        data = await _retry(_do)

        if not data.get("choices"):
            raise RuntimeError(f"empty LLM response: {json.dumps(data)[:300]}")

        content = data["choices"][0]["message"].get("content") or ""
        if content:
            break

        if empty_attempt < EMPTY_RETRIES - 1:
            wait = EMPTY_BACKOFFS[empty_attempt]
            logger.warning(f"[proxy] LLM returned empty content, retry {empty_attempt+1}/{EMPTY_RETRIES} after {wait}s")
            await asyncio.sleep(wait)
        else:
            logger.warning(f"[proxy] LLM returned empty content after {EMPTY_RETRIES} attempts, giving up")
            raise RuntimeError("LLM returned empty content")

    usage = data.get("usage", {})
    prompt_tokens = usage.get("prompt_tokens", 0)
    completion_tokens = usage.get("completion_tokens", 0)
    finish_reason = data["choices"][0].get("finish_reason", "")

    # DeepSeek v4: reasoning_tokens are part of completion_tokens
    details = usage.get("completion_tokens_details", {})
    reasoning_tokens = details.get("reasoning_tokens", 0)
    content_tokens = completion_tokens - reasoning_tokens

    logger.info(f"[proxy] LLM done: tokens={prompt_tokens}+{completion_tokens}"
                f"(reasoning={reasoning_tokens}, content={content_tokens}), "
                f"finish={finish_reason}, content_len={len(content)}")

    if finish_reason == "length":
        logger.warning(f"[proxy] LLM output truncated (max_tokens reached) "
                       f"— reasoning ate {reasoning_tokens}/{completion_tokens} tokens, "
                       f"only {content_tokens} left for content")

    return content, prompt_tokens, completion_tokens


def format_dialog(segments: list) -> str:
    """Format Nexara segments into labeled dialog for LLM input."""
    lines = []
    for seg in segments:
        speaker = seg.get("speaker", "")
        label = "Оператор" if speaker in ("speaker_1", "speaker1", "1", "A") else "Клиент"
        lines.append(f"[{label}]: {seg.get('text', '')}")
    return "\n".join(lines)


def validate_transcript(nexara_result: dict) -> tuple:
    """Returns (ok: bool, msg: str)."""
    segments = nexara_result.get("segments", [])
    if not segments:
        return False, "no segments"
    total_text = sum(len(s.get("text", "")) for s in segments)
    if total_text < 10:
        return False, "transcript too short"
    if nexara_result.get("duration", 0) < 1:
        return False, "duration too short"
    return True, "ok"


def validate_normalized(text: str) -> tuple:
    if not text or len(text.strip()) < 20:
        return False, "normalize too short"
    error_markers = ["я не могу", "ошибка", "не удалось", "unable to process"]
    low = text.lower().strip()
    for marker in error_markers:
        if low.startswith(marker):
            return False, f"LLM refuse: {marker}"
    return True, "ok"


def validate_report(text: str) -> tuple:
    if not text or len(text.strip()) < 50:
        return False, "report too short"
    return True, "ok"


async def process_audio(session: aiohttp.ClientSession, config: dict, audio_data: bytes, filename: str,
                        client_id: int = 0) -> dict:
    """
    Full pipeline: audio → Nexara → normalize → report.
    Each step has independent cache + validator.
    Returns the complete result for the agent.
    """
    # Helper to get LLM params with fallback
    def get_llm(config):
        url = config.get("llm_url", "")
        key = config.get("llm_key", "")
        model = config.get("llm_model", "deepseek-v4-pro")
        temp = config.get("llm_temperature", 0.2)
        return url, key, model, temp

    llm_url, llm_key, llm_model, llm_temp = get_llm(config)
    fb_url = config.get("fallback_llm_url", "")
    fb_key = config.get("fallback_llm_key", "")

    # --- Phase 2: track pipeline steps ---
    if client_id > 0:
        await db.init_pipeline_steps(filename, client_id)

    # --- Step 1: STT (with cache) ---
    cached = await db.get_transcript_cache(filename) if client_id else None
    if cached:
        logger.info(f"[step:stt] cache hit: {filename}")
        nexara_result = cached["transcript"]
        dialog = cached["dialog"]
        duration = cached["duration"]
        if client_id > 0:
            await db.mark_step_status(filename, 'stt', 'done')
    else:
        if client_id > 0:
            await db.mark_step_status(filename, 'stt', 'processing')
        try:
            nexara_result = await call_nexara(
                session, config["nexara_url"], config["nexara_key"], audio_data, filename
            )
            ok, msg = validate_transcript(nexara_result)
            if not ok:
                raise RuntimeError(f"transcript validation failed: {msg}")

            segments = nexara_result.get("segments", [])
            duration = nexara_result.get("duration", 0)
            dialog = format_dialog(segments)

            if not dialog.strip():
                raise RuntimeError("empty transcript — no speech detected")

            # Cache the transcript so retries don't pay for Nexara again
            if client_id:
                try:
                    await db.save_transcript_cache(client_id, filename, nexara_result, dialog, duration)
                    logger.info(f"[step:stt] cached transcript for {filename}")
                except Exception as e:
                    logger.warning(f"[step:stt] failed to cache transcript: {e}")

            if client_id > 0:
                await db.mark_step_status(filename, 'stt', 'done')
        except Exception as e:
            if client_id > 0:
                await db.mark_step_status(filename, 'stt', 'error', str(e))
            raise

    # --- Step 2: Normalize (with cache + validator) ---
    normalize_prompt = config.get("normalize_template", "")
    if not normalize_prompt:
        raise RuntimeError("normalize_template not configured")

    norm_hash = hashlib.md5(normalize_prompt.encode()).hexdigest()[:8]
    norm_in = norm_out = 0

    norm_cached = await db.get_normalize_cache(filename, norm_hash) if client_id else None
    if norm_cached:
        logger.info(f"[step:normalize] cache hit: {filename}")
        normalized = norm_cached["normalized"]
        norm_in = norm_cached["tokens_in"]
        norm_out = norm_cached["tokens_out"]
        if client_id > 0:
            await db.mark_step_status(filename, 'normalize', 'done')
    else:
        if client_id > 0:
            await db.mark_step_status(filename, 'normalize', 'processing')
        try:
            try:
                normalized, norm_in, norm_out = await call_llm(
                    session, llm_url, llm_key, llm_model, llm_temp,
                    normalize_prompt, dialog
                )
            except Exception as e:
                if not fb_url or not fb_key:
                    raise
                logger.warning(f"[step:normalize] primary LLM failed ({e}), trying fallback")
                normalized, norm_in, norm_out = await call_llm(
                    session, fb_url, fb_key, llm_model, llm_temp,
                    normalize_prompt, dialog
                )

            ok, msg = validate_normalized(normalized)
            if not ok:
                raise RuntimeError(f"normalize validation failed: {msg}")

            if client_id:
                try:
                    await db.save_normalize_cache(client_id, filename, norm_hash, normalized, norm_in, norm_out)
                except Exception as e:
                    logger.warning(f"[step:normalize] failed to cache: {e}")

            if client_id > 0:
                await db.mark_step_status(filename, 'normalize', 'done')
        except Exception as e:
            if client_id > 0:
                await db.mark_step_status(filename, 'normalize', 'error', str(e))
            raise

    # --- Step 3: Report (with cache + validator) ---
    report_prompt = config.get("report_template", "")
    if not report_prompt:
        raise RuntimeError("report_template not configured")

    rep_hash = hashlib.md5(report_prompt.encode()).hexdigest()[:8]
    rep_in = rep_out = 0
    report_duration = duration

    rep_cached = await db.get_report_cache(filename, rep_hash) if client_id else None
    if rep_cached:
        logger.info(f"[step:report] cache hit: {filename}")
        report = rep_cached["report"]
        rep_in = rep_cached["tokens_in"]
        rep_out = rep_cached["tokens_out"]
        if client_id > 0:
            await db.mark_step_status(filename, 'report', 'done')
    else:
        if client_id > 0:
            await db.mark_step_status(filename, 'report', 'processing')
        try:
            try:
                report, rep_in, rep_out = await call_llm(
                    session, llm_url, llm_key, llm_model, llm_temp,
                    report_prompt, normalized
                )
            except Exception as e:
                if not fb_url or not fb_key:
                    raise
                logger.warning(f"[step:report] primary LLM failed ({e}), trying fallback")
                report, rep_in, rep_out = await call_llm(
                    session, fb_url, fb_key, llm_model, llm_temp,
                    report_prompt, normalized
                )

            ok, msg = validate_report(report)
            if not ok:
                raise RuntimeError(f"report validation failed: {msg}")

            if client_id:
                try:
                    await db.save_report_cache(client_id, filename, rep_hash, report, rep_in, rep_out, report_duration)
                except Exception as e:
                    logger.warning(f"[step:report] failed to cache: {e}")

            if client_id > 0:
                await db.mark_step_status(filename, 'report', 'done')
        except Exception as e:
            if client_id > 0:
                await db.mark_step_status(filename, 'report', 'error', str(e))
            raise

    total_tokens_in = norm_in + rep_in
    total_tokens_out = norm_out + rep_out

    return {
        "transcript": nexara_result,
        "dialog": dialog,
        "normalized": normalized,
        "report": report,
        "duration": duration,
        "tokens_in": total_tokens_in,
        "tokens_out": total_tokens_out,
    }