Spaces:
Sleeping
Sleeping
| import json | |
| from app.contracts import EngineRequest, EngineResponse, EngineError | |
| from app.hf_client import HFClient | |
| class MicroHintsEngine: | |
| def __init__(self): | |
| self.client = HFClient() | |
| async def process(self, request: EngineRequest) -> EngineResponse: | |
| try: | |
| action = request.action.lower() | |
| if action == "detect_struggle": | |
| return await self._detect_struggle(request) | |
| elif action == "generate_hint_package": | |
| # This acts as the composite action "process_trigger" -> full package | |
| return await self._generate_hint_package(request) | |
| elif action == "score_checks": | |
| return await self._score_checks(request) | |
| else: | |
| raise ValueError(f"Unknown action: {action}") | |
| except Exception as e: | |
| return EngineResponse( | |
| request_id=request.request_id, | |
| ok=False, | |
| status="error", | |
| engine="micro-hints-engine", | |
| action=request.action, | |
| error=EngineError(code="ENGINE_EXECUTION_ERROR", detail=str(e)) | |
| ) | |
| async def _detect_struggle(self, request: EngineRequest) -> EngineResponse: | |
| # In a real system, this would analyze telemetry. | |
| # Here we mock intelligence to decide if a trigger is needed based on input signals. | |
| signals = request.input.refs.get("signals", {}) | |
| prompt = ( | |
| f"Signals: {json.dumps(signals)}\n" | |
| "Analyze if this learner is struggling. " | |
| "Output JSON: 'is_struggling' (bool), 'trigger_confidence' (0.0-1.0), 'error_pattern' (string)." | |
| ) | |
| messages = [{"role": "system", "content": prompt}] | |
| response_text = await self.client.generate(messages) | |
| try: | |
| analysis = json.loads(response_text.replace("```json", "").replace("```", "").strip()) | |
| except: | |
| analysis = {"is_struggling": False, "raw_output": response_text} | |
| return EngineResponse( | |
| request_id=request.request_id, | |
| ok=True, | |
| status="success", | |
| engine="micro-hints-engine", | |
| action="detect_struggle", | |
| result=analysis | |
| ) | |
| async def _generate_hint_package(self, request: EngineRequest) -> EngineResponse: | |
| trigger_data = request.input.refs.get("trigger", {}) | |
| concept_id = trigger_data.get("concept_id", "unknown_concept") | |
| error_pattern = trigger_data.get("error_pattern", "general_confusion") | |
| # 1. Generate Hint + Analogy + Checks in one go (or sequential calls) | |
| prompt = ( | |
| f"Concept: {concept_id}\nError Pattern: {error_pattern}\n" | |
| "Task: Generate a 'MicroHintPackage' to unblock the learner.\n" | |
| "Requirements:\n" | |
| "1. Hint: A 30-second directional hint (not the answer).\n" | |
| "2. Analogy: A 30-second personalized analogy.\n" | |
| "3. Checks: 1-2 comprehension questions (short/numeric).\n" | |
| "Output JSON with keys: 'hint', 'analogy', 'checks' (list of {q, type})." | |
| ) | |
| messages = [{"role": "system", "content": prompt}] | |
| response_text = await self.client.generate(messages) | |
| try: | |
| package_content = json.loads(response_text.replace("```json", "").replace("```", "").strip()) | |
| except: | |
| package_content = {"hint": "Review the concept.", "analogy": "None", "checks": []} | |
| # Add micro-demo stub | |
| package_content["micro_demo"] = { | |
| "type": "interactive_example", | |
| "ref": f"asset://demo/{concept_id.lower().replace(' ', '_')}_01" | |
| } | |
| package_content["concept_id"] = concept_id | |
| return EngineResponse( | |
| request_id=request.request_id, | |
| ok=True, | |
| status="success", | |
| engine="micro-hints-engine", | |
| action="generate_hint_package", | |
| result=package_content | |
| ) | |
| async def _score_checks(self, request: EngineRequest) -> EngineResponse: | |
| responses = request.input.refs.get("responses", []) | |
| prompt = ( | |
| f"Learner Responses: {json.dumps(responses)}\n" | |
| "Evaluate comprehension. Did they pass the check? " | |
| "Output JSON: 'checks_passed' (bool), 'hint_effectiveness' (high/medium/low)." | |
| ) | |
| messages = [{"role": "system", "content": prompt}] | |
| response_text = await self.client.generate(messages) | |
| try: | |
| score_data = json.loads(response_text.replace("```json", "").replace("```", "").strip()) | |
| except: | |
| score_data = {"checks_passed": False, "note": "Failed to parse score."} | |
| return EngineResponse( | |
| request_id=request.request_id, | |
| ok=True, | |
| status="success", | |
| engine="micro-hints-engine", | |
| action="score_checks", | |
| result=score_data | |
| ) | |
| engine = MicroHintsEngine() | |