| import re
|
| import json
|
| from typing import List, Dict, Tuple
|
| from knowledge_base import KnowledgeBase
|
| from retriever import Retriever
|
|
|
| class ITMOChatbot:
|
| def __init__(self):
|
| self.kb = KnowledgeBase()
|
| self.retriever = Retriever()
|
| self.max_history_turns = 3
|
| self.max_context_tokens = 1200
|
| self.relevance_threshold = 0.38
|
|
|
| try:
|
| from transformers import pipeline
|
| self.generator = pipeline('text2text-generation', model='cointegrated/rut5-base-multitask')
|
| except Exception as e:
|
| print(f'Генеративная модель не загружена: {e}')
|
| self.generator = None
|
|
|
| def chat(self, message: str, history: list) -> Tuple[str, float]:
|
| if not message.strip():
|
| return 'Пожалуйста, задайте вопрос.', 0.0
|
|
|
| if not self.kb.is_itmo_query(message):
|
| return self._get_irrelevant_response(), 0.0
|
|
|
| context = self._get_context(message)
|
| if not context:
|
| return 'К сожалению, не нашел релевантной информации в учебных планах ITMO.', 0.0
|
|
|
| response = self._generate_response(message, history, context)
|
| relevance_score = self._calculate_relevance_score(message, context)
|
|
|
| return response, relevance_score
|
|
|
| def recommend_courses(self, profile: dict) -> str:
|
| if not profile.get('semester'):
|
| return 'Пожалуйста, укажите целевой семестр для получения рекомендаций.'
|
|
|
| recommendations = self.kb.recommend(profile)
|
| if not recommendations:
|
| return 'К сожалению, не удалось найти подходящие курсы для вашего профиля.'
|
|
|
| result = '🎯 Рекомендуемые курсы (из официальных учебных планов ITMO):\n\n'
|
| for i, rec in enumerate(recommendations[:7], 1):
|
| result += f'{i}. {rec["name"]} ({rec["semester"]} семестр, {rec["credits"]} кредитов)\n'
|
| result += f' {rec["why"]}\n\n'
|
|
|
| return result
|
|
|
| def _get_context(self, message: str) -> List[Dict]:
|
| try:
|
| results = self.retriever.retrieve(message, k=6, threshold=0.35)
|
|
|
| formatted_results = []
|
| for result in results:
|
| course_id = result.get('course_id')
|
| if course_id:
|
| course = self.kb.get_course_by_id(course_id)
|
| if course:
|
| course['score'] = result.get('score', 0.0)
|
| formatted_results.append(course)
|
| return formatted_results
|
| except Exception as e:
|
| print(f'Ошибка при получении контекста: {e}')
|
| return []
|
|
|
| def _generate_response(self, message: str, history: list, context: List[Dict]) -> str:
|
| if not context:
|
| return 'В предоставленных данных об этом не сказано.'
|
|
|
| prompt = self._build_prompt(message, history, context)
|
|
|
| if self.generator:
|
| try:
|
| response = self.generator(
|
| prompt,
|
| max_new_tokens=180,
|
| temperature=0.4,
|
| do_sample=True
|
| )[0]['generated_text']
|
| return response.strip()
|
| except Exception as e:
|
| print(f'Ошибка генерации: {e}')
|
|
|
| return self._fallback_response(context)
|
|
|
| def _build_prompt(self, message: str, history: list, context: List[Dict]) -> str:
|
| system_prompt = 'Отвечай только по контексту (ниже). Если недостаточно данных — прямо скажи: "в предоставленных данных об этом не сказано". Отвечай кратко и по делу.'
|
|
|
| history_text = ''
|
| if history:
|
| recent_history = history[-self.max_history_turns:]
|
| for turn in recent_history:
|
| history_text += f'Пользователь: {turn[0]}\nБот: {turn[1]}\n'
|
|
|
| context_text = 'Контекст:\n'
|
| for item in context:
|
| context_text += f'- {item["name"]} ({item["semester"]} семестр, {item["credits"]} кредитов): {item["short_desc"]}\n'
|
|
|
| prompt = f'{system_prompt}\n\n{history_text}Контекст:\n{context_text}\nВопрос: {message}'
|
|
|
| if len(prompt) > self.max_context_tokens * 4:
|
| prompt = prompt[:self.max_context_tokens * 4]
|
|
|
| return prompt
|
|
|
| def _fallback_response(self, context: List[Dict]) -> str:
|
| if not context:
|
| return 'В предоставленных данных об этом не сказано.'
|
|
|
| courses = []
|
| for item in context[:3]:
|
| courses.append(f'{item["name"]} ({item["semester"]} семестр, {item["credits"]} кредитов)')
|
|
|
| return f'Найденные курсы: {", ".join(courses)}. Для более подробной информации обратитесь к официальным учебным планам ITMO.'
|
|
|
| def _calculate_relevance_score(self, message: str, context: List[Dict]) -> float:
|
| if not context:
|
| return 0.0
|
|
|
| scores = [item.get('score', 0.0) for item in context]
|
| return sum(scores) / len(scores) if scores else 0.0
|
|
|
| def _get_irrelevant_response(self) -> str:
|
| return '''Похоже, вопрос не относится к магистратурам ITMO и их учебным планам.
|
|
|
| Попробуйте спросить, например:
|
| • "Какие дисциплины по NLP в 1 семестре программы ИИ?"
|
| • "Расскажи о программе AI Product"
|
| • "Какие курсы по машинному обучению есть в программе ИИ?"
|
| • "Сколько кредитов за дисциплину 'Глубокое обучение'?"'''
|
|
|
