import enum import httpx from sqlalchemy.orm import Session from app.core.config import settings as app_settings from app.models.user import SchoolLevel, SystemSettings from app.services.school_level import school_level_label CURRICULUM_JUNIOR = """初中课程标准:代数、几何(全等/相似/勾股)、一次函数与简单二次函数、基础概率统计。 严禁使用:高中导数、向量、解析几何、排列组合进阶、复数、微积分、大学线性代数等。""" CURRICULUM_SENIOR = """高中课程标准:课内函数、三角、向量、解析几何、概率统计、导数(课内范围)等。 严禁使用:大学数学分析、抽象代数、高等几何、超出课内的竞赛高阶技巧。""" CURRICULUM_JUNIOR_OLYMPIAD = """初中奥数培优范围:整数/整除、因数分解、简单数论、代数恒等变形、几何辅助线与全等相似、简单组合计数。 严禁使用:高中及以上方法(导数、向量、解析几何、微积分、复数运算等)。""" CURRICULUM_SENIOR_OLYMPIAD = """高中奥数/竞赛入门范围:课内知识+常规竞赛技巧(不等式、构造、归纳、简单数论等)。 严禁使用:大学数学、超出高中奥数培优体系的 IMO 高阶理论。""" def _curriculum_block(level: SchoolLevel | str | None, olympiad: bool) -> str: label = school_level_label(level) is_senior = level == SchoolLevel.senior_high or level == "senior_high" if olympiad: return CURRICULUM_SENIOR_OLYMPIAD if is_senior else CURRICULUM_JUNIOR_OLYMPIAD return CURRICULUM_SENIOR if is_senior else CURRICULUM_JUNIOR QUESTION_PROMPT = """你是一位{stage}老师。以下是从试卷 OCR 识别出的文字,可能含有噪声。 科目:{subject} 请整理出清晰的题目内容(保留题号、选项、公式),只输出题目正文,不要解释。 OCR 原文: {ocr_text} """ SOLUTION_PROMPT = """你是一位耐心的{stage}{subject}老师。请为以下题目给出详细解法。 【学段要求 — 严禁超纲】 {curriculum} 题目: {question_text} 请按以下结构输出: 1. 考点分析({stage}范围内) 2. 解题步骤(逐步推导,每步说明依据) 3. 易错点提醒 4. 若必须使用超纲方法才能解,请改用{stage}可理解的方法重新解答,不得输出超纲解法。 """ OLYMPIAD_SOLUTION_PROMPT = """你是一位{stage}奥数教练。请为以下奥数题给出详细解题思路与完整解答。 【奥数学段要求 — 严禁超纲】 {curriculum} 题目: {question_text} 请按以下结构输出: 1. 题型与思路切入点({stage}奥数常见技巧) 2. 详细解答步骤 3. 关键技巧总结(仅限{stage}奥数范围) 4. 严禁使用超出上述范围的方法;若题目过难,给出{stage}可接受的培优思路。 """ class AIConfig: def __init__( self, provider: str, ollama_base_url: str, ollama_model: str, openai_base_url: str, openai_model: str, openai_api_key: str | None, ): self.provider = provider self.ollama_base_url = ollama_base_url self.ollama_model = ollama_model self.openai_base_url = openai_base_url self.openai_model = openai_model self.openai_api_key = openai_api_key def load_ai_config(db: Session) -> AIConfig: row = db.get(SystemSettings, 1) if row is None: return AIConfig( provider="ollama", ollama_base_url=app_settings.OLLAMA_BASE_URL, ollama_model=app_settings.OLLAMA_MODEL, openai_base_url=app_settings.OPENAI_BASE_URL, openai_model=app_settings.OPENAI_MODEL, openai_api_key=None, ) return AIConfig( provider=row.ai_provider or "ollama", ollama_base_url=row.ollama_base_url or app_settings.OLLAMA_BASE_URL, ollama_model=row.ollama_model or app_settings.OLLAMA_MODEL, openai_base_url=row.openai_base_url or app_settings.OPENAI_BASE_URL, openai_model=row.openai_model or app_settings.OPENAI_MODEL, openai_api_key=row.openai_api_key, ) async def _ollama_generate(prompt: str, cfg: AIConfig) -> str: url = f"{cfg.ollama_base_url.rstrip('/')}/api/generate" payload = {"model": cfg.ollama_model, "prompt": prompt, "stream": False} async with httpx.AsyncClient(timeout=180.0) as client: response = await client.post(url, json=payload) response.raise_for_status() return (response.json().get("response") or "").strip() async def _openai_generate(prompt: str, cfg: AIConfig) -> str: if not cfg.openai_api_key: raise ValueError("未配置 OpenAI API Key") url = f"{cfg.openai_base_url.rstrip('/')}/chat/completions" headers = {"Authorization": f"Bearer {cfg.openai_api_key}"} payload = { "model": cfg.openai_model, "messages": [{"role": "user", "content": prompt}], "temperature": 0.3, } async with httpx.AsyncClient(timeout=180.0) as client: response = await client.post(url, json=payload, headers=headers) response.raise_for_status() data = response.json() return (data["choices"][0]["message"]["content"] or "").strip() async def generate_text(prompt: str, cfg: AIConfig) -> str: if cfg.provider == "openai": return await _openai_generate(prompt, cfg) return await _ollama_generate(prompt, cfg) async def format_question( cfg: AIConfig, subject: str, ocr_text: str, school_level=None, ) -> str: stage = school_level_label(school_level) prompt = QUESTION_PROMPT.format(stage=stage, subject=subject, ocr_text=ocr_text) return await generate_text(prompt, cfg) async def generate_solution( cfg: AIConfig, subject: str, question_text: str, school_level=None, *, olympiad: bool = False, ) -> str: stage = school_level_label(school_level) curriculum = _curriculum_block(school_level, olympiad) template = OLYMPIAD_SOLUTION_PROMPT if olympiad else SOLUTION_PROMPT prompt = template.format( stage=stage, subject=subject, curriculum=curriculum, question_text=question_text, ) return await generate_text(prompt, cfg)