"""AI interviewer: stateful adaptive sessions (technical / management / hybrid). There is NO fixed question list. Questions come from the competency-keyed exercise generator; after every answer the evaluation contract drives the sequencing policy: - weakness detected (score <= 2) -> targeted follow-up probe from the exercise's follow_up_questions - strong answer (score >= 4) -> escalate difficulty, or switch dimension once difficulty is maxed - shallow-but-polished (score 3 with polish markers but thin concept coverage) -> depth probe on a missed criterion - otherwise -> advance to a new competency The transcript is stored as an Evidence artifact. """ from __future__ import annotations import random from typing import Dict, List, Optional from pydantic import BaseModel, Field from sensorflow.hillclimb import content, evaluate from sensorflow.hillclimb.blueprint import competency_index, load_blueprint from sensorflow.hillclimb.models import (EvaluationResult, Evidence, Store, get_store, new_id, now_iso) MODES = { "technical": [1, 2], "management": [3], "hybrid": [1, 2, 3, 4], } POLISH_MARKERS = ["first", "second", "third", "in summary", "to summarize", "overall", "furthermore", "moreover", "in conclusion", "let me walk", "great question"] class InterviewTurn(BaseModel): index: int question: str question_type: str # opening | probe | depth_probe | escalate | advance competency_id: str exercise_id: str difficulty: int answer: Optional[str] = None evaluation: Optional[EvaluationResult] = None timestamp: str = Field(default_factory=now_iso) class InterviewSession(BaseModel): session_id: str = Field(default_factory=lambda: new_id("iv")) user_id: str = "default" mode: str = "hybrid" seed: int = 7 status: str = "active" # active | complete turns: List[InterviewTurn] = Field(default_factory=list) asked_competencies: List[str] = Field(default_factory=list) probe_count_on_current: int = 0 evidence_id: Optional[str] = None def _pool(mode: str, store: Store) -> List[str]: bp = load_blueprint(store) phases = MODES.get(mode, MODES["hybrid"]) return [c.id for c in bp.competencies if c.phase in phases] def _pick_competency(session: InterviewSession, store: Store, avoid_dimension_of: Optional[str] = None) -> str: bp = load_blueprint(store) idx = competency_index(bp) pool = [c for c in _pool(session.mode, store) if c not in session.asked_competencies] if not pool: pool = _pool(session.mode, store) if avoid_dimension_of and avoid_dimension_of in idx: dim = idx[avoid_dimension_of].dimension switched = [c for c in pool if idx[c].dimension != dim] if switched: pool = switched rng = random.Random(session.seed + len(session.turns) * 131) return rng.choice(sorted(pool)) def _new_exercise_turn(session: InterviewSession, competency_id: str, difficulty: int, question_type: str, store: Store) -> InterviewTurn: ex = content.generate_exercise(competency_id, difficulty=difficulty, seed=session.seed + len(session.turns) * 977, store=store, use_llm=False) session.asked_competencies.append(competency_id) session.probe_count_on_current = 0 return InterviewTurn(index=len(session.turns), question=ex.scenario, question_type=question_type, competency_id=competency_id, exercise_id=ex.exercise_id, difficulty=difficulty) def start_interview(mode: str = "hybrid", user_id: str = "default", seed: int = 7, store: Optional[Store] = None) -> InterviewSession: store = store or get_store() if mode not in MODES: raise ValueError(f"unknown mode '{mode}' (technical|management|hybrid)") session = InterviewSession(user_id=user_id, mode=mode, seed=seed) first_comp = _pick_competency(session, store) session.turns.append(_new_exercise_turn(session, first_comp, difficulty=2, question_type="opening", store=store)) store.put("interviews", session.session_id, session) return session def _is_polished_but_shallow(result: EvaluationResult, answer: str) -> bool: low = answer.lower() polish = sum(1 for m in POLISH_MARKERS if m in low) concept_strengths = [s for s in result.strengths if not s.lower().startswith(("quantifies", "states explicit tradeoff"))] return polish >= 2 and len(concept_strengths) <= 2 def submit_answer(session_id: str, answer: str, use_llm: bool = True, store: Optional[Store] = None) -> InterviewSession: store = store or get_store() raw = store.get("interviews", session_id) if raw is None: raise ValueError(f"unknown interview session '{session_id}'") session = InterviewSession(**raw) if session.status != "active": raise ValueError("interview already complete") current = session.turns[-1] ex = content.get_exercise(current.exercise_id, store) result = evaluate.evaluate_response(ex, answer, use_llm=use_llm) current.answer = answer current.evaluation = result # ------------------------------------------------------ sequencing policy next_turn: Optional[InterviewTurn] = None if result.score <= 2 and session.probe_count_on_current < 2 and ex.follow_up_questions: probe_idx = min(session.probe_count_on_current, len(ex.follow_up_questions) - 1) session.probe_count_on_current += 1 next_turn = InterviewTurn(index=len(session.turns), question=ex.follow_up_questions[probe_idx], question_type="probe", competency_id=current.competency_id, exercise_id=current.exercise_id, difficulty=current.difficulty) elif result.score >= 4: if current.difficulty < 3: next_turn = _new_exercise_turn(session, current.competency_id, difficulty=current.difficulty + 1, question_type="escalate", store=store) else: comp = _pick_competency(session, store, avoid_dimension_of=current.competency_id) next_turn = _new_exercise_turn(session, comp, difficulty=3, question_type="escalate", store=store) elif result.score == 3 and _is_polished_but_shallow(result, answer): missed = result.weaknesses[0] if result.weaknesses else "the mechanism behind your answer" next_turn = InterviewTurn( index=len(session.turns), question=(f"That was well structured, but let's go deeper. Specifically: " f"{missed.rstrip('.').lower()}. Explain the underlying mechanism with a " f"concrete example and real numbers."), question_type="depth_probe", competency_id=current.competency_id, exercise_id=current.exercise_id, difficulty=current.difficulty) else: comp = _pick_competency(session, store) next_turn = _new_exercise_turn(session, comp, difficulty=2, question_type="advance", store=store) session.turns.append(next_turn) store.put("interviews", session.session_id, session) return session def end_interview(session_id: str, store: Optional[Store] = None) -> InterviewSession: store = store or get_store() raw = store.get("interviews", session_id) if raw is None: raise ValueError(f"unknown interview session '{session_id}'") session = InterviewSession(**raw) if session.status == "complete": return session session.status = "complete" answered = [t for t in session.turns if t.evaluation is not None] per_comp: Dict[str, List[int]] = {} for t in answered: per_comp.setdefault(t.competency_id, []).append(t.evaluation.score) avg = (sum(t.evaluation.score for t in answered) / len(answered)) if answered else 0.0 ev = Evidence( user_id=session.user_id, competency_ids=list(per_comp.keys()), artifact_type="interview_transcript", source=f"AI interview ({session.mode})", summary=(f"{len(answered)} answers across {len(per_comp)} competencies, " f"avg score {avg:.1f}/5."), quotes=[t.evaluation.evidence[0] for t in answered if t.evaluation.evidence][:4], score=round(avg, 2), confidence=min(0.85, 0.3 + 0.1 * len(answered)), payload={"session_id": session.session_id, "per_competency": {k: sum(v) / len(v) for k, v in per_comp.items()}, "transcript": [t.model_dump() for t in session.turns]}, ) store.put("evidence", ev.evidence_id, ev) session.evidence_id = ev.evidence_id store.put("interviews", session.session_id, session) return session def get_interview(session_id: str, store: Optional[Store] = None) -> Optional[InterviewSession]: raw = (store or get_store()).get("interviews", session_id) return InterviewSession(**raw) if raw else None