20  Code: Agent 5 — PhD Supervisor

Folder: 02_agents_code/agent_05_phd_supervisor/ — self-contained; run python agent_05_phd_supervisor/main.py --help.

20.1 README

20.1.1 README.md

# Agent 5 — AI PhD Supervisor Agent ("Virtual Research Mentor")

A **long-running, memory-backed mentor**: it keeps a per-student roadmap and an append-only progress log, evaluates weekly updates against the plan, challenges the work like a committee member, and prepares slides/posters/defence Q&A.

```
init     ─► ResearchPlanningAgent  → memory/<student>/roadmap.json + outputs/<student>/roadmap.md
weekly   ─► WeeklyProgressAgent    → progress_log.jsonl + outputs/<student>/week_NN.md
critique ─► CriticalReviewerAgent  → critique_week_NN.md
present  ─► PresentationAgent      → slides.md | poster.md | defence.md
```

## Files

| File | Purpose |
|------|---------|
| `main.py` | Sub-commands: `init`, `weekly`, `critique`, `present`, `status` |
| `agents.py` | Roadmap (structured), weekly evaluation (structured), critic, presentation coach |
| `memory.py` | File-based memory: profile, roadmap, JSONL log, `context_summary()` for prompts |
| `llm.py` | Thin Claude client |
| `memory/` | One folder per student (audit trail the supervisor can read) |
| `outputs/` | Human-readable reports |

## Run

```powershell
python agent_05_phd_supervisor\main.py init --student "Ali" --topic "Multimodal AI for drought resilience in wheat" --years 3
python agent_05_phd_supervisor\main.py weekly --student "Ali" --update "Finished RGB pipeline, accuracy 87%, hyperspectral data delayed"
python agent_05_phd_supervisor\main.py critique --student "Ali"
python agent_05_phd_supervisor\main.py present --student "Ali" --type defence
python agent_05_phd_supervisor\main.py status --student "Ali"
```

## What this agent teaches

1. **Memory & state** — the agent is only as good as the context you give it; `context_summary()` sends roadmap + last N updates, not everything.
2. **Append-only logs** as an audit trail (and as supervision evidence for accreditation).
3. **Structured evaluation** — progress %, status, problems, goals: renderable in a dashboard later.
4. **Responsible AI framing** — every output carries a "supports, not replaces, your supervisor" notice.

## Ideas to extend

- Supervisor dashboard (Streamlit) across students with traffic-light status; reminders via WhatsApp/email; chapter review; mock viva with voice; cohort analytics.

---

**Codanics — Python ka Chilla: Build AI Agents**
Course: https://codanics.com/courses/python-ka-chilla-build-ai-agents/ · Website: https://codanics.com · YouTube: [@codanics](https://www.youtube.com/@codanics)
WhatsApp: https://whatsapp.com/channel/0029Va7nRDq3QxRzGqaQvS3r · Free course: https://www.youtube.com/watch?v=UE3o-7KRNXs&list=PLaQmdklCcX40

20.2 Source files

20.2.1 agents.py

"""
agents.py — the four sub-agents of the PhD Supervisor ("Virtual Research Mentor") Agent.

    ResearchPlanningAgent : profile ─► Roadmap (structured, milestones per semester)
    WeeklyProgressAgent   : memory + update ─► Evaluation (structured: %, status, problems, goals)
    CriticalReviewerAgent : memory ─► critique.md (assumptions, methodology, novelty)
    PresentationAgent     : memory ─► slides / poster / defence Q&A (Markdown)

Codanics — Python ka Chilla: Build AI Agents
https://codanics.com/courses/python-ka-chilla-build-ai-agents/
"""
from __future__ import annotations

from pydantic import BaseModel, Field

from llm import ask, ask_structured

BRAND = ("Codanics — Python ka Chilla: Build AI Agents | "
         "https://codanics.com/courses/python-ka-chilla-build-ai-agents/")

NOTICE = ("\n\n> **Note.** This feedback is generated by an AI research mentor to *support* — not replace — "
          "your supervisor. Verify important decisions with your supervisor and follow your "
          "institution's academic-integrity policy.\n")


# ------------------------------------------------------------------- 1. Roadmap
class Milestone(BaseModel):
    semester: int = Field(description="1..N (two semesters per year)")
    title: str
    deliverables: list[str]
    success_criteria: str
    risks: list[str]


class Roadmap(BaseModel):
    programme_title: str
    aims: list[str] = Field(description="3-4 research aims")
    milestones: list[Milestone]
    publications_plan: list[str] = Field(description="Planned papers/conferences with target semester")
    skills_to_learn: list[str]
    weekly_routine: list[str] = Field(description="Recommended weekly habits, e.g. 'write 500 words'")


class ResearchPlanningAgent:
    SYSTEM = ("You are an experienced PhD supervisor and programme director. Create a realistic, "
              "milestone-based PhD roadmap with two semesters per year, front-loading literature review and "
              "methods, leaving the final year for writing and defence.")

    def run(self, name: str, topic: str, years: int, background: str = "") -> Roadmap:
        return ask_structured(self.SYSTEM,
                              f"Student: {name}\nTopic: {topic}\nProgramme length: {years} years\n"
                              f"Background: {background or 'not provided'}\nCreate the roadmap.",
                              Roadmap, effort="high")


# ------------------------------------------------------------ 2. Weekly progress
class Evaluation(BaseModel):
    progress_percent: int = Field(ge=0, le=100, description="Progress toward the current milestone")
    status: str = Field(description="on track | at risk | behind")
    achievements: list[str]
    problems: list[str] = Field(description="Specific issues, e.g. 'Need more validation experiments'")
    next_week_goals: list[str] = Field(description="3-5 concrete, verifiable goals")
    supervisor_should_know: str = Field(description="One line the human supervisor should see")
    encouragement: str


class WeeklyProgressAgent:
    SYSTEM = ("You are a supportive but rigorous research mentor doing a weekly check-in. Compare the update "
              "with the roadmap and recent history; detect drift, stalls and missing validation; set "
              "concrete goals. Be specific and kind.")

    def run(self, context: str, update: str) -> Evaluation:
        return ask_structured(self.SYSTEM, f"{context}\n\nTHIS WEEK'S UPDATE:\n{update}\n\nEvaluate.",
                              Evaluation, effort="medium")


# ------------------------------------------------------------- 3. Critical review
class CriticalReviewerAgent:
    SYSTEM = ("You are a tough but constructive thesis committee member. Challenge the research assumptions, "
              "methodology, novelty and feasibility based on the roadmap and progress so far. Output Markdown "
              "with sections: Assumptions to test; Methodological weaknesses; Novelty check; Questions the "
              "committee will ask; Recommended actions (prioritised).")

    def run(self, context: str, echo: bool = False) -> str:
        text = ask(self.SYSTEM, context, effort="high", echo=echo)
        return text + NOTICE + f"\n*Generated with the AI PhD Supervisor Agent — {BRAND}*\n"


# ---------------------------------------------------------------- 4. Presentation
class PresentationAgent:
    SYSTEMS = {
        "slides": ("Create a 12-15 slide conference presentation outline in Markdown (one '## Slide N: title' "
                   "per slide with 3-5 bullet points and a speaker note)."),
        "poster": ("Create academic poster content in Markdown: title, authors placeholder, Introduction, "
                   "Methods, Results (describe figures), Conclusions, Future work, References placeholder."),
        "defence": ("Prepare the student for the PhD defence (viva): a 20-minute talk outline, 25 likely "
                    "committee questions grouped by theme with model answers, and 10 tips."),
    }

    def run(self, context: str, kind: str = "slides", echo: bool = False) -> str:
        system = "You are a presentation coach for researchers. " + self.SYSTEMS[kind]
        text = ask(system, context, effort="medium", echo=echo)
        return text + NOTICE + f"\n*Generated with the AI PhD Supervisor Agent — {BRAND}*\n"

20.2.2 llm.py

"""
llm.py — a thin wrapper around the Anthropic Claude API used by this agent.

Two functions are all you need:
    ask(system, user)                  -> free text (streamed)
    ask_structured(system, user, Model) -> a validated Pydantic object (JSON structured output)

Why a wrapper?  Every agent in this project talks to the model the same way.  Keeping the
API details in ONE file per agent means you can swap the model or provider in one place.

Codanics — Python ka Chilla: Build AI Agents
https://codanics.com/courses/python-ka-chilla-build-ai-agents/
"""
from __future__ import annotations

import os
from typing import Type, TypeVar

import anthropic
from dotenv import load_dotenv
from pydantic import BaseModel

load_dotenv()  # reads ANTHROPIC_API_KEY / CLAUDE_MODEL from a .env file if present

MODEL = os.getenv("CLAUDE_MODEL", "claude-opus-5")
T = TypeVar("T", bound=BaseModel)

_client: anthropic.Anthropic | None = None


def client() -> anthropic.Anthropic:
    """Lazily create one shared client (credentials come from the environment)."""
    global _client
    if _client is None:
        _client = anthropic.Anthropic()
    return _client


def _check(msg: anthropic.types.Message) -> None:
    if msg.stop_reason == "refusal":
        detail = getattr(msg, "stop_details", None)
        raise RuntimeError(f"The model refused this request: {detail}")
    if msg.stop_reason == "max_tokens":
        raise RuntimeError("Output was cut off (max_tokens). Increase max_tokens and retry.")


def _text(msg: anthropic.types.Message) -> str:
    return "".join(b.text for b in msg.content if b.type == "text")


def ask(system: str, user: str, *, max_tokens: int = 16000, effort: str = "medium",
        echo: bool = False) -> str:
    """Free-text generation. Streams so long outputs never hit HTTP timeouts."""
    with client().messages.stream(
        model=MODEL,
        max_tokens=max_tokens,
        system=system,
        messages=[{"role": "user", "content": user}],
        thinking={"type": "adaptive"},
        output_config={"effort": effort},
    ) as stream:
        if echo:
            for chunk in stream.text_stream:
                print(chunk, end="", flush=True)
            print()
        msg = stream.get_final_message()
    _check(msg)
    return _text(msg)


def ask_structured(system: str, user: str, model_cls: Type[T], *, max_tokens: int = 16000,
                   effort: str = "medium") -> T:
    """Structured generation: the SDK converts the Pydantic model to a JSON schema, the API
    constrains the output to it, and `parsed_output` is a validated instance of `model_cls`."""
    with client().messages.stream(
        model=MODEL,
        max_tokens=max_tokens,
        system=system,
        messages=[{"role": "user", "content": user}],
        thinking={"type": "adaptive"},
        output_config={"effort": effort},
        output_format=model_cls,
    ) as stream:
        msg = stream.get_final_message()
    _check(msg)
    if msg.parsed_output is None:  # should not happen with a constrained schema; be defensive
        return model_cls.model_validate_json(_text(msg))
    return msg.parsed_output

20.2.3 main.py

"""
AI PhD Supervisor Agent — "Virtual Research Mentor"
====================================================
A long-running mentor with file-based memory per student.

Usage
    python main.py init     --student "Ali" --topic "Multimodal AI for drought resilience in wheat" --years 3
    python main.py weekly   --student "Ali" --update "Finished RGB pipeline, accuracy 87%, hyperspectral data delayed"
    python main.py weekly   --student "Ali" --update-file this_week.md
    python main.py critique --student "Ali"
    python main.py present  --student "Ali" --type slides|poster|defence
    python main.py status   --student "Ali"

Memory: memory/<student>/ (profile.json, roadmap.json, progress_log.jsonl)   Reports: outputs/<student>/

Codanics — Python ka Chilla: Build AI Agents
https://codanics.com/courses/python-ka-chilla-build-ai-agents/
"""
from __future__ import annotations

import sys
from datetime import date
from pathlib import Path

import typer
from rich.console import Console
from rich.panel import Panel
from rich.table import Table
from slugify import slugify

sys.path.insert(0, str(Path(__file__).parent))
if hasattr(sys.stdout, "reconfigure"):  # Windows consoles default to cp1252; we print → ✓ etc.
    sys.stdout.reconfigure(encoding="utf-8")
from agents import (CriticalReviewerAgent, PresentationAgent, ResearchPlanningAgent,  # noqa: E402
                    WeeklyProgressAgent)
from memory import StudentMemory  # noqa: E402

app = typer.Typer(add_completion=False)
console = Console()
OUT_DIR = Path(__file__).parent / "outputs"

BANNER = """[bold cyan]AI PhD Supervisor Agent[/] — Virtual Research Mentor
[dim]Codanics · Python ka Chilla: Build AI Agents · https://codanics.com/courses/python-ka-chilla-build-ai-agents/
YouTube @codanics · WhatsApp https://whatsapp.com/channel/0029Va7nRDq3QxRzGqaQvS3r[/]"""


def out_path(student: str, name: str) -> Path:
    p = OUT_DIR / slugify(student)
    p.mkdir(parents=True, exist_ok=True)
    return p / name


def require(mem: StudentMemory) -> None:
    if not mem.exists():
        console.print(f"[red]No memory for '{mem.student}'. Run: main.py init --student \"{mem.student}\" --topic ...[/]")
        raise typer.Exit(1)


@app.command()
def init(student: str = typer.Option(...), topic: str = typer.Option(...),
         years: int = typer.Option(3), background: str = typer.Option("", help="Degree, skills, prior work")):
    """Create the student's memory and a milestone-based roadmap."""
    console.print(Panel(BANNER, expand=False))
    mem = StudentMemory(student)
    roadmap = ResearchPlanningAgent().run(student, topic, years, background)
    mem.save_profile(name=student, topic=topic, years=years, background=background)
    mem.save_roadmap(roadmap.model_dump_json(indent=2))
    t = Table(title=roadmap.programme_title)
    t.add_column("sem"); t.add_column("milestone"); t.add_column("deliverables", max_width=70)
    for m in roadmap.milestones:
        t.add_row(str(m.semester), m.title, "; ".join(m.deliverables))
    console.print(t)
    md = [f"# PhD Roadmap — {student}\n", f"**Topic:** {topic}  \n**Length:** {years} years\n", "## Aims"]
    md += [f"- {a}" for a in roadmap.aims]
    for m in roadmap.milestones:
        md += [f"\n## Semester {m.semester}: {m.title}", *[f"- {d}" for d in m.deliverables],
               f"\n*Success criteria:* {m.success_criteria}  \n*Risks:* {', '.join(m.risks)}"]
    md += ["\n## Publications plan", *[f"- {p}" for p in roadmap.publications_plan],
           "\n## Skills to learn", *[f"- {s}" for s in roadmap.skills_to_learn],
           "\n## Weekly routine", *[f"- {w}" for w in roadmap.weekly_routine],
           "\n---\n*Codanics — Python ka Chilla: Build AI Agents · "
           "https://codanics.com/courses/python-ka-chilla-build-ai-agents/*"]
    out_path(student, "roadmap.md").write_text("\n".join(md), encoding="utf-8")
    console.print(Panel(f"Memory created in {mem.dir}\nRoadmap saved to outputs/{slugify(student)}/roadmap.md", style="green"))


@app.command()
def weekly(student: str = typer.Option(...), update: str = typer.Option(None),
           update_file: Path = typer.Option(None)):
    """Submit this week's progress update and get a structured evaluation."""
    console.print(Panel(BANNER, expand=False))
    mem = StudentMemory(student); require(mem)
    text = update or (update_file.read_text(encoding="utf-8") if update_file else None)
    if not text:
        raise typer.BadParameter("Provide --update or --update-file")
    ev = WeeklyProgressAgent().run(mem.context_summary(), text)
    mem.append_log({"update": text, "evaluation": ev.model_dump()})
    week = mem.week_number() - 1
    colour = {"on track": "green", "at risk": "yellow", "behind": "red"}.get(ev.status, "white")
    console.print(f"[bold]Week {week}[/] — progress [{colour}]{ev.progress_percent}% ({ev.status})[/]")
    console.print("[bold]Achievements:[/] " + "; ".join(ev.achievements))
    console.print("[bold]Problems:[/]")
    for p in ev.problems:
        console.print(f"  - {p}")
    console.print("[bold]Next week:[/]")
    for g in ev.next_week_goals:
        console.print(f"  - {g}")
    console.print(f"[bold]Supervisor should know:[/] {ev.supervisor_should_know}")
    console.print(f"[italic]{ev.encouragement}[/]")
    report = (f"# Weekly report — {student} — week {week} ({date.today().isoformat()})\n\n**Update:** {text}\n\n"
              f"**Progress:** {ev.progress_percent}% — {ev.status}\n\n## Achievements\n" +
              "\n".join(f"- {a}" for a in ev.achievements) + "\n\n## Problems\n" +
              "\n".join(f"- {p}" for p in ev.problems) + "\n\n## Goals for next week\n" +
              "\n".join(f"- {g}" for g in ev.next_week_goals) +
              f"\n\n**For the supervisor:** {ev.supervisor_should_know}\n\n*{ev.encouragement}*\n\n---\n"
              "*Codanics — Python ka Chilla: Build AI Agents · https://codanics.com/courses/python-ka-chilla-build-ai-agents/*")
    out_path(student, f"week_{week:02d}.md").write_text(report, encoding="utf-8")


@app.command()
def critique(student: str = typer.Option(...), echo: bool = typer.Option(True)):
    """Challenge assumptions, methodology and novelty of the current work."""
    console.print(Panel(BANNER, expand=False))
    mem = StudentMemory(student); require(mem)
    text = CriticalReviewerAgent().run(mem.context_summary(last_n=8), echo=echo)
    p = out_path(student, f"critique_week_{mem.week_number() - 1:02d}.md")
    p.write_text(text, encoding="utf-8")
    console.print(Panel(f"saved {p}", style="green"))


@app.command()
def present(student: str = typer.Option(...),
            type: str = typer.Option("slides", help="slides | poster | defence"),
            echo: bool = typer.Option(True)):
    """Generate conference slides, a poster, or defence preparation."""
    console.print(Panel(BANNER, expand=False))
    mem = StudentMemory(student); require(mem)
    if type not in ("slides", "poster", "defence"):
        raise typer.BadParameter("--type must be slides, poster or defence")
    text = PresentationAgent().run(mem.context_summary(last_n=8), type, echo=echo)
    p = out_path(student, f"{type}.md")
    p.write_text(text, encoding="utf-8")
    console.print(Panel(f"saved {p}", style="green"))


@app.command()
def status(student: str = typer.Option(...)):
    """Show the memory summary (what the agent sees)."""
    mem = StudentMemory(student); require(mem)
    console.print(mem.context_summary())


if __name__ == "__main__":
    app()

20.2.4 memory.py

"""
memory.py — simple, transparent file-based memory for one PhD student.

memory/<student-slug>/
    profile.json        who the student is, topic, start date
    roadmap.json        the 3-year plan (milestones)
    progress_log.jsonl  one JSON line per weekly update + evaluation (append-only audit trail)

The agent never receives the whole history — it receives a compact `context_summary()`:
profile + roadmap + the last N updates.  This keeps prompts small and focused.

Codanics — Python ka Chilla: Build AI Agents
https://codanics.com/courses/python-ka-chilla-build-ai-agents/
"""
from __future__ import annotations

import json
from datetime import date
from pathlib import Path

from slugify import slugify

MEMORY_ROOT = Path(__file__).parent / "memory"


class StudentMemory:
    def __init__(self, student: str):
        self.student = student
        self.dir = MEMORY_ROOT / slugify(student)
        self.dir.mkdir(parents=True, exist_ok=True)
        self.profile_path = self.dir / "profile.json"
        self.roadmap_path = self.dir / "roadmap.json"
        self.log_path = self.dir / "progress_log.jsonl"

    # ---- profile / roadmap ------------------------------------------------------
    def exists(self) -> bool:
        return self.profile_path.exists() and self.roadmap_path.exists()

    def save_profile(self, **fields) -> None:
        fields.setdefault("start_date", date.today().isoformat())
        self.profile_path.write_text(json.dumps(fields, indent=2), encoding="utf-8")

    def profile(self) -> dict:
        return json.loads(self.profile_path.read_text(encoding="utf-8"))

    def save_roadmap(self, roadmap_json: str) -> None:
        self.roadmap_path.write_text(roadmap_json, encoding="utf-8")

    def roadmap(self) -> dict:
        return json.loads(self.roadmap_path.read_text(encoding="utf-8"))

    # ---- progress log ---------------------------------------------------------
    def append_log(self, entry: dict) -> None:
        entry.setdefault("date", date.today().isoformat())
        entry["week"] = self.week_number()
        with self.log_path.open("a", encoding="utf-8") as f:
            f.write(json.dumps(entry, ensure_ascii=False) + "\n")

    def log(self) -> list[dict]:
        if not self.log_path.exists():
            return []
        return [json.loads(l) for l in self.log_path.read_text(encoding="utf-8").splitlines() if l.strip()]

    def week_number(self) -> int:
        return len(self.log()) + 1

    # ---- what the LLM sees ------------------------------------------------------
    def context_summary(self, last_n: int = 4) -> str:
        p, r = self.profile(), self.roadmap()
        recent = self.log()[-last_n:]
        lines = [f"STUDENT: {p.get('name')} | topic: {p.get('topic')} | start: {p.get('start_date')} | "
                 f"programme length: {p.get('years')} years | current week: {self.week_number()}",
                 "ROADMAP:", json.dumps(r, indent=1)[:6000],
                 f"RECENT PROGRESS (last {len(recent)} of {len(self.log())} updates):"]
        for e in recent:
            ev = e.get("evaluation", {})
            lines.append(f"- week {e.get('week')} ({e.get('date')}): {e.get('update')[:400]}\n"
                         f"  progress={ev.get('progress_percent')} status={ev.get('status')} "
                         f"problems={ev.get('problems')} next={ev.get('next_week_goals')}")
        return "\n".join(lines)