| import time |
| from typing import Dict, Any, List |
| from agents.content_phase import generate_quiz_agent, generate_quiz_task |
| from crewai import Crew, Process |
|
|
| |
| |
|
|
| MAX_RETRIES = 3 |
| RETRY_DELAY = 3 |
|
|
| |
| |
| |
|
|
|
|
| def clean_topics(topics: List[Dict[str, Any]]) -> List[Dict[str, Any]]: |
| cleaned_topics = [] |
|
|
| for topic in topics: |
| cleaned_subtopics = [] |
|
|
| for sub in topic.get("subtopics", []): |
| cleaned_subtopics.append( |
| {"title": sub.get("title"), "content": sub.get("generated_content")} |
| ) |
|
|
| cleaned_topics.append( |
| {"topic_title": topic.get("topic_title"), "subtopics": cleaned_subtopics} |
| ) |
|
|
| return cleaned_topics |
|
|
|
|
|
|
| def generate_quiz_for_course( |
| course_data: Dict[str, Any], logger=print |
| ) -> Dict[str, Any]: |
| """ |
| course_data ุดูููุง ุฒู: |
| { |
| "course_name": "...", |
| "course_audience": "...", |
| "results": [ |
| { |
| "unit_name": "...", |
| "unit_outcome": "...", |
| "topics": [...] |
| }, |
| ... |
| ] |
| } |
| """ |
| all_units_quizzes = [] |
| agent = generate_quiz_agent() |
| task = generate_quiz_task(agent) |
| crew_3 = Crew( |
| agents=[agent], |
| tasks=[task], |
| process=Process.sequential |
| ) |
|
|
|
|
| for unit in course_data["results"]: |
| unit_name = unit["unit_name"] |
| cleaned_topics = clean_topics(unit["topics"]) |
|
|
| success = False |
|
|
| for attempt in range(1, MAX_RETRIES + 1): |
| try: |
| logger(f"๐ง Unit '{unit_name}' | Attempt {attempt} started") |
|
|
| result = crew_3.kickoff( |
| inputs={ |
| "course_name": course_data["course_name"], |
| "course_audience": course_data["course_audience"], |
| "unit_name": unit_name, |
| "unit_outcome": unit["unit_outcome"], |
| "topics": cleaned_topics, |
| } |
| ) |
|
|
| if hasattr(result, "dict"): |
| all_units_quizzes.append(result.json_dict) |
| else: |
| all_units_quizzes.append(result) |
|
|
| logger(f"โ
Unit '{unit_name}' succeeded on attempt {attempt}") |
| success = True |
| break |
|
|
| except Exception as e: |
| logger( |
| f"โ Unit '{unit_name}' failed on attempt {attempt} | Error: {str(e)}" |
| ) |
| if attempt < MAX_RETRIES: |
| logger(f"๐ Retrying unit '{unit_name}' after {RETRY_DELAY}s...") |
| time.sleep(RETRY_DELAY) |
|
|
| if not success: |
| logger( |
| f"๐จ Unit '{unit_name}' FAILED after {MAX_RETRIES} attempts โ skipped." |
| ) |
|
|
| final_output = { |
| "course_name": course_data["course_name"], |
| "course_audience": course_data["course_audience"], |
| "units": all_units_quizzes, |
| } |
|
|
| return final_output |
|
|
