"""User-facing chat reply generator. This is the ONLY path through which LLM text reaches a user. Flow: user_message -> _detect_intent() — fixed response for greeting/smalltalk/thanks -> _call_ollama() — LLM for everything else -> guard_reply() — CJK + language + hallucination guard Usage: from chat.responder import generate_reply reply, meta = generate_reply("Привет", trace_id="ctrl-abc-chat") """ from __future__ import annotations import os import re from typing import Any from chat.guard import guard_reply, has_cjk # ── Intent routing ──────────────────────────────────────────────────────────── _PUNCT_RE = re.compile(r"[^\w\s]", re.UNICODE) _SPACE_RE = re.compile(r"\s+") # Exact normalized phrases per intent (after lower + strip punctuation) _INTENT_PATTERNS: dict[str, set[str]] = { "GREETING": { "привет", "здравствуйте", "здравствуй", "добрый день", "добрый вечер", "доброе утро", "hi", "hello", }, "SMALLTALK": { "как дела", "как ты", "что нового", "как поживаешь", "как жизнь", "как ты дела", }, "THANKS": { "спасибо", "благодарю", "спс", "благодарен", "благодарна", "большое спасибо", }, } _INTENT_RESPONSES: dict[str, str] = { "GREETING": "Привет! Чем могу помочь?", "SMALLTALK": "Всё хорошо 🙂 Чем могу помочь?", "THANKS": "Пожалуйста 🙂 Обращайтесь!", } def _normalize(text: str) -> str: """Lowercase + strip punctuation + collapse whitespace.""" return _PUNCT_RE.sub("", text.lower()).strip() def _detect_intent(text: str) -> str | None: """Return intent name if message matches a known pattern, else None.""" normalized = _normalize(text) for intent, patterns in _INTENT_PATTERNS.items(): if normalized in patterns: return intent return None # ── System prompt ───────────────────────────────────────────────────────────── _SYSTEM_PREFIX = ( "Ты — полезный ассистент. " "Отвечай ТОЛЬКО на русском языке. " "Никогда не используй другие языки. " "Давай краткие, точные ответы.\n\n" "ПРАВИЛА:\n" "1. На приветствие ('Привет', 'Здравствуйте' и т.п.) отвечай одной короткой фразой. " "Не упоминай никаких проектов, задач, данных или инструментов.\n" "2. Если пользователь не указал проект или задачу — не выдумывай их. " "Никогда не генерируй цели, датасеты, прогресс, инструменты (pandas, CSV и т.д.) " "или шаги, которые не были явно упомянуты.\n" "3. Если намерение неясно — задай короткий уточняющий вопрос вместо того, " "чтобы предполагать контекст.\n" "4. По умолчанию отвечай коротко. Подробности давай только если явно попросят." ) # Read model + generation settings at import time (env-driven) _CHAT_MODEL: str = ( os.environ.get("CHAT_MODEL") or os.environ.get("OLLAMA_MODEL", "qwen2.5:7b") ) _MAX_TOKENS: int = int(os.environ.get("CHAT_MAX_TOKENS", "150")) _TEMPERATURE: float = float(os.environ.get("CHAT_TEMPERATURE", "0.3")) def _call_ollama(prompt: str, trace_id: str | None = None) -> str: """Raw Ollama call — returns response text or empty string on error.""" from adapters import ollama_adapter result = ollama_adapter.execute( prompt=prompt, model=_CHAT_MODEL, temperature=_TEMPERATURE, max_tokens=_MAX_TOKENS, timeout=30.0, trace_id=trace_id, ) return (result.response or "").strip() def generate_reply( user_message: str, trace_id: str | None = None, ) -> tuple[str, dict[str, Any]]: """Generate a reply for the user. Fast path: known intents (greeting, smalltalk, thanks) return fixed responses without calling the LLM. Slow path: everything else goes through Ollama + full guard pipeline. Returns: (reply_text, meta) meta keys: intent, cjk_detected, fallback_used, lang_incident, model, max_tokens, temperature """ tid = trace_id or "chat-notrace" # ── Fast path: intent routing ───────────────────────────────────────────── intent = _detect_intent(user_message) if intent: print(f"[INTENT] {intent} trace_id={tid} message={user_message!r}") return _INTENT_RESPONSES[intent], { "intent": intent, "cjk_detected": False, "fallback_used": False, "lang_incident": False, "model": "intent_router", "max_tokens": 0, "temperature": 0.0, } # ── Slow path: LLM + guard ──────────────────────────────────────────────── prompt = f"{_SYSTEM_PREFIX}\n\nПользователь: {user_message}\nАссистент:" raw = _call_ollama(prompt, trace_id=tid) raw_has_cjk = has_cjk(raw) final, fallback_used, lang_incident = guard_reply( raw=raw, prompt=prompt, trace_id=tid, retry_fn=lambda p: _call_ollama(p, trace_id=f"{tid}-retry"), user_message=user_message, ) meta: dict[str, Any] = { "intent": None, "cjk_detected": raw_has_cjk, "fallback_used": fallback_used, "lang_incident": lang_incident, "model": _CHAT_MODEL, "max_tokens": _MAX_TOKENS, "temperature": _TEMPERATURE, } return final, meta