Spaces:
Sleeping
Sleeping
File size: 13,481 Bytes
2832da8 f6a6a60 2832da8 f6a6a60 2832da8 f6a6a60 2832da8 f6a6a60 2832da8 f6a6a60 2832da8 f6a6a60 2832da8 f6a6a60 2832da8 f6a6a60 2832da8 f6a6a60 2832da8 f6a6a60 2832da8 |
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 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 |
import json
import asyncio
from typing import Dict, Any, Optional, List
from backend.utils import generate_completions
from backend import config
from backend.db import db
from backend.db_cache import api_cache
import logging
logger = logging.getLogger(__name__)
class ContentGenerator:
"""Service for generating and storing all learning content"""
async def generate_curriculum_from_metadata(
self,
metadata_extraction_id: str,
query: str,
metadata: Dict[str, Any],
user_id: Optional[int] = None
) -> str:
"""Generate curriculum based on extracted metadata"""
# Format curriculum instructions with metadata
instructions = (
config.curriculum_instructions
.replace("{native_language}", metadata['native_language'])
.replace("{target_language}", metadata['target_language'])
.replace("{proficiency}", metadata['proficiency'])
)
# Generate curriculum
logger.info(f"Generating curriculum for {metadata['target_language']} ({metadata['proficiency']})")
curriculum_response = await generate_completions.get_completions(query, instructions)
try:
# Parse curriculum response
curriculum = json.loads(curriculum_response)
except json.JSONDecodeError:
logger.error(f"Failed to parse curriculum response: {curriculum_response[:200]}...")
curriculum = {"lesson_topic": "Language Learning Journey", "sub_topics": []}
# Save curriculum to database
curriculum_id = await db.save_curriculum(
metadata_extraction_id=metadata_extraction_id,
curriculum=curriculum,
user_id=user_id
)
return curriculum_id
async def generate_content_for_lesson(
self,
curriculum_id: str,
lesson_index: int,
lesson: Dict[str, Any],
metadata: Dict[str, Any]
) -> Dict[str, str]:
"""Generate all content types for a single lesson"""
content_ids = {}
lesson_topic = lesson.get('sub_topic', f'Lesson {lesson_index + 1}')
lesson_context = f"{lesson_topic}: {lesson.get('description', '')}"
# Generate flashcards
try:
flashcards_instructions = (
config.flashcard_mode_instructions
.replace("{native_language}", metadata['native_language'])
.replace("{target_language}", metadata['target_language'])
.replace("{proficiency}", metadata['proficiency'])
)
flashcards_response = await api_cache.get_or_set(
category="flashcards",
key_text=lesson_context,
coro=generate_completions.get_completions,
context={
'native_language': metadata['native_language'],
'target_language': metadata['target_language'],
'proficiency': metadata['proficiency'],
'lesson_index': lesson_index
},
prompt=lesson_context,
instructions=flashcards_instructions
)
# Save flashcards
content_ids['flashcards'] = await db.save_learning_content(
curriculum_id=curriculum_id,
content_type='flashcards',
lesson_index=lesson_index,
lesson_topic=lesson_topic,
content=flashcards_response
)
except Exception as e:
logger.error(f"Failed to generate flashcards for lesson {lesson_index}: {e}")
# Generate exercises
try:
exercises_instructions = (
config.exercise_mode_instructions
.replace("{native_language}", metadata['native_language'])
.replace("{target_language}", metadata['target_language'])
.replace("{proficiency}", metadata['proficiency'])
)
exercises_response = await api_cache.get_or_set(
category="exercises",
key_text=lesson_context,
coro=generate_completions.get_completions,
context={
'native_language': metadata['native_language'],
'target_language': metadata['target_language'],
'proficiency': metadata['proficiency'],
'lesson_index': lesson_index
},
prompt=lesson_context,
instructions=exercises_instructions
)
# Save exercises
content_ids['exercises'] = await db.save_learning_content(
curriculum_id=curriculum_id,
content_type='exercises',
lesson_index=lesson_index,
lesson_topic=lesson_topic,
content=exercises_response
)
except Exception as e:
logger.error(f"Failed to generate exercises for lesson {lesson_index}: {e}")
# Generate simulation
try:
simulation_instructions = (
config.simulation_mode_instructions
.replace("{native_language}", metadata['native_language'])
.replace("{target_language}", metadata['target_language'])
.replace("{proficiency}", metadata['proficiency'])
)
simulation_response = await api_cache.get_or_set(
category="simulation",
key_text=lesson_context,
coro=generate_completions.get_completions,
context={
'native_language': metadata['native_language'],
'target_language': metadata['target_language'],
'proficiency': metadata['proficiency'],
'lesson_index': lesson_index
},
prompt=lesson_context,
instructions=simulation_instructions
)
# Save simulation
content_ids['simulation'] = await db.save_learning_content(
curriculum_id=curriculum_id,
content_type='simulation',
lesson_index=lesson_index,
lesson_topic=lesson_topic,
content=simulation_response
)
except Exception as e:
logger.error(f"Failed to generate simulation for lesson {lesson_index}: {e}")
return content_ids
async def generate_all_content_for_curriculum(
self,
curriculum_id: str,
max_concurrent_lessons: int = 3
):
"""Generate all learning content for a curriculum"""
try:
# Update status to generating
await db.update_content_generation_status(
curriculum_id=curriculum_id,
status='generating'
)
# Get curriculum details
curriculum_data = await db.get_curriculum(curriculum_id)
if not curriculum_data:
logger.error(f"Curriculum not found: {curriculum_id}")
await db.update_content_generation_status(
curriculum_id=curriculum_id,
status='failed',
error_message="Curriculum not found"
)
return
# Parse curriculum JSON
try:
curriculum = json.loads(curriculum_data['curriculum_json'])
lessons = curriculum.get('sub_topics', [])
except json.JSONDecodeError:
logger.error(f"Failed to parse curriculum JSON for {curriculum_id}")
await db.update_content_generation_status(
curriculum_id=curriculum_id,
status='failed',
error_message="Failed to parse curriculum JSON"
)
return
# Prepare metadata
metadata = {
'native_language': curriculum_data['native_language'],
'target_language': curriculum_data['target_language'],
'proficiency': curriculum_data['proficiency']
}
logger.info(f"Starting content generation for {len(lessons)} lessons")
# Process lessons in batches to avoid overwhelming the API
for i in range(0, len(lessons), max_concurrent_lessons):
batch = lessons[i:i + max_concurrent_lessons]
batch_indices = list(range(i, min(i + max_concurrent_lessons, len(lessons))))
# Generate content for batch concurrently
tasks = [
self.generate_content_for_lesson(
curriculum_id=curriculum_id,
lesson_index=idx,
lesson=lesson,
metadata=metadata
)
for idx, lesson in zip(batch_indices, batch)
]
results = await asyncio.gather(*tasks, return_exceptions=True)
for idx, result in zip(batch_indices, results):
if isinstance(result, Exception):
logger.error(f"Failed to generate content for lesson {idx}: {result}")
else:
logger.info(f"Generated content for lesson {idx}: {result}")
# Mark curriculum as content generated
await db.mark_curriculum_content_generated(curriculum_id)
await db.update_content_generation_status(
curriculum_id=curriculum_id,
status='completed'
)
logger.info(f"Completed content generation for curriculum {curriculum_id}")
except Exception as e:
logger.error(f"Failed to generate content for curriculum {curriculum_id}: {e}")
await db.update_content_generation_status(
curriculum_id=curriculum_id,
status='failed',
error_message=str(e)
)
async def process_metadata_extraction(
self,
extraction_id: str,
query: str,
metadata: Dict[str, Any],
user_id: Optional[int] = None,
generate_content: bool = True,
skip_curriculum_lookup: bool = False
) -> Dict[str, Any]:
"""Process a metadata extraction by checking for existing curriculum or generating new one"""
if not skip_curriculum_lookup:
# Check for existing curriculum first
existing_curriculum = await db.find_existing_curriculum(
query=query,
native_language=metadata['native_language'],
target_language=metadata['target_language'],
proficiency=metadata['proficiency'],
user_id=user_id
)
if existing_curriculum:
# If we found an exact match for this user, return it
if existing_curriculum.get('user_id') == user_id:
logger.info(f"Found existing curriculum for user {user_id}: {existing_curriculum['id']}")
return {
'curriculum_id': existing_curriculum['id'],
'content_generation_started': False,
'cached': True,
'cache_type': 'user_exact_match'
}
# If we found a similar curriculum from another user, copy it
elif existing_curriculum.get('is_content_generated') == 1:
logger.info(f"Copying existing curriculum {existing_curriculum['id']} for user {user_id}")
curriculum_id = await db.copy_curriculum_for_user(
source_curriculum_id=existing_curriculum['id'],
metadata_extraction_id=extraction_id,
user_id=user_id
)
return {
'curriculum_id': curriculum_id,
'content_generation_started': False,
'cached': True,
'cache_type': 'copied_from_similar'
}
# No suitable existing curriculum found, generate new one
logger.info(f"No existing curriculum found, generating new one for user {user_id}")
curriculum_id = await self.generate_curriculum_from_metadata(
metadata_extraction_id=extraction_id,
query=query,
metadata=metadata,
user_id=user_id
)
result = {
'curriculum_id': curriculum_id,
'content_generation_started': False,
'cached': False,
'cache_type': 'newly_generated'
}
if generate_content:
# Start content generation in background
asyncio.create_task(self.generate_all_content_for_curriculum(curriculum_id))
result['content_generation_started'] = True
return result
# Global content generator instance
content_generator = ContentGenerator() |