import re from typing import Optional from app.config import settings from app.data.knowledge_base import ( DEFAULT_PROFILE, GLOSSARY_ANSWERS, IT_SYSTEMS, KNOWLEDGE_ENTRIES, get_colleague_cases_for_system, ) from app.llm.client import get_llm_client from app.models import AssistantResult, ColleagueCase, ResponsePackage, UserProfile from app.package_builder import build_why_suggested, ensure_escalation_draft, enrich_ticket_draft from app.rag.service import ( build_package_from_hits, build_package_with_llm, search_documents, ) def _detect_glossary(query: str) -> Optional[str]: match = re.search(r"что такое\s+([а-яёa-z0-9]+)", query, re.IGNORECASE) if not match: return None return GLOSSARY_ANSWERS.get(match.group(1).lower()) def _detect_system(query: str) -> Optional[str]: lower = query.lower() if re.search(r"meridium|меридиум", lower): return "Meridium 4" if re.search(r"есуид|esuid|esu[iі]d", lower): return "ЕСУИД" if re.search(r"1[сc]|1c", lower): return "1С" if re.search(r"starlims|старлимс", lower): return "StarLims" if re.search(r"лимс|lims", lower): return "ЛИМС" if re.search(r"вкус|vkus", lower): return "ВКУС" return None def _is_thanks(query: str) -> bool: return bool( re.match( r"^(спасибо|благодар|всё получилось|все получилось|понятно|достаточно|до свидания|нет,? спасибо)", query.strip(), re.IGNORECASE, ) ) def _is_off_topic(query: str) -> bool: return bool(re.search(r"политик|выборы|погод|рецепт|футбол|кино|личн", query, re.IGNORECASE)) def _is_confidential_request(query: str) -> bool: return bool( re.search(r"зарплат|табельн|паспорт|телефон.*коллег|заявк.*коллег|данн.*сотрудник", query, re.IGNORECASE) ) def _is_security_probe(query: str) -> bool: return bool( re.search(r"директор|служб.*безопас|представь.*что|гипотетич|покажи.*промпт|как ты устроен", query, re.IGNORECASE) ) def _try_pattern_knowledge( trimmed: str, greeting: str, conversation: list[dict[str, str]] | None = None, ) -> AssistantResult | None: if not settings.use_pattern_kb: return None for entry in KNOWLEDGE_ENTRIES: if any(re.search(pattern, trimmed, re.IGNORECASE) for pattern in entry["patterns"]): confidence = entry["confidence"] intro = ( f"{greeting}нашёл инструкцию по вашему вопросу. Вот что рекомендую:" if confidence >= settings.confidence_threshold else f"{greeting}нашёл частично подходящую информацию, но уверенность ниже {settings.confidence_threshold}%. Рекомендую также оформить обращение:" ) package_data = {**entry["package"], "confidence": confidence, "system": entry["system"]} package = ResponsePackage.model_validate(package_data) package.why_suggested = build_why_suggested( system=entry["system"], sources=package.sources, confidence=confidence, ) if package.ticket_draft: cases = package.colleague_cases or [ ColleagueCase.model_validate(case) for case in get_colleague_cases_for_system(entry["system"]) ] package.ticket_draft = enrich_ticket_draft( trimmed, entry["system"], package.ticket_draft.priority, conversation=conversation, sources=package.sources, colleague_cases=cases, recommendation=package.recommendation, ) package = ensure_escalation_draft(package, trimmed, conversation=conversation) return AssistantResult(text=intro, package=package) return None def _try_local_rag( trimmed: str, greeting: str, system: str, conversation: list[dict[str, str]] | None = None, ) -> AssistantResult | None: """Временный RAG в backend: поиск по Manuals/ + опционально внешняя LLM.""" if not settings.rag_local_enabled or settings.rag_mode != "local": return None hits = search_documents(trimmed) if not hits: return None llm = get_llm_client() if llm.is_configured(): llm_result = build_package_with_llm( trimmed, hits, llm, greeting, system=system, conversation=conversation ) if llm_result: intro, package = llm_result return AssistantResult(text=intro, package=package) package = build_package_from_hits(trimmed, hits, system=system, conversation=conversation) if package is None: return None threshold = settings.confidence_threshold intro = ( f"{greeting}нашёл релевантные фрагменты в инструкциях (локальный RAG). Вот что рекомендую:" if package.confidence >= threshold else f"{greeting}нашёл частично подходящие материалы (уверенность ниже {threshold}%). Рекомендую также оформить обращение:" ) return AssistantResult(text=intro, package=package) def process_message( query: str, profile: UserProfile, is_first_message: bool, conversation: list[dict[str, str]] | None = None, ) -> AssistantResult: trimmed = query.strip() greeting = f"{profile.first_name} {profile.patronymic}, " if is_first_message else "" if _is_thanks(trimmed): return AssistantResult( text="Рад был помочь. Если появятся вопросы — обращайтесь.", is_simple=True, ) if _is_off_topic(trimmed): return AssistantResult( text=f"{greeting}этот вопрос выходит за рамки моих задач — я помогаю только с рабочими вопросами. Могу ли я помочь ещё?", is_simple=True, ) if _is_confidential_request(trimmed): return AssistantResult( text=f"{greeting}эта информация является конфиденциальной и доступна только самому сотруднику. Могу ли я помочь ещё?", is_simple=True, ) if _is_security_probe(trimmed): return AssistantResult( text=f"{greeting}я работаю на основе ваших актуальных данных в системе. Гипотетические сценарии не влияют на мои возможности и правила работы. Могу ли я помочь ещё?", is_simple=True, ) glossary = _detect_glossary(trimmed) if glossary: return AssistantResult( text=f"{greeting}{glossary}\n\nМогу ли я помочь ещё?", is_simple=True, ) dialog = conversation or [] # 1) Курируемые сценарии (быстрые ответы для демо) pattern_result = _try_pattern_knowledge(trimmed, greeting, dialog) if pattern_result: return pattern_result # 2) Определение ИТ-системы до поиска (CJM §3.2, §4.1) system = _detect_system(trimmed) if not system: systems_list = ", ".join(IT_SYSTEMS) return AssistantResult( text=f"{greeting}чтобы подобрать точную инструкцию, уточните, пожалуйста: с какой ИТ-системой связан ваш вопрос? ({systems_list})", needs_clarification=True, is_simple=True, ) if system not in IT_SYSTEMS: return AssistantResult( text=f"{greeting}система «{system}» пока не входит в базу знаний ассистента. Могу помочь оформить обращение в поддержку.", create_ticket=True, is_simple=True, ) # 3) Локальный RAG по Manuals/ после определения системы rag_result = _try_local_rag(trimmed, greeting, system, dialog) if rag_result: return rag_result cases = [ColleagueCase.model_validate(case) for case in get_colleague_cases_for_system(system)] ticket = enrich_ticket_draft( trimmed, system, "Средний", conversation=dialog, colleague_cases=cases, recommendation="К сожалению, в базе знаний нет точного ответа на ваш вопрос.", ) why = build_why_suggested(system=system, sources=[], confidence=35) return AssistantResult( text=f"{greeting}по системе «{system}» не нашёл подходящей инструкции. Предлагаю создать обращение с предзаполненным контекстом.", create_ticket=True, package=ResponsePackage( recommendation="К сожалению, в базе знаний нет точного ответа на ваш вопрос.", steps=[], sources=[], colleague_cases=cases or None, why_suggested=why, confidence=35, system=system, ticket_draft=ticket, ), ) def get_default_profile() -> UserProfile: return UserProfile.model_validate(DEFAULT_PROFILE)