| """ |
| AI Interview Simulator — OpenEnv Environment |
| ============================================ |
| |
| Implements the full OpenEnv interface: |
| reset() → InterviewObservation |
| step() → StepResult(observation, reward, done, info) |
| state() → InterviewState |
| |
| The environment simulates a real job interview: the agent receives one |
| question at a time and must provide a text response. After every answer |
| a deterministic grader scores it (0.0–1.0) and issues partial reward. |
| The episode ends when all questions have been answered. |
| |
| Usage: |
| env = InterviewEnv(task_id="junior_behavioral") |
| obs = env.reset() |
| while True: |
| action = InterviewAction(response="My answer …") |
| result = env.step(action) |
| if result.done: |
| break |
| """ |
| from __future__ import annotations |
|
|
| import time |
| from typing import Any, Dict, Optional |
|
|
| from interview_env.graders import grade_response |
| from interview_env.models import ( |
| DifficultyLevel, |
| InterviewAction, |
| InterviewObservation, |
| InterviewStage, |
| InterviewState, |
| QAPair, |
| StepResult, |
| ) |
| from interview_env.tasks import TASK_REGISTRY, TaskDefinition, get_task |
|
|
|
|
| class InterviewEnv: |
| """ |
| OpenEnv-compliant AI Interview Simulator environment. |
| |
| Parameters |
| ---------- |
| task_id : str |
| One of: "junior_behavioral", "mid_technical", "senior_system_design" |
| max_response_length : int |
| Characters — responses exceeding this are truncated (soft limit). |
| penalty_per_truncation : float |
| Reward deducted when the agent's response is truncated. |
| """ |
|
|
| VALID_TASKS = list(TASK_REGISTRY.keys()) |
| DEFAULT_TASK = "junior_behavioral" |
|
|
| def __init__( |
| self, |
| task_id: str = DEFAULT_TASK, |
| max_response_length: int = 2000, |
| penalty_per_truncation: float = 0.05, |
| ) -> None: |
| if task_id not in TASK_REGISTRY: |
| raise ValueError( |
| f"Unknown task_id '{task_id}'. " |
| f"Choose from: {self.VALID_TASKS}" |
| ) |
| self._task_id = task_id |
| self._max_response_length = max_response_length |
| self._penalty_per_truncation = penalty_per_truncation |
|
|
| |
| self._task: TaskDefinition = get_task(task_id) |
| self._current_index: int = 0 |
| self._qa_history: list[QAPair] = [] |
| self._per_q_scores: list[float] = [] |
| self._done: bool = False |
| self._total_steps: int = 0 |
| self._start_time: float = time.time() |
| self._cumulative_reward: float = 0.0 |
|
|
| |
| |
| |
|
|
| def reset(self) -> InterviewObservation: |
| """ |
| Reset the environment to the beginning of the interview. |
| Returns the first question as an observation. |
| """ |
| self._task = get_task(self._task_id) |
| self._current_index = 0 |
| self._qa_history = [] |
| self._per_q_scores = [] |
| self._done = False |
| self._total_steps = 0 |
| self._start_time = time.time() |
| self._cumulative_reward = 0.0 |
|
|
| return self._build_observation() |
|
|
| def step(self, action: InterviewAction) -> StepResult: |
| """ |
| Submit the agent's answer to the current interview question. |
| |
| Returns |
| ------- |
| StepResult with: |
| observation — next question (or completion message if done) |
| reward — per-step score in [–0.05, 1.0] |
| done — True when all questions are answered |
| info — detailed grader breakdown |
| """ |
| if self._done: |
| |
| return StepResult( |
| observation=self._build_observation(), |
| reward=0.0, |
| done=True, |
| info={"warning": "Episode already complete. Call reset()."}, |
| ) |
|
|
| self._total_steps += 1 |
| current_q = self._task.questions[self._current_index] |
|
|
| |
| response = action.response |
| truncated = False |
| reward_adj = 0.0 |
| if len(response) > self._max_response_length: |
| response = response[: self._max_response_length] |
| truncated = True |
| reward_adj = -self._penalty_per_truncation |
|
|
| |
| reward_obj = grade_response(response, current_q) |
| step_reward = reward_obj.total_score + reward_adj |
| step_reward = max(0.0, min(1.0, step_reward)) |
|
|
| |
| qa = QAPair( |
| question = current_q.text, |
| answer = response, |
| stage = current_q.stage, |
| question_score = reward_obj.total_score, |
| ) |
| self._qa_history.append(qa) |
| self._per_q_scores.append(reward_obj.total_score) |
| self._cumulative_reward += step_reward |
|
|
| |
| self._current_index += 1 |
| if self._current_index >= len(self._task.questions): |
| self._done = True |
|
|
| obs = self._build_observation() |
| info: Dict[str, Any] = { |
| "question_answered": current_q.text, |
| "grader_breakdown": { |
| "relevance": reward_obj.relevance, |
| "structure": reward_obj.structure, |
| "depth": reward_obj.depth, |
| "professionalism": reward_obj.professionalism, |
| "total_score": reward_obj.total_score, |
| }, |
| "feedback": reward_obj.feedback, |
| "truncated": truncated, |
| "question_number": self._current_index, |
| "total_questions": len(self._task.questions), |
| } |
|
|
| if self._done: |
| final_score = self._final_score() |
| info["episode_summary"] = { |
| "final_score": final_score, |
| "per_question_scores": self._per_q_scores, |
| "passed": final_score >= 0.60, |
| "pass_threshold": 0.60, |
| } |
|
|
| return StepResult( |
| observation = obs, |
| reward = round(step_reward, 4), |
| done = self._done, |
| info = info, |
| ) |
|
|
| def state(self) -> InterviewState: |
| """ |
| Return a serialisable snapshot of the full internal state. |
| Useful for debugging, checkpointing, and OpenEnv validation. |
| """ |
| return InterviewState( |
| task_id = self._task_id, |
| difficulty = self._task.difficulty, |
| job_title = self._task.job_title, |
| company_context = self._task.company_context, |
| questions = [q.text for q in self._task.questions], |
| stages = [q.stage for q in self._task.questions], |
| current_index = self._current_index, |
| qa_history = self._qa_history, |
| per_question_scores = self._per_q_scores, |
| episode_done = self._done, |
| total_steps = self._total_steps, |
| start_time = self._start_time, |
| cumulative_reward = self._cumulative_reward, |
| ) |
|
|
| |
| |
| |
|
|
| @property |
| def task_id(self) -> str: |
| return self._task_id |
|
|
| @property |
| def available_tasks(self) -> list[str]: |
| return self.VALID_TASKS |
|
|
| @property |
| def action_space_description(self) -> str: |
| return ( |
| "InterviewAction(response: str) — " |
| "A free-text string (1–2000 chars) representing the candidate's spoken answer." |
| ) |
|
|
| @property |
| def observation_space_description(self) -> str: |
| return ( |
| "InterviewObservation with fields: " |
| "task_id, job_title, company_context, difficulty, " |
| "current_question, question_number, total_questions, stage, " |
| "previous_qa (list of QAPair), cumulative_score, time_elapsed, interview_complete." |
| ) |
|
|
| |
| |
| |
|
|
| def _build_observation(self) -> InterviewObservation: |
| if self._done or self._current_index >= len(self._task.questions): |
| current_question = ( |
| "Thank you for completing the interview. " |
| "We will be in touch shortly." |
| ) |
| stage = InterviewStage.COMPLETE |
| question_number = len(self._task.questions) |
| interview_complete = True |
| else: |
| q = self._task.questions[self._current_index] |
| current_question = q.text |
| stage = q.stage |
| question_number = self._current_index + 1 |
| interview_complete = False |
|
|
| return InterviewObservation( |
| task_id = self._task_id, |
| job_title = self._task.job_title, |
| company_context = self._task.company_context, |
| difficulty = self._task.difficulty, |
| current_question = current_question, |
| question_number = question_number, |
| total_questions = len(self._task.questions), |
| stage = stage, |
| previous_qa = list(self._qa_history), |
| cumulative_score = round(self._cumulative_reward, 4), |
| time_elapsed = round(time.time() - self._start_time, 2), |
| interview_complete = interview_complete, |
| ) |
|
|
| def _final_score(self) -> float: |
| """Weighted average over all per-question scores.""" |
| if not self._per_q_scores: |
| return 0.0 |
| questions = self._task.questions |
| weighted = sum( |
| s * q.weight |
| for s, q in zip(self._per_q_scores, questions) |
| ) |
| total_w = sum(q.weight for q in questions) |
| return round(weighted / total_w, 4) if total_w > 0 else 0.0 |
|
|
