Files
wehub-resource-sync e4dcfc49aa
Tests / Import Check (Python 3.13) (push) Has been cancelled
Tests / Import Check (Python 3.14) (push) Has been cancelled
Tests / Python Tests (Python 3.11) (push) Has been cancelled
Tests / Python Tests (Python 3.12) (push) Has been cancelled
Tests / Python Tests (Python 3.14) (push) Has been cancelled
Tests / Test Summary (push) Has been cancelled
Tests / Lint and Format (push) Has been cancelled
Tests / Web Node Tests (push) Has been cancelled
Tests / Import Check (Python 3.11) (push) Has been cancelled
Tests / Import Check (Python 3.12) (push) Has been cancelled
Tests / Python Tests (Python 3.13) (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 13:00:43 +08:00

74 lines
2.3 KiB
Python

"""Mastery path loop-capability hooks."""
from __future__ import annotations
from importlib import resources
from typing import Any
from deeptutor.capabilities.mastery.tools import MASTERY_TOOL_NAMES
from deeptutor.capabilities.protocol import PromptBlock
from deeptutor.core.context import UnifiedContext
class MasteryLoopCapability:
"""Turn-scoped integration for mastery-path tutoring.
Reuses the full chat tool surface (rag / read_source / ask_user / … under
the same user toggles as chat) and adds the mastery engine tools on top.
"""
name = "mastery"
owned_tools = MASTERY_TOOL_NAMES
def is_active(self, context: UnifiedContext) -> bool:
return bool(context.metadata.get("mastery_mode"))
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()
updated["_session_id"] = str(context.session_id or "").strip()
updated["_turn_id"] = str(context.metadata.get("turn_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__ = ["MasteryLoopCapability"]