from __future__ import annotations import logging import re from typing import Optional from .llm import LocalLLM from .models import AnswerResponse, BlockKey, QuestionPayload, SessionState from .report import build_markdown_report from .session import SessionManager logger = logging.getLogger(__name__) class InterviewAgent: """LangGraph-inspired orchestrator for the interview flow.""" TOPIC_QUESTIONS = { BlockKey.health: [ ( "health_injuries", "Были ли у вас травмы и есть ли движения, которые вызывают дискомфорт?", "Хотите добавить детали по травмам или дискомфорту?", ), ( "health_medical", "Есть ли медицинские ограничения или диагнозы, которые нужно учитывать?", "Есть ли ещё важные медицинские моменты?", ), ( "health_recovery", "Как вы обычно переносите нагрузку и восстанавливаетесь после тренировок?", "Хотите уточнить восстановление и самочувствие?", ), ], BlockKey.goals: [ ( "goals_primary", "Какая главная цель на ближайшие 8–12 недель?", "Хотите уточнить цель?", ), ( "goals_metrics", "По каким признакам вы поймёте, что цель достигнута?", "Хотите уточнить признаки результата?", ), ( "goals_constraints", "Есть ли то, чего вы точно не хотите в тренировках?", "Есть ли ещё ограничения или нежелательные моменты?", ), ], BlockKey.readiness: [ ( "readiness_schedule", "Сколько раз в неделю готовы тренироваться?", "Хотите уточнить частоту занятий?", ), ( "readiness_location", "Где вы планируете тренироваться?", "Есть ли дополнительные локации?", ), ( "readiness_format", "Какой формат занятий вам удобнее (онлайн/офлайн)?", "Хотите уточнить формат занятий?", ), ], } def __init__( self, session_manager: SessionManager, llm_client: LocalLLM, ) -> None: self._sessions = session_manager self._llm = llm_client async def start(self, state: SessionState) -> Optional[QuestionPayload]: if state.questions: return state.questions[-1] block = self._sessions.current_block(state) return self._create_question(state, block) async def handle_answer( self, state: SessionState, question_id: str, transcript: str, ) -> AnswerResponse: await self._sessions.record_transcript(state.id, question_id, transcript) block = self._sessions.current_block(state) next_question: Optional[QuestionPayload] = None markdown: Optional[str] = None completed_block = False if self._is_block_complete(state, block): moved = await self._sessions.advance_block(state) if moved: completed_block = True next_question = self._create_question( state, self._sessions.current_block(state) ) else: markdown = self._generate_report(state) else: next_question = self._create_question(state, block) return AnswerResponse( transcript=transcript, completedBlock=completed_block, nextQuestion=next_question, markdownReport=markdown, ) def _create_question(self, state: SessionState, block: BlockKey) -> Optional[QuestionPayload]: count = len(self._sessions.questions_for_block(state, block)) if count >= self._sessions.questions_per_block: return None covered_topics = self._covered_topics(state, block) question_text, topic = self._select_question( block=block, covered_topics=covered_topics, index=count, ) question = QuestionPayload( id=f"{block.value}-{count + 1}", block=block, prompt=question_text, topic=topic, ) self._sessions.add_question(state, question) return question def _select_question( self, *, block: BlockKey, covered_topics: set[str], index: int, ) -> tuple[str, str | None]: templates = self.TOPIC_QUESTIONS.get(block, []) if index >= len(templates): return "Поделитесь, пожалуйста, подробностями по этой теме.", "deep_dive" topic, prompt, confirm_prompt = templates[index] if topic in covered_topics: return confirm_prompt, topic return prompt, topic def _is_block_complete(self, state: SessionState, block: BlockKey) -> bool: answered = self._sessions.answered_pairs(state, block) return len(answered) >= self._sessions.questions_per_block def _covered_topics(self, state: SessionState, block: BlockKey) -> set[str]: transcripts = [ state.transcripts[q.id] for q in self._sessions.questions_for_block(state, block) if q.id in state.transcripts ] if not transcripts: return set() text = " ".join(transcripts).lower() topics = self._heuristic_topics(block, text) return topics def _heuristic_topics(self, block: BlockKey, text: str) -> set[str]: topics: set[str] = set() if block == BlockKey.health: injury_keywords = [ "травм", "боль", "болит", "болела", "болят", "огранич", "операц", "перелом", "растяж", "разрыв", "грыж", "сустав", "колен", "плеч", "спин", "шея", "голен", "таз", "связк", ] injury_negations = [ "нет травм", "травм нет", "никаких травм", "без травм", "не травмирован", ] if self._has_any(text, injury_keywords) or self._has_any(text, injury_negations): topics.add("health_injuries") medical_keywords = [ "диагноз", "операц", "давлен", "сердц", "дых", "астм", "лекар", "препарат", "таблет", "обезбол", "противовосп", "кардио", ] if self._has_any(text, medical_keywords): topics.add("health_medical") recovery_keywords = [ "сон", "восстанов", "стресс", "простуд", "устал", "усталость", "болею", "иммун", "перегруз", ] if self._has_any(text, recovery_keywords): topics.add("health_recovery") if block == BlockKey.goals: goal_keywords = [ "похуд", "вес", "масса", "мышц", "сила", "вынослив", "техник", "оли", "форма", "соревн", "hyrox", ] if self._has_any(text, goal_keywords): topics.add("goals_primary") metrics_keywords = [ "кг", "килограмм", "процент", "объем", "см", "сантиметр", "показател", "результат", "цифр", "повтор", "минут", "секунд", "рекорд", "pr", "wod", "1км", "5км", "присед", "тяга", "жим", "рывок", "взятие", ] if self._has_any(text, metrics_keywords): topics.add("goals_metrics") constraints_keywords = [ "без", "не хочу", "не люблю", "не нравится", "doms", "травм", "боль", "огранич", ] if self._has_any(text, constraints_keywords): topics.add("goals_constraints") if block == BlockKey.readiness: frequency_keywords = [ "в неделю", "раз в неделю", "раза в неделю", "раз", "дни", "время", "смен", "работ", "семь", ] frequency_regex = r"\b\d+\s*(раз|р\.)\b" if self._has_any(text, frequency_keywords) or re.search(frequency_regex, text): topics.add("readiness_schedule") equipment_keywords = [ "дом", "дома", "зал", "спортзал", "бокс", "площад", "оборуд", "гантел", "штанг", "турник", "гир", "резин", "коврик", "гребл", "вел", "бег", "дорож", ] if self._has_any(text, equipment_keywords): topics.add("readiness_location") format_keywords = [ "онлайн", "он-лайн", "офлайн", "оффлайн", "смешан", "гибрид", "очно", "индивиду", "групп", "созвон", "обратн", "чат", "голос", "видео", ] if self._has_any(text, format_keywords): topics.add("readiness_format") return topics @staticmethod def _has_any(text: str, keywords: list[str]) -> bool: return any(keyword in text for keyword in keywords) def _generate_report(self, state: SessionState) -> str: if self._llm.available: try: report = self._llm.generate_report(state) if report: return report except Exception as exc: # pragma: no cover - LLM may fail logger.warning("LLM report generation failed: %s", exc) return build_markdown_report(state)