From 0a7db5ebe908d703f1fea5e747e7c3ed414322b5 Mon Sep 17 00:00:00 2001 From: "reza.g" Date: Tue, 29 Sep 2026 02:10:03 +0330 Subject: [PATCH] feat(stt): support multi-model STT with parallel execution and consensus NLU extraction --- ai procurement/app/config.py | 70 +++++++++++++++++++++++-- ai procurement/app/handlers/__init__.py | 29 ++++++---- ai procurement/app/nlu.py | 33 ++++++++++-- ai procurement/app/stt.py | 70 +++++++++++++++++++------ 4 files changed, 171 insertions(+), 31 deletions(-) diff --git a/ai procurement/app/config.py b/ai procurement/app/config.py index 86f655f..7c6acd6 100644 --- a/ai procurement/app/config.py +++ b/ai procurement/app/config.py @@ -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" diff --git a/ai procurement/app/handlers/__init__.py b/ai procurement/app/handlers/__init__.py index 66a1d8b..20adfd7 100644 --- a/ai procurement/app/handlers/__init__.py +++ b/ai procurement/app/handlers/__init__.py @@ -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( diff --git a/ai procurement/app/nlu.py b/ai procurement/app/nlu.py index 44b7f9d..b5d1646 100644 --- a/ai procurement/app/nlu.py +++ b/ai procurement/app/nlu.py @@ -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") diff --git a/ai procurement/app/stt.py b/ai procurement/app/stt.py index 83d9ff8..e72a174 100644 --- a/ai procurement/app/stt.py +++ b/ai procurement/app/stt.py @@ -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 ""