Spaces:
Sleeping
Sleeping
File size: 3,794 Bytes
c21ec99 ceca7a7 c21ec99 5064ff6 ceca7a7 c21ec99 | 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 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 | """
LangGraph state machine for UnMask.
Graph topology:
START
└─► supervisor
├─► [diagnostic/wrapup] socratic_generator
└─► [tutor/assessment] retrieval_planner → socratic_generator
└─► pedagogy_agent
├─► supervisor (loopback: diagnostic→tutoring)
└─► END
"""
import uuid
from typing import Literal
from langgraph.graph import StateGraph, START, END
import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver
from src.state import TutoringState
from src.agents.supervisor import supervisor_agent
from src.nodes.retrieval_planner import retrieval_planner
from src.nodes.socratic_generator import socratic_generator
from src.nodes.pedagogy_agent import pedagogy_agent
def _route_by_supervisor(state: TutoringState) -> Literal["retrieval_planner", "socratic_generator"]:
"""Conditional edge: route to retrieval for tutor/assessment, direct to generator for diagnostic/wrapup."""
agent = state.get("_last_agent", "tutor")
if agent in ("diagnostic", "wrapup"):
return "socratic_generator"
return "retrieval_planner"
def _after_pedagogy(state: TutoringState) -> Literal["supervisor", "__end__"]:
return END
def build_graph() -> StateGraph:
builder = StateGraph(TutoringState)
builder.add_node("supervisor", supervisor_agent)
builder.add_node("retrieval_planner", retrieval_planner)
builder.add_node("socratic_generator", socratic_generator)
builder.add_node("pedagogy_agent", pedagogy_agent)
builder.add_edge(START, "supervisor")
builder.add_conditional_edges(
"supervisor",
_route_by_supervisor,
{
"retrieval_planner": "retrieval_planner",
"socratic_generator": "socratic_generator",
},
)
builder.add_edge("retrieval_planner", "socratic_generator")
builder.add_edge("socratic_generator", "pedagogy_agent")
builder.add_conditional_edges(
"pedagogy_agent",
_after_pedagogy,
{
"supervisor": "supervisor",
END: END,
},
)
return builder
import os as _os
_DB_PATH = str(_os.path.join(_os.getenv("DATA_DIR", "."), "unmask_sessions.db"))
_db_conn = sqlite3.connect(_DB_PATH, check_same_thread=False)
checkpointer = SqliteSaver(_db_conn)
graph = build_graph().compile(checkpointer=checkpointer)
def make_initial_state(session_id: str | None = None) -> TutoringState:
"""Create a fresh session state."""
import yaml
with open("config.yaml") as f:
cfg = yaml.safe_load(f)
return TutoringState(
session_id=session_id or str(uuid.uuid4()),
student_message="",
turn_count=0,
phase="rapport",
elapsed_seconds=0.0,
diagnostic_complete=False,
current_topic=None,
mastery_scores={},
retrieval_mode="context_only",
retrieved_chunks=[],
generated_response="",
_internal_analysis=None,
conversation_history=[],
consecutive_correct=0,
consecutive_incorrect=0,
hints_used=0,
coverage_ratio=0.0,
weak_topics=[],
mistake_log=[],
revisit_scheduled=False,
revisit_topic=None,
_last_revisit_sec=0.0,
last_phase="rapport",
assessment_feedback=None,
visual_hint=None,
study_focus=None,
learning_mode=None,
current_diagnostic_question=None,
current_diagnostic_answer_hint=None,
vlm_image_analyzed=False,
_last_agent=None,
_supervisor_reasoning=None,
)
|