from __future__ import annotations """ AI Courseware AI 接口 (Plan / Prepare / Reflect / Improve)。 按照《AI Courseware AI 接口文档 v1.0》提供 6 个结构化 JSON API： - /ai/courseware/syllabus/generate - /ai/courseware/flow/generate - /ai/courseware/flow/regenerate-partial - /ai/courseware/plan/detail/generate - /ai/courseware/reflection/generate - /ai/courseware/improvement/generate """ import json import time from typing import List, Optional, Any from fastapi import APIRouter, HTTPException from pydantic import BaseModel, Field from api.config import client, DEFAULT_MODEL router = APIRouter(prefix="/ai/courseware", tags=["ai-courseware"]) PROMPT_VERSION = "v1.0" class Meta(BaseModel): """所有成功响应必须携带的 meta 信息。""" model: str model_version: Optional[str] = None prompt_version: str temperature: float tokens_used: int latency_ms: int # --------------------------------------------------------------------------- # 通用 OpenAI 调用辅助 # --------------------------------------------------------------------------- def _call_openai_json( op_name: str, system_prompt: str, user_prompt: str, temperature: float = 0.4, ) -> tuple[dict, Meta]: """调用 OpenAI 并期望得到 JSON 对象；若解析失败则抛 422。""" started = time.time() try: resp = client.chat.completions.create( model=DEFAULT_MODEL, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}, ], temperature=temperature, ) except Exception as e: raise HTTPException(status_code=500, detail=f"{op_name} OpenAI error: {e}") latency_ms = int((time.time() - started) * 1000) model_name = getattr(resp, "model", DEFAULT_MODEL) usage = getattr(resp, "usage", None) tokens_used = int(getattr(usage, "total_tokens", 0) or 0) content = (resp.choices[0].message.content or "").strip() try: data = json.loads(content) if not isinstance(data, dict): raise ValueError("top-level JSON is not an object") except Exception as e: raise HTTPException( status_code=422, detail=f"INVALID_GENERATION for {op_name}: cannot parse JSON ({e}); raw={content[:400]}", ) meta = Meta( model=model_name, model_version=None, prompt_version=PROMPT_VERSION, temperature=temperature, tokens_used=tokens_used, latency_ms=latency_ms, ) return data, meta # --------------------------------------------------------------------------- # 2.1 生成大纲草案 (Generate Syllabus Preview) # --------------------------------------------------------------------------- class SyllabusContext(BaseModel): courseName: str learningOutcome: str studentLevel: str teachingFocus: str courseLength: int = Field(..., ge=1, le=52) class SyllabusGenerateRequest(BaseModel): requestId: str context: SyllabusContext class WeekSyllabus(BaseModel): weekNumber: int title: str learningObjectives: List[str] topics: List[str] class SyllabusGenerateData(BaseModel): syllabus: List[WeekSyllabus] class SyllabusGenerateResponse(BaseModel): data: SyllabusGenerateData meta: Meta @router.post("/syllabus/generate", response_model=SyllabusGenerateResponse) def generate_syllabus(req: SyllabusGenerateRequest): """Phase 1 (Plan)：根据课程元数据生成 Weeks/Modules 大纲草案。""" ctx = req.context sys_prompt = ( "You are an instructional designer. " "Given course metadata, generate a weekly syllabus as JSON. " "Output ONLY a JSON object with key 'syllabus', matching the provided schema." ) user_prompt = f""" Course name: {ctx.courseName} Learning outcome: {ctx.learningOutcome} Student level: {ctx.studentLevel} Teaching focus: {ctx.teachingFocus} Total weeks/modules: {ctx.courseLength} Return JSON: {{ "syllabus": [ {{ "weekNumber": 1, "title": "string", "learningObjectives": ["string", "..."], "topics": ["string", "..."] }}, ... ] }} Make sure the array length equals courseLength ({ctx.courseLength}) and weekNumber starts from 1. """ data_dict, meta = _call_openai_json("syllabus.generate", sys_prompt, user_prompt, temperature=0.4) if "syllabus" not in data_dict: raise HTTPException(status_code=422, detail="INVALID_GENERATION: 'syllabus' missing in JSON.") return SyllabusGenerateResponse(data=SyllabusGenerateData(**data_dict), meta=meta) # --------------------------------------------------------------------------- # 2.2 生成完整教学流程 (Generate Lesson Flow) # --------------------------------------------------------------------------- class ModuleContext(BaseModel): title: str learningObjectives: List[str] topics: List[str] durationMinutes: int = Field(..., ge=10, le=600) class FlowGenerateRequest(BaseModel): requestId: str moduleContext: ModuleContext systemPrompts: Optional[List[str]] = None class LessonStep(BaseModel): type: str title: str estimated_duration: int = Field(..., ge=1) ai_understanding: str class FlowGenerateData(BaseModel): steps: List[LessonStep] class FlowGenerateResponse(BaseModel): data: FlowGenerateData meta: Meta @router.post("/flow/generate", response_model=FlowGenerateResponse) def generate_lesson_flow(req: FlowGenerateRequest): """Phase 2 (Prepare)：为单个 Module 生成完整教学步骤列表。""" mc = req.moduleContext extra_system = "\n".join(req.systemPrompts or []) sys_prompt = ( "You are an AI courseware copilot. " "Generate a detailed lesson flow for one module as JSON. " "Output ONLY a JSON object with key 'steps', each step matching the schema." ) user_prompt = f""" Module title: {mc.title} Learning objectives: {mc.learningObjectives} Topics: {mc.topics} Duration (minutes): {mc.durationMinutes} Additional long-term system prompts (optional, may be empty): {extra_system} Return JSON: {{ "steps": [ {{ "type": "Teacher Explanation | Interactive Quiz | Group Discussion | ...", "title": "string", "estimated_duration": 10, "ai_understanding": "short explanation of why this step exists and what it should achieve" }}, ... ] }} The total sum of estimated_duration should be roughly close to durationMinutes ({mc.durationMinutes}). """ data_dict, meta = _call_openai_json("flow.generate", sys_prompt, user_prompt, temperature=0.5) if "steps" not in data_dict: raise HTTPException(status_code=422, detail="INVALID_GENERATION: 'steps' missing in JSON.") return FlowGenerateResponse(data=FlowGenerateData(**data_dict), meta=meta) # --------------------------------------------------------------------------- # 2.3 局部重构教学流程 (Regenerate Partial Flow) # --------------------------------------------------------------------------- class SimpleStep(BaseModel): id: str title: str duration: int class CurrentFlow(BaseModel): lockedSteps: List[SimpleStep] unlockedSteps: List[SimpleStep] class FlowPartialRequest(BaseModel): requestId: str prompt: str currentFlow: CurrentFlow class CopilotProposedStep(BaseModel): type: str title: str estimated_duration: int ai_understanding: str class FlowPartialData(BaseModel): explanation: str proposedSteps: List[CopilotProposedStep] class FlowPartialResponse(BaseModel): data: FlowPartialData meta: Meta @router.post("/flow/regenerate-partial", response_model=FlowPartialResponse) def regenerate_partial_flow(req: FlowPartialRequest): """Phase 2 (Prepare)：在锁定部分步骤后，根据 Prompt 重写剩余步骤。""" locked = [s.model_dump() for s in req.currentFlow.lockedSteps] unlocked = [s.model_dump() for s in req.currentFlow.unlockedSteps] sys_prompt = ( "You are an AI copilot for lesson flow editing. " "Given lockedSteps and unlockedSteps plus a user prompt, " "propose replacement steps ONLY for unlockedSteps and explain your reasoning." "Return JSON with 'explanation' and 'proposedSteps'." ) user_prompt = f""" User prompt: {req.prompt} Current flow: lockedSteps: {locked} unlockedSteps: {unlocked} Return JSON: {{ "explanation": "short natural language explanation for the teacher UI bubble", "proposedSteps": [ {{ "type": "Teacher Explanation | Interactive Quiz | Group Discussion | ...", "title": "string", "estimated_duration": 10, "ai_understanding": "short explanation of this step" }} ] }} """ data_dict, meta = _call_openai_json( "flow.regenerate-partial", sys_prompt, user_prompt, temperature=0.5 ) if "explanation" not in data_dict or "proposedSteps" not in data_dict: raise HTTPException( status_code=422, detail="INVALID_GENERATION: 'explanation' or 'proposedSteps' missing in JSON.", ) return FlowPartialResponse(data=FlowPartialData(**data_dict), meta=meta) # --------------------------------------------------------------------------- # 2.4 生成/润色实体教案长文 (Generate/Polish Lesson Plan Detail) # --------------------------------------------------------------------------- class PlanDetailRequest(BaseModel): requestId: str finalizedSteps: List[Any] class LessonSection(BaseModel): section_id: str type: str content: str class PlanDetailData(BaseModel): sections: List[LessonSection] class PlanDetailResponse(BaseModel): data: PlanDetailData meta: Meta @router.post("/plan/detail/generate", response_model=PlanDetailResponse) def generate_plan_detail(req: PlanDetailRequest): """Phase 2 (Prepare)：根据最终 Steps 生成/润色详细教案（富文本结构）。""" sys_prompt = ( "You are an expert lesson-plan writer. " "Given finalized lesson steps, generate detailed lesson plan sections as JSON. " "Each section has section_id, type, content (Markdown allowed)." ) user_prompt = f""" Finalized steps (JSON array, opaque to you but describes lesson flow): {json.dumps(req.finalizedSteps, ensure_ascii=False)} Return JSON: {{ "sections": [ {{ "section_id": "sec_obj_01", "type": "Lesson Objectives", "content": "By the end of this lesson, students will be able to..." }}, {{ "section_id": "sec_content_01", "type": "Content", "content": "### 1. Topic heading\\nDetailed explanation..." }} ] }} """ data_dict, meta = _call_openai_json("plan.detail.generate", sys_prompt, user_prompt, temperature=0.4) if "sections" not in data_dict: raise HTTPException(status_code=422, detail="INVALID_GENERATION: 'sections' missing in JSON.") return PlanDetailResponse(data=PlanDetailData(**data_dict), meta=meta) # --------------------------------------------------------------------------- # 2.5 生成反思看板数据 (Generate Reflection Report) # --------------------------------------------------------------------------- class TeachAnnotation(BaseModel): category: str selectedText: Optional[str] = None feedback: Optional[str] = None class QuizAggregations(BaseModel): averageScore: Optional[float] = None lowestTopic: Optional[str] = None class ReflectionRequest(BaseModel): requestId: str teachAnnotations: List[TeachAnnotation] = Field(default_factory=list) quizAggregations: Optional[QuizAggregations] = None class ReflectionUnderstanding(BaseModel): status: str summary: str bulletPoints: Optional[List[str]] = None class ReflectionEngagement(BaseModel): status: str summary: Optional[str] = None class ReflectionDifficulty(BaseModel): status: str challengingTopics: Optional[List[dict]] = None class ReflectionMisconception(BaseModel): status: str issues: Optional[List[dict]] = None class NextLessonSuggestion(BaseModel): actionText: str deepLinkType: Optional[str] = None class ReflectionData(BaseModel): understanding: ReflectionUnderstanding engagement: ReflectionEngagement difficulty: ReflectionDifficulty misconceptions: ReflectionMisconception nextLessonSuggestions: List[NextLessonSuggestion] class ReflectionResponse(BaseModel): data: ReflectionData meta: Meta @router.post("/reflection/generate", response_model=ReflectionResponse) def generate_reflection(req: ReflectionRequest): """Phase 4 (Reflect)：基于反馈与测验数据生成反思看板。""" annotations = [a.model_dump() for a in req.teachAnnotations] qa = req.quizAggregations.model_dump() if req.quizAggregations else {} sys_prompt = ( "You are an AI teaching analytics assistant. " "Given teacher annotations and quiz aggregations, generate a reflection dashboard as JSON " "with understanding, engagement, difficulty, misconceptions, and nextLessonSuggestions." ) user_prompt = f""" Teach annotations: {json.dumps(annotations, ensure_ascii=False)} Quiz aggregations: {json.dumps(qa, ensure_ascii=False)} Return JSON: {{ "understanding": {{ "status": "Low | Medium | High", "summary": "short text", "bulletPoints": ["optional", "..."] }}, "engagement": {{ "status": "Low | Medium | High", "summary": "optional" }}, "difficulty": {{ "status": "Low | Medium | High", "challengingTopics": [{{"topic": "string", "percentStruggle": 45}}] }}, "misconceptions": {{ "status": "Low | Medium | High", "issues": [{{"title": "string", "description": "string", "evidence": "string"}}] }}, "nextLessonSuggestions": [ {{ "actionText": "string", "deepLinkType": "PREPARE_FLOW_EDIT | REFLECT_DASHBOARD | ..." }} ] }} """ data_dict, meta = _call_openai_json("reflection.generate", sys_prompt, user_prompt, temperature=0.4) required = ["understanding", "engagement", "difficulty", "misconceptions", "nextLessonSuggestions"] if any(k not in data_dict for k in required): raise HTTPException( status_code=422, detail=f"INVALID_GENERATION: missing keys in JSON, required={required}", ) return ReflectionResponse(data=ReflectionData(**data_dict), meta=meta) # --------------------------------------------------------------------------- # 2.6 生成跨期改进提案 (Generate Improvement Proposals) # --------------------------------------------------------------------------- class ImprovementRequest(BaseModel): requestId: str reflectionReports: List[Any] class ImprovementProposal(BaseModel): title: str priority: str affectedWeeks: Optional[str] = None evidence: Optional[str] = None rootCause: Optional[str] = None proposedSolution: Optional[str] = None expectedImpact: Optional[str] = None class ImprovementData(BaseModel): proposals: List[ImprovementProposal] class ImprovementResponse(BaseModel): data: ImprovementData meta: Meta @router.post("/improvement/generate", response_model=ImprovementResponse) def generate_improvement(req: ImprovementRequest): """Phase 5 (Improve)：基于过往反思报告生成跨期改进提案。""" reports = json.dumps(req.reflectionReports, ensure_ascii=False) sys_prompt = ( "You are an AI curriculum improvement assistant. " "Given past reflection reports, generate high-level improvement proposals as JSON." ) user_prompt = f""" Past reflection reports (opaque JSON array): {reports} Return JSON: {{ "proposals": [ {{ "title": "Students struggle with gradient descent optimization", "priority": "High Priority | Medium Priority | Low Priority", "affectedWeeks": "Week 2, 3", "evidence": "45% reported difficulty...", "rootCause": "Introduced too quickly without visual math foundation.", "proposedSolution": "Split gradient descent into two lessons: (1) Intuition (2) Math.", "expectedImpact": "Increase scores by 15-20% and reduce frustration." }} ] }} """ data_dict, meta = _call_openai_json("improvement.generate", sys_prompt, user_prompt, temperature=0.4) if "proposals" not in data_dict: raise HTTPException(status_code=422, detail="INVALID_GENERATION: 'proposals' missing in JSON.") return ImprovementResponse(data=ImprovementData(**data_dict), meta=meta)