490 lines
10 KiB
Python
490 lines
10 KiB
Python
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)
|
||
|
||
|