Spaces:
Sleeping
Sleeping
File size: 3,467 Bytes
e266561 | 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 | """
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()
|