lambert/DeepTutor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

docs:update readme

Browse Source
This commit is contained in:
Bingxi Zhao (Frank)
2026-06-14 00:53:57 +08:00
parent e7e38063b8
commit 069ba53a00
22 changed files with 1109 additions and 672 deletions
Download Patch File Download Diff File Expand all files Collapse all files
AGENTS.md
+34 -21
View File
@@ -31,23 +31,25 @@ ignored.
### Level 1 — Tools
Single-function tools the LLM picks on demand. The chat capability auto-mounts
context-gated tools (rag, read_source, read_memory, write_memory, list_notebook,
write_note, web_fetch, github, ask_user); five user-toggleable tools surface in
`/settings/tools`:
Single-function tools the LLM picks on demand. Four user-toggleable tools
surface in `/settings/tools`:
| Tool | Description |
| ---------------- | -------------------------------------------------------- |
| `brainstorm` | Breadth-first idea exploration with rationale |
| `web_search` | Web search with citations |
| `paper_search` | arXiv preprint search |
| `code_execution` | Sandboxed Python (NL intent → code → run) |
| `reason` | Dedicated deep-reasoning LLM call |
| Tool | Description |
| -------------- | --------------------------------------------- |
| `brainstorm` | Breadth-first idea exploration with rationale |
| `web_search` | Web search with citations |
| `paper_search` | arXiv preprint search |
| `reason` | Dedicated deep-reasoning LLM call |
Always-on, context-gated tools: `rag`, `read_source`, `read_memory`,
`write_memory`, `web_fetch`, `list_notebook`, `write_note`, `github`,
`ask_user` (pauses the turn and resumes with the user's reply).
`geogebra_analysis` is parked under `COMING_SOON_TOOL_TYPES`.
The rest are **context-gated**: the chat capability auto-mounts them from
`ToolMountFlags` (presence of a KB, attachments, sandbox availability, …), and
any of them can also be force-enabled via `--tool`. Auto-mounted set: `rag`,
`read_source`, `read_memory`, `write_memory`, `read_skill`, `load_tools`,
`exec`, `code_execution` (sandboxed Python: NL intent → code → run),
`list_notebook`, `write_note`, `web_fetch`, `github`, `cron`,
`ask_user` (pauses the turn and resumes with the user's reply), plus the
mastery-path tools. `geogebra_analysis` is parked under
`COMING_SOON_TOOL_TYPES`.
### Level 2 — Capabilities
@@ -55,7 +57,8 @@ Multi-stage pipelines that own the turn:
| Capability | Stages |
| ---------------- | ----------------------------------------------------- |
| `chat` | thinking → acting → observing → responding (agentic loop, default) |
| `chat` | exploring → responding (single agentic loop, default) |
| `mastery_path` | responding (Guided Learning — chat loop + mastery tools, gated per topic type) |
| `auto` | analyzing → delegating → synthesizing (routes to another capability) |
| `deep_solve` | planning → reasoning → writing |
| `deep_question` | ideation → generation |
@@ -84,11 +87,15 @@ deeptutor run auto "Animate sine wave" # picks the right capability
deeptutor chat
# (inside the REPL: /regenerate or /retry re-runs the last user message)
# Partners (IM-connected companions)
deeptutor partner list
# Knowledge bases, memory, server
deeptutor kb list
deeptutor kb create my-kb --doc textbook.pdf
deeptutor memory show
deeptutor serve --port 8001
deeptutor serve --port 8001 # API server only
deeptutor start # backend + frontend together
```
## Key Files
@@ -106,6 +113,7 @@ deeptutor serve --port 8001
| `deeptutor/core/context.py` | `UnifiedContext` dataclass |
| `deeptutor/tools/builtin/__init__.py` | All built-in tool wrappers |
| `deeptutor/capabilities/` | Built-in capability implementations |
| `deeptutor/app.py` | `DeepTutorApp` — Python SDK facade |
| `deeptutor_cli/main.py` | Typer CLI entry point |
| `deeptutor/api/routers/unified_ws.py` | Unified WebSocket endpoint |
@@ -118,9 +126,14 @@ Requirements files mirror the same dependency groups for Docker/CI installs.
pip install deeptutor — Full app (CLI + Web/API + packaged Web assets)
pip install deeptutor-cli — CLI-only (LLM + RAG + providers + document parsing)
pip install -e . — Source install for development
.[partners] — Full app + partner channel SDKs and MCP client
.[matrix] — Matrix channel for Partners (matrix-nio[e2e]; needs libolm)
.[math-animator] — Manim addon (powers `visualize` Manim renders + `deeptutor animate`)
.[dev] — Full app + test/lint tools
Source extras (.[ extra ], defined in pyproject.toml):
.[cli] — CLI-only dependency set
.[server] — Web/API server dependencies
.[partners] — Partner channel SDKs + MCP client (legacy alias: .[tutorbot])
.[matrix] — Matrix channel for Partners (matrix-nio; needs libolm)
.[matrix-e2e] — Matrix with end-to-end encryption (matrix-nio[e2e])
.[math-animator] — Manim addon (powers `visualize` Manim renders + `deeptutor run math_animator`)
.[dev] — Test / lint tooling
.[all] — Everything above
```
SKILL.md
+96 -40
View File
@@ -6,17 +6,19 @@
Use this skill when the user wants to:
- Set up or configure DeepTutor
- Chat with DeepTutor or run a capability (deep solve, quiz generation, deep research, math animation)
- Chat with DeepTutor or run a capability (deep solve, quiz generation, deep research, math animation, visualize)
- Create, manage, or search knowledge bases
- Manage TutorBot instances
- Create, manage, or run Partners (IM-connected companions)
- Search, install, or manage skills from a hub (ClawHub)
- Inspect or maintain interactive Books
- View or manage learning memory, sessions, or notebooks
- Start the DeepTutor API server
- Start the DeepTutor API server or the full Web app
## Prerequisites
- Python 3.11+
- DeepTutor installed: `pip install deeptutor` for the full Web app, `pip install deeptutor-cli` for CLI-only, or `pip install -e .` from a source checkout
- Run `deeptutor init` for first-time interactive setup (configures LLM, embedding, and search providers under `data/user/settings`)
- Run `deeptutor init` for first-time interactive setup. It walks a guided wizard (ports → LLM → embedding → search → review) and writes the same settings as the Web Settings page under `data/user/settings`. Add `--cli` to skip the ports step for CLI-only use, or `--home <path>` to target a specific workspace.
## Commands
@@ -32,77 +34,123 @@ deeptutor run chat "Explain Fourier transform"
deeptutor run deep_solve "Solve x^2 = 4" --tool rag --kb textbook
deeptutor run deep_question "Linear algebra" --config num_questions=5
deeptutor run deep_research "Attention mechanisms" --kb papers
deeptutor run visualize "Plot the unit circle"
deeptutor run math_animator "Visualize a Fourier series"
# Capabilities accepted by `run` / `chat -c`:
# chat, auto, deep_solve, deep_question, deep_research, visualize, math_animator
# (`auto` inspects the request and routes to the right capability)
# Options for `run`:
# --session <id> Resume existing session
# --tool/-t <name> Enable tool (repeatable): rag, web_search, code_execution, reason, brainstorm, paper_search
# --tool/-t <name> Enable tool (repeatable)
# --kb <name> Knowledge base (repeatable)
# --notebook-ref <ref> Notebook reference (repeatable)
# --notebook-ref <ref> Notebook reference, "<notebook_id>:<rec1>,<rec2>" (repeatable)
# --history-ref <id> Referenced session id (repeatable)
# --language/-l <code> Response language (default: en)
# --config <key=value> Capability config (repeatable)
# --config-json <json> Capability config as JSON
# --format/-f <fmt> Output format: rich | json
# --format/-f <fmt> Output format: rich | json (default: rich)
```
`deeptutor chat` accepts the same `--session / --tool / --kb / --notebook-ref / --history-ref / --language / --config / --config-json` options, plus `--capability/-c <name>` to set the initial capability.
**Tools** for `--tool` / `-t`: user-toggleable tools are `brainstorm`, `web_search`, `paper_search`, `reason`. Context-gated tools (`rag`, `code_execution`, `read_source`, `web_fetch`, `github`, `ask_user`, …) auto-mount when their context is present, but can also be force-enabled with `--tool`. Run `deeptutor plugin list` for the full registered set.
### Knowledge Bases
```bash
deeptutor kb list # List all knowledge bases
deeptutor kb info <name> # Show knowledge base details
deeptutor kb create <name> --doc file.pdf # Create from documents (--doc repeatable)
deeptutor kb add <name> --doc more.pdf # Add documents incrementally
deeptutor kb search <name> "query text" # Search a knowledge base
deeptutor kb set-default <name> # Set as default KB
deeptutor kb delete <name> [--force] # Delete a knowledge base
deeptutor kb list [--format rich|json] # List all knowledge bases
deeptutor kb info <name> # Show knowledge base details (JSON)
deeptutor kb create <name> --doc file.pdf # Create from documents (--doc/-d repeatable)
deeptutor kb create <name> --docs-dir ./papers # ...or from a directory of documents
deeptutor kb add <name> --doc more.pdf # Add documents incrementally
deeptutor kb search <name> "query text" [--mode hybrid] [--format rich|json]
deeptutor kb set-default <name> # Set as default KB
deeptutor kb delete <name> [--force] # Delete a knowledge base
```
### TutorBot
### Partners
Partners are IM-connected learning companions (the former "TutorBot").
```bash
deeptutor bot list # List all TutorBot instances
deeptutor bot create <id> --name "My Tutor" # Create and start a new bot
deeptutor bot start <id> # Start a bot
deeptutor bot stop <id> # Stop a bot
deeptutor partner list # List all partners
deeptutor partner create <id> -n "My Tutor" # Create and start a new partner
# -n/--name <text> Display name
# -s/--soul <md> Soul markdown (the persona)
# -m/--model <id> Model override
deeptutor partner start <id> # Start a partner
deeptutor partner stop <id> # Stop a running partner
```
### Skills
Install and manage skills, including packages from external hubs (ClawHub).
Hub refs use `<hub>:<slug>[@version]` (the hub prefix defaults to `clawhub`).
```bash
deeptutor skill search "flashcards" [--hub clawhub] [--limit 10]
deeptutor skill install clawhub:some-skill[@1.2.0] [--name local-name] [--force] [--allow-unverified]
deeptutor skill list # List local skills (with hub provenance)
deeptutor skill remove <name> # Remove a user-layer skill
```
### Books
Maintenance commands for the BookEngine (authoring/reading is via the Web app).
```bash
deeptutor book list # List all books (flags stale pages)
deeptutor book health <book_id> # Inspect KB drift + log.md health
deeptutor book refresh-fingerprints <book_id> # Re-snapshot KB fingerprints
```
### Memory
```bash
deeptutor memory show [summary|profile|all] # View learning memory
deeptutor memory clear [summary|profile|all] # Clear memory (--force to skip confirm)
deeptutor memory show [<target>] # target: L3 (all global docs, default) | L2 (all surfaces) | a doc name (e.g. profile, chat)
deeptutor memory clear [<target>] # target: all (default) | trace (all L1) | a surface name (clears that surface's L1)
# --force/-f Skip confirmation
```
### Sessions
```bash
deeptutor session list [--limit 20] # List sessions
deeptutor session show <id> # View session messages
deeptutor session open <id> # Resume session in REPL
deeptutor session rename <id> --title "..." # Rename a session
deeptutor session delete <id> # Delete a session
deeptutor session list [--limit 20] # List sessions
deeptutor session show <id> [--format rich|json] # View session messages
deeptutor session open <id> # Resume session in the REPL
deeptutor session rename <id> --title "..." # Rename a session
deeptutor session delete <id> # Delete a session
```
### Notebooks
```bash
deeptutor notebook list # List notebooks
deeptutor notebook create <name> # Create a notebook
deeptutor notebook show <id> # View notebook records
deeptutor notebook add-md <id> <file.md> # Import markdown as record
deeptutor notebook replace-md <id> <rec> <f> # Replace a markdown record
deeptutor notebook remove-record <id> <rec> # Remove a record
deeptutor notebook list # List notebooks
deeptutor notebook create <name> [--description "..."]
deeptutor notebook show <notebook_id> [--format rich|json]
deeptutor notebook add-md <notebook_id> <file.md> [--title "..."] [--type chat|question|research|solve]
deeptutor notebook replace-md <notebook_id> <record_id> <file.md>
deeptutor notebook remove-record <notebook_id> <record_id>
```
### Providers
```bash
deeptutor provider login openai-codex # OAuth login for OpenAI Codex
deeptutor provider login github-copilot # Validate an existing Copilot auth session
```
### System
```bash
deeptutor config show # Print current configuration
deeptutor plugin list # List registered tools and capabilities
deeptutor plugin info <name> # Show tool/capability details
deeptutor provider login <provider> # OAuth login (openai-codex, github-copilot)
deeptutor serve [--port 8001] [--reload] # Start API server
deeptutor config show # Print resolved configuration
deeptutor plugin list # List registered tools and capabilities
deeptutor plugin info <name> # Show a tool/capability's schema + availability
deeptutor serve [--host 0.0.0.0] [--port 8001] [--reload] # Start the API server
deeptutor start [--home <path>] # Launch backend + frontend together
deeptutor init [--cli] [--home <path>] # Create/update workspace settings
```
## REPL Slash Commands
@@ -113,13 +161,16 @@ Inside `deeptutor chat`, use these:
|:---|:---|
| `/quit` | Exit REPL |
| `/session` | Show current session id |
| `/new` | Start a new session |
| `/status` | Print the current REPL state |
| `/new` or `/clear` | Start a new session context |
| `/regenerate` or `/retry` | Re-run the last user message |
| `/tool on\|off <name>` | Toggle a tool |
| `/cap <name>` | Switch capability |
| `/kb <name>\|none` | Set or clear knowledge base |
| `/history add <id>` / `/history clear` | Manage history references |
| `/notebook add <ref>` / `/notebook clear` | Manage notebook references |
| `/refs` | Show active references |
| `/show last\|<n>` | Expand a captured tool result or thinking block |
| `/refs` | Show all active references |
| `/config show\|set\|clear` | Manage capability config |
## Typical Workflows
@@ -128,7 +179,7 @@ Inside `deeptutor chat`, use these:
```bash
cd DeepTutor
pip install -e .
deeptutor init # Interactive guided setup
deeptutor init # Interactive guided setup (add --cli for CLI-only)
```
**Daily learning:**
@@ -146,3 +197,8 @@ deeptutor run chat "Explain Newton's third law" --kb physics --tool rag
```bash
deeptutor run deep_question "Thermodynamics" --kb physics --config num_questions=5
```
**Run the full Web app locally:**
```bash
deeptutor start # backend + frontend; Ctrl+C to stop
```
deeptutor/agents/_shared/tool_composition.py
+19 -10
View File
@@ -25,7 +25,6 @@ from dataclasses import dataclass
from typing import Any
from deeptutor.tools.builtin import BUILTIN_TOOL_NAMES, USER_TOGGLEABLE_TOOL_NAMES
from deeptutor.tools.mastery_tool import MASTERY_TOOL_NAMES
# Tools whose mounting is owned by the pipeline (auto-on under specific
# context conditions), not by the user's composer toggles. Adding a tool
@@ -48,7 +47,7 @@ AUTO_MOUNTED_TOOLS: frozenset[str] = frozenset(
"github",
"cron",
}
).union(MASTERY_TOOL_NAMES)
)
def default_optional_tools(excluded: Iterable[str] = ()) -> list[str]:
@@ -85,7 +84,6 @@ class ToolMountFlags:
has_deferred_tools: bool = False
has_exec: bool = False
has_code: bool = False
has_mastery: bool = False
def compose_enabled_tools(
@@ -94,6 +92,7 @@ def compose_enabled_tools(
requested_tools: list[str] | None,
optional_whitelist: list[str],
mount_flags: ToolMountFlags,
extra_auto_tools: Iterable[str] = (),
) -> list[str]:
"""Compose the per-turn enabled-tool list.
@@ -107,11 +106,11 @@ def compose_enabled_tools(
if a source index exists, ``read_memory`` if memory has content,
``list_notebook`` + ``write_note`` if notebooks exist,
``read_skill`` if the turn carries a skills manifest).
3. Always-on auto-mounts (``web_fetch``, ``github``, ``ask_user``).
3. Plugin-owned auto-mounts supplied by the caller.
4. Always-on auto-mounts (``web_fetch``, ``github``, ``ask_user``).
The result is ordered (no dedup is applied — caller's prerequisite is
that ``optional_whitelist`` excludes ``AUTO_MOUNTED_TOOLS``, which
:func:`default_optional_tools` guarantees).
The result is ordered and deduplicated. ``optional_whitelist`` is still
expected to exclude ``AUTO_MOUNTED_TOOLS`` via :func:`default_optional_tools`.
"""
composed: list[str] = [
tool.name
@@ -135,14 +134,24 @@ def compose_enabled_tools(
composed.append("exec")
if mount_flags.has_code:
composed.append("code_execution")
if mount_flags.has_mastery:
composed.extend(MASTERY_TOOL_NAMES)
composed.extend(str(name) for name in extra_auto_tools if str(name).strip())
composed.append("write_memory")
composed.append("web_fetch")
composed.append("github")
composed.append("ask_user")
composed.append("cron")
return composed
return _ordered_unique(composed)
def _ordered_unique(names: Iterable[str]) -> list[str]:
seen: set[str] = set()
result: list[str] = []
for name in names:
if name in seen:
continue
seen.add(name)
result.append(name)
return result
def user_has_memory() -> bool:
deeptutor/agents/chat/agent_loop.py
+51 -1
View File
@@ -31,6 +31,7 @@ import re
from typing import TYPE_CHECKING, Any
from deeptutor.capabilities._shared import emit_capability_result
from deeptutor.core.agentic.tool_dispatch import DispatchOutcome
from deeptutor.core.context import UnifiedContext
from deeptutor.core.stream_bus import StreamBus
from deeptutor.core.trace import build_trace_metadata, merge_trace_metadata, new_call_id
@@ -162,13 +163,21 @@ class AgentLoop:
state = AgentLoopState()
async with self.stream.stage(LOOP_STAGE, source="chat"):
seed_block = await self.pipeline._retrieve_kb_seed_block(self.context, self.stream)
plugin_seed = self.pipeline._plugin_pre_loop_seed(self.context)
seed_block = "\n\n".join(
block for block in (seed_block.strip(), plugin_seed.strip()) if block
)
messages = self.pipeline._build_loop_messages(
context=self.context,
enabled_tools=self.enabled_tools,
kb_seed=seed_block,
include_tool_manifest=bool(self.tool_schemas),
)
outcome = await self._run_loop(messages=messages, state=state)
outcome = await self._run_loop(
messages=messages,
state=state,
checkpoint_boundary=len(messages),
)
if state.sources:
await self.stream.sources(
@@ -200,6 +209,7 @@ class AgentLoop:
*,
messages: list[dict[str, Any]],
state: AgentLoopState,
checkpoint_boundary: int,
) -> LoopOutcome:
"""Run rounds of one LLM call + tool dispatch over *messages*.
@@ -288,6 +298,12 @@ class AgentLoop:
# ``role=tool`` message; the next round sees them in-protocol.
continue
checkpoint_boundary = self._fold_context_checkpoint(
messages=messages,
dispatch=dispatch,
checkpoint_boundary=checkpoint_boundary,
)
if dispatch.terminate:
payload = dispatch.terminate_payload or {}
await self.pipeline._emit_terminator_final_response(self.stream, payload)
@@ -299,6 +315,26 @@ class AgentLoop:
# Round budget ran out while still requesting tools — force a finish.
return await self._forced_finish(messages, state)
def _fold_context_checkpoint(
self,
*,
messages: list[dict[str, Any]],
dispatch: DispatchOutcome,
checkpoint_boundary: int,
) -> int:
summary = _last_context_checkpoint_summary(dispatch)
if not summary:
return checkpoint_boundary
prefix = messages[:checkpoint_boundary]
prefix.append(
{
"role": "system",
"content": f"[Context checkpoint]\n{summary}",
}
)
messages[:] = prefix
return len(messages)
async def _forced_finish(
self,
messages: list[dict[str, Any]],
@@ -590,6 +626,20 @@ def _message_content_chars(message: dict[str, Any]) -> int:
return 0
def _last_context_checkpoint_summary(dispatch: DispatchOutcome) -> str:
summary = ""
for tool_message in dispatch.tool_messages:
tool_call_id = str(tool_message.get("tool_call_id") or "")
metadata = dispatch.tool_metadata_by_id.get(tool_call_id) or {}
checkpoint = metadata.get("_context_checkpoint")
if not isinstance(checkpoint, dict):
continue
candidate = str(checkpoint.get("summary") or "").strip()
if candidate:
summary = candidate
return summary
def _error_text(exc: Exception) -> str:
response = getattr(exc, "response", None)
body = (
deeptutor/agents/chat/agentic_pipeline.py
+34 -60
View File
@@ -33,6 +33,7 @@ from deeptutor.core.trace import (
merge_trace_metadata,
new_call_id,
)
from deeptutor.loop_plugins import LoopPlugin, active_loop_plugins
from deeptutor.runtime.registry.deferred_tools import (
DeferredToolLoader,
render_deferred_tools_manifest,
@@ -47,7 +48,6 @@ from deeptutor.services.llm import (
)
from deeptutor.services.llm.context_window import resolve_effective_context_window
from deeptutor.services.prompt import get_prompt_manager
from deeptutor.tools.mastery_tool import MASTERY_TOOL_NAMES
logger = logging.getLogger(__name__)
@@ -63,42 +63,6 @@ DEFAULT_MAX_ROUNDS = 8
CONTEXT_WINDOW_GUARD_RATIO = 0.9
_DispatchOutcome = DispatchOutcome
# The mastery-tutor playbook, injected as a system block only when the turn is
# in mastery mode (set by MasteryPathCapability). The intelligence — what to
# teach, how to question — lives here at the loop's exit; the hard gate and the
# spaced-repetition arithmetic live in the engine the mastery tools call.
_MASTERY_SYSTEM_EN = """\
[Mastery Tutor mode]
You are a one-on-one mastery tutor. The learner works through a map of objectives, each behind a HARD mastery gate: an objective counts as "mastered" only once its gate clears, and you must not move on until it does.
FIRST on every turn, call `mastery_status`. It returns the next objective to work on, any question awaiting an answer, due reviews, and the full map. Trust it to choose the objective — never guess what comes next.
Then act on the objective:
- No objectives yet? Design a path from the learner's materials (use `rag` / `read_source` when materials are attached) and call `mastery_build`. Tag each knowledge point: memory (facts), procedure (step-by-step skills), concept (ideas to understand), design (open-ended judgement).
- `probe` (untouched): briefly check whether the learner already knows it before teaching — let them test out, don't lecture what they have already mastered.
- memory / procedure objectives: register the question + its answer with `mastery_quiz`, then ALWAYS present it with the `ask_user` tool so the learner answers on an interactive card — never write the choices as plain numbered text. For multiple choice, give each `ask_user` option a short label (A / B / C …) and set the matching label as `mastery_quiz`'s `expected_answer`; for open questions use `ask_user` free text. When the answer comes back, score it with `mastery_grade`. Keep working the same objective until `mastery_grade` reports `mastered: true`.
- concept / design objectives: ask the learner to explain the idea in their own words, judge it, and record the result with `mastery_assess` (`passed: true` only when the explanation truly shows understanding).
- `review`: a spaced-repetition item is due — quiz it again to refresh it.
- `complete`: congratulate the learner and summarise what they have mastered.
Teach from the learner's own materials when available. Keep each turn focused on one objective. Be warm and encouraging, but hold the bar — clearing the gate is the point, not moving fast."""
_MASTERY_SYSTEM_ZH = """\
[精通导师模式]
你是一对一的掌握式导师。学习者沿着一张知识点地图前进，每个知识点都有一道硬性掌握门槛：只有门槛达成，该知识点才算"已掌握"，在此之前你绝不能推进到下一个。
每一轮都要先调用 `mastery_status`。它会返回当前要攻克的知识点、是否有待批改的作答、到期复习项，以及整张地图。请信任它来决定学什么——绝不要自己猜下一个知识点。
然后针对该知识点行动：
- 还没有任何知识点？根据学习者的材料设计一条路径（材料已挂载时用 `rag` / `read_source`），调用 `mastery_build`。给每个知识点标类型：memory（记忆/事实）、procedure（程序/步骤技能）、concept（概念/需理解）、design（设计/开放判断）。
- `probe`（未触碰）：先简短探查学习者是否已经会了再教——允许其"测试通过"直接跳过，不要重复讲他已掌握的内容。
- memory / procedure 类：先用 `mastery_quiz` 登记题目与答案，然后**始终用 `ask_user` 工具**把题目呈现成可点选的卡片让学习者作答——绝不要把选项写成纯文字的 1./2./3.。选择题给每个 `ask_user` 选项一个短标签（A / B / C …），并把正确标签设为 `mastery_quiz` 的 `expected_answer`；简答题用 `ask_user` 的自由输入。收到作答后用 `mastery_grade` 批改。在 `mastery_grade` 返回 `mastered: true` 之前，持续打磨同一个知识点。
- concept / design 类：让学习者用自己的话解释该概念，你来判断，并用 `mastery_assess` 记录结果（只有解释确实体现理解时才 `passed: true`）。
- `review`：有到期的间隔复习项——再考一次以巩固。
- `complete`：祝贺学习者并总结其已掌握的内容。
有材料时优先用学习者自己的材料来教。每一轮聚焦一个知识点。态度温暖鼓励，但守住门槛——目标是达成掌握，而非求快。"""
def _read_int(cfg: Any, *, key: str, default: int) -> int:
if isinstance(cfg, dict):
@@ -307,7 +271,7 @@ class AgenticChatPipeline:
),
notebook_manifest=self._build_notebook_manifest(),
workspace_note=self._workspace_system_note(context),
mastery_note=self._mastery_system_note(context),
plugin_blocks=self._plugin_system_blocks(context),
include_tool_manifest=include_tool_manifest,
)
@@ -477,10 +441,39 @@ class AgenticChatPipeline:
has_deferred_tools=getattr(self, "_deferred_loader", None) is not None,
has_exec=getattr(self, "_exec_enabled", False),
has_code=getattr(self, "_exec_enabled", False),
has_mastery=self._mastery_active(context),
),
extra_auto_tools=self._plugin_tool_names(context),
)
def _active_loop_plugins(self, context: UnifiedContext) -> tuple[LoopPlugin, ...]:
return active_loop_plugins(context)
def _plugin_tool_names(self, context: UnifiedContext) -> tuple[str, ...]:
names: list[str] = []
for plugin in self._active_loop_plugins(context):
names.extend(plugin.tool_types(context))
return tuple(names)
def _plugin_system_blocks(self, context: UnifiedContext):
blocks = []
for plugin in self._active_loop_plugins(context):
block = plugin.system_block(
context,
language=self.language,
prompts=self._prompts,
)
if block is not None:
blocks.append(block)
return blocks
def _plugin_pre_loop_seed(self, context: UnifiedContext) -> str:
seeds = [
seed.strip()
for plugin in self._active_loop_plugins(context)
if (seed := plugin.pre_loop_seed(context))
]
return "\n\n".join(seed for seed in seeds if seed)
def _build_llm_tool_schemas(
self,
enabled_tools: list[str],
@@ -717,10 +710,6 @@ class AgenticChatPipeline:
get_path_service().get_task_workspace("chat", workspace_key) if workspace_key else None
)
exec_dir = task_dir / "exec" if task_dir is not None else None
if tool_name in MASTERY_TOOL_NAMES:
# The active path id is server-owned; the model never picks which
# learner's progress a mastery tool reads or writes.
kwargs["_mastery_path_id"] = self._mastery_path_id(context)
if tool_name == "rag":
kwargs.setdefault("mode", "hybrid")
elif tool_name == "load_tools":
@@ -808,6 +797,8 @@ class AgenticChatPipeline:
mime = getattr(first_image, "mime_type", "") or "image/png"
kwargs["image_base64"] = f"data:{mime};base64,{raw_b64}"
kwargs["language"] = context.language or "zh"
for plugin in self._active_loop_plugins(context):
kwargs = plugin.augment_kwargs(tool_name, kwargs, context)
return kwargs
def _retrieve_trace_metadata(
@@ -1069,23 +1060,6 @@ class AgenticChatPipeline:
cleaned = "".join(ch if ch.isalnum() or ch in {"-", "_"} else "_" for ch in raw)
return cleaned.strip("_") or "direct"
@staticmethod
def _mastery_active(context: UnifiedContext) -> bool:
"""Whether this turn runs in mastery-tutor mode (set by the capability)."""
return bool(context.metadata.get("mastery_mode"))
@staticmethod
def _mastery_path_id(context: UnifiedContext) -> str:
return str(context.metadata.get("mastery_path_id") or "").strip()
def _mastery_system_note(self, context: UnifiedContext) -> str:
if not self._mastery_active(context):
return ""
override = self._t("mastery.system")
if override:
return override
return _MASTERY_SYSTEM_ZH if self.language == "zh" else _MASTERY_SYSTEM_EN
def _kb_system_note(self, context: UnifiedContext) -> str:
kbs = self._selected_kbs(context)
if not kbs:
deeptutor/agents/chat/prompt_blocks.py
+7 -21
View File
@@ -2,26 +2,13 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
from deeptutor.core.context import UnifiedContext
from deeptutor.loop_plugins.protocol import PromptBlock
from deeptutor.services.prompt.language import append_language_directive
@dataclass(frozen=True, slots=True)
class PromptBlock:
"""One named prompt fragment.
Blocks are deliberately simple strings. The value is in naming and
ordering: runtime code can find, replace, omit, or budget one category
without editing a monolithic prompt.
"""
name: str
content: str
class ChatPromptAssembler:
"""Build system prompts from explicit, category-named blocks."""
@@ -38,7 +25,7 @@ class ChatPromptAssembler:
deferred_tools_manifest: str = "",
notebook_manifest: str = "",
workspace_note: str = "",
mastery_note: str = "",
plugin_blocks: list[PromptBlock] | None = None,
include_tool_manifest: bool = True,
) -> str:
blocks = self.blocks(
@@ -48,7 +35,7 @@ class ChatPromptAssembler:
deferred_tools_manifest=deferred_tools_manifest,
notebook_manifest=notebook_manifest,
workspace_note=workspace_note,
mastery_note=mastery_note,
plugin_blocks=plugin_blocks,
include_tool_manifest=include_tool_manifest,
)
joined = "\n\n---\n\n".join(
@@ -65,7 +52,7 @@ class ChatPromptAssembler:
deferred_tools_manifest: str = "",
notebook_manifest: str = "",
workspace_note: str = "",
mastery_note: str = "",
plugin_blocks: list[PromptBlock] | None = None,
include_tool_manifest: bool = True,
) -> list[PromptBlock]:
blocks: list[PromptBlock] = [
@@ -73,10 +60,9 @@ class ChatPromptAssembler:
PromptBlock("runtime_policy", self._t("runtime_policy")),
PromptBlock("loop", self._t("loop.system")),
]
# Mastery-tutor playbook sits high so it frames the whole turn when a
# mastery path is active; empty (and omitted) for ordinary chat.
if mastery_note:
blocks.append(PromptBlock("mastery_tutor", mastery_note))
# Plugin playbooks sit high so they frame the whole turn when active;
# empty blocks are omitted by ``system_prompt``'s join.
blocks.extend(plugin_blocks or [])
if context.persona_context:
blocks.append(PromptBlock("persona_style", context.persona_context))
if context.memory_context:
deeptutor/agents/research/pipeline.py
-5
View File
@@ -837,11 +837,6 @@ class ResearchPipeline:
usage=self.usage,
stream_body_live=False,
eager_sub_trace=True,
# Reasoning models may emit a native ``<think>...</think>``
# planning pass without the requested ``THINK`` label. Treat
# that as a real THINK iteration so the next round can perform
# the tool call instead of burning the budget on label repair.
implicit_think_label=LABEL_THINK,
)
except Exception as exc:
logger.exception("Research block %s failed: %s", block.block_id, exc)
deeptutor/capabilities/mastery_path.py
+1 -1
View File
@@ -21,7 +21,7 @@ from deeptutor.agents.chat.agentic_pipeline import AgenticChatPipeline
from deeptutor.core.capability_protocol import BaseCapability, CapabilityManifest
from deeptutor.core.context import UnifiedContext
from deeptutor.core.stream_bus import StreamBus
from deeptutor.tools.mastery_tool import MASTERY_TOOL_NAMES
from deeptutor.loop_plugins.mastery import MASTERY_TOOL_NAMES
_UNSAFE_ID_CHARS = re.compile(r"[^A-Za-z0-9_-]")
deeptutor/loop_plugins/__init__.py
+12
View File
@@ -0,0 +1,12 @@
"""Turn-scoped chat loop plugins.
Each plugin lives in its own subpackage under :mod:`deeptutor.loop_plugins`.
The chat loop imports only the generic registry/protocol from this package;
feature-specific prompts, tools, and kwargs injection stay inside each plugin
subpackage.
"""
from deeptutor.loop_plugins.protocol import LoopPlugin, PromptBlock
from deeptutor.loop_plugins.registry import LOOP_PLUGINS, active_loop_plugins
__all__ = ["LOOP_PLUGINS", "LoopPlugin", "PromptBlock", "active_loop_plugins"]
deeptutor/loop_plugins/mastery/__init__.py
+6
View File
@@ -0,0 +1,6 @@
"""Mastery path loop plugin."""
from deeptutor.loop_plugins.mastery.plugin import MasteryLoopPlugin
from deeptutor.loop_plugins.mastery.tools import MASTERY_TOOL_NAMES, MASTERY_TOOL_TYPES
__all__ = ["MASTERY_TOOL_NAMES", "MASTERY_TOOL_TYPES", "MasteryLoopPlugin"]
deeptutor/loop_plugins/mastery/plugin.py
+69
View File
@@ -0,0 +1,69 @@
"""Mastery path loop plugin hooks."""
from __future__ import annotations
from importlib import resources
from typing import Any
from deeptutor.core.context import UnifiedContext
from deeptutor.loop_plugins.mastery.tools import MASTERY_TOOL_NAMES
from deeptutor.loop_plugins.protocol import PromptBlock
class MasteryLoopPlugin:
"""Turn-scoped integration for mastery-path tutoring."""
name = "mastery"
def is_active(self, context: UnifiedContext) -> bool:
return bool(context.metadata.get("mastery_mode"))
def tool_types(self, context: UnifiedContext) -> tuple[str, ...]:
return MASTERY_TOOL_NAMES if self.is_active(context) else ()
def system_block(
self,
context: UnifiedContext,
*,
language: str,
prompts: dict[str, Any],
) -> PromptBlock | None:
if not self.is_active(context):
return None
override = _prompt_text(prompts, ("mastery", "system"))
content = override or _load_system_prompt(language)
return PromptBlock("mastery_tutor", content)
def augment_kwargs(
self,
tool_name: str,
kwargs: dict[str, Any],
context: UnifiedContext,
) -> dict[str, Any]:
if self.is_active(context) and tool_name in MASTERY_TOOL_NAMES:
updated = dict(kwargs)
updated["_mastery_path_id"] = str(context.metadata.get("mastery_path_id") or "").strip()
return updated
return kwargs
def pre_loop_seed(self, context: UnifiedContext) -> str:
_ = context
return ""
def _prompt_text(prompts: dict[str, Any], path: tuple[str, ...]) -> str:
value: Any = prompts
for key in path:
if not isinstance(value, dict):
return ""
value = value.get(key)
return value if isinstance(value, str) and value else ""
def _load_system_prompt(language: str) -> str:
lang = "zh" if language.lower().startswith("zh") else "en"
prompt = resources.files(__package__).joinpath("prompts", lang, "system.md")
return prompt.read_text(encoding="utf-8").strip()
__all__ = ["MasteryLoopPlugin"]
deeptutor/loop_plugins/mastery/prompts/en/system.md
+14
View File
@@ -0,0 +1,14 @@
[Mastery Tutor mode]
You are a one-on-one mastery tutor. The learner works through a map of objectives, each behind a HARD mastery gate: an objective counts as "mastered" only once its gate clears, and you must not move on until it does.
FIRST on every turn, call `mastery_status`. It returns the next objective to work on, any question awaiting an answer, due reviews, and the full map. Trust it to choose the objective — never guess what comes next.
Then act on the objective:
- No objectives yet? Design a path from the learner's materials (use `rag` / `read_source` when materials are attached) and call `mastery_build`. Tag each knowledge point: memory (facts), procedure (step-by-step skills), concept (ideas to understand), design (open-ended judgement).
- `probe` (untouched): briefly check whether the learner already knows it before teaching. A test-out is not a silent skip — record its result through the gate (`mastery_assess` for concept / design, `mastery_quiz` + `mastery_grade` for memory / procedure) before advancing. Never move past an objective the engine hasn't marked mastered.
- memory / procedure objectives: register the question + its answer with `mastery_quiz`, then ALWAYS present it with the `ask_user` tool so the learner answers on an interactive card — never write the choices as plain numbered text. For multiple choice, give each `ask_user` option a short label (A / B / C …) and set the matching label as `mastery_quiz`'s `expected_answer`; for open questions use `ask_user` free text. When the answer comes back, score it with `mastery_grade`. Keep working the same objective until `mastery_grade` reports `mastered: true`.
- concept / design objectives: ask the learner to explain the idea in their own words, judge it, and record the result with `mastery_assess` (`passed: true` only when the explanation truly shows understanding).
- `review`: a spaced-repetition item is due — quiz it again to refresh it.
- `complete`: congratulate the learner and summarise what they have mastered.
Teach from the learner's own materials when available. Keep each turn focused on one objective. Be warm and encouraging, but hold the bar — clearing the gate is the point, not moving fast.
deeptutor/loop_plugins/mastery/prompts/zh/system.md
+14
View File
@@ -0,0 +1,14 @@
[精通导师模式]
你是一对一的掌握式导师。学习者沿着一张知识点地图前进，每个知识点都有一道硬性掌握门槛：只有门槛达成，该知识点才算"已掌握"，在此之前你绝不能推进到下一个。
每一轮都要先调用 `mastery_status`。它会返回当前要攻克的知识点、是否有待批改的作答、到期复习项，以及整张地图。请信任它来决定学什么——绝不要自己猜下一个知识点。
然后针对该知识点行动：
- 还没有任何知识点？根据学习者的材料设计一条路径（材料已挂载时用 `rag` / `read_source`），调用 `mastery_build`。给每个知识点标类型：memory（记忆/事实）、procedure（程序/步骤技能）、concept（概念/需理解）、design（设计/开放判断）。
- `probe`（未触碰）：先简短探查学习者是否已经会了再教。"测试通过"不等于直接跳过——仍要用门工具记录结果（concept / design 用 `mastery_assess`，memory / procedure 用 `mastery_quiz` + `mastery_grade`）再推进；绝不要越过引擎尚未标记为"已掌握"的知识点。
- memory / procedure 类：先用 `mastery_quiz` 登记题目与答案，然后**始终用 `ask_user` 工具**把题目呈现成可点选的卡片让学习者作答——绝不要把选项写成纯文字的 1./2./3.。选择题给每个 `ask_user` 选项一个短标签（A / B / C …），并把正确标签设为 `mastery_quiz` 的 `expected_answer`；简答题用 `ask_user` 的自由输入。收到作答后用 `mastery_grade` 批改。在 `mastery_grade` 返回 `mastered: true` 之前，持续打磨同一个知识点。
- concept / design 类：让学习者用自己的话解释该概念，你来判断，并用 `mastery_assess` 记录结果（只有解释确实体现理解时才 `passed: true`）。
- `review`：有到期的间隔复习项——再考一次以巩固。
- `complete`：祝贺学习者并总结其已掌握的内容。
有材料时优先用学习者自己的材料来教。每一轮聚焦一个知识点。态度温暖鼓励，但守住门槛——目标是达成掌握，而非求快。
deeptutor/loop_plugins/mastery/tools.py
+521
View File
@@ -0,0 +1,521 @@
"""Mastery Path tools — the seam between the chat-loop tutor and the pure
mastery engine (:mod:`deeptutor.learning`).
These five tools are auto-mounted only when a mastery path is active on the
turn (via the chat loop mastery plugin). The chat agent loop IS the tutor;
these tools let it read the gate and record outcomes, while the pedagogy —
what to teach, how to question, when to explain — stays the model's job. The
arithmetic (mastery, gate, spaced repetition) stays in the engine.
The active path id is injected server-side by the pipeline as
``_mastery_path_id``; the model never supplies it. Each call constructs a
fresh store + service (matching the REST router) so concurrent turns can't
race on a shared object.
"""
from __future__ import annotations
import json
from typing import TYPE_CHECKING, Any
import uuid
from deeptutor.core.tool_protocol import BaseTool, ToolDefinition, ToolParameter, ToolResult
# ``learning.models`` and ``learning.policy`` only depend on pydantic — safe to
# import at module load. ``learning.service`` / ``storage`` / ``scheduler``
# reach the path service (and so the runtime + tool registry), so importing
# them here would close an import cycle through the built-in registry. They
# are imported lazily inside the call paths instead (same pattern as the other
# builtin tools).
from deeptutor.learning.models import (
KnowledgePoint,
KnowledgeType,
LearningModule,
PendingQuestion,
)
from deeptutor.learning.policy import (
QUALITATIVE_TYPES,
display_mastery,
find_knowledge_point,
gate_threshold,
is_mastered,
map_summary,
next_objective,
)
if TYPE_CHECKING:
from deeptutor.learning.service import LearningService
# Tool names the pipeline mounts together when a mastery path is active. Kept
# here so the mount policy and the registration list can't disagree.
MASTERY_TOOL_NAMES: tuple[str, ...] = (
"mastery_status",
"mastery_quiz",
"mastery_grade",
"mastery_assess",
"mastery_build",
)
_QUESTION_TYPES = ("choice", "short", "open")
_ALLOWED_KP_TYPES = {t.value for t in KnowledgeType}
def _new_service() -> LearningService:
from deeptutor.learning.service import LearningService
from deeptutor.learning.storage import LearningStore
return LearningService(LearningStore())
def _resolve_path_id(kwargs: dict[str, Any]) -> str:
return str(kwargs.get("_mastery_path_id") or "").strip()
def _json_result(payload: dict[str, Any], *, meta_key: str, success: bool = True) -> ToolResult:
return ToolResult(
content=json.dumps(payload, ensure_ascii=False),
success=success,
metadata={meta_key: payload},
)
def _no_path_result() -> ToolResult:
return ToolResult(
content="No mastery path is active on this turn; mastery tools are unavailable.",
success=False,
)
class MasteryStatusTool(BaseTool):
"""Read the current objective + map snapshot. Call FIRST every turn."""
def get_definition(self) -> ToolDefinition:
return ToolDefinition(
name="mastery_status",
description=(
"Read the learner's mastery path: the next objective to work on "
"(decided by a hard mastery gate), any question awaiting an "
"answer, due reviews, and a map of every objective's status "
"(new / learning / mastered). Call this FIRST on every mastery "
"turn — it tells you what to do; never guess the next objective."
),
parameters=[],
)
async def execute(self, **kwargs: Any) -> ToolResult:
path_id = _resolve_path_id(kwargs)
if not path_id:
return _no_path_result()
service = _new_service()
progress = service.get_or_create(path_id)
if not any(module.knowledge_points for module in progress.modules):
return _json_result(
{
"status": "empty",
"message": (
"No mastery path has been built yet. Design one from the "
"learner's materials and call mastery_build."
),
},
meta_key="mastery_status",
)
payload = {
"status": "active",
"next": next_objective(progress).to_dict(),
"map": map_summary(progress),
}
return _json_result(payload, meta_key="mastery_status")
class MasteryQuizTool(BaseTool):
"""Register an objective-type question; the engine holds the answer."""
def get_definition(self) -> ToolDefinition:
return ToolDefinition(
name="mastery_quiz",
description=(
"Pose a question for a MEMORY or PROCEDURE objective and register "
"its expected answer with the engine (so grading is deterministic "
"and you never re-state the answer later). After calling this, "
"present the question with the ask_user tool so the learner answers "
"on an interactive card (for choices, give ask_user options short "
"labels like A/B/C and set the correct label as expected_answer); "
"then call mastery_grade with their answer. For CONCEPT / DESIGN "
"objectives use mastery_assess instead."
),
parameters=[
ToolParameter(
name="knowledge_point_id",
type="string",
description="Objective id from mastery_status (verbatim).",
),
ToolParameter(
name="question",
type="string",
description="The question text shown to the learner.",
),
ToolParameter(
name="expected_answer",
type="string",
description="The correct answer, used only server-side for grading.",
),
ToolParameter(
name="question_type",
type="string",
description=(
"'choice' (exact match), 'short' (exact / fuzzy for ≤30 "
"chars), or 'open' (keyword overlap). Default 'short'."
),
required=False,
default="short",
enum=list(_QUESTION_TYPES),
),
ToolParameter(
name="options",
type="array",
description="Choice labels, when question_type='choice'.",
required=False,
items={"type": "string"},
),
],
)
async def execute(self, **kwargs: Any) -> ToolResult:
path_id = _resolve_path_id(kwargs)
if not path_id:
return _no_path_result()
kp_id = str(kwargs.get("knowledge_point_id") or "").strip()
question = str(kwargs.get("question") or "").strip()
expected = str(kwargs.get("expected_answer") or "").strip()
if not kp_id or not question or not expected:
return ToolResult(
content="mastery_quiz needs knowledge_point_id, question, and expected_answer.",
success=False,
)
q_type = str(kwargs.get("question_type") or "short").strip().lower()
if q_type not in _QUESTION_TYPES:
q_type = "short"
options = [str(o) for o in (kwargs.get("options") or []) if str(o).strip()]
service = _new_service()
progress = service.get_or_create(path_id)
kp, module_id, _ = find_knowledge_point(progress, kp_id)
if kp is None:
return ToolResult(
content=f"Unknown objective {kp_id!r}; call mastery_status for valid ids.",
success=False,
)
pending = PendingQuestion(
question_id=uuid.uuid4().hex,
knowledge_point_id=kp_id,
module_id=module_id,
prompt=question,
question_type=q_type,
expected_answer=expected,
options=options,
)
service.set_pending_question(progress, pending)
return _json_result(
{
"status": "registered",
"knowledge_point_id": kp_id,
"question": question,
"options": options,
"instruction": (
"Present this question with the ask_user tool (use its options "
"for multiple choice; the option labels must match the "
"expected_answer you registered), then call mastery_grade with "
"the learner's answer."
),
},
meta_key="mastery_quiz",
)
class MasteryGradeTool(BaseTool):
"""Grade the learner's answer to the pending question (deterministic)."""
def get_definition(self) -> ToolDefinition:
return ToolDefinition(
name="mastery_grade",
description=(
"Grade the learner's answer to the question you registered with "
"mastery_quiz. Grading is deterministic against the stored "
"expected answer; this updates mastery, advances spaced "
"repetition, and tells you whether the objective's gate is now "
"cleared. Then give the learner feedback."
),
parameters=[
ToolParameter(
name="answer",
type="string",
description="The learner's answer, verbatim.",
),
],
)
async def execute(self, **kwargs: Any) -> ToolResult:
path_id = _resolve_path_id(kwargs)
if not path_id:
return _no_path_result()
from deeptutor.learning.scheduler import SpacedRepetitionScheduler
answer = str(kwargs.get("answer") or "")
service = _new_service()
scheduler = SpacedRepetitionScheduler()
progress = service.get_or_create(path_id)
pending = progress.pending_question
if pending is None:
return ToolResult(
content="No question is awaiting an answer. Pose one with mastery_quiz first.",
success=False,
)
is_correct = service.grade_and_record(
progress,
question_id=pending.question_id,
knowledge_point_id=pending.knowledge_point_id,
module_id=pending.module_id,
user_answer=answer,
expected_answer=pending.expected_answer,
question_type=pending.question_type,
scheduler=scheduler,
)
service.clear_pending_question(progress)
kp, _, _ = find_knowledge_point(progress, pending.knowledge_point_id)
mastered = bool(kp and is_mastered(progress, kp))
payload = {
"is_correct": is_correct,
"knowledge_point_id": pending.knowledge_point_id,
"mastery": round(display_mastery(progress, kp), 3) if kp else 0.0,
"threshold": round(gate_threshold(kp.type), 3) if kp else 0.0,
"mastered": mastered,
"next": next_objective(progress).to_dict(),
}
return _json_result(payload, meta_key="mastery_grade")
class MasteryAssessTool(BaseTool):
"""Record the qualitative (CONCEPT / DESIGN) gate from a Feynman check."""
def get_definition(self) -> ToolDefinition:
return ToolDefinition(
name="mastery_assess",
description=(
"Record your judgement of a CONCEPT or DESIGN objective after the "
"learner explains it in their own words (a Feynman-style check). "
"Pass passed=true only when the explanation is correct and "
"complete enough to count as mastery — this is the gate for these "
"objective types. For MEMORY / PROCEDURE objectives use "
"mastery_quiz + mastery_grade instead."
),
parameters=[
ToolParameter(
name="knowledge_point_id",
type="string",
description="Objective id from mastery_status (verbatim).",
),
ToolParameter(
name="passed",
type="boolean",
description="True if the explanation demonstrates mastery.",
),
ToolParameter(
name="feedback",
type="string",
description="Short note on what was strong or missing (stored as evidence).",
required=False,
),
],
)
async def execute(self, **kwargs: Any) -> ToolResult:
path_id = _resolve_path_id(kwargs)
if not path_id:
return _no_path_result()
kp_id = str(kwargs.get("knowledge_point_id") or "").strip()
if not kp_id:
return ToolResult(content="mastery_assess needs a knowledge_point_id.", success=False)
passed = bool(kwargs.get("passed"))
feedback = str(kwargs.get("feedback") or "").strip()
service = _new_service()
progress = service.get_or_create(path_id)
kp, _, _ = find_knowledge_point(progress, kp_id)
if kp is None:
return ToolResult(
content=f"Unknown objective {kp_id!r}; call mastery_status for valid ids.",
success=False,
)
if kp.type not in QUALITATIVE_TYPES:
return ToolResult(
content=(
f"Objective {kp.name!r} is a {kp.type.value} type — gate it with "
"mastery_quiz + mastery_grade, not mastery_assess."
),
success=False,
)
service.record_qualitative(progress, kp_id, passed=passed, evidence=feedback)
payload = {
"knowledge_point_id": kp_id,
"passed": passed,
"mastered": is_mastered(progress, kp),
"mastery": round(display_mastery(progress, kp), 3),
"next": next_objective(progress).to_dict(),
}
return _json_result(payload, meta_key="mastery_assess")
class MasteryBuildTool(BaseTool):
"""Create / extend the skill map from objectives the tutor designed."""
def get_definition(self) -> ToolDefinition:
return ToolDefinition(
name="mastery_build",
description=(
"Create or extend the learner's mastery path. Design modules and "
"their knowledge points from the learner's materials (use rag / "
"read_source first when materials are attached) and pass them "
"here. Each knowledge point needs a 'type': memory (facts), "
"procedure (step-by-step skills), concept (ideas to understand), "
"or design (open-ended judgement). Use mode='replace' to start "
"fresh or 'append' to add to an existing path."
),
parameters=[
ToolParameter(
name="modules",
type="array",
description=(
"Ordered modules: each {name, knowledge_points: [{name, "
"type}]}. type is one of memory/procedure/concept/design."
),
items={
"type": "object",
"properties": {
"name": {"type": "string"},
"knowledge_points": {
"type": "array",
"items": {
"type": "object",
"properties": {
"name": {"type": "string"},
"type": {
"type": "string",
"enum": sorted(_ALLOWED_KP_TYPES),
},
},
"required": ["name"],
},
},
},
"required": ["name", "knowledge_points"],
},
),
ToolParameter(
name="mode",
type="string",
description="'replace' (default) starts fresh; 'append' adds modules.",
required=False,
default="replace",
enum=["replace", "append"],
),
],
)
async def execute(self, **kwargs: Any) -> ToolResult:
path_id = _resolve_path_id(kwargs)
if not path_id:
return _no_path_result()
mode = str(kwargs.get("mode") or "replace").strip().lower()
if mode not in {"replace", "append"}:
mode = "replace"
service = _new_service()
progress = service.get_or_create(path_id)
offset = len(progress.modules) if mode == "append" else 0
new_modules, error = _parse_modules(kwargs.get("modules"), path_id, offset)
if error:
return ToolResult(content=error, success=False)
combined = (list(progress.modules) + new_modules) if mode == "append" else new_modules
service.replace_modules(progress, combined)
progress.pending_question = None # a rebuilt map invalidates any open question
if combined:
progress.current_module_id = combined[0].id
progress.current_kp_index = 0
service.save(progress)
kp_count = sum(len(m.knowledge_points) for m in new_modules)
return _json_result(
{
"status": "built",
"mode": mode,
"modules_added": len(new_modules),
"knowledge_points_added": kp_count,
"map": map_summary(progress),
},
meta_key="mastery_build",
)
def _parse_modules(
raw_modules: Any, path_id: str, offset: int
) -> tuple[list[LearningModule], str | None]:
"""Validate the model-designed module tree into engine models.
Ids are generated server-side (``<path>_m<i>_kp<j>``) so the model never
controls storage keys; unknown knowledge types fall back to 'concept'.
"""
if not isinstance(raw_modules, list) or not raw_modules:
return [], "mastery_build needs a non-empty 'modules' array."
modules: list[LearningModule] = []
for i, raw in enumerate(raw_modules):
if not isinstance(raw, dict):
continue
index = offset + i
name = str(raw.get("name") or "").strip()[:200]
if not name:
continue
module_id = f"{path_id}_m{index}"
kps: list[KnowledgePoint] = []
for j, raw_kp in enumerate(raw.get("knowledge_points") or []):
if not isinstance(raw_kp, dict):
continue
kp_name = str(raw_kp.get("name") or "").strip()[:200]
if len(kp_name) < 2:
continue
kp_type = str(raw_kp.get("type") or "concept").strip().lower()
if kp_type not in _ALLOWED_KP_TYPES:
kp_type = "concept"
kps.append(
KnowledgePoint(
id=f"{module_id}_kp{j}",
name=kp_name,
type=KnowledgeType(kp_type),
module_id=module_id,
)
)
if not kps:
continue
modules.append(LearningModule(id=module_id, name=name, order=index, knowledge_points=kps))
if not modules:
return [], "No valid modules: each module needs a name and at least one knowledge point."
return modules, None
MASTERY_TOOL_TYPES: tuple[type[BaseTool], ...] = (
MasteryStatusTool,
MasteryQuizTool,
MasteryGradeTool,
MasteryAssessTool,
MasteryBuildTool,
)
__all__ = [
"MASTERY_TOOL_NAMES",
"MASTERY_TOOL_TYPES",
"MasteryStatusTool",
"MasteryQuizTool",
"MasteryGradeTool",
"MasteryAssessTool",
"MasteryBuildTool",
]
deeptutor/loop_plugins/protocol.py
+51
View File
@@ -0,0 +1,51 @@
"""Protocol shared by the chat loop and loop plugins."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Protocol
from deeptutor.core.context import UnifiedContext
@dataclass(frozen=True, slots=True)
class PromptBlock:
"""One named prompt fragment contributed to the loop system prompt."""
name: str
content: str
class LoopPlugin(Protocol):
"""Optional per-turn extension point for the chat agent loop."""
name: str
def is_active(self, context: UnifiedContext) -> bool:
"""Whether this plugin participates in the current turn."""
def tool_types(self, context: UnifiedContext) -> tuple[str, ...]:
"""Tool names this plugin auto-mounts for the current turn."""
def system_block(
self,
context: UnifiedContext,
*,
language: str,
prompts: dict[str, Any],
) -> PromptBlock | None:
"""Optional system prompt block contributed by the plugin."""
def augment_kwargs(
self,
tool_name: str,
kwargs: dict[str, Any],
context: UnifiedContext,
) -> dict[str, Any]:
"""Inject server-owned private kwargs for plugin tools."""
def pre_loop_seed(self, context: UnifiedContext) -> str:
"""Optional text appended to the initial user message seed."""
__all__ = ["LoopPlugin", "PromptBlock"]
deeptutor/loop_plugins/registry.py
+17
View File
@@ -0,0 +1,17 @@
"""Built-in loop plugin registry."""
from __future__ import annotations
from deeptutor.core.context import UnifiedContext
from deeptutor.loop_plugins.mastery import MasteryLoopPlugin
from deeptutor.loop_plugins.protocol import LoopPlugin
LOOP_PLUGINS: tuple[LoopPlugin, ...] = (MasteryLoopPlugin(),)
def active_loop_plugins(context: UnifiedContext) -> tuple[LoopPlugin, ...]:
"""Return active plugins for this turn in stable registry order."""
return tuple(plugin for plugin in LOOP_PLUGINS if plugin.is_active(context))
__all__ = ["LOOP_PLUGINS", "active_loop_plugins"]
deeptutor/tools/builtin/__init__.py
+3 -3
View File
@@ -8,8 +8,8 @@ import logging
from typing import Any
from deeptutor.core.tool_protocol import BaseTool, ToolDefinition, ToolParameter, ToolResult
from deeptutor.loop_plugins.mastery import MASTERY_TOOL_TYPES
from deeptutor.tools.exec_tool import ExecTool
from deeptutor.tools.mastery_tool import MASTERY_TOOL_TYPES
from deeptutor.tools.prompting import load_prompt_hints
logger = logging.getLogger(__name__)
@@ -1452,8 +1452,8 @@ BUILTIN_TOOL_TYPES: tuple[type[BaseTool], ...] = (
GithubTool,
AskUserTool,
CronTool,
# Mastery Path tools — auto-mounted only when a path is active on the turn
# (see ToolMountFlags.has_mastery); the chat agent loop drives them.
# Mastery Path tools — globally registered so schemas/API stay stable;
# the chat loop plugin decides when to auto-mount them for a turn.
*MASTERY_TOOL_TYPES,
)
deeptutor/tools/mastery_tool.py
+11 -508
View File
@@ -1,518 +1,21 @@
"""Mastery Path tools — the seam between the chat-loop tutor and the pure
mastery engine (:mod:`deeptutor.learning`).
"""Compatibility exports for mastery path tools.
These five tools are auto-mounted only when a mastery path is active on the
turn (see ``ToolMountFlags.has_mastery``). The chat agent loop IS the tutor;
these tools let it read the gate and record outcomes, while the pedagogy —
what to teach, how to question, when to explain — stays the model's job. The
arithmetic (mastery, gate, spaced repetition) stays in the engine.
The active path id is injected server-side by the pipeline as
``_mastery_path_id``; the model never supplies it. Each call constructs a
fresh store + service (matching the REST router) so concurrent turns can't
race on a shared object.
The mastery loop plugin owns the implementation under
``deeptutor.loop_plugins.mastery.tools``. This module keeps the historical
import path stable for the built-in tool registry, capability manifests, and
external users.
"""
from __future__ import annotations
import json
from typing import TYPE_CHECKING, Any
import uuid
from deeptutor.core.tool_protocol import BaseTool, ToolDefinition, ToolParameter, ToolResult
# ``learning.models`` and ``learning.policy`` only depend on pydantic — safe to
# import at module load. ``learning.service`` / ``storage`` / ``scheduler``
# reach the path service (and so the runtime + tool registry), so importing
# them here would close an import cycle: tool_registry -> builtin ->
# mastery_tool -> service -> path_service -> runtime -> tool_registry. They are
# imported lazily inside the call paths instead (same pattern as the other
# builtin tools).
from deeptutor.learning.models import (
KnowledgePoint,
KnowledgeType,
LearningModule,
PendingQuestion,
)
from deeptutor.learning.policy import (
QUALITATIVE_TYPES,
display_mastery,
find_knowledge_point,
gate_threshold,
is_mastered,
map_summary,
next_objective,
)
if TYPE_CHECKING:
from deeptutor.learning.service import LearningService
# Tool names the pipeline mounts together when a mastery path is active. Kept
# here so the mount policy and the registration list can't disagree.
MASTERY_TOOL_NAMES: tuple[str, ...] = (
"mastery_status",
"mastery_quiz",
"mastery_grade",
"mastery_assess",
"mastery_build",
)
_QUESTION_TYPES = ("choice", "short", "open")
_ALLOWED_KP_TYPES = {t.value for t in KnowledgeType}
def _new_service() -> LearningService:
from deeptutor.learning.service import LearningService
from deeptutor.learning.storage import LearningStore
return LearningService(LearningStore())
def _resolve_path_id(kwargs: dict[str, Any]) -> str:
return str(kwargs.get("_mastery_path_id") or "").strip()
def _json_result(payload: dict[str, Any], *, meta_key: str, success: bool = True) -> ToolResult:
return ToolResult(
content=json.dumps(payload, ensure_ascii=False),
success=success,
metadata={meta_key: payload},
)
def _no_path_result() -> ToolResult:
return ToolResult(
content="No mastery path is active on this turn; mastery tools are unavailable.",
success=False,
)
class MasteryStatusTool(BaseTool):
"""Read the current objective + map snapshot. Call FIRST every turn."""
def get_definition(self) -> ToolDefinition:
return ToolDefinition(
name="mastery_status",
description=(
"Read the learner's mastery path: the next objective to work on "
"(decided by a hard mastery gate), any question awaiting an "
"answer, due reviews, and a map of every objective's status "
"(new / learning / mastered). Call this FIRST on every mastery "
"turn — it tells you what to do; never guess the next objective."
),
parameters=[],
)
async def execute(self, **kwargs: Any) -> ToolResult:
path_id = _resolve_path_id(kwargs)
if not path_id:
return _no_path_result()
service = _new_service()
progress = service.get_or_create(path_id)
if not any(module.knowledge_points for module in progress.modules):
return _json_result(
{
"status": "empty",
"message": (
"No mastery path has been built yet. Design one from the "
"learner's materials and call mastery_build."
),
},
meta_key="mastery_status",
)
payload = {
"status": "active",
"next": next_objective(progress).to_dict(),
"map": map_summary(progress),
}
return _json_result(payload, meta_key="mastery_status")
class MasteryQuizTool(BaseTool):
"""Register an objective-type question; the engine holds the answer."""
def get_definition(self) -> ToolDefinition:
return ToolDefinition(
name="mastery_quiz",
description=(
"Pose a question for a MEMORY or PROCEDURE objective and register "
"its expected answer with the engine (so grading is deterministic "
"and you never re-state the answer later). After calling this, "
"present the question with the ask_user tool so the learner answers "
"on an interactive card (for choices, give ask_user options short "
"labels like A/B/C and set the correct label as expected_answer); "
"then call mastery_grade with their answer. For CONCEPT / DESIGN "
"objectives use mastery_assess instead."
),
parameters=[
ToolParameter(
name="knowledge_point_id",
type="string",
description="Objective id from mastery_status (verbatim).",
),
ToolParameter(
name="question",
type="string",
description="The question text shown to the learner.",
),
ToolParameter(
name="expected_answer",
type="string",
description="The correct answer, used only server-side for grading.",
),
ToolParameter(
name="question_type",
type="string",
description=(
"'choice' (exact match), 'short' (exact / fuzzy for ≤30 "
"chars), or 'open' (keyword overlap). Default 'short'."
),
required=False,
default="short",
enum=list(_QUESTION_TYPES),
),
ToolParameter(
name="options",
type="array",
description="Choice labels, when question_type='choice'.",
required=False,
items={"type": "string"},
),
],
)
async def execute(self, **kwargs: Any) -> ToolResult:
path_id = _resolve_path_id(kwargs)
if not path_id:
return _no_path_result()
kp_id = str(kwargs.get("knowledge_point_id") or "").strip()
question = str(kwargs.get("question") or "").strip()
expected = str(kwargs.get("expected_answer") or "").strip()
if not kp_id or not question or not expected:
return ToolResult(
content="mastery_quiz needs knowledge_point_id, question, and expected_answer.",
success=False,
)
q_type = str(kwargs.get("question_type") or "short").strip().lower()
if q_type not in _QUESTION_TYPES:
q_type = "short"
options = [str(o) for o in (kwargs.get("options") or []) if str(o).strip()]
service = _new_service()
progress = service.get_or_create(path_id)
kp, module_id, _ = find_knowledge_point(progress, kp_id)
if kp is None:
return ToolResult(
content=f"Unknown objective {kp_id!r}; call mastery_status for valid ids.",
success=False,
)
pending = PendingQuestion(
question_id=uuid.uuid4().hex,
knowledge_point_id=kp_id,
module_id=module_id,
prompt=question,
question_type=q_type,
expected_answer=expected,
options=options,
)
service.set_pending_question(progress, pending)
return _json_result(
{
"status": "registered",
"knowledge_point_id": kp_id,
"question": question,
"options": options,
"instruction": (
"Present this question with the ask_user tool (use its options "
"for multiple choice; the option labels must match the "
"expected_answer you registered), then call mastery_grade with "
"the learner's answer."
),
},
meta_key="mastery_quiz",
)
class MasteryGradeTool(BaseTool):
"""Grade the learner's answer to the pending question (deterministic)."""
def get_definition(self) -> ToolDefinition:
return ToolDefinition(
name="mastery_grade",
description=(
"Grade the learner's answer to the question you registered with "
"mastery_quiz. Grading is deterministic against the stored "
"expected answer; this updates mastery, advances spaced "
"repetition, and tells you whether the objective's gate is now "
"cleared. Then give the learner feedback."
),
parameters=[
ToolParameter(
name="answer",
type="string",
description="The learner's answer, verbatim.",
),
],
)
async def execute(self, **kwargs: Any) -> ToolResult:
path_id = _resolve_path_id(kwargs)
if not path_id:
return _no_path_result()
from deeptutor.learning.scheduler import SpacedRepetitionScheduler
answer = str(kwargs.get("answer") or "")
service = _new_service()
scheduler = SpacedRepetitionScheduler()
progress = service.get_or_create(path_id)
pending = progress.pending_question
if pending is None:
return ToolResult(
content="No question is awaiting an answer. Pose one with mastery_quiz first.",
success=False,
)
is_correct = service.grade_and_record(
progress,
question_id=pending.question_id,
knowledge_point_id=pending.knowledge_point_id,
module_id=pending.module_id,
user_answer=answer,
expected_answer=pending.expected_answer,
question_type=pending.question_type,
scheduler=scheduler,
)
service.clear_pending_question(progress)
kp, _, _ = find_knowledge_point(progress, pending.knowledge_point_id)
mastered = bool(kp and is_mastered(progress, kp))
payload = {
"is_correct": is_correct,
"knowledge_point_id": pending.knowledge_point_id,
"mastery": round(display_mastery(progress, kp), 3) if kp else 0.0,
"threshold": round(gate_threshold(kp.type), 3) if kp else 0.0,
"mastered": mastered,
"next": next_objective(progress).to_dict(),
}
return _json_result(payload, meta_key="mastery_grade")
class MasteryAssessTool(BaseTool):
"""Record the qualitative (CONCEPT / DESIGN) gate from a Feynman check."""
def get_definition(self) -> ToolDefinition:
return ToolDefinition(
name="mastery_assess",
description=(
"Record your judgement of a CONCEPT or DESIGN objective after the "
"learner explains it in their own words (a Feynman-style check). "
"Pass passed=true only when the explanation is correct and "
"complete enough to count as mastery — this is the gate for these "
"objective types. For MEMORY / PROCEDURE objectives use "
"mastery_quiz + mastery_grade instead."
),
parameters=[
ToolParameter(
name="knowledge_point_id",
type="string",
description="Objective id from mastery_status (verbatim).",
),
ToolParameter(
name="passed",
type="boolean",
description="True if the explanation demonstrates mastery.",
),
ToolParameter(
name="feedback",
type="string",
description="Short note on what was strong or missing (stored as evidence).",
required=False,
),
],
)
async def execute(self, **kwargs: Any) -> ToolResult:
path_id = _resolve_path_id(kwargs)
if not path_id:
return _no_path_result()
kp_id = str(kwargs.get("knowledge_point_id") or "").strip()
if not kp_id:
return ToolResult(content="mastery_assess needs a knowledge_point_id.", success=False)
passed = bool(kwargs.get("passed"))
feedback = str(kwargs.get("feedback") or "").strip()
service = _new_service()
progress = service.get_or_create(path_id)
kp, _, _ = find_knowledge_point(progress, kp_id)
if kp is None:
return ToolResult(
content=f"Unknown objective {kp_id!r}; call mastery_status for valid ids.",
success=False,
)
if kp.type not in QUALITATIVE_TYPES:
return ToolResult(
content=(
f"Objective {kp.name!r} is a {kp.type.value} type — gate it with "
"mastery_quiz + mastery_grade, not mastery_assess."
),
success=False,
)
service.record_qualitative(progress, kp_id, passed=passed, evidence=feedback)
payload = {
"knowledge_point_id": kp_id,
"passed": passed,
"mastered": is_mastered(progress, kp),
"mastery": round(display_mastery(progress, kp), 3),
"next": next_objective(progress).to_dict(),
}
return _json_result(payload, meta_key="mastery_assess")
class MasteryBuildTool(BaseTool):
"""Create / extend the skill map from objectives the tutor designed."""
def get_definition(self) -> ToolDefinition:
return ToolDefinition(
name="mastery_build",
description=(
"Create or extend the learner's mastery path. Design modules and "
"their knowledge points from the learner's materials (use rag / "
"read_source first when materials are attached) and pass them "
"here. Each knowledge point needs a 'type': memory (facts), "
"procedure (step-by-step skills), concept (ideas to understand), "
"or design (open-ended judgement). Use mode='replace' to start "
"fresh or 'append' to add to an existing path."
),
parameters=[
ToolParameter(
name="modules",
type="array",
description=(
"Ordered modules: each {name, knowledge_points: [{name, "
"type}]}. type is one of memory/procedure/concept/design."
),
items={
"type": "object",
"properties": {
"name": {"type": "string"},
"knowledge_points": {
"type": "array",
"items": {
"type": "object",
"properties": {
"name": {"type": "string"},
"type": {
"type": "string",
"enum": sorted(_ALLOWED_KP_TYPES),
},
},
"required": ["name"],
},
},
},
"required": ["name", "knowledge_points"],
},
),
ToolParameter(
name="mode",
type="string",
description="'replace' (default) starts fresh; 'append' adds modules.",
required=False,
default="replace",
enum=["replace", "append"],
),
],
)
async def execute(self, **kwargs: Any) -> ToolResult:
path_id = _resolve_path_id(kwargs)
if not path_id:
return _no_path_result()
mode = str(kwargs.get("mode") or "replace").strip().lower()
if mode not in {"replace", "append"}:
mode = "replace"
service = _new_service()
progress = service.get_or_create(path_id)
offset = len(progress.modules) if mode == "append" else 0
new_modules, error = _parse_modules(kwargs.get("modules"), path_id, offset)
if error:
return ToolResult(content=error, success=False)
combined = (list(progress.modules) + new_modules) if mode == "append" else new_modules
service.replace_modules(progress, combined)
progress.pending_question = None # a rebuilt map invalidates any open question
if combined:
progress.current_module_id = combined[0].id
progress.current_kp_index = 0
service.save(progress)
kp_count = sum(len(m.knowledge_points) for m in new_modules)
return _json_result(
{
"status": "built",
"mode": mode,
"modules_added": len(new_modules),
"knowledge_points_added": kp_count,
"map": map_summary(progress),
},
meta_key="mastery_build",
)
def _parse_modules(
raw_modules: Any, path_id: str, offset: int
) -> tuple[list[LearningModule], str | None]:
"""Validate the model-designed module tree into engine models.
Ids are generated server-side (``<path>_m<i>_kp<j>``) so the model never
controls storage keys; unknown knowledge types fall back to 'concept'.
"""
if not isinstance(raw_modules, list) or not raw_modules:
return [], "mastery_build needs a non-empty 'modules' array."
modules: list[LearningModule] = []
for i, raw in enumerate(raw_modules):
if not isinstance(raw, dict):
continue
index = offset + i
name = str(raw.get("name") or "").strip()[:200]
if not name:
continue
module_id = f"{path_id}_m{index}"
kps: list[KnowledgePoint] = []
for j, raw_kp in enumerate(raw.get("knowledge_points") or []):
if not isinstance(raw_kp, dict):
continue
kp_name = str(raw_kp.get("name") or "").strip()[:200]
if len(kp_name) < 2:
continue
kp_type = str(raw_kp.get("type") or "concept").strip().lower()
if kp_type not in _ALLOWED_KP_TYPES:
kp_type = "concept"
kps.append(
KnowledgePoint(
id=f"{module_id}_kp{j}",
name=kp_name,
type=KnowledgeType(kp_type),
module_id=module_id,
)
)
if not kps:
continue
modules.append(
LearningModule(id=module_id, name=name, order=index, knowledge_points=kps)
)
if not modules:
return [], "No valid modules: each module needs a name and at least one knowledge point."
return modules, None
MASTERY_TOOL_TYPES: tuple[type[BaseTool], ...] = (
MasteryStatusTool,
MasteryQuizTool,
MasteryGradeTool,
from deeptutor.loop_plugins.mastery.tools import (
MASTERY_TOOL_NAMES,
MASTERY_TOOL_TYPES,
MasteryAssessTool,
MasteryBuildTool,
MasteryGradeTool,
MasteryQuizTool,
MasteryStatusTool,
)
__all__ = [
"MASTERY_TOOL_NAMES",
"MASTERY_TOOL_TYPES",
start_backend.bat → scripts/start_backend.bat
+2 -1
View File
@@ -2,7 +2,8 @@
REM DeepTutor Backend Startup Script
REM Activates virtual environment and starts the backend API server
cd /d "%~dp0"
REM Move to the project root (this script lives in scripts/)
cd /d "%~dp0.."
REM Set UTF-8 encoding for stdout to support emoji characters
set PYTHONIOENCODING=utf-8
start_frontend.bat → scripts/start_frontend.bat
+2 -1
View File
@@ -2,7 +2,8 @@
REM DeepTutor Frontend Startup Script
REM Starts the frontend Next.js development server
cd /d "%~dp0"
REM Move to the project root (this script lives in scripts/)
cd /d "%~dp0.."
echo Starting DeepTutor Frontend...
echo Frontend will be available at: http://localhost:3782
echo Press Ctrl+C to stop the server.
tests/agents/chat/test_agent_loop.py
+121
View File
@@ -13,6 +13,7 @@ from deeptutor.core.context import Attachment, UnifiedContext
from deeptutor.core.stream import StreamEvent, StreamEventType
from deeptutor.core.stream_bus import StreamBus
from deeptutor.core.tool_protocol import ToolResult
from deeptutor.loop_plugins.mastery import MASTERY_TOOL_NAMES
async def _collect_bus_events(bus: StreamBus) -> tuple[list[StreamEvent], asyncio.Task[Any]]:
@@ -404,6 +405,84 @@ async def test_tool_round_then_finish(monkeypatch: pytest.MonkeyPatch) -> None:
assert result.metadata["response"] == "Found what was needed."
@pytest.mark.asyncio
async def test_context_checkpoint_folds_completed_tool_rounds(
monkeypatch: pytest.MonkeyPatch,
) -> None:
class _CheckpointRegistry(_Registry):
async def execute(self, name: str, **kwargs):
self.executed.append({"name": name, "kwargs": kwargs})
query = str(kwargs.get("query") or "")
return ToolResult(
content=f"noisy tool result for {query}",
success=True,
metadata={"_context_checkpoint": {"summary": f"checkpoint: {query}"}},
)
registry = _CheckpointRegistry()
client = _ScriptedChatClient(
[
[
_llm_chunk(content="Searching step one."),
_llm_chunk(
tool_calls=[
{
"id": "call-1",
"name": "web_search",
"arguments": json.dumps({"query": "step one"}),
}
]
),
],
[
_llm_chunk(content="Searching step two."),
_llm_chunk(
tool_calls=[
{
"id": "call-2",
"name": "web_search",
"arguments": json.dumps({"query": "step two"}),
}
]
),
],
[_llm_chunk(content="Final from checkpoints.")],
]
)
pipeline = AgenticChatPipeline(language="en")
pipeline.registry = registry
monkeypatch.setattr(pipeline, "_compose_enabled_tools", lambda _context: ["web_search"])
monkeypatch.setattr(pipeline, "_build_openai_client", lambda: client)
events = await _run(
pipeline,
UnifiedContext(
session_id="s1",
user_message="Research this",
enabled_tools=["web_search"],
),
)
assert client.call_count == 3
second_round = client.calls[1]["messages"]
assert any(
m.get("role") == "system" and "checkpoint: step one" in str(m.get("content"))
for m in second_round
)
assert not any(m.get("role") == "tool" for m in second_round)
assert not any("Searching step one." in str(m.get("content")) for m in second_round)
third_round = client.calls[2]["messages"]
checkpoint_text = "\n".join(
str(m.get("content") or "") for m in third_round if m.get("role") == "system"
)
assert "checkpoint: step one" in checkpoint_text
assert "checkpoint: step two" in checkpoint_text
assert not any(m.get("role") == "tool" for m in third_round)
assert not any("noisy tool result" in str(m.get("content")) for m in third_round)
result = _result(events)
assert result.metadata["response"] == "Final from checkpoints."
@pytest.mark.asyncio
async def test_ask_user_available_every_round(monkeypatch: pytest.MonkeyPatch) -> None:
"""The single loop offers the full tool belt — including ask_user — on
@@ -609,6 +688,48 @@ def test_compose_enabled_tools_injects_rag_when_kb_selected(
assert "web_search" in pipeline._compose_enabled_tools(context)
def test_compose_enabled_tools_mounts_mastery_plugin_only_in_mastery_mode(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(
"deeptutor.services.memory.get_memory_store",
lambda: SimpleNamespace(read_raw=lambda *_args, **_kwargs: ""),
)
monkeypatch.setattr(
"deeptutor.services.notebook.get_notebook_manager",
lambda: SimpleNamespace(list_notebooks=lambda: []),
)
pipeline = AgenticChatPipeline.__new__(AgenticChatPipeline)
pipeline._deferred_loader = None
pipeline._exec_enabled = False
pipeline.registry = SimpleNamespace(
get_enabled=lambda selected: [SimpleNamespace(name=n) for n in selected]
)
ordinary = UnifiedContext(user_message="hi")
mastery = UnifiedContext(
user_message="teach me",
metadata={"mastery_mode": True, "mastery_path_id": "path-a"},
)
ordinary_tools = pipeline._compose_enabled_tools(ordinary)
mastery_tools = pipeline._compose_enabled_tools(mastery)
assert not set(MASTERY_TOOL_NAMES).intersection(ordinary_tools)
assert set(MASTERY_TOOL_NAMES).issubset(mastery_tools)
def test_augment_tool_kwargs_injects_mastery_path_id() -> None:
pipeline = AgenticChatPipeline.__new__(AgenticChatPipeline)
context = UnifiedContext(
user_message="teach",
metadata={"mastery_mode": True, "mastery_path_id": "book-1"},
)
augmented = pipeline._augment_tool_kwargs("mastery_status", {}, context)
assert augmented["_mastery_path_id"] == "book-1"
def test_augment_tool_kwargs_injects_geogebra_image() -> None:
pipeline = AgenticChatPipeline.__new__(AgenticChatPipeline)
pipeline.language = "zh"
tests/agents/chat/test_language_prompts.py
+24
View File
@@ -53,6 +53,30 @@ def test_agentic_chat_final_prompt_uses_selected_language(
assert "You are DeepTutor" in en_prompt
def test_mastery_plugin_system_prompt_uses_localized_fallback(
monkeypatch: pytest.MonkeyPatch,
) -> None:
class FakeRegistry:
def build_prompt_text(self, *_args, **_kwargs) -> str:
return "- tool"
monkeypatch.setattr(
"deeptutor.agents.chat.agentic_pipeline.get_tool_registry",
lambda: FakeRegistry(),
)
from deeptutor.core.context import UnifiedContext
ctx = UnifiedContext(metadata={"mastery_mode": True, "mastery_path_id": "p1"})
zh_prompt = AgenticChatPipeline(language="zh")._build_system_prompt([], ctx)
en_prompt = AgenticChatPipeline(language="en")._build_system_prompt([], ctx)
assert "## mastery_tutor" in zh_prompt
assert "精通导师模式" in zh_prompt
assert "## mastery_tutor" in en_prompt
assert "Mastery Tutor mode" in en_prompt
def test_legacy_chat_agent_system_prompt_uses_selected_language() -> None:
zh_messages = ChatAgent(language="zh", config={}).build_messages(
message="解释梯度下降",