Spaces:
Sleeping
Sleeping
| from typing import List, Dict, Any | |
| from app.contracts import EngineRequest, EngineResponse, ErrorDetail | |
| from app.hf_client import HFClient | |
| # Internal constants for Mode enum (keep simple strings) | |
| MODE_TEXT = "text" | |
| MODE_AUDIO = "audio" | |
| MODE_VIDEO = "video" | |
| MODE_VISUAL = "visual" | |
| class MultiModeEngine: | |
| def __init__(self): | |
| self.hf_client = HFClient() | |
| async def run(self, request: EngineRequest) -> EngineResponse: | |
| try: | |
| if request.action == "generate_content": | |
| return await self._generate_content(request) | |
| else: | |
| return self._error_response(request, "INVALID_ACTION", f"Action {request.action} not supported") | |
| except Exception as e: | |
| # Catch-all to ensure no crashes | |
| return self._error_response(request, "ENGINE_CRASH", str(e)) | |
| async def _generate_content(self, request: EngineRequest) -> EngineResponse: | |
| # Extract context - Engine expects orchestrated context | |
| concept_data = request.context.get("concept_data") | |
| if not concept_data: | |
| return self._error_response(request, "MISSING_CONTEXT", "concept_data is required in context") | |
| learner_ctx = request.context.get("learner_context", {}) | |
| # 1. Blueprint Generation (using HF Logic) | |
| concept_id = concept_data.get("id", "UNKNOWN") | |
| blueprint_prompt = f""" | |
| create an instructional blueprint for the concept: {concept_data.get('definition', concept_id)}. | |
| Include 3 key steps and 1 misconception. | |
| Format as clear text. | |
| """ | |
| blueprint_text = await self.hf_client.generate_text(blueprint_prompt) | |
| # 2. Mode Selection (Logic from previous implementation, simplified) | |
| modes = self._select_modes(learner_ctx) | |
| # 3. Render Artifacts | |
| artifacts = {} | |
| if MODE_TEXT in modes: | |
| artifacts[MODE_TEXT] = {"content": blueprint_text} | |
| if MODE_AUDIO in modes: | |
| # Mock Audio generation or call HF ASR/TTS if available | |
| artifacts[MODE_AUDIO] = {"url": f"asset://audio/{concept_id}.mp3"} | |
| if MODE_VIDEO in modes: | |
| artifacts[MODE_VIDEO] = {"url": f"asset://video/{concept_id}.mp4"} | |
| if MODE_VISUAL in modes: | |
| artifacts[MODE_VISUAL] = {"url": f"asset://visual/{concept_id}.png"} | |
| return EngineResponse( | |
| request_id=request.request_id, | |
| ok=True, | |
| status="success", | |
| engine=request.engine, | |
| action=request.action, | |
| result={ | |
| "concept_id": concept_id, | |
| "blueprint_summary": blueprint_text[:100] + "...", | |
| "artifacts": artifacts | |
| } | |
| ) | |
| def _select_modes(self, context: Dict[str, Any]) -> List[str]: | |
| # Context extraction | |
| device = context.get("device_type", "desktop") | |
| connectivity = context.get("connectivity", "high") | |
| modes = [MODE_TEXT] | |
| if connectivity == "high": | |
| modes.extend([MODE_AUDIO, MODE_VIDEO, MODE_VISUAL]) | |
| elif connectivity == "medium": | |
| modes.extend([MODE_AUDIO, MODE_VISUAL]) | |
| # else low -> text only | |
| if device == "mobile" and MODE_AUDIO not in modes: | |
| if connectivity != "low": modes.append(MODE_AUDIO) | |
| return list(set(modes)) | |
| def _error_response(self, request: EngineRequest, code: str, detail: str) -> EngineResponse: | |
| return EngineResponse( | |
| request_id=request.request_id, | |
| ok=False, | |
| status="error", | |
| engine=request.engine, | |
| action=request.action, | |
| error=ErrorDetail(code=code, detail=detail) | |
| ) | |