service.py

# -*- coding: utf-8 -*-
"""
================================================================================
文件名称：service.py
文件定位：核心业务逻辑层 / 总控 Agent 与 6 个子 Agent 的调度桥梁
职责说明：
    1. 持有 AgentManager 实例，统一管理 6 阶段 Agent 流水线
    2. 被 main.py 直接调用，提供 step_1 到 step_6 共 6 个业务方法
    3. 每个方法内部：构建增强提示词 → 调用对应 Agent → 解析结果 → 返回结构化字典
    4. 实现 API 未配置时的演示降级模式（Mock 数据），确保系统无 API Key 也能展示完整流程
    5. 大模型与代码职责分离：代码负责任务编排、状态管理、异常降级；大模型负责语义分析
================================================================================
"""

import os
import json
import logging
import asyncio
from typing import Dict, Any, Optional
from datetime import datetime

# ------------------------------------------------------------------------------
# 本地模块导入
# ------------------------------------------------------------------------------
from agent_manager import AgentManager, BaseAgent, KimiClient, AgentOutput
from prompts import PromptBuilder, TextPreprocessor, EntityExtractor, StructuredConstraints, AntiHallucination

# ------------------------------------------------------------------------------
# 日志配置
# ------------------------------------------------------------------------------
logger = logging.getLogger("StudyService")

# 全局配置
AGENT_TIMEOUT_SEC = 25       # 单步超时25秒
MAX_RETRY_TIMES = 2          # 最大重试2次
RETRY_INTERVAL_SEC = 1       # 重试间隔1秒


# ==============================================================================
# 演示模式数据：当 MOONSHOT_API_KEY 未配置时，返回预设的 Mock 数据，
# 确保前端能完整展示 6 个 Agent 的处理流程与界面效果
# ==============================================================================

DEMO_RESULTS: Dict[str, Dict[str, Any]] = {
    "agent_1_parse": {
        "knowledge_points": [
            {"name": "导数的几何意义", "domain": "数学-微积分", "importance": "high"},
            {"name": "链式法则", "domain": "数学-微积分", "importance": "high"},
            {"name": "隐函数求导", "domain": "数学-微积分", "importance": "medium"},
            {"name": "高阶导数", "domain": "数学-微积分", "importance": "medium"},
            {"name": "导数在极值问题中的应用", "domain": "数学-微积分", "importance": "high"},
        ],
        "weak_points": [
            {"name": "链式法则多层嵌套时符号容易遗漏", "reason": "概念模糊", "suggestion": "逐步拆解复合函数，标注每一层的内函数与外函数"},
            {"name": "隐函数求导时忘记对 y 求导后乘 dy/dx", "reason": "公式记错", "suggestion": "强化隐函数求导公式的记忆与专项练习"},
            {"name": "极值判定中二阶导数测试的条件掌握不牢", "reason": "逻辑跳跃", "suggestion": "整理极值判定流程图，明确一阶与二阶条件的逻辑关系"},
        ],
        "error_prone_points": [
            {"name": "复合函数求导遗漏内层导数", "error_type": "链式法则应用不完整", "correct_way": "逐层识别内函数，每层都乘上对应的导数"},
            {"name": "驻点与极值点概念混淆", "error_type": "概念混淆", "correct_way": "明确：驻点是 f'(x)=0 的点，极值点还需满足单调性变化"},
            {"name": "导数计算中符号错误", "error_type": "计算粗心", "correct_way": "求导后反向验证：积分还原检查是否正确"},
        ],
        "summary": "该生在导数核心概念上具备基础理解，但在链式法则的复杂嵌套和隐函数求导中暴露出系统性薄弱，需要加强复合结构拆解训练。",
    },
    "agent_2_profile": {
        "ability_tags": [
            {"tag": "基础概念理解", "level": "medium", "evidence": "能正确表述导数定义"},
            {"tag": "复杂计算能力", "level": "weak", "evidence": "链式法则多层嵌套频繁出错"},
            {"tag": "逻辑推理", "level": "medium", "evidence": "极值判定逻辑基本正确但条件记混"},
            {"tag": "细心程度", "level": "weak", "evidence": "符号错误和遗漏内层导数反复出现"},
        ],
        "shortcoming_profile": {
            "concept_gaps": ["链式法则的递归结构理解", "隐函数求导的完整步骤"],
            "skill_gaps": ["多层复合函数拆解", "符号管理"],
            "severity": "medium",
        },
        "learning_level": {
            "level": "中级",
            "score": 68,
            "reason": "基础概念掌握尚可，但复杂运算和符号管理处于薄弱状态，整体处于中等偏下水平",
        },
        "growth_suggestions": [
            {"action": "每天完成 5 道链式法则专项题，限时 15 分钟", "target": "提升复合函数拆解熟练度", "duration": "2周"},
            {"action": "建立个人错题本，分类记录符号错误和遗漏类型", "target": "降低粗心错误率", "duration": "持续"},
            {"action": "每周复盘一次本周错题，总结规律", "target": "形成系统性纠错机制", "duration": "4周"},
        ],
        "summary": "该生为中等水平学习者，概念理解尚可但运算执行层面存在明显短板，建议通过专项训练与错题管理实现突破。",
    },
    "agent_3_guide": {
        "learning_path": [
            {"phase": "短期(1周内)", "goals": ["熟练掌握链式法则 3 层以内嵌套", "隐函数求导零失误"], "actions": ["完成教材 3.2 节全部例题", "每天 10 道链式法则速算"]},
            {"phase": "中期(2-4周)", "goals": ["导数应用题（极值、最值）正确率 90% 以上"], "actions": ["分类刷极值应用题 30 道", "整理极值判定决策树"]},
            {"phase": "长期(1学期)", "goals": ["微积分整体章节考试达到良好"], "actions": ["建立完整章节知识图谱", "每月一次模拟自测"]},
        ],
        "knowledge_guidance": [
            {"topic": "链式法则", "guidance": "将复合函数想象为'俄罗斯套娃'，每一层都要'揭开'并乘上对应的导数。建议画图标注各层函数名。", "key_question": "如果 f(g(h(x))) 有三层，最终的导数表达式中有几个乘积项？分别是什么？"},
            {"topic": "隐函数求导", "guidance": "记住核心原则：遇到 y 的函数时，先对 y 求导，再乘上 dy/dx。建议写出完整的'd/dx'步骤。", "key_question": "对方程 x² + y² = 1 两边求导时，y² 项的处理与 x² 有什么不同？"},
        ],
        "practice_strategy": [
            {"method": "错题重做+变式训练", "description": "同一道错题，改变系数或函数形式后重新做一遍", "frequency": "每周 3 次"},
            {"method": "限时速算", "description": "基础求导题限时完成，培养速度和准确率", "frequency": "每天 1 次"},
        ],
        "resource_recommendation": [
            {"type": "教材章节", "description": "《高等数学》第三章 导数与微分，重点 3.2-3.4 节"},
            {"type": "视频讲解", "description": "复合函数求导专题视频，重点学习'拆解思路'而非只记公式"},
        ],
        "warning": "本系统仅提供学习思路引导和方法建议，严禁直接输出作业答案或代写内容。学生应独立完成作业，系统仅作为学习辅助工具。",
    },
    "agent_4_ethics": {
        "risk_level": "low",
        "risk_score": 25,
        "ai_trace": {
            "detected": False,
            "indicators": [],
            "confidence": "low",
        },
        "academic_integrity": {
            "issues": [
                {"type": "引用不规范", "location": "后半段", "severity": "low", "suggestion": "如引用教材定理，建议标注出处"},
            ],
        },
        "logic_flaws": [
            {"type": "论证跳跃", "description": "从链式法则直接跳到极值应用，缺少过渡说明", "suggestion": "补充'基于导数可以分析函数单调性，进而判定极值'的过渡句"},
        ],
        "language_issues": [
            {"type": "语句冗长", "example": "在求解这个复合函数的导数的时候需要注意的是", "fix": "求解复合函数导数时，需注意："},
        ],
        "overall_suggestion": "整体学术规范性良好，逻辑基本通顺。建议加强段落过渡和引用标注，提升学术写作规范度。",
    },
    "agent_5_polish": {
        "polished_fragments": [
            {"original": "在求解这个复合函数的导数的时候需要注意的是", "polished": "求解复合函数导数时，需注意：", "change_reason": "精简冗余表达，提升语句紧凑度"},
            {"original": "用链式法则求导然后代入", "polished": "先应用链式法则求导，再代入具体值计算", "change_reason": "拆分动作，逻辑更清晰"},
        ],
        "structure_suggestions": [
            {"aspect": "段落衔接", "current": "知识点之间缺少过渡", "suggested": "在导数定义与链式法则之间增加'基于导数定义可推导出...'的过渡段"},
            {"aspect": "示例分布", "current": "前半段示例多，后半段缺少", "suggested": "在隐函数求导和高阶导数部分各增加 1 个示例"},
        ],
        "format_guidelines": [
            {"item": "公式排版", "standard": "独立行居中，编号如 (1)(2)", "status": "需改进"},
            {"item": "标题层级", "standard": "一级标题用'一、'，二级用'（一）'", "status": "符合"},
        ],
        "terminology_check": [
            {"term": "导数", "usage_correct": True, "suggestion": "使用正确"},
            {"term": "微分", "usage_correct": True, "suggestion": "注意与导数概念区分，此处使用恰当"},
        ],
        "overall_quality_score": 72,
        "summary": "文本整体质量良好，概念表达准确。主要改进空间在于语句精简度和段落衔接，润色后学术规范性将显著提升。",
    },
    "agent_6_report": {
        "report_title": "个人学习成长报告",
        "generated_at": "2026-04-30",
        "overall_assessment": {
            "learning_level": "中级",
            "risk_level": "low",
            "quality_score": 72,
            "key_strengths": ["基础概念理解扎实", "学习态度认真（文本内容充实）"],
            "key_improvements": ["复杂运算熟练度不足", "学术写作规范性待提升"],
        },
        "stage_summary": [
            {"stage": "学情解析", "key_findings": "掌握导数核心概念 5 个，发现薄弱点 3 个、易错点 3 个"},
            {"stage": "学情画像", "key_findings": "中等水平（68 分），逻辑推理中等，细心程度待加强"},
            {"stage": "分层导学", "key_findings": "制定短/中/长期路径，重点突破链式法则与隐函数求导"},
            {"stage": "学术伦理", "key_findings": "风险等级低（25 分），逻辑与语言存在轻微问题"},
            {"stage": "规范整编", "key_findings": "整体质量 72 分，语句精简和段落衔接为主要改进方向"},
        ],
        "growth_plan": {
            "short_term": ["完成链式法则 3 层嵌套训练", "隐函数求导零失误"],
            "medium_term": ["极值应用题正确率达 90%", "建立错题本机制"],
            "long_term": ["微积分整体章节达良好水平", "形成自主学习能力"],
        },
        "report_html": "",
    },
}

# 为 Agent 6 生成完整的 HTML 报告（演示模式）
DEMO_RESULTS["agent_6_report"]["report_html"] = """
<div style="font-family:'Microsoft YaHei',sans-serif; line-height:1.8; color:#333;">
  <h2 style="color:#1a5276; border-bottom:2px solid #3498db; padding-bottom:10px;">个人学习成长报告</h2>
  <p style="color:#666; font-size:14px;">生成日期：2026-04-30</p>
  
  <h3 style="color:#2874a6;">一、综合评估</h3>
  <table style="width:100%; border-collapse:collapse; margin:10px 0;">
    <tr style="background:#eaf2f8;"><td style="padding:8px; border:1px solid #bbb;"><strong>学习层级</strong></td><td style="padding:8px; border:1px solid #bbb;">中级</td></tr>
    <tr><td style="padding:8px; border:1px solid #bbb;"><strong>学术风险</strong></td><td style="padding:8px; border:1px solid #bbb;"><span style="color:green;">低 (25分)</span></td></tr>
    <tr style="background:#eaf2f8;"><td style="padding:8px; border:1px solid #bbb;"><strong>文本质量</strong></td><td style="padding:8px; border:1px solid #bbb;">72分</td></tr>
  </table>
  
  <h3 style="color:#2874a6;">二、核心优势</h3>
  <ul><li>基础概念理解扎实，导数定义表述准确</li><li>学习态度认真，提交文本内容充实完整</li></ul>
  
  <h3 style="color:#2874a6;">三、重点提升方向</h3>
  <ul><li>链式法则多层嵌套的熟练度不足</li><li>学术写作中语句精简度和段落衔接待优化</li></ul>
  
  <h3 style="color:#2874a6;">四、成长计划</h3>
  <p><strong>短期（1周内）：</strong>完成链式法则专项训练，隐函数求导零失误</p>
  <p><strong>中期（2-4周）：</strong>极值应用题正确率达90%，建立错题分类本</p>
  <p><strong>长期（1学期）：</strong>微积分整体章节达良好水平，形成自主学习能力</p>
  
  <h3 style="color:#2874a6;">五、各阶段分析摘要</h3>
  <table style="width:100%; border-collapse:collapse; margin:10px 0; font-size:14px;">
    <tr style="background:#3498db; color:white;"><th style="padding:8px; border:1px solid #bbb;">阶段</th><th style="padding:8px; border:1px solid #bbb;">关键发现</th></tr>
    <tr><td style="padding:8px; border:1px solid #bbb;">学情解析</td><td style="padding:8px; border:1px solid #bbb;">掌握5个核心知识点，发现3个薄弱点、3个易错点</td></tr>
    <tr style="background:#f8f9fa;"><td style="padding:8px; border:1px solid #bbb;">学情画像</td><td style="padding:8px; border:1px solid #bbb;">中等水平（68分），逻辑推理中等，细心程度待加强</td></tr>
    <tr><td style="padding:8px; border:1px solid #bbb;">分层导学</td><td style="padding:8px; border:1px solid #bbb;">制定短/中/长期路径，重点突破链式法则与隐函数</td></tr>
    <tr style="background:#f8f9fa;"><td style="padding:8px; border:1px solid #bbb;">学术伦理</td><td style="padding:8px; border:1px solid #bbb;">风险等级低，逻辑与语言存在轻微问题</td></tr>
    <tr><td style="padding:8px; border:1px solid #bbb;">规范整编</td><td style="padding:8px; border:1px solid #bbb;">整体质量72分，语句精简和段落衔接为主要改进方向</td></tr>
  </table>
  
  <p style="margin-top:20px; padding:10px; background:#eaf2f8; border-left:4px solid #3498db; font-size:13px;">
    <strong>声明：</strong>本报告由 AI 辅助系统基于您提交的学习文本自动生成，仅供参考。学习提升仍需依靠自身努力与教师指导。
  </p>
</div>
"""


# ==============================================================================
# 核心业务服务类
# ==============================================================================

class StudyAssistantService:
    """
    学情分析业务服务层
    对外接口：6 个 step_xxx 方法（与 main.py 中的 pipeline_steps 一一对应）
    内部机制：
        - 正常模式：调用 Kimi2.6 API 执行真实分析
        - 演示模式（DEMO_MODE=True）：返回预设 Mock 数据，无需 API Key 即可运行
    """

    def __init__(self):
        self.agent_manager = AgentManager()
        self.demo_mode = not bool(os.environ.get("MOONSHOT_API_KEY", "").strip())
        if self.demo_mode:
            logger.warning("[StudyService] 未检测到 MOONSHOT_API_KEY，进入演示模式（返回 Mock 数据）")
        else:
            logger.info("[StudyService] 检测到 API Key，启用真实大模型分析模式")

    async def _safe_agent_run(self, agent, user_input, context):
        """带超时+重试的安全执行"""
        last_err = None
        for retry_cnt in range(MAX_RETRY_TIMES + 1):
            try:
                # 超时控制
                output = await asyncio.wait_for(
                    agent.run(user_input, context),
                    timeout=AGENT_TIMEOUT_SEC
                )
                return output
            except asyncio.TimeoutError:
                last_err = Exception(f"Agent执行超时{AGENT_TIMEOUT_SEC}s")
                logger.warning(f"[AgentRun] 第{retry_cnt+1}次执行超时")
            except Exception as e:
                last_err = e
                logger.warning(f"[AgentRun] 第{retry_cnt+1}次执行异常: {str(e)}")

            # 不是最后一次，间隔重试
            if retry_cnt < MAX_RETRY_TIMES:
                await asyncio.sleep(RETRY_INTERVAL_SEC)

        # 重试耗尽，返回失败占位
        return AgentOutput(
            agent_name=agent.name,
            status="failed",
            result={},
            raw_response="",
            cost_ms=0,
            error_msg=str(last_err)
        )

    def _fix_result_struct(self, agent_name: str, result: Dict[str, Any]) -> Dict[str, Any]:
        """结构校验不通过时，兜底补全字段，保证前端不报错"""
        try:
            valid, err = StructuredConstraints.validate_output(agent_name, result)
            if not valid:
                logger.warning(f"[{agent_name}] 结构校验失败，进行字段兜底兼容: {err}")
            return result
        except Exception as e:
            logger.error(f"[{agent_name}] 结构校验异常: {e}")
            return result

    # ==========================================================================
    # 统一的单步执行引擎（内部方法，被 6 个 step 方法复用）
    # ==========================================================================

    async def _execute_single_step(self, step: int, context: Dict[str, Any]) -> Dict[str, Any]:
        """
        执行单个 Agent 步骤的统一引擎
        增加：超时控制、自动重试、结构兜底、全局异常捕获
        """
        agent_key = f"agent_{step}_" + ["parse", "profile", "guide", "ethics", "polish", "report"][step - 1]

        # 演示模式：直接返回预设 Mock 数据
        if self.demo_mode:
            logger.info(f"[StudyService][Demo] Step {step} 返回 Mock 数据")
            mock_data = DEMO_RESULTS.get(agent_key, {})
            return {
                "agent_name": agent_key,
                "status": "success",
                "result": mock_data,
                "cost_ms": 150,
                "demo_mode": True,
            }

        try:
            # 真实模式：调用大模型
            agent = self.agent_manager.agents[step - 1]
            input_text = context.get("input_text", "")
            text_type = context.get("text_type", "homework")

            base_user_input = self.agent_manager._build_agent_input(agent, input_text, text_type, context)

            enhanced_system_prompt, enhanced_user_input = PromptBuilder.build_for_agent(
                agent_name=agent.name,
                base_system_prompt=agent.system_prompt,
                user_input=base_user_input,
                context=context,
            )

            original_system_prompt = agent.system_prompt
            agent.system_prompt = enhanced_system_prompt

            try:
                # 带超时+重试执行
                output: AgentOutput = await self._safe_agent_run(agent, enhanced_user_input, context)

                # 防幻觉检测
                if output.status == "success" and output.raw_response:
                    passed, warning = AntiHallucination.quick_check_output(output.raw_response, agent.name)
                    if not passed:
                        logger.warning(f"[StudyService] Step {step} 防幻觉检测警告: {warning}")

                # 结构兜底修复
                if output.status == "success" and isinstance(output.result, dict):
                    output.result = self._fix_result_struct(agent.name, output.result)

                return {
                    "agent_name": output.agent_name,
                    "status": output.status,
                    "result": output.result,
                    "cost_ms": output.cost_ms,
                    "error_msg": output.error_msg,
                }

            finally:
                agent.system_prompt = original_system_prompt

        except Exception as e:
            logger.error(f"[StudyService] Step {step} 全局未知异常: {str(e)}")
            # 全局异常兜底返回，不中断流水线
            return {
                "agent_name": agent_key,
                "status": "failed",
                "result": {},
                "cost_ms": 0,
                "error_msg": f"步骤执行异常: {str(e)}",
                "demo_mode": False
            }

    # ==========================================================================
    # 6 个对外业务方法（与 main.py pipeline_steps 严格对应）
    # ==========================================================================

    async def step_1_parse_knowledge(self, context: Dict[str, Any]) -> Dict[str, Any]:
        return await self._execute_single_step(1, context)

    async def step_2_build_profile(self, context: Dict[str, Any]) -> Dict[str, Any]:
        return await self._execute_single_step(2, context)

    async def step_3_guide_learning(self, context: Dict[str, Any]) -> Dict[str, Any]:
        return await self._execute_single_step(3, context)

    async def step_4_ethics_check(self, context: Dict[str, Any]) -> Dict[str, Any]:
        return await self._execute_single_step(4, context)

    async def step_5_polish_text(self, context: Dict[str, Any]) -> Dict[str, Any]:
        return await self._execute_single_step(5, context)

    async def step_6_generate_report(self, context: Dict[str, Any]) -> Dict[str, Any]:
        return await self._execute_single_step(6, context)

    # ==========================================================================
    # 便捷方法：一次性执行完整 6 阶段流水线（供扩展使用）
    # ==========================================================================

    async def run_full_pipeline(
        self,
        input_text: str,
        text_type: str,
        progress_callback=None,
    ) -> Dict[str, Any]:
        context = {
            "input_text": input_text,
            "text_type": text_type,
            "task_id": "batch_" + datetime.now().strftime("%Y%m%d%H%M%S"),
        }

        step_methods = [
            self.step_1_parse_knowledge,
            self.step_2_build_profile,
            self.step_3_guide_learning,
            self.step_4_ethics_check,
            self.step_5_polish_text,
            self.step_6_generate_report,
        ]

        results = {}
        for i, method in enumerate(step_methods, start=1):
            agent_key = f"agent_{i}_" + ["parse", "profile", "guide", "ethics", "polish", "report"][i - 1]
            if progress_callback:
                progress_callback(i, agent_key)

            result = await method(context)
            results[agent_key] = result
            context[f"{agent_key}_result"] = result.get("result", {})

            # 不直接break，继续走完流程，前端友好展示
            if result.get("status") != "success":
                logger.error(f"[StudyService] 流水线 Step {i} 执行失败，继续后续步骤")

        return {
            "step_results": results,
            "context": context,
            "completed_steps": len(results),
            "demo_mode": self.demo_mode,
        }