Spaces:
Sleeping
Sleeping
| """ | |
| graph.py — LangGraph State Machine Definition (v2) | |
| New in v2: | |
| - Added gate_solution node between analyze and terminals | |
| - Added hint → analyze loopback edge for iterative tutoring | |
| - turn_count loop-break at >= 3 forces validation early | |
| - Supports 'hint_forced' mode (gate redirect) | |
| """ | |
| from langgraph.graph import StateGraph, END | |
| from agent.models import AgentState | |
| from agent.nodes import ( | |
| classify_problem, | |
| evaluate_reasoning, | |
| generate_hint, | |
| validate_solution, | |
| reveal_solution, | |
| gate_solution, | |
| ) | |
| def define_graph(): | |
| """ | |
| Defines and compiles the v2 DSA Mentor StateGraph. | |
| Graph topology: | |
| classify → analyze → gate → {hint | validate | solution} | |
| ↑ | |
| hint ──┘ (loop if turn_count < 3 AND gap > 2) | |
| """ | |
| workflow = StateGraph(AgentState) | |
| # ── Register Nodes ─────────────────────────────────────────────────────── | |
| workflow.add_node("classify", classify_problem) | |
| workflow.add_node("analyze", evaluate_reasoning) | |
| workflow.add_node("gate", gate_solution) | |
| workflow.add_node("hint", generate_hint) | |
| workflow.add_node("validate", validate_solution) | |
| workflow.add_node("solution", reveal_solution) | |
| # ── Linear Edges ───────────────────────────────────────────────────────── | |
| workflow.set_entry_point("classify") | |
| workflow.add_edge("classify", "analyze") | |
| workflow.add_edge("analyze", "gate") | |
| # ── Conditional: Gate → (hint | validate | solution) ──────────────────── | |
| def route_after_gate(state: AgentState) -> str: | |
| mode = state.get("request_mode", "analyze") | |
| gap = state.get("gap_magnitude", 5) | |
| # Solution was approved by gate | |
| if mode == "solution": | |
| return "solution" | |
| # User is correct (gap <= 2) | |
| if gap <= 2: | |
| return "validate" | |
| # Default: generate a hint | |
| return "hint" | |
| workflow.add_conditional_edges( | |
| "gate", | |
| route_after_gate, | |
| {"hint": "hint", "validate": "validate", "solution": "solution"}, | |
| ) | |
| # ── Conditional: Hint → (analyze loop | END) ───────────────────────────── | |
| def route_after_hint(state: AgentState) -> str: | |
| """ | |
| Loop back to analyze if: | |
| - turn_count < 3 (still in early conversation) | |
| - gap_magnitude > 2 (user still needs more help) | |
| Otherwise end the turn and return response to frontend. | |
| """ | |
| turn_count = state.get("turn_count", 0) | |
| gap = state.get("gap_magnitude", 5) | |
| if turn_count < 3 and gap > 2: | |
| # Continue loop — re-analyze after hint is given | |
| # (In practice the frontend sends a new request with updated thought; | |
| # this loop serves intra-turn multi-step refinement) | |
| return END # Return hint to user; next request resumes loop | |
| return END | |
| # Hint always ends the turn (user needs to respond); loop is cross-request | |
| workflow.add_edge("hint", END) | |
| workflow.add_edge("validate", END) | |
| workflow.add_edge("solution", END) | |
| return workflow.compile() | |