Spaces:
Sleeping
Sleeping
File size: 3,226 Bytes
03faf26 d50ee26 03faf26 d50ee26 03faf26 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 | from typing import Any, Dict, List, Optional, TypedDict
from langgraph.graph import END, StateGraph
from utils.gemini import generate_interview_question
class InterviewGraphState(TypedDict, total=False):
role_title: str
skills: List[str]
previous_questions: List[str]
previous_answer: Optional[str]
question_count: int
max_questions: int
current_difficulty: str
next_difficulty: str
question_stage: str
is_complete: bool
question_data: Dict[str, Any]
FOUNDATION_QUESTION_LIMIT = 0
def _difficulty_for_question_number(question_number: int, foundation_limit: int = FOUNDATION_QUESTION_LIMIT) -> str:
if question_number <= 5:
return "medium"
return "hard"
async def _check_completion(state: InterviewGraphState) -> InterviewGraphState:
question_count = int(state.get("question_count", 0))
max_questions = int(state.get("max_questions", 10))
return {"is_complete": question_count >= max_questions}
def _route_after_completion(state: InterviewGraphState) -> str:
return "end" if state.get("is_complete") else "difficulty"
async def _set_next_difficulty(state: InterviewGraphState) -> InterviewGraphState:
question_count = int(state.get("question_count", 0))
# We are generating the next question, so use question_count + 1.
next_question_number = question_count + 1
stage = "foundation" if next_question_number <= FOUNDATION_QUESTION_LIMIT else "deep"
return {
"next_difficulty": _difficulty_for_question_number(next_question_number),
"question_stage": stage,
}
async def _generate_question(state: InterviewGraphState) -> InterviewGraphState:
role_title = state.get("role_title", "Software Developer")
skills = state.get("skills", ["general"])
previous_questions = state.get("previous_questions", [])
previous_answer = state.get("previous_answer")
difficulty = state.get("next_difficulty", state.get("current_difficulty", "medium"))
question_stage = state.get("question_stage", "deep")
question_data = await generate_interview_question(
skills=skills,
role_title=role_title,
previous_questions=previous_questions,
previous_answer=previous_answer,
difficulty=difficulty,
question_stage=question_stage,
foundation_limit=FOUNDATION_QUESTION_LIMIT,
)
return {
"question_data": question_data,
"current_difficulty": question_data.get("difficulty", difficulty),
}
def _build_graph():
graph = StateGraph(InterviewGraphState)
graph.add_node("check", _check_completion)
graph.add_node("difficulty", _set_next_difficulty)
graph.add_node("generate", _generate_question)
graph.set_entry_point("check")
graph.add_conditional_edges(
"check",
_route_after_completion,
{
"end": END,
"difficulty": "difficulty",
},
)
graph.add_edge("difficulty", "generate")
graph.add_edge("generate", END)
return graph.compile()
_INTERVIEW_GRAPH = _build_graph()
async def run_interview_graph(state: InterviewGraphState) -> InterviewGraphState:
result = await _INTERVIEW_GRAPH.ainvoke(state)
return result
|