src/graph/nodes.py - working - Yan_Yu/Nereus

src/graph/nodes.py
460 строк16 KB
yan
refactor: consolidate state models, improve logging, and harden infrastructure
02 авг 2026, 20:28
02 авг 2026, 20:28b4c023c
"""Stub agent nodes for the Nereus LangGraph workflow.

Each node is a function that takes AgentState and returns a dict
with updates to the state. These are PLACEHOLDER implementations
that will be replaced with real AI agent logic in future steps.
"""

import hashlib
import logging
import random
from typing import Dict, Any

from src.graph.state import AgentState
from src.config.settings import settings
from src.database.models import BlockStatus

logger = logging.getLogger(__name__)


# =========================================================
# Stub data: Roadmaps for common skills
# =========================================================

_STUB_ROADMAPS = {
    "python": [
        {
            "title": "Основы синтаксиса и типы данных",
            "description": "Переменные, типы данных, базовые операции",
            "order_index": 0,
        },
        {
            "title": "Функции и модули",
            "description": "Определение функций, аргументы, импорты, модули",
            "order_index": 1,
        },
        {
            "title": "Работа с файлами и исключениями",
            "description": "Чтение/запись файлов, обработка исключений",
            "order_index": 2,
        },
        {
            "title": "Объектно-ориентированное программирование",
            "description": "Классы, наследование, инкапсуляция, полиморфизм",
            "order_index": 3,
        },
        {
            "title": "Работа с внешними API и данными",
            "description": "JSON, HTTP-запросы, работа с базами данных",
            "order_index": 4,
        },
    ],
    "javascript": [
        {
            "title": "Основы JavaScript и переменные",
            "description": "let, const, var, типы данных, операторы",
            "order_index": 0,
        },
        {
            "title": "Функции и область видимости",
            "description": "Function declarations, arrow functions, closures",
            "order_index": 1,
        },
        {
            "title": "DOM и события",
            "description": "Manipulation, event handling, event delegation",
            "order_index": 2,
        },
        {
            "title": "Асинхронный JavaScript",
            "description": "Callbacks, Promises, async/await, fetch API",
            "order_index": 3,
        },
    ],
    "docker": [
        {
            "title": "Основы Docker и контейнеризация",
            "description": "Что такое контейнеры, образы, Dockerfile",
            "order_index": 0,
        },
        {
            "title": "Docker Compose",
            "description": "Мульт контейнерные приложения, docker-compose.yml",
            "order_index": 1,
        },
        {
            "title": "Сети и тома в Docker",
            "description": "Networking, volumes, binds",
            "order_index": 2,
        },
        {
            "title": "Оптимизация образов и best practices",
            "description": "Multi-stage builds, минимизация образов",
            "order_index": 3,
        },
    ],
}

# Default roadmap for unknown skills
_DEFAULT_ROADMAP = [
    {
        "title": "Введение и основы",
        "description": "Базовые концепции и терминология",
        "order_index": 0,
    },
    {
        "title": "Средний уровень",
        "description": "Расширенные концепции и паттерны",
        "order_index": 1,
    },
    {
        "title": "Практическое применение",
        "description": "Реальные задачи и проекты",
        "order_index": 2,
    },
    {
        "title": "Продвинутые темы",
        "description": "Оптимизация, best practices, архитектура",
        "order_index": 3,
    },
]


def _get_roadmap_for_skill(skill: str) -> list:
    """Get a stub roadmap for the given skill."""
    skill_lower = skill.lower()
    for key, roadmap in _STUB_ROADMAPS.items():
        if key in skill_lower or skill_lower in key:
            return roadmap
    return _DEFAULT_ROADMAP


# =========================================================
# Stub materials for each block
# =========================================================

def _generate_stub_materials(block_title: str) -> list:
    """Generate stub learning materials for a block."""
    return [
        {
            "type": "text",
            "title": f"Теория: {block_title}",
            "content": f"Подробное объяснение темы '{block_title}'. "
                       f"Здесь будут теоретические материалы, "
                       f"объясняющие ключевые концепции и принципы.",
        },
        {
            "type": "exercise",
            "title": f"Практическое задание: {block_title}",
            "description": f"Выполните практическое задание по теме '{block_title}'. "
                           f"Напишите код/скрипт, демонстрирующий понимание материала.",
        },
        {
            "type": "link",
            "title": f"Дополнительные ресурсы: {block_title}",
            "url": f"https://example.com/docs/{block_title.replace(' ', '-').lower()}",
        },
    ]


# =========================================================
# Stub assessment generation
# =========================================================

def _generate_stub_assessment(block_title: str, current_level: str) -> Dict[str, Any]:
    """Generate stub assessment (theory + practice scores).

    Uses a pseudo-random approach based on current_level to make
    assessments somewhat realistic:
    - beginner: lower scores (0.3 - 0.7)
    - intermediate: medium scores (0.5 - 0.85)
    - advanced: higher scores (0.7 - 1.0)
    """
    level_scores = {
        "beginner": (0.3, 0.7),
        "intermediate": (0.5, 0.85),
        "advanced": (0.7, 1.0),
    }

    low, high = level_scores.get(current_level, (0.4, 0.8))

    # Use a seeded random for reproducibility (deterministic via hashlib)
    seed_bytes = hashlib.sha256(block_title.encode()).digest()
    seed_int = int.from_bytes(seed_bytes[:4], byteorder="big") % 10000
    rng = random.Random(seed_int)

    theory_score = round(rng.uniform(low, high), 2)
    practice_score = round(rng.uniform(low, high), 2)

    # Weighted average (practice: 60%, theory: 40%)
    overall = round(theory_score * 0.4 + practice_score * 0.6, 2)

    passing_threshold = settings.passing_score

    if overall >= passing_threshold:
        feedback = (
            f"Отличная работа по блоку '{block_title}'! "
            f"Вы успешно освоили материал. Баллы: теория {theory_score}, "
            f"практика {practice_score}, общий {overall}."
        )
    else:
        feedback = (
            f"Нужна доработка по блоку '{block_title}'. "
            f"Текущий балл {overall} ниже порогового {passing_threshold}. "
            f"Рекомендуется повторное изучение материалов и повторная проверка."
        )

    return {
        "theory_score": theory_score,
        "practice_score": practice_score,
        "overall_block_score": overall,
        "block_passed": overall >= passing_threshold,
        "assessment_feedback": feedback,
    }


# =========================================================
# Agent Nodes
# =========================================================

def coach_node(state: AgentState) -> Dict[str, Any]:
    """Coach agent: collects user context and builds the roadmap.

    This is a STUB implementation. In the future, this will use
    an LLM to analyze user input and dynamically generate a
    personalized roadmap.

    Args:
        state: Current agent state with user input.

    Returns:
        Dict with roadmap_blocks and initial state updates.
    """
    logger.info("Analyzing user context and building roadmap...")

    skill = state.get("target_skill", "general")
    roadmap = _get_roadmap_for_skill(skill)

    logger.info(
        "Skill: %s | Level: %s | Goals: %s | Hours/day: %s | Deadline: %s days",
        skill,
        state.get("current_level", "unknown"),
        state.get("goals", "N/A"),
        state.get("hours_per_day", 0),
        state.get("deadline_days", 0),
    )
    logger.info("Generated %d blocks for roadmap.", len(roadmap))

    return {
        "roadmap_blocks": roadmap,
        "current_block_index": 0,
        "block_status": BlockStatus.PENDING.value,
        "scores_history": [],
    }


def tutor_node(state: AgentState) -> Dict[str, Any]:
    """Tutor agent: provides learning materials for the current block.

    This is a STUB implementation. In the future, this will:
    - Query ChromaDB for relevant materials (RAG)
    - Generate personalized explanations via LLM
    - Track progress in the database

    Args:
        state: Current agent state.

    Returns:
        Dict with materials, current_block, and rag_results.
    """
    logger.info("Preparing learning materials for current block...")

    roadmap_blocks = state.get("roadmap_blocks", [])
    current_index = state.get("current_block_index", 0)

    if current_index >= len(roadmap_blocks):
        logger.error("Block index %d out of range.", current_index)
        return state

    current_block = roadmap_blocks[current_index]
    block_title = current_block.get("title", "Unknown Block")

    logger.info(
        "Current block: '%s' | Index: %d/%d",
        block_title,
        current_index + 1,
        len(roadmap_blocks),
    )

    # Generate stub materials
    materials = _generate_stub_materials(block_title)

    # Simulate RAG search (would query ChromaDB in production)
    rag_results = [
        {
            "id": f"rag_{i}",
            "score": round(0.7 + i * 0.1, 3),
            "text": f"Relevant context snippet for '{block_title}' - result {i + 1}",
        }
        for i in range(2)
    ]

    logger.info("Generated %d materials.", len(materials))
    logger.info("Retrieved %d RAG results.", len(rag_results))

    return {
        "current_block": block_title,
        "block_status": BlockStatus.IN_PROGRESS.value,
        "materials": materials,
        "rag_results": rag_results,
    }


def examiner_node(state: AgentState) -> Dict[str, Any]:
    """Examiner agent: assesses the user's understanding of the current block.

    This is a STUB implementation. In the future, this will:
    - Generate theory questions via LLM
    - Evaluate practical code submissions via LLM
    - Store results in the database

    Args:
        state: Current agent state.

    Returns:
        Dict with assessment scores and pass/fail decision.
    """
    logger.info("Assessing user's understanding...")

    current_block = state.get("current_block", "Unknown Block")
    current_level = state.get("current_level", "beginner")

    logger.info("Block: '%s' | User level: %s", current_block, current_level)

    # Generate stub assessment
    assessment = _generate_stub_assessment(current_block, current_level)

    logger.info(
        "Theory: %s | Practice: %s | Overall: %s | Passed: %s",
        assessment["theory_score"],
        assessment["practice_score"],
        assessment["overall_block_score"],
        assessment["block_passed"],
    )
    logger.info("Feedback: %s", assessment["assessment_feedback"])

    return {
        "theory_score": assessment["theory_score"],
        "practice_score": assessment["practice_score"],
        "overall_block_score": assessment["overall_block_score"],
        "block_passed": assessment["block_passed"],
        "assessment_feedback": assessment["assessment_feedback"],
        "scores_history": state.get("scores_history", []) + [assessment["overall_block_score"]],
    }


def deep_dive_node(state: AgentState) -> Dict[str, Any]:
    """Deep dive agent: provides additional study for weak areas.

    This is a STUB implementation. In the future, this will:
    - Analyze weak points from examiner feedback
    - Generate targeted remedial materials
    - Allow re-study and re-assessment

    Args:
        state: Current agent state.

    Returns:
        Dict with deep dive materials and reset block status.
    """
    logger.info("Preparing remedial study materials...")

    current_block = state.get("current_block", "Unknown Block")
    overall_score = state.get("overall_block_score", 0)

    logger.info(
        "Block: '%s' | Previous score: %s (below threshold)",
        current_block,
        overall_score,
    )
    logger.info("Generating additional study materials...")

    # Provide additional materials for the same block
    deep_materials = [
        {
            "type": "text",
            "title": f"Углублённое изучение: {current_block} (дополнительно)",
            "content": f"Расширенное объяснение '{current_block}' с примерами "
                       f"и пояснениями для закрепления материала.",
        },
        {
            "type": "exercise",
            "title": f"Дополнительная практика: {current_block}",
            "description": f"Ещё одно практическое задание для отработки "
                           f"навыков по теме '{current_block}'.",
        },
    ]

    # Simulate RAG results with better matches
    rag_results = [
        {
            "id": f"deep_dive_{i}",
            "score": round(0.85 + i * 0.05, 3),
            "text": f"Targeted remedial context for '{current_block}'",
        }
        for i in range(2)
    ]

    logger.info("Generated %d remedial materials.", len(deep_materials))
    logger.info("Redirecting to tutor for material review...")

    return {
        "materials": deep_materials,
        "rag_results": rag_results,
        "current_block": current_block,
        "deep_dive_needed": True,
        "deep_dive_blocks": [current_block],
        "block_status": BlockStatus.IN_PROGRESS.value,
    }


def finalize_node(state: AgentState) -> Dict[str, Any]:
    """Finalize agent: wraps up the learning process and generates a summary.

    Args:
        state: Current agent state with all scores.

    Returns:
        Dict with final_result and total_score.
    """
    logger.info("Generating learning summary...")

    scores = state.get("scores_history", [])
    user_name = state.get("user_name", "User")
    skill = state.get("target_skill", "unknown")
    roadmap_blocks = state.get("roadmap_blocks", [])

    # Calculate total average score
    total_score = round(sum(scores) / len(scores), 2) if scores else 0.0

    # Generate summary
    completed_blocks = len(scores)
    passed_blocks = sum(1 for s in scores if s >= settings.passing_score)

    summary = (
        f"Обучение по '{skill}' завершено! "
        f"Пользователь: {user_name}. "
        f"Изучено блоков: {completed_blocks}/{len(roadmap_blocks)}. "
        f"Пройдено: {passed_blocks}. "
        f"Средний балл: {total_score}."
    )

    logger.info("Total blocks: %d | Passed: %d/%d | Average: %s",
                completed_blocks, passed_blocks, completed_blocks, total_score)
    logger.info("Summary: %s", summary)

    return {
        "final_result": summary,
        "total_score": total_score,
        "block_status": BlockStatus.COMPLETED.value,
    }