CrewAI 实战解析：程序2考试系统与混合架构GREEN BOOK · PROGRAMMER AI CODING · 课时 06

─────────────────────────────────────────
  阶段1: 组卷
  • 传统业务：从题库随机抽题
  • 逻辑：min(题库总数, 配置数量)
─────────────────────────────────────────
  阶段2: 学生答题
  • 传统业务：收集学生输入（A/B/C/D）
─────────────────────────────────────────
  阶段3: 自动批改
  • 传统业务：对比答案，计算分数
  • 无需AI：选择题答案确定
─────────────────────────────────────────
  阶段4: AI深度分析（可选，用户询问）
  • AI决策：Manager协调2个Agent
    - Agent1: 答案解析专家
    - Agent2: 学习诊断教练
─────────────────────────────────────────

# ✗ 错误的想法：把所有阶段都交给Crew
class ExamSystem:
    def run_exam(self):
        # 创建4个Task
        task1 = create_paper_task(...)      # 组卷
        task2 = create_answer_task(...)     # 收集答案
        task3 = create_grading_task(...)    # 批改
        task4 = create_analysis_task(...)   # 分析

        # 顺序执行
        crew = Crew(
            agents=[paper_agent, answer_agent, grading_agent, analysis_agent],
            tasks=[task1, task2, task3, task4],
            process=Process.sequential
        )

        result = crew.kickoff()

class ExamSystem:
    """考试系统主程序"""

    def __init__(self):
        self.config = load_config()
        self.db = QuestionDatabase()
        self.console = Console()

        # 只在需要AI时才初始化Crew
        self.llm = create_llm()
        self.crew = ExerciseCrew(self.llm)

    def run(self):
        """完整考试流程"""
        # 阶段1-3：传统代码（不用Crew）
        self.create_exam_paper()      # 传统：随机抽题
        self.collect_answers()        # 传统：终端输入
        grading_result = self.grade_answers()  # 传统：对比答案

        # 显示分数
        self._display_grading_result(grading_result)

        # 询问是否需要AI分析
        need_analysis = Confirm.ask("是否需要AI进行答案解析和学习建议？")

        if need_analysis:
            # 阶段4：启用Crew（Hierarchical模式）
            explanation, advice = self.crew.run_analysis(grading_result)
            self.console.print(f"答案解析：\n{explanation}")
            self.console.print(f"学习建议：\n{advice}")

class ExamSystem:
    def create_exam_paper(self) -> bool:
        """阶段1：组卷 - 核心算法：random.sample()"""
        exam_count = self.config.exam_count
        total_count = self.db.get_question_count()

        if total_count < exam_count:
            return False

        # 随机抽题
        self.exam_questions = self.db.random_select_questions(exam_count)
        return True

    def collect_answers(self):
        """阶段2：收集答案 - 核心算法：输入验证"""
        self.student_answers = []

        for i, question in enumerate(self.exam_questions, 1):
            self._display_question(i, question)

            # 验证输入
            while True:
                answer = Prompt.ask("请输入答案（A/B/C/D）").upper()
                if answer in ['A', 'B', 'C', 'D']:
                    self.student_answers.append(answer)
                    break

    def grade_answers(self) -> dict:
        """阶段3：批改 - 核心算法：字符串比较"""
        total_score = 0
        details = []

        # 逐题对比
        for question, student_answer in zip(self.exam_questions, self.student_answers):
            is_correct = (student_answer == question.answer)
            score = 10 if is_correct else 0
            total_score += score

            details.append({
                'is_correct': is_correct,
                'score': score
            })

        return {'total_score': total_score, 'details': details}

# crews/exercise_crew.py

class ExerciseCrew:
    def run_analysis(self, grading_result: dict):
        """
        执行AI分析（Hierarchical模式）
        只在这里调用LLM！
        """
        # 创建2个Agent
        explanation_agent = create_explanation_agent(self.llm)
        advisor_agent = create_learning_advisor(self.llm)

        # 创建2个Task（可以并行）
        explanation_task = create_explanation_task(explanation_agent, grading_result)
        advice_task = create_learning_advice_task(advisor_agent, grading_result)

        # Hierarchical模式：Manager协调2个Agent并行工作
        crew = Crew(
            agents=[explanation_agent, advisor_agent],
            tasks=[explanation_task, advice_task],
            process=Process.hierarchical,  # 并行执行
            manager_llm=self.llm,          # Manager使用的LLM
            verbose=True
        )

        result = crew.kickoff()

        # 返回两个Agent的输出
        return result.explanation, result.advice

# crews/exercise_crew.py

class ExerciseCrew:
    def run_analysis(self, grading_result: dict):
        """执行AI分析（Hierarchical模式）"""
        # 创建2个专家Agent
        explanation_agent = create_explanation_agent(self.llm)
        advisor_agent = create_learning_advisor(self.llm)

        # 创建2个Task
        explanation_task = create_explanation_task(explanation_agent, grading_result)
        advice_task = create_learning_advice_task(advisor_agent, grading_result)

        # Hierarchical模式：Manager协调
        crew = Crew(
            agents=[explanation_agent, advisor_agent],
            tasks=[explanation_task, advice_task],
            process=Process.hierarchical,  # 关键：启用Hierarchical
            manager_llm=self.llm,          # Manager使用的LLM
            verbose=True
        )

        result = crew.kickoff()
        return result.explanation, result.advice