feat(stt): support multi-model STT with parallel execution and consensus NLU extraction

2026-09-29 02:10:03 +03:30
parent da68fc683f
commit 0a7db5ebe9
4 changed files with 171 additions and 31 deletions
@@ -37,9 +37,30 @@ class Settings(BaseSettings):
    LLM_MODEL: str = "gpt-4o-mini"
    LLM_TIMEOUT_SECONDS: float = 60.0

-    # STT provider — any OpenAI-compatible /audio/transcriptions endpoint.
-    # Falls back to the LLM provider when unset. Note: some providers
-    # (e.g. DeepSeek) have no audio API — voice stays disabled gracefully.
+    # STT providers — support multiple models/APIs in parallel.
+    # 1. JSON list: [{"name": "m1", "base_url": "...", "api_key": "...", "model": "..."}, ...]
+    STT_PROVIDERS_JSON: str = ""
+
+    # 2. Individual provider variables (up to 3 distinct STT models)
+    STT_PROVIDER_1_NAME: str = "model-1"
+    STT_PROVIDER_1_BASE_URL: str = ""
+    STT_PROVIDER_1_API_KEY: str = ""
+    STT_PROVIDER_1_MODEL: str = ""
+    STT_PROVIDER_1_LANGUAGE: str = "fa"
+
+    STT_PROVIDER_2_NAME: str = "model-2"
+    STT_PROVIDER_2_BASE_URL: str = ""
+    STT_PROVIDER_2_API_KEY: str = ""
+    STT_PROVIDER_2_MODEL: str = ""
+    STT_PROVIDER_2_LANGUAGE: str = "fa"
+
+    STT_PROVIDER_3_NAME: str = "model-3"
+    STT_PROVIDER_3_BASE_URL: str = ""
+    STT_PROVIDER_3_API_KEY: str = ""
+    STT_PROVIDER_3_MODEL: str = ""
+    STT_PROVIDER_3_LANGUAGE: str = "fa"
+
+    # Legacy single STT provider fallback:
    STT_BASE_URL: str = ""
    STT_API_KEY: str = ""
    STT_MODEL: str = "whisper-1"
@@ -53,6 +74,49 @@ class Settings(BaseSettings):
    def stt_api_key(self) -> str:
        return self.STT_API_KEY or self.AI_API_KEY

+    @property
+    def stt_providers(self) -> list[dict]:
+        import json
+        if self.STT_PROVIDERS_JSON:
+            try:
+                data = json.loads(self.STT_PROVIDERS_JSON)
+                if isinstance(data, list) and len(data) > 0:
+                    return data
+            except Exception:
+                pass
+
+        # Check explicit STT_PROVIDER_1 / 2 / 3
+        custom_list: list[dict] = []
+        for idx in (1, 2, 3):
+            b_url = getattr(self, f"STT_PROVIDER_{idx}_BASE_URL")
+            key = getattr(self, f"STT_PROVIDER_{idx}_API_KEY")
+            model = getattr(self, f"STT_PROVIDER_{idx}_MODEL")
+            name = getattr(self, f"STT_PROVIDER_{idx}_NAME")
+            lang = getattr(self, f"STT_PROVIDER_{idx}_LANGUAGE") or "fa"
+            if b_url and key:
+                custom_list.append({
+                    "name": name or f"model-{idx}",
+                    "base_url": b_url,
+                    "api_key": key,
+                    "model": model or self.STT_MODEL,
+                    "language": lang,
+                    "timeout": 30.0,
+                })
+        if custom_list:
+            return custom_list
+
+        # Fallback to single legacy STT provider if set
+        if self.stt_base_url and self.stt_api_key:
+            return [{
+                "name": "default",
+                "base_url": self.stt_base_url,
+                "api_key": self.stt_api_key,
+                "model": self.STT_MODEL,
+                "language": self.STT_LANGUAGE,
+                "timeout": 30.0,
+            }]
+        return []
+
    # PostgreSQL — same instance/database as the api service.
    DATABASE_URL: str = "postgresql+asyncpg://postgres:postgres@localhost:5432/procurement"

@@ -14,8 +14,8 @@ from app import drafts as draft_store
 from app.config import settings
 from app.nlu import understand
 from app.runtime import BotContext, LinkedUser
-from app.stt import transcribe
-from app.texts import CANCELLED, LINK_START, NOTHING_TO_COMMIT, STT_FAILED, UNKNOWN_HINT, VOICE_EMPTY
+from app.stt import transcribe, transcribe_multi
+from app.texts import CANCELLED, LINK_START, NOTHING_TO_COMMIT, STT_FAILED, UNKNOWN_HINT, VOICE_EMPTY, esc

 from . import commands, callbacks, quote, supplier, material, queries, decisions

@@ -40,6 +40,7 @@ async def handle_update(ctx: BotContext, update: dict) -> None:
 async def handle_message(ctx: BotContext, chat_id: int, message: dict) -> None:
    text = (message.get("text") or "").strip()
    source = "text"
+    voice_transcripts: list[dict] | None = None

    if text.startswith("/"):
        await commands.handle_command(ctx, chat_id, text)
@@ -51,17 +52,20 @@ async def handle_message(ctx: BotContext, chat_id: int, message: dict) -> None:
        file_id = (message.get("voice") or message.get("audio") or message.get("document"))["file_id"]
        try:
            audio = await ctx.bale.download_file(file_id)
-            text = await transcribe(settings, audio,
-                                    mime_type=(message.get("voice") or {}).get("mime_type"),
-                                    filename=(message.get("voice") or {}).get("file_name"))
+            voice_transcripts = await transcribe_multi(
+                settings,
+                audio,
+                mime_type=(message.get("voice") or {}).get("mime_type"),
+                filename=(message.get("voice") or {}).get("file_name"),
+            )
        except Exception as exc:
            detail = "هوش مصنوعی تنظیم نشده" if "not configured" in str(exc) else str(exc)[:200]
            await ctx.send(chat_id, STT_FAILED.format(detail=esc(detail)))
            return
-        if not text:
+        if not voice_transcripts:
            await ctx.send(chat_id, VOICE_EMPTY)
            return
-        await ctx._log(chat_id, "in", "voice", text)
+        text = voice_transcripts[0]["text"]

    # Pending multi-step interactions take the message first.
    flow = ctx.pending.get(chat_id)
@@ -91,9 +95,16 @@ async def handle_message(ctx: BotContext, chat_id: int, message: dict) -> None:
            return

    draft_summary = _draft_summary(target_draft) if target_draft else None
-    result = await understand(settings, text, draft_summary)
+    input_payload = voice_transcripts if voice_transcripts else text
+    result = await understand(settings, input_payload, draft_summary)
    intent, patch = result.get("intent"), result.get("patch") or {}
-    logger.info("chat %s intent=%s conf=%s (is_edit=%s)", chat_id, intent, result.get("confidence"), target_draft is not None)
+    consensus_text = result.get("consensus_text") or text
+
+    if source == "voice":
+        await ctx._log(chat_id, "in", "voice", consensus_text)
+
+    logger.info("chat %s intent=%s conf=%s consensus=%r (is_edit=%s)",
+                chat_id, intent, result.get("confidence"), consensus_text[:40], target_draft is not None)

    if intent in ("register_quote", "update_draft"):
        await quote.handle_entities(
@@ -51,12 +51,19 @@ SYSTEM_PROMPT = """تو دستیار تدارکات یک ERP فارسی هستی
 یک JSON و فقط JSON برگردان (بدون توضیح اضافه) با این شکل:

 {
+  "consensus_text": "متن تجمیع‌شده و اصلاح‌شده کاربر (فارسی روان و بدون غلط شنیداری)",
  "intent": یکی از {intents},
  "patch": {شکل partial هر فعالیتی که کاربر داده},
  "confidence": 0.0..1.0
 }

 قوانین:
+- اگر ورودی شامل چند نسخه تبدیل گفتار به متن (STT) از یک پیام صوتی کاربر است:
+  ۱. این نسخه‌ها از چند مدل مختلف برای افزایش دقت به دست آمده‌اند. آن‌ها را با هم مقایسه و مقابله کن.
+  ۲. خطاهای شنیداری هر نسخه را بر مبنای منطق تدارکات (نام کالاها، برندها، مشخصات فنی و واحدهای اندازه‌گیری) تصحیح و حل تعارض کن.
+  ۳. متن نهایی اصلاح و تجمیع‌شده را در فیلد consensus_text قرار بده.
+  ۴. استخراج قصد (intent) و موجودیت‌ها (patch) را بر اساس همین متن دقیق انجام بده.
+- اگر ورودی یک متن عادی است، همان متن را در consensus_text قرار بده.
 - اگر پیام کاربر اطلاعات جدیدی به پیش‌نویسِ باز اضافه می‌کند یا اصلاح می‌کند → intent="update_draft" و patch شامل فقط فیلدهای جدید/تغییریافته.
 - اگر عبارت تأیید نهایی است (مثل «ثبت کن»، «همینه، ثبتش کن») → intent="commit_draft".
 - اگر لغو است («بی‌خیال»، «لغو») → intent="discard_draft".
@@ -113,12 +120,28 @@ def _extract_json(raw: str) -> dict | None:
        return None


-async def understand(settings: Settings, text: str, draft_summary: str | None = None) -> dict:
-    """Return {"intent": ..., "patch": {...}, "confidence": ...}; never raises."""
-    fallback = {"intent": "unknown", "patch": {}, "confidence": 0.0}
+async def understand(settings: Settings, text: str | list[dict], draft_summary: str | None = None) -> dict:
+    """Return {"consensus_text": ..., "intent": ..., "patch": {...}, "confidence": ...}; never raises."""
+    fallback = {"consensus_text": "", "intent": "unknown", "patch": {}, "confidence": 0.0}
    if not settings.AI_BASE_URL or not settings.AI_API_KEY:
        logger.warning("AI provider not configured; cannot understand message")
        return fallback
+
+    if isinstance(text, list):
+        first_text = text[0].get("text") if text else ""
+        lines = ["پیام صوتی کاربر توسط چند مدل تبدیل گفتار به متن (STT) به صورت زیر پیاده‌سازی شده است:"]
+        for idx, item in enumerate(text, start=1):
+            name = item.get("name") or f"مدل {idx}"
+            t = normalize_digits(item.get("text") or "")
+            lines.append(f"- نسخه {idx} ({name}): «{t}»")
+        lines.append("\nبا تلفیق این نسخه‌ها و تصحیح خطاهای شنیداری، پیام را تحلیل کن و consensus_text را به همراه intent و patch بساز.")
+        user_content = "\n".join(lines)
+    else:
+        user_content = normalize_digits(str(text))
+        first_text = str(text)
+
+    fallback["consensus_text"] = first_text
+
    try:
        async with httpx.AsyncClient(timeout=settings.LLM_TIMEOUT_SECONDS) as client:
            resp = await client.post(
@@ -128,7 +151,7 @@ async def understand(settings: Settings, text: str, draft_summary: str | None =
                    "model": settings.LLM_MODEL,
                    "messages": [
                        {"role": "system", "content": _build_system_prompt(draft_summary)},
-                        {"role": "user", "content": normalize_digits(text)},
+                        {"role": "user", "content": user_content},
                    ],
                    "temperature": 0.1,
                    "response_format": {"type": "json_object"},
@@ -143,6 +166,8 @@ async def understand(settings: Settings, text: str, draft_summary: str | None =
            logger.warning("unparseable LLM output: %r", content[:200])
            return fallback
        parsed.setdefault("patch", {})
+        if not parsed.get("consensus_text"):
+            parsed["consensus_text"] = first_text
        return parsed
    except (httpx.HTTPError, json.JSONDecodeError, KeyError, TypeError):
        logger.exception("NLU failure")
@@ -37,11 +37,45 @@ def _to_mp3(raw: bytes, suffix: str) -> bytes:
        return dst.read_bytes()


-async def transcribe(settings: Settings, audio: bytes, mime_type: str | None = None,
-                     filename: str | None = None) -> str:
-    """Return the Persian transcript for a voice note (empty string on silence)."""
-    if not settings.stt_base_url or not settings.stt_api_key:
-        raise RuntimeError("STT provider is not configured (PROCUREMENT_BOT_STT_BASE_URL / _AI_BASE_URL)")
+async def _call_single_stt(client: httpx.AsyncClient, provider: dict, converted_mp3: bytes) -> dict | None:
+    name = provider.get("name") or "stt"
+    base_url = (provider.get("base_url") or "").rstrip("/")
+    api_key = provider.get("api_key") or ""
+    model = provider.get("model") or "whisper-1"
+    language = provider.get("language") or "fa"
+    timeout = float(provider.get("timeout") or _STT_TIMEOUT)
+
+    url = f"{base_url}/audio/transcriptions"
+    try:
+        resp = await client.post(
+            url,
+            headers={"Authorization": f"Bearer {api_key}"},
+            files={"file": ("voice.mp3", converted_mp3, "audio/mpeg")},
+            data={"model": model, "language": language},
+            timeout=timeout,
+        )
+        if resp.status_code == 200:
+            txt = str((resp.json() or {}).get("text") or "").strip()
+            if txt:
+                logger.info("STT provider %s succeeded: %r", name, txt[:60])
+                return {"name": name, "text": txt, "model": model}
+        else:
+            logger.warning("STT provider %s returned HTTP %s: %s", name, resp.status_code, resp.text[:200])
+    except Exception as exc:
+        logger.warning("STT provider %s failed: %s", name, exc)
+    return None
+
+
+async def transcribe_multi(settings: Settings, audio: bytes, mime_type: str | None = None,
+                           filename: str | None = None) -> list[dict]:
+    """Transcribe audio with all configured STT providers in parallel.
+
+    Returns a list of candidate dicts: [{"name": str, "text": str, "model": str}, ...]
+    Empty list if all failed or silence.
+    """
+    providers = settings.stt_providers
+    if not providers:
+        raise RuntimeError("No STT providers are configured (PROCUREMENT_BOT_STT_*)")

    suffix = Path(filename or "").suffix
    if not suffix and mime_type:
@@ -49,14 +83,20 @@ async def transcribe(settings: Settings, audio: bytes, mime_type: str | None = N
                  "audio/mp4": ".m4a", "audio/wav": ".wav"}.get(mime_type.split(";")[0].strip(), ".ogg")
    converted = await asyncio.to_thread(_to_mp3, audio, suffix)

-    url = f"{settings.stt_base_url.rstrip('/')}/audio/transcriptions"
    async with httpx.AsyncClient(timeout=_STT_TIMEOUT) as client:
-        resp = await client.post(
-            url,
-            headers={"Authorization": f"Bearer {settings.stt_api_key}"},
-            files={"file": ("voice.mp3", converted, "audio/mpeg")},
-            data={"model": settings.STT_MODEL, "language": settings.STT_LANGUAGE},
-        )
-    if resp.status_code != 200:
-        raise RuntimeError(f"STT provider returned HTTP {resp.status_code}: {resp.text[:300]}")
-    return str((resp.json() or {}).get("text") or "").strip()
+        tasks = [_call_single_stt(client, p, converted) for p in providers]
+        results = await asyncio.gather(*tasks, return_exceptions=True)
+
+    valid_transcripts: list[dict] = []
+    for r in results:
+        if isinstance(r, dict) and r.get("text"):
+            valid_transcripts.append(r)
+
+    return valid_transcripts
+
+
+async def transcribe(settings: Settings, audio: bytes, mime_type: str | None = None,
+                     filename: str | None = None) -> str:
+    """Return the Persian transcript for a voice note (empty string on silence)."""
+    transcripts = await transcribe_multi(settings, audio, mime_type=mime_type, filename=filename)
+    return transcripts[0]["text"] if transcripts else ""