scripts/expert.py

"""Gemini/DeepSeek-backed expert for concept extraction and student dialogue."""

import json
import os


EXPERT_SYSTEM_PROMPT = """You are a knowledgeable subject-matter expert on the \
study material provided below. A student is going to question, probe, and \
challenge you in order to deepen their own understanding. You are not a tutor \
who quizzes them; they lead the inquiry.

Rules:
- Answer directly and substantively, grounded in the material. Do not pad.
- Match the student's depth. A plain factual question gets a plain answer; a \
question that probes mechanism or challenges an assumption gets a deeper, \
reasoned response.
- When the student challenges you or surfaces a genuine edge case, limitation, \
or counterexample, acknowledge it honestly and engage with it. Do not be \
defensive, and do not pretend the material is more complete than it is.
- If they ask something the material does not cover, say so plainly and reason \
from first principles, flagging that you are going beyond the source.
- Never quiz the student back or end with "does that make sense?". They are the \
one asking the questions.
- You may use light markdown (bold, lists) when it aids clarity.

Study material:
---
{material}
---
"""

CONCEPT_PROMPT = """From the study material below, extract the {n} most \
important distinct concepts or topics a student would want to understand. \
Return a JSON object with a single key "concepts" whose value is an array of \
short concept names (2-5 words each), ordered from most to least central.

Material:
---
{material}
---
Return only the JSON object."""


class Expert:
    """Concept extraction and expert dialogue, via DeepSeek or Gemini."""

    def __init__(self, product_config):
        """Initialise the client for the configured provider."""
        self.cfg = product_config
        self.provider = getattr(product_config, "provider", "deepseek")
        if self.provider == "deepseek":
            self._init_deepseek()
        elif self.provider == "gemini":
            self._init_gemini()
        else:
            raise ValueError(f"Unknown provider: {self.provider}")

    def _init_deepseek(self):
        """Set up the DeepSeek (OpenAI-compatible) client."""
        from openai import OpenAI

        api_key = os.environ.get("DEEPSEEK_API_KEY")
        if not api_key:
            raise SystemExit(
                "DEEPSEEK_API_KEY is not set. Add it to .env, or set "
                "ProductConfig.provider = 'gemini' to use Gemini instead."
            )
        self.client = OpenAI(api_key=api_key, base_url="https://api.deepseek.com")
        self.model = self.cfg.deepseek_model

    def _init_gemini(self):
        """Set up the Gemini client."""
        from google import genai

        api_key = os.environ.get("GEMINI_API_KEY")
        if not api_key:
            raise SystemExit("GEMINI_API_KEY is not set. Add it to .env.")
        self.client = genai.Client(api_key=api_key)
        self.model = self.cfg.gemini_model

    # ---- concept extraction ----

    def extract_concepts(self, material_text):
        """Ask the model for the key concepts in the material."""
        material = material_text[: self.cfg.max_context_chars]
        prompt = CONCEPT_PROMPT.format(n=self.cfg.num_concepts, material=material)
        if self.provider == "deepseek":
            raw = self._deepseek_json(prompt)
        else:
            raw = self._gemini_json(prompt)
        try:
            parsed = json.loads(raw)
            concepts = parsed.get("concepts", []) if isinstance(parsed, dict) else []
        except (json.JSONDecodeError, TypeError):
            concepts = []
        return [c.strip() for c in concepts if isinstance(c, str) and c.strip()]

    def _deepseek_json(self, prompt):
        """One DeepSeek call returning a JSON object string."""
        response = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"},
            temperature=0.3,
        )
        return response.choices[0].message.content

    def _gemini_json(self, prompt):
        """One Gemini call returning a JSON object string."""
        from google.genai import types

        response = self.client.models.generate_content(
            model=self.model,
            contents=prompt,
            config=types.GenerateContentConfig(
                response_mime_type="application/json",
                temperature=0.3,
            ),
        )
        return response.text

    # ---- expert dialogue ----

    def answer(self, question, material_text, history):
        """Answer a student question given material context and chat history."""
        material = material_text[: self.cfg.max_context_chars]
        system = EXPERT_SYSTEM_PROMPT.format(material=material)
        if self.provider == "deepseek":
            return self._deepseek_answer(system, question, history)
        return self._gemini_answer(system, question, history)

    def _deepseek_answer(self, system, question, history):
        """Send question to DeepSeek and return the answer."""
        messages = [{"role": "system", "content": system}]
        for turn in history[-8:]:
            role = "user" if turn["role"] == "student" else "assistant"
            messages.append({"role": role, "content": turn["text"]})
        messages.append({"role": "user", "content": question})
        response = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            temperature=0.7,
        )
        return response.choices[0].message.content.strip()

    def _gemini_answer(self, system, question, history):
        """Send question to Gemini and return the answer."""
        from google.genai import types

        transcript = []
        for turn in history[-8:]:
            speaker = "Student" if turn["role"] == "student" else "Expert"
            transcript.append(f"{speaker}: {turn['text']}")
        transcript.append(f"Student: {question}")
        conversation = "\n\n".join(transcript)
        response = self.client.models.generate_content(
            model=self.model,
            contents=conversation,
            config=types.GenerateContentConfig(
                system_instruction=system,
                temperature=0.7,
            ),
        )
        return response.text.strip()