init

.gitignore

@@ -0,0 +1,32 @@
+# --- Python ---
+__pycache__/
+*.py[cod]
+*$py.class
+
+# --- Virtual env ---
+.venv/
+venv/
+ENV/
+env/
+
+# --- OS / IDE ---
+.DS_Store
+Thumbs.db
+.idea/
+.vscode/
+*.iml
+
+# --- Logs ---
+*.log
+
+# --- Gradio / cache ---
+.gradio/
+.cache/
+
+# --- Local data (do NOT commit exam results / sessions) ---
+data/sessions.json
+data/results.json
+
+# If you ever store secrets locally:
+.env
+.env.*

README.md

@@ -0,0 +1,43 @@
+---
+title: AI Examiner Agent
+emoji: 📝
+colorFrom: blue
+colorTo: green
+sdk: gradio
+sdk_version: "4.0.0"
+app_file: app.py
+pinned: false
+---
+
+# AI Examiner Agent
+
+Міні-екзаменатор (LLM-агент з tools): питає ім’я та email, обирає 2–3 теми, веде короткий діалог по темах, виставляє оцінку (0..10) та зберігає історію іспиту у файли.
+
+## Функціонал
+
+- Збір **name + email**
+- Tool `start_exam(email, name) -> list[str]`:
+  - перевіряє, чи є студент у `data/students.json`
+  - обирає 2–3 теми з `data/topics.json`
+  - створює сесію у `data/sessions.json`
+- Tool `get_next_topic() -> str`:
+  - повертає наступну тему з черги сесії
+- Tool `end_exam(email, score, history)`:
+  - зберігає результат у `data/results.json`
+  - `history` зберігається у форматі `{role, content, datetime}`
+
+## Структура проєкту
+
+```text
+.
+├── app.py
+├── agent.py
+├── llm.py
+├── tools.py
+├── requirements.txt
+└── data
+    ├── students.json
+    ├── topics.json
+    ├── sessions.json
+    └── results.json
+```

agent.py

@@ -0,0 +1,248 @@
+from __future__ import annotations
+
+import json
+import re
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from typing import Dict, List, Optional, Tuple
+
+from llm import LLMError, chat_completion
+from tools import (
+    Message,
+    StudentNotFoundError,
+    end_exam,
+    get_last_session_id,
+    get_next_topic,
+    set_current_session,
+    start_exam,
+)
+
+EMAIL_RE = re.compile(r"([a-zA-Z0-9_.+\-]+@[a-zA-Z0-9\-]+\.[a-zA-Z0-9\-.]+)")
+
+
+def _utc_now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _looks_like_idk(text: str) -> bool:
+    t = text.strip().lower()
+    triggers = ["не знаю", "не пам'ятаю", "не памятаю", "не можу", "no idea", "idk", "не впевнений", "не впевнена"]
+    return any(x in t for x in triggers)
+
+
+@dataclass
+class ExamAgent:
+    stage: str = "collect_identity"  
+    name: Optional[str] = None
+    email: Optional[str] = None
+
+    session_id: Optional[str] = None
+    topics_total: int = 0
+    current_topic: Optional[str] = None
+
+    max_questions_per_topic: int = 3
+    questions_in_topic: int = 0
+
+    topic_scores: Dict[str, float] = field(default_factory=dict)
+    history: List[Message] = field(default_factory=list)
+
+    def _log(self, role: str, content: str) -> None:
+        self.history.append(Message(role=role, content=content, datetime=_utc_now_iso()))
+
+    def initial_message(self) -> str:
+        msg = "Привіт! Я екзаменатор. Як тебе звати?"
+        self._log("system", msg)
+        return msg
+
+    def _bind_session(self) -> None:
+        if self.session_id:
+            set_current_session(self.session_id)
+
+    def _start_exam_tools(self) -> Tuple[bool, str]:
+        self._log("tool_call", f"start_exam(email={self.email}, name={self.name})")
+        try:
+            topics = start_exam(self.email or "", self.name or "")
+        except StudentNotFoundError:
+            self.session_id = None
+            self.topics_total = 0
+            self.current_topic = None
+            self.stage = "collect_identity"
+            msg = (
+                "Я не знайшов(ла) студента з таким email у списку. "
+                "Перевір, будь ласка, email і надішли його ще раз."
+            )
+            self._log("system", msg)
+            return False, msg
+
+        self.session_id = get_last_session_id()
+        self.topics_total = len(topics)
+
+        msg = f"Добре, {self.name}. Починаємо іспит. Тем буде {self.topics_total}."
+        self._log("system", msg)
+        return True, msg
+
+    def _next_topic(self) -> Optional[str]:
+        self._bind_session()
+        topic = get_next_topic()
+        if topic:
+            self.current_topic = topic
+            self.questions_in_topic = 0
+            return topic
+        self.current_topic = None
+        return None
+
+    def _ask_question(self, api_key: str, model: str, base_url: str) -> str:
+        assert self.current_topic is not None
+
+        sys = "Ти екзаменатор. Питай коротко українською. Одне питання за раз."
+        user = f"Тема: {self.current_topic}\nЗадай наступне питання."
+        try:
+            q = chat_completion(
+                api_key=api_key,
+                model=model,
+                base_url=base_url,
+                messages=[{"role": "system", "content": sys}, {"role": "user", "content": user}],
+            ).strip()
+        except LLMError:
+            q = f"Поясни ключові поняття та приклад по темі: {self.current_topic}."
+
+        self.questions_in_topic += 1
+        self._log("system", q)
+        return q
+
+    def _evaluate_answer(self, api_key: str, model: str, base_url: str, answer: str) -> Dict[str, object]:
+        assert self.current_topic is not None
+
+        if _looks_like_idk(answer):
+            return {"score": 0.0, "action": "next_topic", "note": "student_idk", "feedback": ""}
+
+        sys = "Оціни відповідь студента для однієї теми. Поверни ТІЛЬКИ JSON."
+        user = {
+            "topic": self.current_topic,
+            "answer": answer,
+            "json_schema": {"score": "0..10", "action": "ask_followup|next_topic", "feedback": "string"},
+        }
+
+        try:
+            raw = chat_completion(
+                api_key=api_key,
+                model=model,
+                base_url=base_url,
+                messages=[{"role": "system", "content": sys}, {"role": "user", "content": json.dumps(user, ensure_ascii=False)}],
+            ).strip()
+            data = json.loads(raw)
+
+            score = float(data.get("score", 0.0))
+            score = max(0.0, min(10.0, score))
+
+            action = str(data.get("action", "ask_followup")).strip()
+            if action not in {"ask_followup", "next_topic"}:
+                action = "ask_followup"
+
+            feedback = str(data.get("feedback", "")).strip()
+            return {"score": score, "action": action, "feedback": feedback}
+        except Exception:
+            length = len(answer.strip())
+            score = 3.0 if length < 80 else 6.0 if length < 250 else 7.5
+            action = "next_topic" if score >= 7.0 else "ask_followup"
+            return {"score": score, "action": action, "feedback": "Оцінка приблизна (fallback)."}
+
+    def _finalize(self) -> str:
+        avg = round((sum(self.topic_scores.values()) / max(1, len(self.topic_scores))) if self.topic_scores else 0.0, 1)
+
+        strong = [t for t, s in self.topic_scores.items() if s >= 7.0]
+        weak = [t for t, s in self.topic_scores.items() if s < 7.0]
+
+        feedback_lines = []
+        if strong:
+            feedback_lines.append("Добре вийшло по темах: " + ", ".join(strong) + ".")
+        if weak:
+            feedback_lines.append("Варто підтягнути: " + ", ".join(weak) + ".")
+        if not feedback_lines:
+            feedback_lines.append("Дякую! Є над чим попрацювати — продовжуй практикуватися.")
+
+        msg = f"Іспит завершено. Оцінка: {avg}/10.\n" + "\n".join(feedback_lines)
+        self._log("system", msg)
+
+        self._log("tool_call", f"end_exam(email={self.email}, score={avg}, history=[...])")
+        end_exam(self.email or "", avg, [m.to_dict() for m in self.history])
+
+        self.stage = "finished"
+        return msg
+
+    def step(self, user_text: str, api_key: str, model: str, base_url: str) -> str:
+        user_text = (user_text or "").strip()
+        self._log("user", user_text)
+
+        if self.stage == "collect_identity":
+            if not self.name:
+                self.name = user_text if user_text else None
+                if not self.name:
+                    msg = "Як тебе звати?"
+                    self._log("system", msg)
+                    return msg
+                msg = f"Приємно познайомитись, {self.name}! Тепер напиши свій email."
+                self._log("system", msg)
+                return msg
+
+            if not self.email:
+                m = EMAIL_RE.search(user_text)
+                self.email = m.group(1) if m else None
+                if not self.email:
+                    msg = "Не бачу коректного email. Спробуй ще раз (наприклад: name@example.com)."
+                    self._log("system", msg)
+                    return msg
+
+                ok, start_msg = self._start_exam_tools()
+                if not ok:
+                    self.email = None
+                    return start_msg
+
+                self.stage = "exam"
+                topic = self._next_topic()
+                if not topic:
+                    return self._finalize()
+
+                intro = f"Тема 1/{self.topics_total}: {topic}"
+                self._log("system", intro)
+                q = self._ask_question(api_key, model, base_url)
+                return intro + "\n" + q
+
+        if self.stage == "exam":
+            if not self.current_topic:
+                return self._finalize()
+
+            eval_res = self._evaluate_answer(api_key, model, base_url, user_text)
+            score = float(eval_res.get("score", 0.0))
+            action = str(eval_res.get("action", "ask_followup"))
+            feedback = str(eval_res.get("feedback", "")).strip()
+
+            prev = self.topic_scores.get(self.current_topic, 0.0)
+            self.topic_scores[self.current_topic] = max(prev, score)
+
+            idk = (eval_res.get("note") == "student_idk")
+            too_many = self.questions_in_topic >= self.max_questions_per_topic
+            good_enough = score >= 7.0
+
+            if idk or good_enough or too_many or action == "next_topic":
+                note = f"Коментар: {feedback}\n" if feedback else ""
+                next_t = self._next_topic()
+                if not next_t:
+                    return note + self._finalize()
+
+                done = len({t for t in self.topic_scores.keys()})
+                header = f"{note}Переходимо далі.\nТема {min(done+1, self.topics_total)}/{self.topics_total}: {next_t}"
+                self._log("system", header)
+                q = self._ask_question(api_key, model, base_url)
+                return header + "\n" + q
+
+            follow = "Ок. Уточню:"
+            self._log("system", follow)
+            q = self._ask_question(api_key, model, base_url)
+            if feedback:
+                return f"Коментар: {feedback}\n{follow}\n{q}"
+            return f"{follow}\n{q}"
+
+        msg = "Іспит уже завершено. Якщо хочеш — натисни Reset і почнемо заново."
+        self._log("system", msg)
+        return msg

app.py

@@ -0,0 +1,87 @@
+from __future__ import annotations
+
+import gradio as gr
+from agent import ExamAgent
+
+
+def new_agent() -> ExamAgent:
+    return ExamAgent()
+
+
+def on_load():
+    agent = new_agent()
+    first = agent.initial_message()
+    chat_messages = [{"role": "assistant", "content": first}]
+    return agent, chat_messages
+
+
+def on_reset():
+    return on_load()
+
+
+def on_user_message(
+    agent: ExamAgent,
+    chat_messages,
+    user_text: str,
+    api_key: str,
+    model: str,
+    base_url: str,
+):
+    if agent is None:
+        agent = new_agent()
+
+    if chat_messages is None:
+        chat_messages = []
+
+    user_text = (user_text or "").strip()
+    if not user_text:
+        return agent, chat_messages, ""
+
+    chat_messages.append({"role": "user", "content": user_text})
+
+    try:
+        reply = agent.step(user_text, api_key=api_key, model=model, base_url=base_url)
+    except Exception as e:
+        reply = f"Сталася помилка: {e}"
+
+    chat_messages.append({"role": "assistant", "content": reply})
+    return agent, chat_messages, ""
+
+
+with gr.Blocks(title="AI Examiner Agent") as demo:
+    gr.Markdown(
+      "# AI Examiner Agent\n"
+      "Сервіс проводить міні-іспит: питає ім’я та email, обирає 2–3 теми, ставить питання, "
+      "оцінює відповіді та зберігає результат у файли."
+)
+
+    with gr.Row():
+        api_key = gr.Textbox(label="LLM API Key", type="password", placeholder="Встав ключ тут")
+        model = gr.Textbox(label="Model", value="gpt-4o-mini")
+        base_url = gr.Textbox(label="Base URL", value="https://api.openai.com")
+
+    chatbot = gr.Chatbot(label="Exam Chat", height=420)
+
+    with gr.Row():
+        user_in = gr.Textbox(label="Твоє повідомлення", placeholder="Напиши відповідь…", scale=4)
+        send = gr.Button("Send", scale=1)
+        reset = gr.Button("Reset", scale=1)
+
+    agent_state = gr.State()
+
+    demo.load(on_load, outputs=[agent_state, chatbot])
+    reset.click(on_reset, outputs=[agent_state, chatbot])
+
+    send.click(
+        on_user_message,
+        inputs=[agent_state, chatbot, user_in, api_key, model, base_url],
+        outputs=[agent_state, chatbot, user_in],
+    )
+    user_in.submit(
+        on_user_message,
+        inputs=[agent_state, chatbot, user_in, api_key, model, base_url],
+        outputs=[agent_state, chatbot, user_in],
+    )
+
+if __name__ == "__main__":
+    demo.launch()

data/students.json

@@ -0,0 +1,4 @@
+[
+  { "email": "student1@example.com", "name": "Student One" },
+  { "email": "student2@example.com", "name": "Student Two" }
+]

data/topics.json

@@ -0,0 +1,12 @@
+[
+  "HDFS: реплікація та читання",
+  "Git: push/SSH і типові проблеми",
+  "REST API: CRUD та валідація",
+  "MongoDB: індекси під патерни доступу",
+  "CI/CD: пайплайни та деплой",
+  "Моніторинг і логування сервісів",
+  "LLM агенти: tools і state",
+  "RAG: базова архітектура",
+  "HTTP: статус-коди та помилки",
+  "Docker: образи і змінні середовища"
+]

llm.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+import json
+from typing import Any, Dict, List, Optional
+import requests
+
+
+class LLMError(RuntimeError):
+    pass
+
+
+def chat_completion(
+    api_key: str,
+    messages: List[Dict[str, str]],
+    model: str = "gpt-4o-mini",
+    base_url: str = "https://api.openai.com",
+    timeout_s: int = 45,
+) -> str:
+    """
+    Minimal OpenAI-compatible Chat Completions call.
+    Expects provider that supports:
+      POST {base_url}/v1/chat/completions
+    """
+    if not api_key or not api_key.strip():
+        raise LLMError("LLM API key is empty.")
+
+    url = base_url.rstrip("/") + "/v1/chat/completions"
+    headers = {
+        "Authorization": f"Bearer {api_key.strip()}",
+        "Content-Type": "application/json",
+    }
+    payload: Dict[str, Any] = {
+        "model": model,
+        "messages": messages,
+        "temperature": 0.3,
+    }
+
+    try:
+        r = requests.post(url, headers=headers, json=payload, timeout=timeout_s)
+    except Exception as e:
+        raise LLMError(f"Request failed: {e}") from e
+
+    if r.status_code >= 400:
+        raise LLMError(f"LLM HTTP {r.status_code}: {r.text[:400]}")
+
+    data = r.json()
+    try:
+        return data["choices"][0]["message"]["content"]
+    except Exception:
+        raise LLMError(f"Unexpected LLM response shape: {json.dumps(data)[:600]}")

requirements.txt

@@ -0,0 +1,2 @@
+gradio>=4.0.0
+requests>=2.31.0

tools.py

@@ -0,0 +1,185 @@
+from __future__ import annotations
+
+import json
+import random
+import threading
+import uuid
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+import contextvars
+
+DATA_DIR = Path(__file__).parent / "data"
+STUDENTS_PATH = DATA_DIR / "students.json"
+TOPICS_PATH = DATA_DIR / "topics.json"
+SESSIONS_PATH = DATA_DIR / "sessions.json"
+RESULTS_PATH = DATA_DIR / "results.json"
+
+_FILE_LOCK = threading.Lock()
+
+_current_session_id: contextvars.ContextVar[Optional[str]] = contextvars.ContextVar(
+    "current_session_id", default=None
+)
+
+_LAST_SESSION_ID: Optional[str] = None
+
+
+class StudentNotFoundError(Exception):
+    pass
+
+
+def _utc_now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _read_json(path: Path, default: Any) -> Any:
+    with _FILE_LOCK:
+        if not path.exists():
+            path.parent.mkdir(parents=True, exist_ok=True)
+            path.write_text(json.dumps(default, ensure_ascii=False, indent=2), encoding="utf-8")
+            return default
+        txt = path.read_text(encoding="utf-8").strip()
+        if not txt:
+            return default
+        return json.loads(txt)
+
+
+def _write_json(path: Path, data: Any) -> None:
+    with _FILE_LOCK:
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
+
+
+def ensure_data_files() -> None:
+    _read_json(STUDENTS_PATH, [])
+    _read_json(TOPICS_PATH, [])
+    _read_json(SESSIONS_PATH, {})
+    _read_json(RESULTS_PATH, [])
+
+
+@dataclass
+class Message:
+    role: str  
+    content: str
+    datetime: str
+
+    def to_dict(self) -> Dict[str, str]:
+        return {"role": self.role, "content": self.content, "datetime": self.datetime}
+
+
+def set_current_session(session_id: Optional[str]) -> None:
+    _current_session_id.set(session_id)
+
+
+def get_last_session_id() -> Optional[str]:
+    return _LAST_SESSION_ID
+
+
+def _load_students() -> List[Dict[str, str]]:
+    return _read_json(STUDENTS_PATH, [])
+
+
+def _load_topics() -> List[str]:
+    return _read_json(TOPICS_PATH, [])
+
+
+def _load_sessions() -> Dict[str, Any]:
+    return _read_json(SESSIONS_PATH, {})
+
+
+def _save_sessions(sessions: Dict[str, Any]) -> None:
+    _write_json(SESSIONS_PATH, sessions)
+
+
+def _append_result(record: Dict[str, Any]) -> None:
+    results = _read_json(RESULTS_PATH, [])
+    if not isinstance(results, list):
+        results = []
+    results.append(record)
+    _write_json(RESULTS_PATH, results)
+
+
+def start_exam(email: str, name: str) -> List[str]:
+    global _LAST_SESSION_ID
+
+    ensure_data_files()
+
+    students = _load_students()
+    found = next((s for s in students if s.get("email", "").strip().lower() == email.strip().lower()), None)
+    if not found:
+        raise StudentNotFoundError(f"Student not found for email: {email}")
+
+    topics = _load_topics()
+    if len(topics) < 2:
+        topics = ["General knowledge: topic A", "General knowledge: topic B", "General knowledge: topic C"]
+
+    k = 2 if len(topics) < 3 else random.choice([2, 3])
+    selected = random.sample(topics, k=min(k, len(topics)))
+
+    sessions = _load_sessions()
+    session_id = str(uuid.uuid4())
+    sessions[session_id] = {
+        "session_id": session_id,
+        "email": email,
+        "name": name,
+        "topics_queue": selected,
+        "topics_all": selected,
+        "started_at": _utc_now_iso(),
+        "finished_at": None,
+        "status": "active",
+    }
+    _save_sessions(sessions)
+
+    _LAST_SESSION_ID = session_id
+    set_current_session(session_id)
+
+    return selected
+
+
+def get_next_topic() -> str:
+    """
+    IMPORTANT FIX:
+    Gradio callbacks may run in a different context, so contextvar can be empty.
+    We fallback to _LAST_SESSION_ID.
+    """
+    ensure_data_files()
+
+    session_id = _current_session_id.get() or _LAST_SESSION_ID
+    if not session_id:
+        return ""
+
+    sessions = _load_sessions()
+    sess = sessions.get(session_id)
+    if not sess or sess.get("status") != "active":
+        return ""
+
+    queue = sess.get("topics_queue", [])
+    if not queue:
+        return ""
+
+    next_topic = queue.pop(0)
+    sess["topics_queue"] = queue
+    sessions[session_id] = sess
+    _save_sessions(sessions)
+    return str(next_topic)
+
+
+def end_exam(email: str, score: float, history: List[Dict[str, str]]) -> None:
+    ensure_data_files()
+
+    sessions = _load_sessions()
+    for sid, sess in list(sessions.items()):
+        if sess.get("email", "").strip().lower() == email.strip().lower() and sess.get("status") == "active":
+            sess["status"] = "finished"
+            sess["finished_at"] = _utc_now_iso()
+            sessions[sid] = sess
+    _save_sessions(sessions)
+
+    record = {
+        "email": email,
+        "score": float(score),
+        "history": history,
+        "saved_at": _utc_now_iso(),
+    }
+    _append_result(record)