api/clare_core.py

# clare_core.py
import re
import math
from typing import List, Dict, Tuple, Optional

from docx import Document

from .config import (
    client,
    DEFAULT_MODEL,
    EMBEDDING_MODEL,
    DEFAULT_COURSE_TOPICS,
    CLARE_SYSTEM_PROMPT,
    LEARNING_MODE_INSTRUCTIONS,
)
from langsmith import traceable
from langsmith.run_helpers import set_run_metadata



# ---------- syllabus 解析 ----------
def parse_syllabus_docx(file_path: str, max_lines: int = 15) -> List[str]:
    """
    非常简单的 syllabus 解析：取前若干个非空段落当作主题行。
    只是为了给 Clare 一些课程上下文，不追求超精确结构。
    """
    topics: List[str] = []
    try:
        doc = Document(file_path)
        for para in doc.paragraphs:
            text = para.text.strip()
            if not text:
                continue
            topics.append(text)
            if len(topics) >= max_lines:
                break
    except Exception as e:
        topics = [f"[Error parsing syllabus: {e}]"]

    return topics


# ---------- 简单“弱项”检测 ----------
WEAKNESS_KEYWORDS = [
    "don't understand",
    "do not understand",
    "not understand",
    "not sure",
    "confused",
    "hard to",
    "difficult",
    "struggle",
    "不会",
    "不懂",
    "看不懂",
    "搞不清",
    "很难",
]

# ---------- 简单“掌握”检测 ----------
MASTERY_KEYWORDS = [
    "got it",
    "makes sense",
    "now i see",
    "i see",
    "understand now",
    "clear now",
    "easy",
    "no problem",
    "没问题",
    "懂了",
    "明白了",
    "清楚了",
]


def update_weaknesses_from_message(message: str, weaknesses: List[str]) -> List[str]:
    lower_msg = message.lower()
    if any(k in lower_msg for k in WEAKNESS_KEYWORDS):
        weaknesses = weaknesses or []
        weaknesses.append(message)
    return weaknesses


def update_cognitive_state_from_message(
    message: str,
    state: Optional[Dict[str, int]],
) -> Dict[str, int]:
    """
    简单认知状态统计：
    - 遇到困惑类关键词 → confusion +1
    - 遇到掌握类关键词 → mastery +1
    """
    if state is None:
        state = {"confusion": 0, "mastery": 0}

    lower_msg = message.lower()
    if any(k in lower_msg for k in WEAKNESS_KEYWORDS):
        state["confusion"] = state.get("confusion", 0) + 1
    if any(k in lower_msg for k in MASTERY_KEYWORDS):
        state["mastery"] = state.get("mastery", 0) + 1
    return state


def describe_cognitive_state(state: Optional[Dict[str, int]]) -> str:
    if not state:
        return "unknown"
    confusion = state.get("confusion", 0)
    mastery = state.get("mastery", 0)
    if confusion >= 2 and confusion >= mastery + 1:
        return "student shows signs of HIGH cognitive load (often confused)."
    elif mastery >= 2 and mastery >= confusion + 1:
        return "student seems COMFORTABLE; material may be slightly easy."
    else:
        return "mixed or uncertain cognitive state."


# ---------- Session Memory ----------
def build_session_memory_summary(
    history: List[Tuple[str, str]],
    weaknesses: Optional[List[str]],
    cognitive_state: Optional[Dict[str, int]],
    max_questions: int = 4,
    max_weaknesses: int = 3,
) -> str:
    """
    只在本次会话内使用的“记忆摘要”：
    - 最近几条学生提问
    - 最近几条学生觉得难的问题
    - 当前的认知状态描述
    """
    parts: List[str] = []

    # 最近几条提问（只取 student）
    if history:
        recent_qs = [u for (u, _a) in history[-max_questions:]]
        trimmed_qs = []
        for q in recent_qs:
            q = q.strip()
            if len(q) > 120:
                q = q[:117] + "..."
            trimmed_qs.append(q)
        if trimmed_qs:
            parts.append("Recent student questions: " + " | ".join(trimmed_qs))

    # 最近几条“弱项”
    if weaknesses:
        recent_weak = weaknesses[-max_weaknesses:]
        trimmed_weak = []
        for w in recent_weak:
            w = w.strip()
            if len(w) > 120:
                w = w[:117] + "..."
            trimmed_weak.append(w)
        parts.append("Recent difficulties mentioned by the student: " + " | ".join(trimmed_weak))

    # 当前认知状态
    if cognitive_state:
        parts.append("Current cognitive state: " + describe_cognitive_state(cognitive_state))

    if not parts:
        return (
            "No prior session memory. You can treat this as an early stage of the conversation; "
            "start with simple explanations and ask a quick check-up question."
        )

    return " | ".join(parts)


# ---------- 语言检测（用于 Auto 模式） ----------
def detect_language(message: str, preference: str) -> str:
    """
    preference:
      - 'English' → 强制英文
      - '中文' → 强制中文
      - 'Auto' → 检测文本是否包含中文字符
    """
    if preference in ("English", "中文"):
        return preference
    # Auto 模式下简单检测是否含有中文字符
    if re.search(r"[\u4e00-\u9fff]", message):
        return "中文"
    return "English"


def get_empty_input_prompt(lang: str) -> str:
    """
    空输入时的友好提示，根据语言返回中/英文。
    """
    if lang == "中文":
        return "请先输入一个问题或想法，再按回车发送，我才能帮到你哦。"
    # 默认英文
    return "Please type a question or some text before sending, then hit Enter."


def build_error_message(
    e: Exception,
    lang: str,
    op: str = "chat",
) -> str:
    """
    针对不同操作类型（普通对话 / quiz / summary）和语言，生成友好的错误提示。
    不把原始异常直接暴露给学生，只在后台打印。
    """
    if lang == "中文":
        prefix = {
            "chat": "抱歉，刚刚在和模型对话时出现了一点问题。",
            "quiz": "抱歉，刚刚在生成测验题目时出现了一点问题。",
            "summary": "抱歉，刚刚在生成总结时出现了一点问题。",
        }.get(op, "抱歉，刚刚出现了一点问题。")
        return prefix + " 请稍后再试一次，或者换个问法试试。"

    # 默认英文
    prefix_en = {
        "chat": "Sorry, I ran into a problem while talking to the model.",
        "quiz": "Sorry, there was a problem while generating the quiz.",
        "summary": "Sorry, there was a problem while generating the summary.",
    }.get(op, "Sorry, something went wrong just now.")
    return prefix_en + " Please try again in a moment or rephrase your request."


# ---------- Session 状态展示 ----------
def render_session_status(
    learning_mode: str,
    weaknesses: Optional[List[str]],
    cognitive_state: Optional[Dict[str, int]],
) -> str:
    lines: List[str] = []
    lines.append("### Session status\n")
    lines.append(f"- Learning mode: **{learning_mode}**")
    lines.append(f"- Cognitive state: {describe_cognitive_state(cognitive_state)}")

    if weaknesses:
        lines.append("- Recent difficulties (last 3):")
        for w in weaknesses[-3:]:
            lines.append(f"  - {w}")
    else:
        lines.append("- Recent difficulties: *(none yet)*")

    return "\n".join(lines)


# ---------- Same Question Check helpers ----------
def _normalize_text(text: str) -> str:
    """
    将文本转为小写、去除标点和多余空格，用于简单相似度计算。
    """
    text = text.lower().strip()
    text = re.sub(r"[^\w\s]", " ", text)
    text = re.sub(r"\s+", " ", text)
    return text


def _jaccard_similarity(a: str, b: str) -> float:
    tokens_a = set(a.split())
    tokens_b = set(b.split())
    if not tokens_a or not tokens_b:
        return 0.0
    return len(tokens_a & tokens_b) / len(tokens_a | tokens_b)


def cosine_similarity(a: List[float], b: List[float]) -> float:
    if not a or not b or len(a) != len(b):
        return 0.0
    dot = sum(x * y for x, y in zip(a, b))
    norm_a = math.sqrt(sum(x * x for x in a))
    norm_b = math.sqrt(sum(y * y for y in b))
    if norm_a == 0 or norm_b == 0:
        return 0.0
    return dot / (norm_a * norm_b)

@traceable(run_type="embedding", name="get_embedding")
def get_embedding(text: str) -> Optional[List[float]]:
    """
    调用 OpenAI Embedding API，将文本编码为向量。
    """
    try:
        resp = client.embeddings.create(
            model=EMBEDDING_MODEL,
            input=[text],
        )
        return resp.data[0].embedding
    except Exception as e:
        # 打到 Space 的 log，便于排查
        print(f"[Embedding error] {repr(e)}")
        return None


def find_similar_past_question(
    message: str,
    history: List[Tuple[str, str]],
    jaccard_threshold: float = 0.65,
    embedding_threshold: float = 0.85,
    max_turns_to_check: int = 6,
) -> Optional[Tuple[str, str, float]]:
    """
    在最近若干轮历史对话中查找与当前问题相似的既往问题。
    两级检测：先 Jaccard，再 Embedding。
    返回 (past_question, past_answer, similarity_score) 或 None
    """
    norm_msg = _normalize_text(message)
    if not norm_msg:
        return None

    # 1) Jaccard
    best_sim_j = 0.0
    best_pair_j: Optional[Tuple[str, str]] = None
    checked = 0

    for user_q, assistant_a in reversed(history):
        checked += 1
        if checked > max_turns_to_check:
            break

        norm_hist_q = _normalize_text(user_q)
        if not norm_hist_q:
            continue

        if norm_msg == norm_hist_q:
            return user_q, assistant_a, 1.0

        sim_j = _jaccard_similarity(norm_msg, norm_hist_q)
        if sim_j > best_sim_j:
            best_sim_j = sim_j
            best_pair_j = (user_q, assistant_a)

    if best_pair_j and best_sim_j >= jaccard_threshold:
        return best_pair_j[0], best_pair_j[1], best_sim_j

    # 2) Embedding 语义相似度
    if not history:
        return None

    msg_emb = get_embedding(message)
    if msg_emb is None:
        return None

    best_sim_e = 0.0
    best_pair_e: Optional[Tuple[str, str]] = None
    checked = 0

    for user_q, assistant_a in reversed(history):
        checked += 1
        if checked > max_turns_to_check:
            break

        hist_emb = get_embedding(user_q)
        if hist_emb is None:
            continue

        sim_e = cosine_similarity(msg_emb, hist_emb)
        if sim_e > best_sim_e:
            best_sim_e = sim_e
            best_pair_e = (user_q, assistant_a)

    if best_pair_e and best_sim_e >= embedding_threshold:
        return best_pair_e[0], best_pair_e[1], best_sim_e

    return None


@traceable(run_type="llm", name="safe_chat_completion")
def safe_chat_completion(
    model_name: str,
    messages: List[Dict[str, str]],
    lang: str,
    op: str = "chat",
    temperature: float = 0.5,
) -> str:
    """
    统一安全调用 OpenAI Chat Completion：
    - 最多尝试 2 次
    - 每次请求 timeout = 20 秒
    - 第一次用学生选择的模型；出错后，如果不是 DEFAULT_MODEL，则自动回退到 DEFAULT_MODEL 再试一次
    - 所有异常都会打印到后台 log，但对学生只返回友好的中/英文错误文案
    """
    preferred_model = model_name or DEFAULT_MODEL
    last_error: Optional[Exception] = None

    for attempt in range(2):
        # 第一次用学生指定模型，第二次（如果需要）切到默认模型
        if attempt == 0:
            current_model = preferred_model
        else:
            current_model = DEFAULT_MODEL

        try:
            resp = client.chat.completions.create(
                model=current_model,
                messages=messages,
                temperature=temperature,
                timeout=20,  # 20 秒超时
            )
            return resp.choices[0].message.content
        except Exception as e:
            print(
                f"[safe_chat_completion][{op}] attempt {attempt+1} "
                f"failed with model={current_model}: {repr(e)}"
            )
            last_error = e

            # 如果已经用的是默认模型，或者已经是第二次尝试，就跳出循环
            if current_model == DEFAULT_MODEL or attempt == 1:
                break

    # 两次都失败，返回友好的错误文案
    return build_error_message(last_error or Exception("unknown error"), lang, op)


# ---------- 构建 messages ----------
def build_messages(
    user_message: str,
    history: List[Tuple[str, str]],
    language_preference: str,
    learning_mode: str,
    doc_type: str,
    course_outline: Optional[List[str]],
    weaknesses: Optional[List[str]],
    cognitive_state: Optional[Dict[str, int]],
    rag_context: Optional[str] = None,   # 新增：RAG 检索结果
) -> List[Dict[str, str]]:
    messages: List[Dict[str, str]] = [
        {"role": "system", "content": CLARE_SYSTEM_PROMPT}
    ]

    # 学习模式
    if learning_mode in LEARNING_MODE_INSTRUCTIONS:
        mode_instruction = LEARNING_MODE_INSTRUCTIONS[learning_mode]
        messages.append(
            {
                "role": "system",
                "content": f"Current learning mode: {learning_mode}. {mode_instruction}",
            }
        )

    # 课程大纲
    topics = course_outline if course_outline else DEFAULT_COURSE_TOPICS
    topics_text = " | ".join(topics)
    messages.append(
        {
            "role": "system",
            "content": (
                "Here is the course syllabus context. Use this to stay aligned "
                "with the course topics when answering: "
                + topics_text
            ),
        }
    )

    # 上传文件类型提示
    if doc_type and doc_type != "Syllabus":
        messages.append(
            {
                "role": "system",
                "content": (
                    f"The student also uploaded a {doc_type} document as supporting material. "
                    "You do not see the full content directly, but you may assume it is relevant "
                    "to the same course and topics."
                ),
            }
        )

    # 学生弱项提示
    if weaknesses:
        weak_text = " | ".join(weaknesses[-5:])
        messages.append(
            {
                "role": "system",
                "content": (
                    "The student seems to struggle with the following questions or topics. "
                    "Be extra gentle and clear when these appear: " + weak_text
                ),
            }
        )

    # 认知状态提示
    if cognitive_state:
        confusion = cognitive_state.get("confusion", 0)
        mastery = cognitive_state.get("mastery", 0)
        if confusion >= 2 and confusion >= mastery + 1:
            messages.append(
                {
                    "role": "system",
                    "content": (
                        "The student is currently under HIGH cognitive load. "
                        "Use simpler language, shorter steps, and more concrete examples. "
                        "Avoid long derivations in a single answer, and check understanding "
                        "frequently."
                    ),
                }
            )
        elif mastery >= 2 and mastery >= confusion + 1:
            messages.append(
                {
                    "role": "system",
                    "content": (
                        "The student seems comfortable with the material. "
                        "You may increase difficulty slightly, introduce deeper follow-up "
                        "questions, and connect concepts across topics."
                    ),
                }
            )
        else:
            messages.append(
                {
                    "role": "system",
                    "content": (
                        "The student's cognitive state is mixed or uncertain. "
                        "Keep explanations clear and moderately paced, and probe for "
                        "understanding with short questions."
                    ),
                }
            )

    # 语言偏好控制
    if language_preference == "English":
        messages.append(
            {"role": "system", "content": "Please answer in English."}
        )
    elif language_preference == "中文":
        messages.append(
            {"role": "system", "content": "请用中文回答学生的问题。"}
        )

    # Session 内记忆摘要
    session_memory_text = build_session_memory_summary(
        history=history,
        weaknesses=weaknesses,
        cognitive_state=cognitive_state,
    )
    messages.append(
        {
            "role": "system",
            "content": (
                "Here is a short summary of this session's memory (only within the current chat; "
                "it is not persisted across sessions). Use it to stay consistent with the "
                "student's previous questions, difficulties, and cognitive state: "
                + session_memory_text
            ),
        }
    )

    # RAG 检索结果
    if rag_context:
        messages.append(
            {
                "role": "system",
                "content": (
                    "Here are some relevant excerpts from the course materials. "
                    "Use them as the primary factual grounding when answering the student's question. "
                    "If there is any conflict between these excerpts and your prior knowledge, "
                    "prefer the excerpts.\n\n"
                    + rag_context
                ),
            }
        )

    # 历史对话
    for user, assistant in history:
        messages.append({"role": "user", "content": user})
        if assistant is not None:
            messages.append({"role": "assistant", "content": assistant})

    # 当前输入
    messages.append({"role": "user", "content": user_message})
    return messages

# 装饰器
@traceable(run_type="chain", name="chat_with_clare")
def chat_with_clare(
    message: str,
    history: List[Tuple[str, str]],
    model_name: str,
    language_preference: str,
    learning_mode: str,
    doc_type: str,
    course_outline: Optional[List[str]],
    weaknesses: Optional[List[str]],
    cognitive_state: Optional[Dict[str, int]],
    rag_context: Optional[str] = None,
) -> Tuple[str, List[Tuple[str, str]]]:
    try:
        set_run_metadata(
            learning_mode=learning_mode,
            language_preference=language_preference,
            doc_type=doc_type,
        )
    except Exception as e:
        print(f"[LangSmith metadata error in chat_with_clare] {repr(e)}")


    # 构建 messages
    messages = build_messages(
        user_message=message,
        history=history,
        language_preference=language_preference,
        learning_mode=learning_mode,
        doc_type=doc_type,
        course_outline=course_outline,
        weaknesses=weaknesses,
        cognitive_state=cognitive_state,
        rag_context=rag_context,
    )

    # 统一安全调用
    answer = safe_chat_completion(
        model_name=model_name,
        messages=messages,
        lang=language_preference,
        op="chat",
        temperature=0.5,
    )

    history = history + [(message, answer)]
    return answer, history


# ---------- 导出对话为 Markdown ----------
def export_conversation(
    history: List[Tuple[str, str]],
    course_outline: List[str],
    learning_mode_val: str,
    weaknesses: List[str],
    cognitive_state: Optional[Dict[str, int]],
) -> str:
    lines: List[str] = []
    lines.append("# Clare – Conversation Export\n")
    lines.append(f"- Learning mode: **{learning_mode_val}**\n")
    lines.append("- Course topics (short): " + "; ".join(course_outline[:5]) + "\n")
    lines.append(f"- Cognitive state snapshot: {describe_cognitive_state(cognitive_state)}\n")

    if weaknesses:
        lines.append("- Observed student difficulties:\n")
        for w in weaknesses[-5:]:
            lines.append(f"  - {w}\n")
    lines.append("\n---\n\n")

    for user, assistant in history:
        lines.append(f"**Student:** {user}\n\n")
        lines.append(f"**Clare:** {assistant}\n\n")
        lines.append("---\n\n")

    return "".join(lines)


# ---------- 生成 3 个 quiz 题目 ----------
from langsmith import traceable

@traceable(run_type="chain", name="generate_quiz_from_history")
def generate_quiz_from_history(
    history: List[Tuple[str, str]],
    course_outline: List[str],
    weaknesses: List[str],
    cognitive_state: Optional[Dict[str, int]],
    model_name: str,
    language_preference: str,
) -> str:
    conversation_text = ""
    for user, assistant in history[-8:]:
        conversation_text += f"Student: {user}\nClare: {assistant}\n"

    topics_text = "; ".join(course_outline[:8])
    weakness_text = "; ".join(weaknesses[-5:]) if weaknesses else "N/A"
    cog_text = describe_cognitive_state(cognitive_state)

    messages = [
        {"role": "system", "content": CLARE_SYSTEM_PROMPT},
        {
            "role": "system",
            "content": (
                "Now your task is to create a **short concept quiz** for the student. "
                "Based on the conversation and course topics, generate **3 questions** "
                "(a mix of multiple-choice and short-answer is fine). After listing the "
                "questions, provide an answer key at the end under a heading 'Answer Key'. "
                "Number the questions Q1, Q2, Q3. Adjust the difficulty according to the "
                "student's cognitive state."
            ),
        },
        {
            "role": "system",
            "content": f"Course topics: {topics_text}",
        },
        {
            "role": "system",
            "content": f"Student known difficulties: {weakness_text}",
        },
        {
            "role": "system",
            "content": f"Student cognitive state: {cog_text}",
        },
        {
            "role": "user",
            "content": (
                "Here is the recent conversation between you and the student:\n\n"
                + conversation_text
                + "\n\nPlease create the quiz now."
            ),
        },
    ]

    if language_preference == "中文":
        messages.append(
            {
                "role": "system",
                "content": "请用中文给出问题和答案。",
            }
        )

    quiz_text = safe_chat_completion(
        model_name=model_name,
        messages=messages,
        lang=language_preference,
        op="quiz",
        temperature=0.5,
    )
    return quiz_text


# ---------- 概念总结（知识点摘要） ----------
@traceable(run_type="chain", name="summarize_conversation")
def summarize_conversation(
    history: List[Tuple[str, str]],
    course_outline: List[str],
    weaknesses: List[str],
    cognitive_state: Optional[Dict[str, int]],
    model_name: str,
    language_preference: str,
) -> str:
    conversation_text = ""
    for user, assistant in history[-10:]:
        conversation_text += f"Student: {user}\nClare: {assistant}\n"

    topics_text = "; ".join(course_outline[:8])
    weakness_text = "; ".join(weaknesses[-5:]) if weaknesses else "N/A"
    cog_text = describe_cognitive_state(cognitive_state)

    messages = [
        {"role": "system", "content": CLARE_SYSTEM_PROMPT},
        {
            "role": "system",
            "content": (
                "Your task now is to produce a **concept-only summary** of this tutoring "
                "session. Only include knowledge points, definitions, key formulas, "
                "examples, and main takeaways. Do **not** include any personal remarks, "
                "jokes, or off-topic chat. Write in clear bullet points. This summary "
                "should be suitable for the student to paste into their study notes. "
                "Take into account what the student struggled with and their cognitive state."
            ),
        },
        {
            "role": "system",
            "content": f"Course topics context: {topics_text}",
        },
        {
            "role": "system",
            "content": f"Student known difficulties: {weakness_text}",
        },
        {
            "role": "system",
            "content": f"Student cognitive state: {cog_text}",
        },
        {
            "role": "user",
            "content": (
                "Here is the recent conversation between you and the student:\n\n"
                + conversation_text
                + "\n\nPlease summarize only the concepts and key ideas learned."
            ),
        },
    ]

    if language_preference == "中文":
        messages.append(
            {
                "role": "system",
                "content": "请用中文给出要点总结，只保留知识点和结论，使用条目符号。"
            }
        )

    summary_text = safe_chat_completion(
        model_name=model_name,
        messages=messages,
        lang=language_preference,
        op="summary",
        temperature=0.4,
    )
    return summary_text