Okia Services

All business logic is in /app/services/.

InterviewService

File: interview.py

Orchestrates the entire interview flow.

Methods

Method	Parameters	Returns	Description
`create_session`	candidate info, job details, settings	`OkiaSession`	Create new interview
`get_session_by_token`	`token: str`	`OkiaSession`	Find session by token
`get_session_by_id`	`session_id: UUID`	`OkiaSession`	Find session by ID
`start_interview`	`session: OkiaSession`	`dict`	Start and return greeting
`process_candidate_answer`	`session, answer`	`dict`	Process answer, generate next question
`complete_interview`	`session: OkiaSession`	`dict`	Complete and generate report
`pause_interview`	`session: OkiaSession`	`bool`	Pause timer
`resume_interview`	`session: OkiaSession`	`bool`	Resume timer
`send_invitation_sms`	`session: OkiaSession`	`bool`	Send SMS invite

Session Token Generation

python

import secrets

@staticmethod
def generate_session_token() -> str:
    """Generate a secure session token."""
    return secrets.token_urlsafe(32)  # ~256 bits of entropy

Answer Processing Flow

python

async def process_candidate_answer(
    db: AsyncSession,
    session: OkiaSession,
    answer: str,
) -> Dict[str, Any]:
    # 1. Save candidate's answer
    answer_msg = OkiaMessage(
        session_id=session.id,
        role=MessageRole.CANDIDATE,
        message_type=MessageType.ANSWER,
        content=answer,
        sequence_number=len(session.messages) + 1,
    )
    db.add(answer_msg)
    
    # 2. Score the answer against last question
    last_question = self._get_last_question(session)
    score_result = await claude_service.score_answer(
        question=last_question.content,
        answer=answer,
        question_metadata={...}
    )
    
    # 3. Calculate composite score
    composite = scoring_service.calculate_answer_composite(
        relevance=score_result.get("relevance_score", 70),
        depth=score_result.get("depth_score", 70),
        clarity=score_result.get("clarity_score", 70),
        technical=score_result.get("technical_score"),
    )
    
    # 4. Save score
    score = OkiaScore(
        session_id=session.id,
        message_id=answer_msg.id,
        composite_score=composite,
        ...
    )
    db.add(score)
    
    # 5. Check if interview should end
    session.question_count += 1
    if session.question_count >= session.max_questions:
        return await self.complete_interview(db, session)
    
    # 6. Generate next question
    next_question = await self._generate_next_question(db, session)
    
    return {
        "answer_id": str(answer_msg.id),
        "score": {"composite_score": composite, ...},
        "next_message": next_question,
    }

Webhook Notification

python

async def _call_completion_webhook(
    session: OkiaSession, 
    report: OkiaReport
) -> None:
    if not session.webhook_url:
        return

    payload = {
        "session_id": str(session.id),
        "status": "completed",
        "overall_score": int(report.overall_score),
        "grade": report.grade,
        "scores": {
            "technical": int(report.technical_score),
            "communication": int(report.communication_score),
            "problem_solving": int(report.problem_solving_score),
            "cultural_fit": int(report.cultural_fit_score),
        },
        "metadata": session.webhook_metadata or {},
    }

    headers = {"Content-Type": "application/json"}
    if session.webhook_secret:
        headers["X-Webhook-Secret"] = session.webhook_secret

    async with httpx.AsyncClient(timeout=30.0) as client:
        response = await client.post(
            session.webhook_url,
            json=payload,
            headers=headers,
        )
        response.raise_for_status()

ClaudeService

File: claude.py

Handles all Claude AI interactions.

Configuration

python

import anthropic

client = anthropic.AsyncAnthropic(api_key=settings.anthropic_api_key)
MODEL = settings.claude_model  # "claude-sonnet-4-20250514"

Methods

Method	Parameters	Returns	Description
`generate_greeting`	`candidate_name, job_title, time_limit`	`str`	Personalized greeting
`generate_question`	`session_context, candidate_info, job_context, history`	`dict`	Context-aware question
`score_answer`	`question, answer, metadata`	`dict`	Score with rationale
`generate_closing`	`candidate_name, highlight`	`str`	Closing message
`generate_report`	`all_qa, all_scores, duration`	`dict`	Full interview report

Question Generation

python

async def generate_question(
    session_context: dict,
    candidate_info: dict,
    job_context: dict,
    conversation_history: list,
) -> dict:
    system_prompt = """You are Okia, an AI interviewer for YeboJobs. 
    Generate the next interview question based on:
    - The job requirements
    - The candidate's background
    - Previous Q&A (avoid repetition)
    - Topics not yet covered
    
    Return JSON:
    {
      "question": "...",
      "category": "technical|behavioral|situational|experience",
      "assesses": "specific skill being evaluated",
      "expected_elements": ["key points to look for"],
      "suggestions": ["quick response options for candidate"]
    }
    """
    
    message = await client.messages.create(
        model=MODEL,
        max_tokens=1000,
        system=system_prompt,
        messages=[{
            "role": "user",
            "content": f"""
            Job: {job_context['title']} ({job_context['category']})
            Experience level: {job_context['experience_level']}
            Candidate CV summary: {candidate_info['cv_summary']}
            Questions asked so far: {session_context['questions_asked']}
            Topics covered: {session_context['topics_covered']}
            
            Conversation so far:
            {format_conversation(conversation_history)}
            
            Generate the next question.
            """
        }],
    )
    
    return json.loads(message.content[0].text)

Answer Scoring

python

async def score_answer(
    question: str,
    answer: str,
    question_metadata: dict,
) -> dict:
    system_prompt = """Score this interview answer on:
    - relevance_score (0-100): How well it addresses the question
    - depth_score (0-100): Level of detail and examples
    - clarity_score (0-100): Communication quality
    - technical_score (0-100, optional): Technical accuracy if applicable
    
    Return JSON:
    {
      "relevance_score": 85,
      "depth_score": 75,
      "clarity_score": 90,
      "technical_score": 80,
      "scoring_rationale": "...",
      "strengths": ["..."],
      "improvements": ["..."],
      "notable_quote": "memorable part of answer (if any)"
    }
    """
    
    message = await client.messages.create(
        model=MODEL,
        max_tokens=800,
        system=system_prompt,
        messages=[{
            "role": "user",
            "content": f"""
            Question: {question}
            Category: {question_metadata.get('category')}
            Assessing: {question_metadata.get('assesses')}
            Expected elements: {question_metadata.get('expected_elements')}
            
            Candidate's Answer:
            {answer}
            
            Score this answer.
            """
        }],
    )
    
    return json.loads(message.content[0].text)

Report Generation

python

async def generate_report(
    candidate_name: str,
    job_title: str,
    job_category: str,
    all_qa: list,
    all_scores: list,
    duration_minutes: int,
) -> dict:
    system_prompt = """Generate a comprehensive interview report.
    
    Return JSON:
    {
      "technical_score": 0-100,
      "communication_score": 0-100,
      "problem_solving_score": 0-100,
      "cultural_fit_score": 0-100,
      "overall_score": 0-100,
      "grade": "A+|A|B+|B|C|D|F",
      "recommendation": "strong_hire|hire|consider|no_hire",
      "executive_summary": "2-3 sentences",
      "strengths_analysis": ["detailed strength 1", "..."],
      "weaknesses_analysis": ["area for improvement 1", "..."],
      "technical_assessment": "paragraph",
      "behavioral_assessment": "paragraph",
      "demonstrated_skills": ["skill1", "skill2"],
      "skill_gaps": ["gap1", "gap2"],
      "interview_highlights": [{"question": "...", "highlight": "..."}],
      "red_flags": ["concern1" or empty],
      "follow_up_questions": ["question for next round"]
    }
    """
    
    message = await client.messages.create(
        model=MODEL,
        max_tokens=2000,
        system=system_prompt,
        messages=[{
            "role": "user",
            "content": f"""
            Candidate: {candidate_name}
            Position: {job_title} ({job_category})
            Duration: {duration_minutes} minutes
            
            Full Interview Transcript:
            {format_qa(all_qa)}
            
            Individual Scores:
            {format_scores(all_scores)}
            
            Generate the complete interview report.
            """
        }],
    )
    
    return json.loads(message.content[0].text)

ScoringService

File: scoring.py

Handles score calculations and grading.

Methods

Method	Parameters	Returns	Description
`calculate_answer_composite`	`relevance, depth, clarity, technical?`	`int`	Weighted composite
`calculate_category_scores`	`all_scores, messages`	`dict`	Scores by category
`calculate_overall_score`	`category_scores`	`int`	Final score
`get_grade`	`score: int`	`str`	A+ to F grade
`get_recommendation`	`score: int`	`str`	Hire recommendation
`get_recommendation_label`	`recommendation`	`str`	Human-readable

Composite Score Calculation

python

def calculate_answer_composite(
    relevance: int,
    depth: int,
    clarity: int,
    technical: Optional[int] = None,
) -> int:
    """Calculate weighted composite score for a single answer."""
    if technical is not None:
        # Technical question: weight technical higher
        weights = {
            "relevance": 0.25,
            "depth": 0.25,
            "clarity": 0.20,
            "technical": 0.30,
        }
        return round(
            relevance * weights["relevance"] +
            depth * weights["depth"] +
            clarity * weights["clarity"] +
            technical * weights["technical"]
        )
    else:
        # Non-technical question
        weights = {
            "relevance": 0.35,
            "depth": 0.35,
            "clarity": 0.30,
        }
        return round(
            relevance * weights["relevance"] +
            depth * weights["depth"] +
            clarity * weights["clarity"]
        )

Grading Scale

python

def get_grade(score: int) -> str:
    """Convert score to letter grade."""
    if score >= 95: return "A+"
    if score >= 90: return "A"
    if score >= 85: return "B+"
    if score >= 80: return "B"
    if score >= 70: return "C"
    if score >= 60: return "D"
    return "F"

def get_recommendation(score: int) -> str:
    """Get hiring recommendation based on score."""
    if score >= 85: return "strong_hire"
    if score >= 75: return "hire"
    if score >= 60: return "consider"
    return "no_hire"

def get_recommendation_label(recommendation: str) -> str:
    """Human-readable recommendation."""
    labels = {
        "strong_hire": "Strongly Recommend",
        "hire": "Recommend for Hire",
        "consider": "Consider for Next Round",
        "no_hire": "Not Recommended",
    }
    return labels.get(recommendation, "Unknown")

SMSService

File: sms.py

Handles SMS notifications via Twilio.

Methods

Method	Parameters	Returns	Description
`send_interview_invitation`	`phone, name, job_title, url`	`str` (SID)	Invite SMS
`send_score_notification`	`phone, name, score, recommendation`	`str` (SID)	Result SMS

SMS Templates

python

async def send_interview_invitation(
    to_number: str,
    candidate_name: str,
    job_title: str,
    interview_url: str,
) -> Optional[str]:
    message_body = f"""Hi {candidate_name}!

You've been invited to an AI interview for: {job_title}

Click to start: {interview_url}

This link expires in 24 hours. Good luck!

- YeboJobs Team"""
    
    message = await client.messages.create_async(
        body=message_body,
        from_=settings.twilio_phone_number,
        to=to_number,
    )
    return message.sid

async def send_score_notification(
    to_number: str,
    candidate_name: str,
    score: int,
    recommendation: str,
) -> Optional[str]:
    grade = scoring_service.get_grade(score)
    
    message_body = f"""Hi {candidate_name}!

Your interview has been reviewed.
Score: {score}/100 (Grade: {grade})

Log in to YeboJobs to see your detailed feedback and track your applications.

- YeboJobs Team"""
    
    message = await client.messages.create_async(
        body=message_body,
        from_=settings.twilio_phone_number,
        to=to_number,
    )
    return message.sid

UserStatsService

File: user_stats.py

Updates YeboScore based on interview performance.

python

async def update_stats_for_session(
    db: AsyncSession,
    session: OkiaSession,
    report: OkiaReport,
) -> None:
    """Update user's YeboScore after interview completion."""
    if not session.user_id:
        return
    
    # Find or create YeboScore
    yebo_score = await db.execute(
        select(YeboScore).where(YeboScore.user_id == session.user_id)
    )
    yebo_score = yebo_score.scalar_one_or_none()
    
    if not yebo_score:
        yebo_score = YeboScore(user_id=session.user_id)
        db.add(yebo_score)
    
    # Update general interview score
    yebo_score.general_score = report.overall_score
    yebo_score.general_grade = report.grade
    yebo_score.last_interview_at = datetime.utcnow()
    
    # Recalculate overall YeboScore
    yebo_score.overall_score = calculate_overall_yebo_score(yebo_score)
    yebo_score.tier = get_tier_from_score(yebo_score.overall_score)
    yebo_score.last_calculated_at = datetime.utcnow()
    
    await db.commit()

API

Backend API

Okia AI Service

Frontend

Student Dashboard

Teacher Dashboard

Parent Dashboard

School Admin Dashboard

Okia Services ​

InterviewService ​

Methods ​

Session Token Generation ​

Answer Processing Flow ​

Webhook Notification ​

ClaudeService ​

Configuration ​

Methods ​

Question Generation ​

Answer Scoring ​

Report Generation ​

ScoringService ​

Methods ​

Composite Score Calculation ​

Grading Scale ​

SMSService ​

Methods ​

SMS Templates ​

UserStatsService ​

Okia Services

InterviewService

Methods

Session Token Generation

Answer Processing Flow

Webhook Notification

ClaudeService

Configuration

Methods

Question Generation

Answer Scoring

Report Generation

ScoringService

Methods

Composite Score Calculation

Grading Scale

SMSService

Methods

SMS Templates

UserStatsService