# ClareVoice FastAPI server: React UI + same backend as app.py (Weaviate + FAISS). # Run: uvicorn server:app --host 0.0.0.0 --port 7860 import os import re import time import concurrent.futures from collections import defaultdict from typing import Dict, List, Any, Optional from fastapi import FastAPI, UploadFile, File, Form, Request from fastapi.responses import FileResponse, JSONResponse, Response from fastapi.staticfiles import StaticFiles from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel from config import ( DEFAULT_MODEL, DEFAULT_COURSE_TOPICS, USE_WEAVIATE_DIRECT, GENAI_COURSES_SPACE, WEAVIATE_URL, WEAVIATE_API_KEY, WEAVIATE_COLLECTION, WEAVIATE_EMBEDDING, EMBEDDING_MODEL, ) from clare_core import ( detect_language, chat_with_clare, update_weaknesses_from_message, update_cognitive_state_from_message, generate_hint_for_question, grade_quiz_answers, render_session_status, export_conversation, summarize_conversation, generate_quiz_for_external, ) from rag_engine import build_rag_chunks_from_file, retrieve_relevant_chunks from syllabus_utils import extract_course_topics_from_file from tts_podcast import ( text_to_speech, build_podcast_script_from_history, build_podcast_script_from_summary, generate_podcast_audio, ) MODULE10_PATH = os.path.join(os.path.dirname(__file__), "module10_responsible_ai.pdf") MODULE10_DOC_TYPE = "Literature Review / Paper" # Preload Module 10 (same as app.py) preloaded_topics: List[str] = [] preloaded_chunks: List[Dict] = [] if os.path.exists(MODULE10_PATH): try: class _FileObj: name = MODULE10_PATH preloaded_topics = extract_course_topics_from_file(_FileObj(), MODULE10_DOC_TYPE) or [] preloaded_chunks = build_rag_chunks_from_file(MODULE10_PATH, MODULE10_DOC_TYPE) or [] print("[server] Module 10 preloaded.") except Exception as e: print(f"[server] Module 10 preload failed: {e}") if not preloaded_topics: preloaded_topics = list(DEFAULT_COURSE_TOPICS) _WEAVIATE_EMBED_MODEL = None def _get_weaviate_embed_model(): """与建索引一致：WEAVIATE_EMBEDDING=openai 时用 OpenAI，否则用 HuggingFace。""" global _WEAVIATE_EMBED_MODEL if _WEAVIATE_EMBED_MODEL is None: if WEAVIATE_EMBEDDING == "openai": from llama_index.embeddings.openai import OpenAIEmbedding _WEAVIATE_EMBED_MODEL = OpenAIEmbedding(model=EMBEDDING_MODEL) else: from llama_index.embeddings.huggingface import HuggingFaceEmbedding _WEAVIATE_EMBED_MODEL = HuggingFaceEmbedding( model_name="sentence-transformers/all-MiniLM-L6-v2" ) return _WEAVIATE_EMBED_MODEL def _retrieve_from_weaviate(question: str, top_k: int = 5, timeout_sec: float = 45.0) -> str: if not USE_WEAVIATE_DIRECT or len(question.strip()) < 5: return "" def _call(): try: import weaviate from weaviate.classes.init import Auth from llama_index.core import Settings, VectorStoreIndex from llama_index.vector_stores.weaviate import WeaviateVectorStore Settings.embed_model = _get_weaviate_embed_model() client = weaviate.connect_to_weaviate_cloud( cluster_url=WEAVIATE_URL, auth_credentials=Auth.api_key(WEAVIATE_API_KEY), ) try: if not client.is_ready(): return "" vs = WeaviateVectorStore(weaviate_client=client, index_name=WEAVIATE_COLLECTION) index = VectorStoreIndex.from_vector_store(vs) nodes = index.as_retriever(similarity_top_k=top_k).retrieve(question) return "\n\n---\n\n".join(n.get_content() for n in nodes) if nodes else "" finally: client.close() except Exception as e: print(f"[weaviate] retrieve failed: {repr(e)}") return "" try: with concurrent.futures.ThreadPoolExecutor(max_workers=1) as ex: return ex.submit(_call).result(timeout=timeout_sec) except concurrent.futures.TimeoutError: print(f"[weaviate] timeout after {timeout_sec}s") return "" def _retrieve_from_genai_courses(question: str, top_k: int = 5, timeout_sec: float = 25.0) -> str: if not GENAI_COURSES_SPACE or len(question.strip()) < 5: return "" def _call(): try: from gradio_client import Client c = Client(GENAI_COURSES_SPACE) return (c.predict(question, api_name="/retrieve") or "").strip() except Exception as e: print(f"[genai_courses] failed: {repr(e)}") return "" try: with concurrent.futures.ThreadPoolExecutor(max_workers=1) as ex: return ex.submit(_call).result(timeout=timeout_sec) except concurrent.futures.TimeoutError: return "" def format_references(rag_chunks: List[Dict], max_files: int = 2, max_sections_per_file: int = 3) -> str: if not rag_chunks: return "\n".join(["**References:**", "- (No RAG context used. Answer is based on the model's general knowledge.)"]) chunks = list(rag_chunks) chunks.sort(key=lambda c: float(c.get("_rag_score", 0.0)), reverse=True) refs_by_file: Dict[str, List[str]] = defaultdict(list) for chunk in chunks: file_name = chunk.get("source_file") or "module10_responsible_ai.pdf" section = chunk.get("section") or "Related section" score = chunk.get("_rag_score") score_str = f" (score={float(score):.2f})" if score is not None else "" entry = section + score_str if entry not in refs_by_file[file_name]: refs_by_file[file_name].append(entry) if not refs_by_file: return "\n".join(["**References:**", "- (No RAG context used.)"]) lines = ["**References (RAG context used):**"] for i, (file_name, sections) in enumerate(refs_by_file.items()): if i >= max_files: break lines.append(f"- *{file_name}* — {'; '.join(sections[:max_sections_per_file])}") return "\n".join(lines) def is_academic_query(message: str) -> bool: if not message or not message.strip(): return False m = " ".join(message.strip().lower().split()) smalltalk = {"hi", "hello", "hey", "thanks", "thank", "ok", "okay", "bye", "goodbye", "haha", "lol"} tokens = m.split() if "?" not in m and all(t in smalltalk for t in tokens): return False meta = ["who are you", "what are you", "what is your name", "what can you do", "what is clare"] if any(p in m for p in meta): return False if len(tokens) <= 2 and "?" not in m: return False return True MODULE10_DOC_TYPE = "Literature Review / Paper" MICRO_QUIZ_INSTRUCTION = ( "We are running a short micro-quiz session based ONLY on **Module 10 – " "Responsible AI (Alto, 2024, Chapter 12)** and the pre-loaded materials.\n\n" "Step 1 – Before asking any content question:\n" "• First ask me which quiz style I prefer right now:\n" " - (1) Multiple-choice questions\n" " - (2) Short-answer / open-ended questions\n" "• Ask me explicitly: \"Which quiz style do you prefer now: 1) Multiple-choice or 2) Short-answer? " "Please reply with 1 or 2.\"\n" "• Do NOT start a content question until I have answered 1 or 2.\n\n" "Step 2 – After I choose the style:\n" "• If I choose 1 (multiple-choice):\n" " - Ask ONE multiple-choice question at a time, based on Module 10 concepts.\n" " - Provide 3–4 options (A, B, C, D) and make only one option clearly correct.\n" "• If I choose 2 (short-answer):\n" " - Ask ONE short-answer question at a time, also based on Module 10 concepts.\n" " - Do NOT show the answer when you ask the question.\n\n" "Step 3 – For each answer I give:\n" "• Grade my answer (correct / partially correct / incorrect).\n" "• Give a brief explanation and the correct answer.\n" "• Then ask if I want another question of the SAME style.\n\n" "Please start by asking me which quiz style I prefer (1 = multiple-choice, 2 = short-answer). " "Do not ask any content question before I choose." ) # ---------------------------- # Session store (in-memory) # ---------------------------- SESSIONS: Dict[str, Dict[str, Any]] = {} def _get_session(user_id: str) -> Dict[str, Any]: if user_id not in SESSIONS: SESSIONS[user_id] = { "user_id": user_id, "name": "", "history": [], "weaknesses": [], "cognitive_state": {"confusion": 0, "mastery": 0}, "course_outline": list(preloaded_topics) if preloaded_topics else list(DEFAULT_COURSE_TOPICS), "rag_chunks": list(preloaded_chunks) if preloaded_chunks else [], "model_name": DEFAULT_MODEL, "uploaded_files": [], "profile_bio": "", "init_answers": {}, "init_dismiss_until": 0, } return SESSIONS[user_id] # ---------------------------- # App # ---------------------------- app = FastAPI(title="ClareVoice API") app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) WEB_DIR = os.path.join(os.path.dirname(__file__), "web", "build") WEB_INDEX = os.path.join(WEB_DIR, "index.html") WEB_ASSETS = os.path.join(WEB_DIR, "assets") if os.path.isdir(WEB_ASSETS): app.mount("/assets", StaticFiles(directory=WEB_ASSETS), name="assets") if os.path.isdir(WEB_DIR): app.mount("/static", StaticFiles(directory=WEB_DIR), name="static") # ---------------------------- # Request models # ---------------------------- class LoginReq(BaseModel): name: str user_id: str class ChatReq(BaseModel): user_id: str message: str learning_mode: str = "Concept Explainer" language_preference: str = "Auto" doc_type: str = "Syllabus" class QuizStartReq(BaseModel): user_id: str language_preference: str = "Auto" doc_type: str = MODULE10_DOC_TYPE learning_mode: str = "quiz" class ExportReq(BaseModel): user_id: str learning_mode: str class SummaryReq(BaseModel): user_id: str learning_mode: str language_preference: str = "Auto" class TtsReq(BaseModel): user_id: str text: str voice: Optional[str] = "nova" class PodcastReq(BaseModel): user_id: str source: str = "summary" # "summary" | "conversation" voice: Optional[str] = "nova" title: Optional[str] = None # 自定义播客标题，用于开场白 class QuizGenerateContext(BaseModel): """教学上下文信息（用于 Generate，完整格式）""" courseId: Optional[int] = None moduleId: Optional[int] = None topics: Optional[List[str]] = None class QuizConfigurations(BaseModel): """生成配置（完整格式）""" questionCount: Optional[int] = None questionTypes: Optional[List[str]] = None language: Optional[str] = None # EN/CN 或 en/zh class QuizGenerateReq(BaseModel): """供外部网站调用的 Quiz 生成请求（支持简化格式和完整格式）""" # 简化格式字段 topic: Optional[str] = None num_questions: Optional[int] = None language: Optional[str] = None # en | zh # 完整格式字段（向后兼容） requestId: Optional[str] = None context: Optional[QuizGenerateContext] = None configurations: Optional[QuizConfigurations] = None class QuestionOption(BaseModel): """题目选项""" label: Optional[str] = None content: Optional[str] = None key: Optional[str] = None # 兼容简化格式 text: Optional[str] = None # 兼容简化格式 class QuestionContext(BaseModel): """题目上下文（用于 Hint）""" content: str options: Optional[List[QuestionOption]] = None class HintRequest(BaseModel): """Hint 请求""" requestId: Optional[str] = None questionContext: QuestionContext type: str = "HINT" language: Optional[str] = None # en | zh class CorePoint(BaseModel): """Rubric 核心要点""" point: str weight: float = 0.0 class CommonError(BaseModel): """Rubric 常见错误及扣分""" description: str deduction: float = 0.0 class AnswerScope(BaseModel): """Rubric 允许的回答范围""" minLength: Optional[int] = None maxLength: Optional[int] = None language: Optional[str] = None requireExample: Optional[bool] = None class QuestionRubric(BaseModel): """简答题题目 Rubric（动态传入）""" corePoints: Optional[List[dict]] = None # [{ "point": str, "weight": number }] acceptableSynonyms: Optional[List[str]] = None commonErrors: Optional[List[dict]] = None # [{ "description": str, "deduction": number }] answerScope: Optional[AnswerScope] = None class UserAnswer(BaseModel): """用户答案""" questionContent: Optional[str] = None questionId: Optional[str] = None userChoiceLabel: Optional[str] = None correctChoiceLabel: Optional[str] = None isCorrect: Optional[bool] = None userTextAnswer: Optional[str] = None referenceAnswer: Optional[str] = None rubric: Optional[QuestionRubric] = None # 简答题本题评分标准 class QuizContext(BaseModel): """测验上下文（用于 Grade）""" title: Optional[str] = None totalScore: Optional[float] = None maxScore: Optional[float] = None class GradeRequest(BaseModel): """判卷请求""" requestId: Optional[str] = None quizContext: Optional[QuizContext] = None userAnswers: List[UserAnswer] language: Optional[str] = None # en | zh class FeedbackReq(BaseModel): user_id: str rating: str run_id: Optional[str] = None assistant_message_id: Optional[str] = None assistant_text: str = "" user_text: Optional[str] = None comment: Optional[str] = None refs: Optional[List] = None tags: Optional[List] = None timestamp_ms: Optional[int] = None learning_mode: Optional[str] = None doc_type: Optional[str] = None # ---------------------------- # Routes # ---------------------------- @app.get("/") def index(): if os.path.exists(WEB_INDEX): return FileResponse(WEB_INDEX) return JSONResponse({"detail": "web/build not found. Build frontend first."}, status_code=500) @app.post("/api/login") def login(req: LoginReq): user_id = (req.user_id or "").strip() name = (req.name or "").strip() if not user_id or not name: return JSONResponse({"ok": False, "error": "Missing name/user_id"}, status_code=400) sess = _get_session(user_id) sess["name"] = name return {"ok": True, "user": {"name": name, "user_id": user_id}} @app.post("/api/chat") def chat(req: ChatReq): user_id = (req.user_id or "").strip() msg = (req.message or "").strip() if not user_id: return JSONResponse({"error": "Missing user_id"}, status_code=400) sess = _get_session(user_id) if not msg: return { "reply": "", "session_status_md": render_session_status( req.learning_mode, sess["weaknesses"], sess["cognitive_state"] ), "refs": [], "latency_ms": 0.0, "run_id": None, } t0 = time.time() resolved_lang = detect_language(msg, req.language_preference) sess["weaknesses"] = update_weaknesses_from_message(msg, sess["weaknesses"]) sess["cognitive_state"] = update_cognitive_state_from_message(msg, sess["cognitive_state"]) rag_context_text = "" rag_used_chunks: List[Dict] = [] if is_academic_query(msg): rag_context_text, rag_used_chunks = retrieve_relevant_chunks(msg, sess["rag_chunks"] or []) course_chunks = "" course_source = "" if USE_WEAVIATE_DIRECT: course_chunks = _retrieve_from_weaviate(msg) course_source = "Weaviate Cloud (GENAI COURSES)" elif GENAI_COURSES_SPACE: course_chunks = _retrieve_from_genai_courses(msg) course_source = "GenAICoursesDB" if course_chunks and course_source: rag_context_text = (rag_context_text or "") + "\n\n[来自 GENAI 课程知识库]\n\n" + course_chunks rag_used_chunks = list(rag_used_chunks or []) + [ {"source_file": course_source, "section": "retrieve (GENAI COURSES dataset)", "_rag_score": 1.0} ] try: answer, new_history = chat_with_clare( message=msg, history=sess["history"], model_name=sess["model_name"], language_preference=resolved_lang, learning_mode=req.learning_mode, doc_type=req.doc_type, course_outline=sess["course_outline"], weaknesses=sess["weaknesses"], cognitive_state=sess["cognitive_state"], rag_context=rag_context_text, ) except Exception as e: print(f"[chat] error: {repr(e)}") return JSONResponse({"error": f"chat failed: {repr(e)}"}, status_code=500) sess["history"] = new_history total_ms = (time.time() - t0) * 1000.0 ref_text = format_references(rag_used_chunks) if is_academic_query(msg) else "" if ref_text and new_history: last_u, last_a = new_history[-1] if "References (RAG context used):" not in (last_a or ""): answer = f"{last_a or ''}\n\n{ref_text}" refs = [{"source_file": c.get("source_file"), "section": c.get("section")} for c in (rag_used_chunks or [])] if not refs: refs = [{"source_file": "No RAG", "section": "Answer based on model general knowledge."}] return { "reply": answer, "session_status_md": render_session_status(req.learning_mode, sess["weaknesses"], sess["cognitive_state"]), "refs": refs, "latency_ms": total_ms, "run_id": None, } @app.post("/api/quiz/start") def quiz_start(req: QuizStartReq): user_id = (req.user_id or "").strip() if not user_id: return JSONResponse({"error": "Missing user_id"}, status_code=400) sess = _get_session(user_id) resolved_lang = detect_language(MICRO_QUIZ_INSTRUCTION, req.language_preference) quiz_ctx_text, _ = retrieve_relevant_chunks("Module 10 quiz", sess["rag_chunks"] or []) try: answer, new_history = chat_with_clare( message=MICRO_QUIZ_INSTRUCTION, history=sess["history"], model_name=sess["model_name"], language_preference=resolved_lang, learning_mode=req.learning_mode, doc_type=req.doc_type, course_outline=sess["course_outline"], weaknesses=sess["weaknesses"], cognitive_state=sess["cognitive_state"], rag_context=quiz_ctx_text, ) except Exception as e: print(f"[quiz] error: {repr(e)}") return JSONResponse({"error": str(e)}, status_code=500) sess["history"] = new_history return { "reply": answer, "session_status_md": render_session_status(req.learning_mode, sess["weaknesses"], sess["cognitive_state"]), "refs": [], "latency_ms": 0.0, "run_id": None, } @app.post("/api/upload") async def upload( user_id: str = Form(...), doc_type: str = Form(...), file: UploadFile = File(...), ): user_id = (user_id or "").strip() doc_type = (doc_type or "").strip() if not user_id: return JSONResponse({"ok": False, "error": "Missing user_id"}, status_code=400) if not file or not file.filename: return JSONResponse({"ok": False, "error": "Missing file"}, status_code=400) sess = _get_session(user_id) safe_name = os.path.basename(file.filename).replace("..", "_") tmp_path = os.path.join("/tmp", safe_name) content = await file.read() with open(tmp_path, "wb") as f: f.write(content) if doc_type == "Syllabus": class _F: pass fo = _F() fo.name = tmp_path try: sess["course_outline"] = extract_course_topics_from_file(fo, doc_type) except Exception as e: print(f"[upload] syllabus parse error: {repr(e)}") try: new_chunks = build_rag_chunks_from_file(tmp_path, doc_type) or [] sess["rag_chunks"] = (sess["rag_chunks"] or []) + new_chunks except Exception as e: print(f"[upload] rag build error: {repr(e)}") new_chunks = [] sess.setdefault("uploaded_files", []).append({ "filename": safe_name, "doc_type": doc_type, "added_chunks": len(new_chunks), "ts": int(time.time()), }) return {"ok": True, "added_chunks": len(new_chunks), "status_md": f"✅ Loaded base reading + uploaded {doc_type} file."} @app.post("/api/feedback") def api_feedback(req: FeedbackReq): user_id = (req.user_id or "").strip() if not user_id: return JSONResponse({"ok": False, "error": "Missing user_id"}, status_code=400) if (req.rating or "").strip().lower() not in ("helpful", "not_helpful"): return JSONResponse({"ok": False, "error": "Invalid rating"}, status_code=400) return {"ok": True} @app.post("/api/export") def api_export(req: ExportReq): user_id = (req.user_id or "").strip() if not user_id: return JSONResponse({"error": "Missing user_id"}, status_code=400) sess = _get_session(user_id) md = export_conversation( sess["history"], sess["course_outline"], req.learning_mode, sess["weaknesses"], sess["cognitive_state"], ) return {"markdown": md} @app.post("/api/summary") def api_summary(req: SummaryReq): user_id = (req.user_id or "").strip() if not user_id: return JSONResponse({"error": "Missing user_id"}, status_code=400) sess = _get_session(user_id) lang = (req.language_preference or "Auto").strip() if lang == "简体中文": lang = "中文" md = summarize_conversation( sess["history"], sess["course_outline"], sess["weaknesses"], sess["cognitive_state"], sess["model_name"], lang, ) return {"markdown": md} @app.post("/api/tts") def api_tts(req: TtsReq): user_id = (req.user_id or "").strip() if not user_id: return JSONResponse({"error": "Missing user_id"}, status_code=400) text = (req.text or "").strip() if not text: return JSONResponse({"error": "Missing text"}, status_code=400) try: audio_bytes = text_to_speech(text, voice=req.voice or "nova") except Exception as e: print(f"[tts] error: {repr(e)}") return JSONResponse({"error": str(e)}, status_code=500) if not audio_bytes: return JSONResponse({"error": "No audio generated"}, status_code=500) return Response(content=audio_bytes, media_type="audio/mpeg") @app.post("/api/podcast") def api_podcast(req: PodcastReq): user_id = (req.user_id or "").strip() if not user_id: return JSONResponse({"error": "Missing user_id"}, status_code=400) sess = _get_session(user_id) source = (req.source or "summary").lower() voice = req.voice or "nova" title = (req.title or "").strip() or None try: if source == "conversation": script = build_podcast_script_from_history( sess["history"], intro_title=title or "Clare Learning Summary", ) else: md = summarize_conversation( sess["history"], sess["course_outline"], sess["weaknesses"], sess["cognitive_state"], sess["model_name"], "Auto", ) script = build_podcast_script_from_summary( md, intro_title=title or "Clare Summary Podcast", ) audio_bytes = generate_podcast_audio(script, voice=voice) except Exception as e: print(f"[podcast] error: {repr(e)}") return JSONResponse({"error": str(e)}, status_code=500) if not audio_bytes: return JSONResponse({"error": "No audio generated"}, status_code=500) return Response(content=audio_bytes, media_type="audio/mpeg") # 可选：设置 QUIZ_API_KEY 后，外部调用 /api/quiz/generate 需在 Header 带 X-API-Key QUIZ_API_KEY = (os.getenv("QUIZ_API_KEY") or "").strip() @app.post("/api/quiz/generate") def quiz_generate(req: QuizGenerateReq, request: Request): """供外部网站调用的 AI Quiz 生成接口（无会话，无需 user_id）。 支持两种格式： 1. 简化格式：{"topic": "...", "num_questions": 3, "language": "en"} 2. 完整格式：{"requestId": "...", "context": {...}, "configurations": {...}} """ if QUIZ_API_KEY: key = request.headers.get("X-API-Key") or request.headers.get("Authorization", "").replace("Bearer ", "") if (key or "").strip() != QUIZ_API_KEY: return JSONResponse( status_code=429, content={"code": 429, "error": {"type": "RATE_LIMIT", "reason": "missing_or_invalid_api_key"}}, ) # 解析请求参数（支持两种格式） topic = None num_questions = 3 language = "en" use_full_format = False # 检查是否使用完整格式 if req.context is not None or req.configurations is not None: use_full_format = True # 从完整格式中提取参数 if req.configurations: if req.configurations.questionCount is not None: num_questions = req.configurations.questionCount if req.configurations.language: lang = req.configurations.language.strip().upper() language = "zh" if lang in ("CN", "中文", "ZH") else "en" if req.context and req.context.topics: topic = ", ".join(req.context.topics) elif req.topic: topic = req.topic else: # 使用简化格式 if req.topic: topic = req.topic if req.num_questions is not None: num_questions = req.num_questions if req.language: language = req.language topic = (topic or "").strip() if not topic: return JSONResponse( status_code=422, content={"code": 422, "error": {"type": "INVALID_GENERATION", "reason": "topic_required"}}, ) t0 = time.time() try: questions, tokens_used = generate_quiz_for_external( topic=topic, num_questions=num_questions, language=language, ) except ValueError as e: return JSONResponse( status_code=422, content={"code": 422, "error": {"type": "INVALID_GENERATION", "reason": str(e).replace(" ", "_")}}, ) except Exception as e: print(f"[quiz_generate] error: {repr(e)}") return JSONResponse( status_code=500, content={"code": 500, "error": {"type": "MODEL_ERROR", "reason": "generation_failed"}}, ) latency_ms = (time.time() - t0) * 1000.0 meta = { "model": DEFAULT_MODEL, "model_version": "", "prompt_version": "quiz_generate_v1", "temperature": 0.4, "tokens_used": tokens_used, "latency_ms": round(latency_ms, 2), } # 转换题目格式以匹配接口文档 formatted_questions = [] for q in questions: formatted_q = { "question_id": q.get("question_id", ""), "type": q.get("type", "SINGLE_CHOICE"), "content": q.get("question_text", q.get("content", "")), # 支持两种字段名 "correct_answers": q.get("correct_answers", []), } # 转换 options 格式 if "options" in q and q["options"]: formatted_q["options"] = [] for opt in q["options"]: # 支持两种格式：{key, text} 或 {label, content} formatted_q["options"].append({ "label": opt.get("key", opt.get("label", "")), "content": opt.get("text", opt.get("content", "")), }) if "explanation" in q: formatted_q["explanation"] = q["explanation"] formatted_questions.append(formatted_q) # 根据请求格式返回相应格式的响应 if use_full_format: # 完整格式：使用 data 包装 return { "data": { "questions": formatted_questions, }, "meta": meta, } else: # 简化格式：直接返回 questions return { "questions": formatted_questions, "meta": meta, } @app.post("/api/quiz/hint") def quiz_hint(req: HintRequest, request: Request): """生成题目提示（不泄露答案）。""" if QUIZ_API_KEY: key = request.headers.get("X-API-Key") or request.headers.get("Authorization", "").replace("Bearer ", "") if (key or "").strip() != QUIZ_API_KEY: return JSONResponse( status_code=429, content={"code": 429, "error": {"type": "RATE_LIMIT", "reason": "missing_or_invalid_api_key"}}, ) question_content = (req.questionContext.content or "").strip() if not question_content: return JSONResponse( status_code=422, content={"code": 422, "error": {"type": "INVALID_GENERATION", "reason": "question_content_required"}}, ) language = (req.language or "en").strip().lower() if language not in ("en", "zh", "中文"): language = "en" # 转换选项格式 options = [] if req.questionContext.options: for opt in req.questionContext.options: options.append({ "label": opt.label or opt.key or "", "content": opt.content or opt.text or "", }) t0 = time.time() try: hint_text, tokens_used = generate_hint_for_question( question_content=question_content, options=options if options else None, language=language, ) except Exception as e: print(f"[quiz_hint] error: {repr(e)}") return JSONResponse( status_code=500, content={"code": 500, "error": {"type": "MODEL_ERROR", "reason": "hint_generation_failed"}}, ) latency_ms = (time.time() - t0) * 1000.0 meta = { "model": DEFAULT_MODEL, "model_version": "", "prompt_version": "quiz_hint_v1", "temperature": 0.6, "tokens_used": tokens_used, "latency_ms": round(latency_ms, 2), } return { "data": { "hint": hint_text, }, "meta": meta, } @app.post("/api/quiz/grade") def quiz_grade(req: GradeRequest, request: Request): """智能判卷：对用户提交的答案进行评分和反馈。""" if QUIZ_API_KEY: key = request.headers.get("X-API-Key") or request.headers.get("Authorization", "").replace("Bearer ", "") if (key or "").strip() != QUIZ_API_KEY: return JSONResponse( status_code=429, content={"code": 429, "error": {"type": "RATE_LIMIT", "reason": "missing_or_invalid_api_key"}}, ) if not req.userAnswers or len(req.userAnswers) == 0: return JSONResponse( status_code=422, content={"code": 422, "error": {"type": "INVALID_GENERATION", "reason": "user_answers_required"}}, ) language = (req.language or "en").strip().lower() if language not in ("en", "zh", "中文"): language = "en" # 转换用户答案格式（含简答题 rubric） user_answers = [] for ans in req.userAnswers: rubric_dict = None if ans.rubric: r = ans.rubric rubric_dict = { "corePoints": r.corePoints, "acceptableSynonyms": r.acceptableSynonyms, "commonErrors": r.commonErrors, "answerScope": r.answerScope.model_dump() if r.answerScope else None, } user_answers.append({ "questionContent": ans.questionContent or "", "questionId": ans.questionId, "userChoiceLabel": ans.userChoiceLabel, "correctChoiceLabel": ans.correctChoiceLabel, "isCorrect": ans.isCorrect, "userTextAnswer": ans.userTextAnswer, "referenceAnswer": ans.referenceAnswer, "rubric": rubric_dict, }) t0 = time.time() try: grading_result, tokens_used = grade_quiz_answers( user_answers=user_answers, language=language, ) except Exception as e: print(f"[quiz_grade] error: {repr(e)}") return JSONResponse( status_code=500, content={"code": 500, "error": {"type": "MODEL_ERROR", "reason": "grading_failed"}}, ) latency_ms = (time.time() - t0) * 1000.0 meta = { "model": DEFAULT_MODEL, "model_version": "", "prompt_version": "quiz_grade_v1", "temperature": 0.4, "tokens_used": tokens_used, "latency_ms": round(latency_ms, 2), } return { "data": grading_result, "meta": meta, } @app.get("/api/memoryline") def memoryline(user_id: str): _ = _get_session((user_id or "").strip()) return {"next_review_label": "T+7", "progress_pct": 0.4} @app.get("/api/profile/status") def profile_status(user_id: str): user_id = (user_id or "").strip() if not user_id: return JSONResponse({"error": "Missing user_id"}, status_code=400) sess = _get_session(user_id) bio = (sess.get("profile_bio") or "").strip() need_init = len(bio) <= 50 and (int(time.time()) >= int(sess.get("init_dismiss_until") or 0)) return {"need_init": need_init, "bio_length": len(bio)} @app.get("/health") def health(): return {"status": "ok"} @app.get("/{full_path:path}") def spa_fallback(full_path: str): if full_path.startswith("api/") or full_path.startswith("assets/") or full_path.startswith("static/"): return JSONResponse({"detail": "Not Found"}, status_code=404) if os.path.exists(WEB_INDEX): return FileResponse(WEB_INDEX) return JSONResponse({"detail": "web/build not found"}, status_code=500)