Hugging Face's logo

Spaces:
aliroya32
/
ContiAI
Sleeping

App Files Community
ContiAI
File size: 3,362 Bytes
325b94c
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
import time
from typing import Dict, Any, List
from agents.content_phase import generate_quiz_agent, generate_quiz_task
from crewai import Crew, Process

# ู‡ู†ุง ุฃู†ุช ุฃุตู„ุงู‹ ู…ุฌู‡ู‘ุฒ quiz_agent ูˆ generate_quiz_task
# from your_agents_module import quiz_agent, generate_quiz_task

MAX_RETRIES = 3
RETRY_DELAY = 3  # ุซูˆุงู†ูŠ ุจูŠู† ูƒู„ ู…ุญุงูˆู„ุฉ

# ======================================================
# 1๏ธโƒฃ Clean Topics Function (ู†ูุณ ุงู„ู„ูŠ ุนู†ุฏูƒ ุชู‚ุฑูŠุจู‹ุง)
# ======================================================


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