Socratic Questioning v0

self_hosted_env/voice_agent/services/agent_service.py

@@ -9,6 +9,7 @@ from core import StudentNationality, Models
 from services.pgvector_service import PGVectorService
 from services.openai_service import OpenAIService
 from services.chat_database_service import ChatDatabaseService
+from services.pedagogy_service import PedagogyService
  
 
 logger = logging.getLogger(__name__)
@@ -77,6 +78,10 @@ class AgentService:
         self.db_service = ChatDatabaseService()
         self.pgvector = PGVectorService() if use_pgvector else None
         
+        # Initialize pedagogy service for Socratic questioning
+        self.pedagogy_service = PedagogyService()
+        logger.info("AgentService initialized with PedagogyService for Socratic questioning")
+
     def is_available(self) -> bool:
         return self.client is not None
 
@@ -161,6 +166,17 @@ class AgentService:
             formatted_base_prompt = base_system_prompt.format(student_name=student_name)
             subject_specific_prompt = f"{formatted_base_prompt}\n\nإنت بتدرّس مادة {subject} للطفل {student_name}."
             
+            # Add Socratic questioning instructions if applicable (grades 4-6)
+            socratic_instructions = self.pedagogy_service.get_socratic_instructions(
+                student_info['grade'], 
+                student_info['nationality']
+            )
+            if socratic_instructions:
+                subject_specific_prompt += f"\n\n{socratic_instructions}"
+                logger.info(f"Added Socratic questioning instructions for grade {student_info['grade']}, nationality: {student_info['nationality']}")
+            else:
+                logger.debug(f"No Socratic instructions for grade {student_info['grade']}")
+            
             logger.info(f"Using nationality: {nationality}, subject: {subject}, and student name: {student_name} for student: {student_id}")
 
             # Prepare messages
@@ -287,6 +303,15 @@ class AgentService:
             formatted_base_prompt = base_system_prompt.format(student_name=student_name)
             subject_specific_prompt = f"{formatted_base_prompt}\n\nإنت بتدرّس مادة {subject} للطفل {student_name}."
             
+            # Add Socratic questioning instructions if applicable
+            socratic_instructions = self.pedagogy_service.get_socratic_instructions(
+                student_info['grade'], 
+                student_info['nationality']
+            )
+            if socratic_instructions:
+                subject_specific_prompt += f"\n\n{socratic_instructions}"
+                logger.info(f"Added Socratic instructions when updating subject context for grade {student_info['grade']}")
+            
             self.add_message_to_history(student_id, subject_specific_prompt, "system")
             
             logger.info(f"Updated subject context to {subject} for student {student_id} ({student_name})")

self_hosted_env/voice_agent/services/pedagogy_service.py

+import logging
+from typing import Dict, List
+from enum import Enum
+
+logger = logging.getLogger(__name__)
+
+class QuestionType(Enum):
+    CLARIFICATION = "clarification"
+    APPLICATION = "application"
+    REFLECTION = "reflection"
+
+class PedagogyService:
+    """Service class for managing pedagogical strategies for elementary students (grades 4-6)"""
+    
+    def __init__(self):
+        # Egyptian Arabic Socratic Question Libraries
+        self.egyptian_questions = {
+            QuestionType.CLARIFICATION: [
+                "تقصد إيه بالكلمة دي في المثال؟",
+                "ممكن تشرح بكلمة أبسط؟",
+                "إيه معنى المصطلح دا هنا؟",
+                "تحب أشرحها بجملة تانية؟",
+                "فهمت كدا ولا تحب أوضح أكتر؟"
+            ],
+            QuestionType.APPLICATION: [
+                "تفتكر ينفع نفس الفكرة في البيت أو المدرسة؟",
+                "ممكن تقوللي موقف حصل معاك شبه دا؟",
+                "إزاي ممكن نستعمل دا في حياتنا اليومية؟",
+                "تفتكر لو كنا في الصحرا كنا نعمل إيه عشان نتأقلم؟"
+            ],
+            QuestionType.REFLECTION: [
+                "ليه الكائن دا محتاج الجزء دا من جسمه عشان يعيش؟",
+                "إزاي نتاكد إن المعلومة دي صح؟",
+                "إيه السبب اللي يخلي دا يحصل؟",
+                "تفتكر فيه حيوان تاني متأقلم أحسن للصحرا؟",
+                "إيه اللي ممكن يحصل لو الظروف اتغيّرت؟"
+            ]
+        }
+
+        # Saudi Arabic Socratic Question Libraries
+        self.saudi_questions = {
+            QuestionType.CLARIFICATION: [
+                "وش تقصد بالكلمة هذي في الشرح؟",
+                "تقدر تقولها بكلمة أبسط؟",
+                "وش معنى المصطلح هنا؟",
+                "تحب أقولها بجملة ثانية؟",
+                "فهمت كذا أو تحتاج أوضح أكثر؟"
+            ],
+            QuestionType.APPLICATION: [
+                "تظن ممكن نستخدم الفكرة هذي في البيت أو المدرسة؟",
+                "صار معك موقف يشبه كذا؟",
+                "وين ممكن نشوف الشي هذا في حياتنا اليومية؟",
+                "لو كنا في الصحراء وش ممكن نسوي عشان نتأقلم؟"
+            ],
+            QuestionType.REFLECTION: [
+                "ليش الحيوان يعتمد على هالجزء من جسمه عشان يعيش؟",
+                "كيف نتاكد إن الجواب صحيح؟",
+                "وش السبب اللي يخلي هذا يصير؟",
+                "تظن فيه حيوان متأقلم أكثر للصحرا؟",
+                "وش اللي ممكن يصير لو تغيرت الظروف؟"
+            ]
+        }
+
+        logger.info("PedagogyService initialized with Socratic questioning libraries")
+    
+    def _get_question_library(self, nationality: str) -> Dict[QuestionType, List[str]]:
+        """Get the appropriate question library based on nationality"""
+        nationality_lower = nationality.lower().strip()
+        
+        if nationality_lower == 'saudi':
+            return self.saudi_questions
+        elif nationality_lower == 'egyptian':
+            return self.egyptian_questions
+        else:
+            # Default to Egyptian if nationality is not recognized
+            logger.warning(f"Unknown nationality '{nationality}', defaulting to Egyptian question library")
+            return self.egyptian_questions
+    
+    def _format_questions_for_prompt(self, questions: Dict[QuestionType, List[str]]) -> str:
+        """Format question libraries for inclusion in system prompt"""
+        formatted_questions = []
+        
+        clarification_list = ", ".join([f'"{q}"' for q in questions[QuestionType.CLARIFICATION]])
+        formatted_questions.append(f"- Clarification: [{clarification_list}]")
+        
+        application_list = ", ".join([f'"{q}"' for q in questions[QuestionType.APPLICATION]])
+        formatted_questions.append(f"- Application: [{application_list}]")
+        
+        reflection_list = ", ".join([f'"{q}"' for q in questions[QuestionType.REFLECTION]])
+        formatted_questions.append(f"- Reflection: [{reflection_list}]")
+        
+        return "\n".join(formatted_questions)
+    
+    def get_socratic_instructions(self, grade: int, nationality: str) -> str:
+        """Get Socratic questioning instructions for the system prompt"""
+        if grade not in [4, 5, 6]:
+            logger.debug(f"Grade {grade} not in target range (4-6), skipping Socratic instructions")
+            return ""
+        
+        questions = self._get_question_library(nationality)
+        formatted_questions = self._format_questions_for_prompt(questions)
+        
+        if nationality.lower().strip() == 'saudi':
+            instruction_text = f"""بعد ما تشرح الفكرة للطفل، قيّم إذا يحتاج يتعمّق أكثر:
+- لو الشرح كافي ومفهوم → لا تضيف أي سؤال.
+- لو تشوف الطفل يستفيد من تفكير أو مثال إضافي → اختر سؤال واحد فقط من القائمة تحت.
+ممكن تحط السؤال في النص أو في الآخر حسب ما يكون طبيعي.
+
+الأسئلة الممكنة:
+{formatted_questions}"""
+        else:  # Egyptian or default
+            instruction_text = f"""بعد ما تشرح الفكرة للطفل، فكّر هل محتاج يتعمّق أكتر:
+- لو الشرح واضح وكافي → خلّص من غير أي سؤال.
+- لو هيفيده تفكير أو مثال زيادة → اختار سؤال واحد بس من اللي تحت.
+ممكن تحط السؤال في النص أو في الآخر، حسب ما يبان طبيعي.
+
+الأسئلة الممكنة:
+{formatted_questions}"""
+        
+        logger.info(f"Generated Socratic instructions for grade {grade}, nationality: {nationality}")
+        return instruction_text
+    
+    def get_available_question_types(self) -> List[str]:
+        return [qtype.value for qtype in QuestionType]
+    
+    def get_questions_by_type(self, nationality: str, question_type: QuestionType) -> List[str]:
+        questions = self._get_question_library(nationality)
+        return questions.get(question_type, [])