src/agents/primary_chatbot/flow.py

from langgraph.graph import StateGraph, START, END
from langgraph.graph.state import CompiledStateGraph
from langgraph.prebuilt import tools_condition

from .func import State, trim_history, entry, build_lesson_plan, change_lesson
from src.utils.helper import create_tool_node_with_fallback
from .tools import EntryExtractor, extract_lesson_content, ChangeLesson
from src.utils.logger import logger


class LessonPlanDesignAgent:
    def __init__(self):
        self.builder = StateGraph(State)
        self.lesson_plan_design_tools = [extract_lesson_content, ChangeLesson]

    @staticmethod
    def check_existed_entry_info(state: State):
        if (
            not state["class_number"]
            or not state["subject_name"]
            or not state["lesson_name"]
        ):
            return "entry"
        else:
            return "build_lesson_plan"

    @staticmethod
    def after_entry(state: State):
        if (
            not state["class_number"]
            or not state["subject_name"]
            or not state["lesson_name"]
        ):
            return END
        else:
            return "build_lesson_plan"

    @staticmethod
    def after_build_lesson_plan(state: State):
        logger.info("Build lesson plan response")
        route = tools_condition(state, "build_lesson_plan_response")
        if route == END:
            return END
        tool_calls = state["build_lesson_plan_response"][-1].tool_calls
        if tool_calls:
            if tool_calls[0]["name"] == "ChangeLesson":
                logger.info("Change Lesson")
                return "change_lesson"
            logger.info("Extract Lesson Content")
            return "lesson_plan_design_tools"
        raise ValueError("No tool calls found")

    def node(self):
        self.builder.add_node("trim_history", trim_history)
        self.builder.add_node("entry", entry)
        self.builder.add_node("build_lesson_plan", build_lesson_plan)
        self.builder.add_node("change_lesson", change_lesson)
        self.builder.add_node(
            "lesson_plan_design_tools",
            create_tool_node_with_fallback(self.lesson_plan_design_tools),
        )

    def edge(self):
        self.builder.add_edge(START, "trim_history")
        self.builder.add_conditional_edges(
            "trim_history",
            self.check_existed_entry_info,
            {
                "entry": "entry",
                "build_lesson_plan": "build_lesson_plan",
            },
        )
        self.builder.add_conditional_edges(
            "entry",
            self.after_entry,
            {
                END: END,
                "build_lesson_plan": "build_lesson_plan",
            },
        )
        self.builder.add_conditional_edges(
            "build_lesson_plan",
            self.after_build_lesson_plan,
            {
                END: END,
                "change_lesson": "change_lesson",
                "lesson_plan_design_tools": "lesson_plan_design_tools",
            },
        )
        self.builder.add_edge("change_lesson", "build_lesson_plan")
        self.builder.add_edge("lesson_plan_design_tools", "build_lesson_plan")

    def __call__(self) -> CompiledStateGraph:
        self.node()
        self.edge()
        return self.builder.compile()


lesson_plan_design_agent = LessonPlanDesignAgent()()

#