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
215 lines
7.9 KiB
Python
215 lines
7.9 KiB
Python
"""Stable application-layer facade for DeepTutor entry points."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
import importlib.util
|
|
import json
|
|
from typing import Any, AsyncIterator
|
|
|
|
from deeptutor.runtime.registry.capability_registry import get_capability_registry
|
|
from deeptutor.services.notebook import get_notebook_manager
|
|
from deeptutor.services.session import get_session_store, get_turn_runtime_manager
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class TurnRequest:
|
|
"""Stable turn payload used by adapters such as the CLI package."""
|
|
|
|
content: str
|
|
capability: str = "chat"
|
|
session_id: str | None = None
|
|
tools: list[str] = field(default_factory=list)
|
|
knowledge_bases: list[str] = field(default_factory=list)
|
|
language: str = "en"
|
|
config: dict[str, Any] = field(default_factory=dict)
|
|
notebook_references: list[dict[str, Any]] = field(default_factory=list)
|
|
history_references: list[str] = field(default_factory=list)
|
|
attachments: list[dict[str, Any]] = field(default_factory=list)
|
|
skills: list[str] = field(default_factory=list)
|
|
|
|
def to_payload(self) -> dict[str, Any]:
|
|
return {
|
|
"content": self.content,
|
|
"capability": self.capability,
|
|
"session_id": self.session_id,
|
|
"tools": list(self.tools),
|
|
"knowledge_bases": list(self.knowledge_bases),
|
|
"language": self.language,
|
|
"config": dict(self.config),
|
|
"notebook_references": list(self.notebook_references),
|
|
"history_references": list(self.history_references),
|
|
"attachments": list(self.attachments),
|
|
"skills": list(self.skills),
|
|
}
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class CapabilityAvailability:
|
|
"""Availability result for optional capabilities."""
|
|
|
|
name: str
|
|
available: bool
|
|
install_hint: str = ""
|
|
|
|
|
|
class DeepTutorApp:
|
|
"""Facade around runtime, session, notebook, and capability contracts."""
|
|
|
|
def __init__(self) -> None:
|
|
self.runtime = get_turn_runtime_manager()
|
|
self.store = get_session_store()
|
|
self.notebooks = get_notebook_manager()
|
|
self.capabilities = get_capability_registry()
|
|
|
|
def resolve_capability(self, value: str | None) -> str:
|
|
requested = str(value or "chat").strip() or "chat"
|
|
manifests = self.capabilities.get_manifests()
|
|
for manifest in manifests:
|
|
if manifest["name"] == requested:
|
|
return requested
|
|
aliases = {str(alias).strip() for alias in manifest.get("cli_aliases", [])}
|
|
if requested in aliases:
|
|
return str(manifest["name"])
|
|
available = ", ".join(sorted(manifest["name"] for manifest in manifests))
|
|
raise ValueError(f"Unknown capability `{requested}`. Available: {available}")
|
|
|
|
def get_capability_contracts(self) -> list[dict[str, Any]]:
|
|
contracts = []
|
|
for manifest in self.capabilities.get_manifests():
|
|
contracts.append(
|
|
{
|
|
**manifest,
|
|
"availability": self.get_capability_availability(manifest["name"]).__dict__,
|
|
}
|
|
)
|
|
return contracts
|
|
|
|
def get_capability_contract(self, value: str) -> dict[str, Any]:
|
|
resolved = self.resolve_capability(value)
|
|
for manifest in self.capabilities.get_manifests():
|
|
if manifest["name"] == resolved:
|
|
return {
|
|
**manifest,
|
|
"availability": self.get_capability_availability(resolved).__dict__,
|
|
}
|
|
raise ValueError(f"Capability not found: {resolved}")
|
|
|
|
def get_capability_availability(self, capability: str) -> CapabilityAvailability:
|
|
resolved = self.resolve_capability(capability)
|
|
if resolved == "math_animator":
|
|
available = importlib.util.find_spec("manim") is not None
|
|
return CapabilityAvailability(
|
|
name=resolved,
|
|
available=available,
|
|
install_hint=(
|
|
""
|
|
if available
|
|
else "Install with `pip install -e '.[math-animator]'` "
|
|
"or `pip install -r requirements/math-animator.txt`."
|
|
),
|
|
)
|
|
return CapabilityAvailability(name=resolved, available=True)
|
|
|
|
async def start_turn(
|
|
self, request: TurnRequest | dict[str, Any]
|
|
) -> tuple[dict[str, Any], dict[str, Any]]:
|
|
if isinstance(request, dict):
|
|
request = TurnRequest(**request)
|
|
resolved_capability = self.resolve_capability(request.capability)
|
|
session, turn = await self.runtime.start_turn(
|
|
{
|
|
**request.to_payload(),
|
|
"capability": resolved_capability,
|
|
}
|
|
)
|
|
await self.store.update_session_preferences(
|
|
session["id"],
|
|
{
|
|
"language": request.language,
|
|
"notebook_references": request.notebook_references,
|
|
"history_references": request.history_references,
|
|
},
|
|
)
|
|
return session, turn
|
|
|
|
async def stream_turn(self, turn_id: str, after_seq: int = 0) -> AsyncIterator[dict[str, Any]]:
|
|
async for item in self.runtime.subscribe_turn(turn_id, after_seq=after_seq):
|
|
yield item
|
|
|
|
async def cancel_turn(self, turn_id: str) -> bool:
|
|
return await self.runtime.cancel_turn(turn_id)
|
|
|
|
async def submit_user_reply(
|
|
self,
|
|
turn_id: str,
|
|
text: str | None = None,
|
|
*,
|
|
answers: list[dict[str, Any]] | None = None,
|
|
) -> bool:
|
|
"""Deliver the user's reply to a turn paused on ``ask_user``."""
|
|
return await self.runtime.submit_user_reply(turn_id, text=text, answers=answers)
|
|
|
|
async def regenerate_last_turn(
|
|
self,
|
|
session_id: str,
|
|
overrides: dict[str, Any] | None = None,
|
|
) -> tuple[dict[str, Any], dict[str, Any]]:
|
|
return await self.runtime.regenerate_last_turn(session_id, overrides=overrides)
|
|
|
|
async def list_sessions(self, limit: int = 50, offset: int = 0) -> list[dict[str, Any]]:
|
|
return await self.store.list_sessions(limit=limit, offset=offset)
|
|
|
|
async def get_session(self, session_id: str) -> dict[str, Any] | None:
|
|
return await self.store.get_session_with_messages(session_id)
|
|
|
|
async def rename_session(self, session_id: str, title: str) -> bool:
|
|
return await self.store.update_session_title(session_id, title)
|
|
|
|
async def delete_session(self, session_id: str) -> bool:
|
|
return await self.store.delete_session(session_id)
|
|
|
|
async def get_active_turn(self, session_id: str) -> dict[str, Any] | None:
|
|
return await self.store.get_active_turn(session_id)
|
|
|
|
def list_notebooks(self) -> list[dict[str, Any]]:
|
|
return self.notebooks.list_notebooks()
|
|
|
|
def create_notebook(
|
|
self,
|
|
name: str,
|
|
description: str = "",
|
|
*,
|
|
color: str = "#3B82F6",
|
|
icon: str = "book",
|
|
) -> dict[str, Any]:
|
|
return self.notebooks.create_notebook(
|
|
name=name,
|
|
description=description,
|
|
color=color,
|
|
icon=icon,
|
|
)
|
|
|
|
def get_notebook(self, notebook_id: str) -> dict[str, Any] | None:
|
|
return self.notebooks.get_notebook(notebook_id)
|
|
|
|
def add_record(self, **kwargs: Any) -> dict[str, Any]:
|
|
return self.notebooks.add_record(**kwargs)
|
|
|
|
def update_record(
|
|
self, notebook_id: str, record_id: str, **kwargs: Any
|
|
) -> dict[str, Any] | None:
|
|
return self.notebooks.update_record(notebook_id, record_id, **kwargs)
|
|
|
|
def remove_record(self, notebook_id: str, record_id: str) -> bool:
|
|
return self.notebooks.remove_record(notebook_id, record_id)
|
|
|
|
def get_records_by_references(
|
|
self, notebook_references: list[dict[str, Any]]
|
|
) -> list[dict[str, Any]]:
|
|
return self.notebooks.get_records_by_references(notebook_references)
|
|
|
|
|
|
def dumps_json(value: Any) -> str:
|
|
return json.dumps(value, ensure_ascii=False, indent=2, default=str)
|