Spaces:
Sleeping
Sleeping
File size: 5,409 Bytes
c0a7f25 0f1954e c0a7f25 0f1954e c0a7f25 |
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 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 |
from fastapi import APIRouter, status, Depends, BackgroundTasks, HTTPException
from fastapi.responses import JSONResponse
from src.utils.logger import logger
from pydantic import BaseModel, Field
from typing import List, Dict, Any, Optional
from src.agents.lesson_practice.flow import lesson_practice_agent
from src.apis.models.lesson_models import Lesson, LessonResponse, LessonDetailResponse
import json
import os
import uuid
from datetime import datetime
router = APIRouter(prefix="/lesson", tags=["AI"])
class LessonPracticeRequest(BaseModel):
unit: str = Field(..., description="Unit of the lesson")
vocabulary: list = Field(..., description="Vocabulary for the lesson")
key_structures: list = Field(..., description="Key structures for the lesson")
practice_questions: list = Field(
..., description="Practice questions for the lesson"
)
student_level: str = Field("beginner", description="Student's level of English")
query: str = Field(..., description="User query for the lesson")
session_id: str = Field(..., description="Session ID for the lesson")
# Helper function to load lessons from JSON file
def load_lessons_from_file() -> List[Lesson]:
"""Load lessons from the JSON file"""
try:
lessons_file_path = os.path.join(os.path.dirname(__file__), "..", "..", "data", "lessons.json")
if not os.path.exists(lessons_file_path):
logger.warning(f"Lessons file not found at {lessons_file_path}")
return []
with open(lessons_file_path, 'r', encoding='utf-8') as file:
lessons_data = json.load(file)
# Convert to Lesson objects
lessons = []
for lesson_data in lessons_data:
try:
lesson = Lesson(**lesson_data)
lessons.append(lesson)
except Exception as e:
logger.error(f"Error parsing lesson {lesson_data.get('id', 'unknown')}: {str(e)}")
continue
return lessons
except Exception as e:
logger.error(f"Error loading lessons: {str(e)}")
return []
@router.get("/all", response_model=LessonResponse)
async def get_all_lessons():
"""
Get all available lessons
Returns:
LessonResponse: Contains list of all lessons and total count
"""
try:
lessons = load_lessons_from_file()
return LessonResponse(
lessons=lessons,
total=len(lessons)
)
except Exception as e:
logger.error(f"Error retrieving lessons: {str(e)}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Failed to retrieve lessons"
)
@router.get("/{lesson_id}", response_model=LessonDetailResponse)
async def get_lesson_by_id(lesson_id: str):
"""
Get a specific lesson by ID
Args:
lesson_id (str): The unique identifier of the lesson
Returns:
LessonDetailResponse: Contains the lesson details
"""
try:
lessons = load_lessons_from_file()
# Find the lesson with the specified ID
lesson = next((l for l in lessons if l.id == lesson_id), None)
if not lesson:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Lesson with ID '{lesson_id}' not found"
)
return LessonDetailResponse(lesson=lesson)
except HTTPException:
raise
except Exception as e:
logger.error(f"Error retrieving lesson {lesson_id}: {str(e)}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Failed to retrieve lesson"
)
@router.get("/search/unit/{unit_name}")
async def search_lessons_by_unit(unit_name: str):
"""
Search lessons by unit name (case-insensitive partial match)
Args:
unit_name (str): Part of the unit name to search for
Returns:
LessonResponse: Contains list of matching lessons
"""
try:
lessons = load_lessons_from_file()
# Filter lessons by unit name (case-insensitive partial match)
matching_lessons = [
lesson for lesson in lessons
if unit_name.lower() in lesson.unit.lower()
]
return LessonResponse(
lessons=matching_lessons,
total=len(matching_lessons)
)
except Exception as e:
logger.error(f"Error searching lessons by unit '{unit_name}': {str(e)}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Failed to search lessons"
)
@router.post("/chat")
async def chat(request: LessonPracticeRequest):
response = await lesson_practice_agent().ainvoke(
{
"unit": request.unit,
"vocabulary": request.vocabulary,
"key_structures": request.key_structures,
"practice_questions": request.practice_questions,
"student_level": request.student_level,
"messages": [request.query],
},
{"configurable": {"thread_id": request.session_id}},
)
return JSONResponse(
content=response["messages"][-1].content, status_code=status.HTTP_200_OK
)
|