Skip to content

A
AI Tutor
Commit 12111380 authored by SalmaMohammedHamedMustafa's avatar SalmaMohammedHamedMustafa
Browse files
Options

better db connectin hadeling for long lasting deplyment

parent a4501a9c
Hide whitespace changes
Inline Side-by-side
Showing with 685 additions and 341 deletions
self_hosted_env/voice_agent/captain-definition 0 → 100644
View file @ 12111380
{
"schemaVersion": 2,
"dockerfilePath": "./Dockerfile",
"containerHttpPort": "8000",
"env": {
"POSTGRES_HOST": "srv-captain--postgres",
"MINIO_HOST": "srv-captain--minio"
}
}
self_hosted_env/voice_agent/core/enums.py
View file @ 12111380
......@@ -17,7 +17,7 @@ class StudentNationality(str, Enum):
class Models(str, Enum):
chat = "gpt-5"
chat = "gpt-4o-mini"
tts = "gpt-4o-mini-tts"
embedding = "text-embedding-3-small"
transcription = "gpt-4o-transcribe"
self_hosted_env/voice_agent/main.py
View file @ 12111380
import os
from fastapi import FastAPI, UploadFile, File, Form, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, Response
from fastapi.staticfiles import StaticFiles
from typing import Optional
import uvicorn
import base64
from pathlib import Path
# Import your existing modules
from core import AppConfig, StudentNationality
from repositories import StorageRepository, MinIOStorageRepository
from handlers import AudioMessageHandler, TextMessageHandler
......@@ -36,19 +42,28 @@ class DIContainer:
def create_app() -> FastAPI:
app = FastAPI(title="Unified Chat API with Local Agent")
# Fixed CORS configuration for CapRover
app.add_middleware(
CORSMiddleware,
allow_origins=[
"http://teamtestingdocker.caprover.al-arcade.com:8000",
"http://localhost:8000", # For local development
],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
expose_headers=["X-Response-Text"],
)
CORSMiddleware,
allow_origins=[
"https://voice-agent.caprover.al-arcade.com",
"http://voice-agent.caprover.al-arcade.com",
"http://localhost:8000", # For local development
"http://127.0.0.1:8000",
"*" # Allow all origins for testing - remove in production
],
allow_credentials=True,
allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"],
allow_headers=[
"Accept",
"Accept-Language",
"Content-Language",
"Content-Type",
"Authorization",
"X-Response-Text"
],
expose_headers=["X-Response-Text"],
)
# Initialize dependencies
container = DIContainer()
......@@ -59,94 +74,238 @@ def create_app() -> FastAPI:
print("OpenAI Service Available:", container.openai_service.is_available())
print("Agent Service Available:", container.agent_service.is_available())
from fastapi.responses import FileResponse
# Serve static files if the directory exists
static_path = Path("static")
if static_path.exists():
app.mount("/static", StaticFiles(directory=static_path), name="static")
@app.get("/chat-interface")
async def serve_audio_recorder():
return FileResponse("static/audio-recorder.html")
"""Serve the audio recorder HTML file"""
try:
# Try to serve from static directory first
static_file = Path("static/audio-recorder.html")
if static_file.exists():
return FileResponse(static_file)
# Fallback to current directory
current_file = Path("audio-recorder.html")
if current_file.exists():
return FileResponse(current_file)
# If no file found, return an error
raise HTTPException(status_code=404, detail="Audio recorder interface not found")
except Exception as e:
print(f"Error serving audio recorder: {e}")
raise HTTPException(status_code=500, detail=f"Error serving interface: {str(e)}")
@app.post("/chat")
async def chat_handler(
file: Optional[UploadFile] = File(None),
text: Optional[str] = Form(None),
student_id: str = Form("student_001") # Default student_id, but can be overridden
student_id: str = Form("student_001")
):
"""
Handles incoming chat messages (either text or audio).
Generates responses locally using the agent service.
"""
if not student_id.strip():
raise HTTPException(status_code=400, detail="Student ID is required")
return container.chat_service.process_message(student_id=student_id, file=file, text=text)
try:
if not student_id.strip():
raise HTTPException(status_code=400, detail="Student ID is required")
print(f"Processing message for student: {student_id}")
print(f"Text: {text}")
print(f"File: {file.filename if file else 'None'}")
result = container.chat_service.process_message(
student_id=student_id,
file=file,
text=text
)
print(f"Chat service result: {result}")
return result
except Exception as e:
print(f"Error in chat handler: {str(e)}")
raise HTTPException(status_code=500, detail=f"Chat processing error: {str(e)}")
@app.get("/get-audio-response")
async def get_audio_response():
"""Fetches the agent's text and audio response."""
return container.response_service.get_agent_response()
"""Fetches the agent's text and audio response with proper CORS headers."""
try:
print("Getting audio response...")
result = container.response_service.get_agent_response()
if hasattr(result, 'status_code'):
# This is already a Response object from response_service
print(f"Response headers: {dict(result.headers)}")
return result
else:
# If it's not a Response object, create one
print("Creating new response object")
response_text = result.get('text', 'No response available')
# Encode the response text in base64 for the header
encoded_text = base64.b64encode(response_text.encode('utf-8')).decode('utf-8')
# Create response with proper headers
response = Response(
content=result.get('audio', b''),
media_type="audio/mpeg",
headers={
"X-Response-Text": encoded_text,
"Access-Control-Expose-Headers": "X-Response-Text",
"Access-Control-Allow-Origin": "*",
"Access-Control-Allow-Methods": "GET, POST, OPTIONS",
"Access-Control-Allow-Headers": "*"
}
)
print(f"Created response with headers: {dict(response.headers)}")
return response
except Exception as e:
print(f"Error getting audio response: {str(e)}")
raise HTTPException(status_code=500, detail=f"Audio response error: {str(e)}")
@app.options("/chat")
async def chat_options():
"""Handle preflight CORS requests for chat endpoint"""
return Response(
headers={
"Access-Control-Allow-Origin": "*",
"Access-Control-Allow-Methods": "GET, POST, OPTIONS",
"Access-Control-Allow-Headers": "*",
"Access-Control-Max-Age": "86400"
}
)
@app.options("/get-audio-response")
async def audio_response_options():
"""Handle preflight CORS requests for audio response endpoint"""
return Response(
headers={
"Access-Control-Allow-Origin": "*",
"Access-Control-Allow-Methods": "GET, POST, OPTIONS",
"Access-Control-Allow-Headers": "*",
"Access-Control-Expose-Headers": "X-Response-Text",
"Access-Control-Max-Age": "86400"
}
)
@app.get("/health")
async def health_check():
"""Health check endpoint with agent service status"""
health_status = container.health_service.get_health_status()
# Add agent service status
health_status.update({
"openai_service_status": "available" if container.openai_service.is_available() else "unavailable",
"agent_service_status": "available" if container.agent_service.is_available() else "unavailable"
})
return health_status
try:
health_status = container.health_service.get_health_status()
# Add agent service status
health_status.update({
"openai_service_status": "available" if container.openai_service.is_available() else "unavailable",
"agent_service_status": "available" if container.agent_service.is_available() else "unavailable",
"minio_endpoint": container.config.minio_endpoint,
"minio_bucket": container.config.minio_bucket
})
return health_status
except Exception as e:
print(f"Health check error: {e}")
return {"status": "error", "message": str(e)}
# Agent management endpoints
@app.get("/conversation/stats")
async def get_conversation_stats(student_id: str = "student_001"):
"""Get conversation statistics"""
return container.chat_service.get_agent_stats(student_id)
try:
return container.chat_service.get_agent_stats(student_id)
except Exception as e:
print(f"Stats error: {e}")
raise HTTPException(status_code=500, detail=str(e))
@app.post("/conversation/clear")
async def clear_conversation(student_id: str = Form("student_001")):
"""Clear conversation history"""
return container.chat_service.clear_conversation(student_id)
try:
return container.chat_service.clear_conversation(student_id)
except Exception as e:
print(f"Clear conversation error: {e}")
raise HTTPException(status_code=500, detail=str(e))
@app.post("/agent/system-prompt")
async def set_system_prompt(request: dict):
"""Update the agent's system prompt"""
prompt = request.get("prompt", "")
if not prompt:
raise HTTPException(status_code=400, detail="System prompt cannot be empty")
return container.chat_service.set_system_prompt(prompt)
try:
prompt = request.get("prompt", "")
if not prompt:
raise HTTPException(status_code=400, detail="System prompt cannot be empty")
return container.chat_service.set_system_prompt(prompt)
except Exception as e:
print(f"Set system prompt error: {e}")
raise HTTPException(status_code=500, detail=str(e))
@app.get("/agent/system-prompt")
async def get_system_prompt():
"""Get the current system prompt"""
return {
"system_prompt": container.agent_service.system_prompt,
"status": "success"
}
try:
return {
"system_prompt": container.agent_service.system_prompt,
"status": "success"
}
except Exception as e:
print(f"Get system prompt error: {e}")
raise HTTPException(status_code=500, detail=str(e))
@app.get("/conversation/export")
async def export_conversation(student_id: str = "student_001"):
"""Export conversation history"""
history = container.agent_service.export_conversation(student_id)
return {
"student_id": student_id,
"messages": history,
"total_messages": len(history)
}
try:
history = container.agent_service.export_conversation(student_id)
return {
"student_id": student_id,
"messages": history,
"total_messages": len(history)
}
except Exception as e:
print(f"Export conversation error: {e}")
raise HTTPException(status_code=500, detail=str(e))
@app.post("/conversation/import")
async def import_conversation(request: dict):
"""Import conversation history"""
student_id = request.get("student_id", "student_001")
messages = request.get("messages", [])
if not messages:
raise HTTPException(status_code=400, detail="Messages list cannot be empty")
container.agent_service.import_conversation(messages, student_id)
return {
"status": "success",
"message": f"Imported {len(messages)} messages to conversation {student_id}"
}
try:
student_id = request.get("student_id", "student_001")
messages = request.get("messages", [])
if not messages:
raise HTTPException(status_code=400, detail="Messages list cannot be empty")
container.agent_service.import_conversation(messages, student_id)
return {
"status": "success",
"message": f"Imported {len(messages)} messages to conversation {student_id}"
}
except Exception as e:
print(f"Import conversation error: {e}")
raise HTTPException(status_code=500, detail=str(e))
@app.get("/debug/test-response")
async def debug_test_response():
"""Debug endpoint to test response generation"""
try:
# Test basic response
test_text = "This is a test response"
encoded_text = base64.b64encode(test_text.encode('utf-8')).decode('utf-8')
return Response(
content=b"test audio data",
media_type="audio/mpeg",
headers={
"X-Response-Text": encoded_text,
"Access-Control-Expose-Headers": "X-Response-Text"
}
)
except Exception as e:
print(f"Debug test error: {e}")
raise HTTPException(status_code=500, detail=str(e))
@app.get("/")
async def root():
......@@ -155,21 +314,22 @@ def create_app() -> FastAPI:
"service": "Unified Chat API with Local Agent",
"version": "2.1.0",
"description": "Unified backend for audio/text chat with a local AI agent.",
"status": "running",
"deployment": "CapRover",
"features": [
"Local AI agent responses using OpenAI GPT",
"Audio transcription using OpenAI Whisper",
"Audio transcription using OpenAI Whisper",
"Text-to-speech using OpenAI TTS",
"Conversation history management",
"Student-specific conversations"
"Student-specific conversations",
"CORS enabled for cross-origin requests"
],
"endpoints": {
"chat": "/chat (accepts audio or text with student_id, generates local agent response)",
"get_audio_response": "/get-audio-response (fetches agent's audio and text response)",
"conversation_stats": "/conversation/stats (get conversation statistics)",
"clear_conversation": "/conversation/clear (clear conversation history)",
"set_system_prompt": "/agent/system-prompt (update agent system prompt)",
"export_conversation": "/conversation/export (export conversation history)",
"health": "/health (service health check)"
"chat_interface": "/chat-interface (HTML interface)",
"chat": "/chat (accepts audio or text with student_id)",
"get_audio_response": "/get-audio-response (fetches agent's audio and text)",
"health": "/health (service health check)",
"debug": "/debug/test-response (test response generation)"
}
}
......@@ -177,3 +337,12 @@ def create_app() -> FastAPI:
# Application entry point
app = create_app()
if __name__ == "__main__":
# For development
uvicorn.run(
"main:app",
host="0.0.0.0",
port=int(os.environ.get("PORT", 8000)),
reload=True
)
\ No newline at end of file
self_hosted_env/voice_agent/services/agent_service.py
View file @ 12111380
......@@ -15,29 +15,29 @@ from services.pedagogy_service import PedagogyService
logger = logging.getLogger(__name__)
SYSTEM_PROMPTS: Dict[StudentNationality, str] = {
StudentNationality.EGYPTIAN: """
إنت مُدرّس لطفل في ابتدائي اسمه {student_name}.
رد باللهجة المصريّة الطبيعيّة كأنك بتكلم {student_name} قصادك.
خلي الكلام بسيط، واضح، وقريب من ودنه.
الجمل قصيرة ومترابطة، مش مقطّعة.
اشرح كأنك بتحكي له حكاية أو بتورّيه حاجة من الحياة حوالينا، مش بتقرأ من كتاب.
ممكن تذكر اسم {student_name} مرّة واحدة في أول الرد فقط.
بعد كده ممنوع تكرار الاسم في نفس الرد، حتى في الأسئلة الختامية.
ممنوع تستعمل أي ألقاب زي "يا بطل" أو "يا شاطر"، الاسم الأول بس.
ولو الرد قصير جدًا (جملة أو اتنين)، ممكن تستغنى عن ذكر الاسم خالص.
لو فيه مصطلح صعب، فسّره بكلمة أسهل.
لو فيه رمز كيميائي زي H2O أو CO2، اكتبه زي ما هو.
الأرقام العادية اكتبها بالحروف العربي زي "اتنين" أو "تلاتة".
استخدم التشكيل بس على الكلمات اللي ممكن الـTTS ينطقها غلط أو يحصل فيها لبس، مش على كل كلمة.
لو {student_name} مكتوب بالإنجليزي، اكتبه دايمًا بالعربي في ردودك.
الهدف: رد قصير يعلّم ويوصل المعلومة، ويبان إن في مُعلّم بيشرح للطفل مش كتاب بيتقري.
إنت مُدرّس لطفل في ابتدائي اسمه {student_name} في الصف {grade}.
رَدّ باللهجة المصريّة الطبيعيّة كأنّك بتكَلِّم {student_name} قصادك.
خَلّي الكلام بسيط، واضح، وقَريب من وُدنه.
الجُمَل قُصَيَّرة ومُترابطة، مِش مَقطَّعة.
اشرح كأنّك بتحكي له حِكاية أو بتورّيه حاجَة من الحَياة حَوالينا، مِش بتِقرا من كتاب.
مُمكِن تِذكُر اسم {student_name} مَرّة واحِدة في أوِّل الرّد فَقَط.
بعد كِدا مَمنوع تِكرار الاسم في نَفس الرّد، حَتّى في الأسئِلة الخِتاميّة.
مَمنوع تِستَعمِل أي ألقاب زي "يا بَطَل" أو "يا شاطِر"، الاسم الأوَّل بَس.
ولو الرّد قُصَيَّر جِدًّا (جُملة أو اتنِين)، مُمكِن تِستَغنَى عن ذِكر الاسم خالِص.
لو فيه مُصطَلَح صَعب، فَسَّره بِكِلمة أسهَل.
لو فيه رَمز كيمياوي زي H2O أو CO2، اكتُبه زي ما هو.
الأرقام العادِيّة اكتُبها بالحُروف العربي زي "اِتنِين" أو "تَلاتة".
اِستَخدِم التَّشكِيل الكامِل على كُلّ الكَلام عَشان يِطْلَع بالصِّوت زي نُطق اللّهجة المِصريّة الطَّبيعي.
لو {student_name} مكتوب بالإنجليزي، اكتُبه دايمًا بالعَربي في رُدودك.
لَمّا تِذكُر الصف {grade}، قُله بالطريقة الطبيعيّة زي ما الأطفال بيقولوها: الصف 4 = سنة رابعة ابتدائي، الصف 5 = سنة خامسة ابتدائي، وهكذا.
الهَدف: رَد قُصَيَّر يِعلِّم ويُوصَّل المَعلومة، ويِبان إن فيه مُعلِّم بيِشرَح للطفل مِش كتاب بيتقري.
""",
StudentNationality.SAUDI: """
إنت معلّم لطفل في ابتدائي اسمه {student_name}.
رد باللهجة السعوديّة الطبيعيّة، كأنك تشرح له قدّامك.
إنت معلّم لطفل في ابتدائي اسمه {student_name} في الصف {grade}.
رَد باللهجة السعوديّة الطبيعيّة، كأنك تشرح له قدّامك.
خل الشرح واضح وسهل، لكن لا يكون ناشف.
اشرح كأنك تسولف معه وتشبّه بأشياء من حياته اليومية.
......@@ -52,17 +52,15 @@ SYSTEM_PROMPTS: Dict[StudentNationality, str] = {
استخدم التشكيل بس على الكلمات اللي ممكن الـTTS يخبّص فيها أو يقرأها خطأ، واترك الباقي بدون تشكيل عشان يطلع طبيعي.
لو {student_name} مكتوب بالإنجليزي، اكتبه دايمًا بالعربي في ردودك.
الهدف: رد مبسّط، قريب، ويبيّن إن المعلّم يشرح للطفل، مو يقرأ من كتاب.
لما تذكر الصف {grade}، قُلها بالطريقة اللي الطفل متعود يسمعها: الصف 4 = رابع ابتدائي، الصف 5 = خامس ابتدائي، وهكذا.
الهدف: رد مبسّط، قريب، ويبيّن إن المعلّم يشرح للطفل، مو يقرأ من كتاب.
"""
}
class AgentService:
"""Service class for handling AI agent conversations using database memory"""
......@@ -163,7 +161,10 @@ class AgentService:
# Create subject-specific system prompt with first name only
base_system_prompt = SYSTEM_PROMPTS.get(nationality, SYSTEM_PROMPTS[StudentNationality.EGYPTIAN])
formatted_base_prompt = base_system_prompt.format(student_name=student_name)
formatted_base_prompt = base_system_prompt.format(
student_name=student_name,
grade=student_info['grade']
)
subject_specific_prompt = f"{formatted_base_prompt}\n\nإنت بتدرّس مادة {subject} للطفل {student_name}."
# Add Socratic questioning instructions if applicable (grades 4-6)
......@@ -258,7 +259,6 @@ class AgentService:
logger.error(f"Error generating AI response: {e}")
raise HTTPException(status_code=500, detail=f"AI response generation failed: {str(e)}")
def search_similar(self, query_embedding: List[float], student_id: str,
subject: str = "chemistry", top_k: int = 3):
"""Search similar content with student-specific filtering"""
......@@ -300,7 +300,11 @@ class AgentService:
nationality = nationality_mapping.get(nationality_lower, StudentNationality.EGYPTIAN)
base_system_prompt = SYSTEM_PROMPTS.get(nationality, SYSTEM_PROMPTS[StudentNationality.EGYPTIAN])
# Format the prompt with student name
formatted_base_prompt = base_system_prompt.format(student_name=student_name)
formatted_base_prompt = base_system_prompt.format(
student_name=student_name,
grade=student_info['grade']
)
subject_specific_prompt = f"{formatted_base_prompt}\n\nإنت بتدرّس مادة {subject} للطفل {student_name}."
# Add Socratic questioning instructions if applicable
......@@ -321,12 +325,81 @@ class AgentService:
logger.error(f"Error updating subject context: {e}")
return False
def export_conversation(self, student_id: str) -> List[Dict[str, str]]:
"""Export conversation history for a student"""
return self.get_conversation_history(student_id)
def import_conversation(self, messages: List[Dict[str, str]], student_id: str):
"""Import conversation history for a student"""
# Clear existing history first
self.db_service.clear_history(student_id)
# Import messages in order
for message in messages:
role = message.get("role", "user")
content = message.get("content", "")
if content:
self.add_message_to_history(student_id, content, role)
def clear_conversation(self, student_id: str) -> Dict[str, str]:
"""Clear conversation history for a student"""
try:
self.db_service.clear_history(student_id)
return {
"status": "success",
"message": f"Conversation cleared for student {student_id}"
}
except Exception as e:
logger.error(f"Error clearing conversation: {e}")
return {
"status": "error",
"message": f"Failed to clear conversation: {str(e)}"
}
def get_agent_stats(self, student_id: str) -> Dict:
"""Get conversation statistics for a student"""
try:
history = self.get_conversation_history(student_id)
user_messages = [msg for msg in history if msg['role'] == 'user']
assistant_messages = [msg for msg in history if msg['role'] == 'assistant']
system_messages = [msg for msg in history if msg['role'] == 'system']
return {
"student_id": student_id,
"total_messages": len(history),
"user_messages": len(user_messages),
"assistant_messages": len(assistant_messages),
"system_messages": len(system_messages),
"conversation_active": len(history) > 0
}
except Exception as e:
logger.error(f"Error getting agent stats: {e}")
return {
"student_id": student_id,
"error": str(e)
}
def set_system_prompt(self, prompt: str) -> Dict[str, str]:
"""Set a new system prompt (this is a placeholder - actual implementation would depend on requirements)"""
# This method would need to be implemented based on your specific requirements
# for how system prompts should be managed globally vs per student
return {
"status": "success",
"message": "System prompt updated"
}
@property
def system_prompt(self) -> str:
"""Get the current system prompt"""
# Return a default system prompt - this could be made more sophisticated
return "Default system prompt for educational AI assistant"
def close(self):
"""Close database connections"""
"""Close database connection pools"""
if self.db_service:
self.db_service.close()
self.db_service.close_pool()
if self.pgvector:
self.pgvector.close()
self.pgvector.close_pool()
# ----------------- Test -----------------
......
self_hosted_env/voice_agent/services/chat_database_service.py
View file @ 12111380
import os
import psycopg2
from psycopg2.extras import RealDictCursor
from psycopg2.pool import ThreadedConnectionPool
from typing import List, Dict, Optional, Tuple
import logging
......@@ -8,10 +9,12 @@ logger = logging.getLogger(__name__)
class ChatDatabaseService:
"""Simple service for managing chat history in PostgreSQL"""
"""Simple service for managing chat history in PostgreSQL with connection pooling"""
def __init__(self):
self.conn = psycopg2.connect(
self.pool = ThreadedConnectionPool(
minconn=1,
maxconn=20,
host=os.getenv("POSTGRES_HOST", "postgres"),
user=os.getenv("POSTGRES_USER"),
password=os.getenv("POSTGRES_PASSWORD"),
......@@ -20,104 +23,132 @@ class ChatDatabaseService:
def get_student_nationality(self, student_id: str) -> Optional[str]:
"""Get student nationality from database"""
with self.conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute(
"SELECT nationality FROM students WHERE student_id = %s",
(student_id,)
)
result = cur.fetchone()
return result["nationality"] if result else None
conn = self.pool.getconn()
try:
with conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute(
"SELECT nationality FROM students WHERE student_id = %s",
(student_id,)
)
result = cur.fetchone()
return result["nationality"] if result else None
finally:
self.pool.putconn(conn)
def get_student_info(self, student_id: str) -> Optional[Dict]:
"""Get complete student information from database"""
with self.conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute(
"""
SELECT student_id, student_name, grade, language, nationality
FROM students
WHERE student_id = %s
""",
(student_id,)
)
result = cur.fetchone()
if result:
return {
'student_id': result['student_id'],
'student_name': result['student_name'],
'grade': result['grade'], # This is now an integer
'is_arabic': result['language'], # Convert language boolean to is_arabic
'nationality': result['nationality']
}
return None
conn = self.pool.getconn()
try:
with conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute(
"""
SELECT student_id, student_name, grade, language, nationality
FROM students
WHERE student_id = %s
""",
(student_id,)
)
result = cur.fetchone()
if result:
return {
'student_id': result['student_id'],
'student_name': result['student_name'],
'grade': result['grade'], # This is now an integer
'is_arabic': result['language'], # Convert language boolean to is_arabic
'nationality': result['nationality']
}
return None
finally:
self.pool.putconn(conn)
def get_student_grade_and_language(self, student_id: str) -> Optional[Tuple[int, bool]]:
"""Get student grade and language preference"""
with self.conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute(
"SELECT grade, language FROM students WHERE student_id = %s",
(student_id,)
)
result = cur.fetchone()
if result:
return (result["grade"], result["language"])
return None
conn = self.pool.getconn()
try:
with conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute(
"SELECT grade, language FROM students WHERE student_id = %s",
(student_id,)
)
result = cur.fetchone()
if result:
return (result["grade"], result["language"])
return None
finally:
self.pool.putconn(conn)
def get_chat_history(self, student_id: str, limit: int = 20) -> List[Dict[str, str]]:
"""Get chat history for a student, returns in chronological order"""
with self.conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute(
"""
SELECT role, content
FROM chat_history
WHERE student_id = %s
ORDER BY created_at DESC
LIMIT %s;
""",
(student_id, limit)
)
results = cur.fetchall()
# Return in chronological order (oldest first)
return [{"role": row["role"], "content": row["content"]} for row in reversed(results)]
conn = self.pool.getconn()
try:
with conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute(
"""
SELECT role, content
FROM chat_history
WHERE student_id = %s
ORDER BY created_at DESC
LIMIT %s;
""",
(student_id, limit)
)
results = cur.fetchall()
# Return in chronological order (oldest first)
return [{"role": row["role"], "content": row["content"]} for row in reversed(results)]
finally:
self.pool.putconn(conn)
def add_message(self, student_id: str, role: str, content: str):
"""Add a message to chat history"""
with self.conn.cursor() as cur:
cur.execute(
"""
INSERT INTO chat_history (student_id, role, content)
VALUES (%s, %s, %s);
""",
(student_id, role, content)
)
self.conn.commit()
conn = self.pool.getconn()
try:
with conn.cursor() as cur:
cur.execute(
"""
INSERT INTO chat_history (student_id, role, content)
VALUES (%s, %s, %s);
""",
(student_id, role, content)
)
conn.commit()
finally:
self.pool.putconn(conn)
def clear_history(self, student_id: str):
"""Clear chat history for a student"""
with self.conn.cursor() as cur:
cur.execute(
"DELETE FROM chat_history WHERE student_id = %s",
(student_id,)
)
self.conn.commit()
conn = self.pool.getconn()
try:
with conn.cursor() as cur:
cur.execute(
"DELETE FROM chat_history WHERE student_id = %s",
(student_id,)
)
conn.commit()
finally:
self.pool.putconn(conn)
def limit_history(self, student_id: str, max_messages: int = 40):
"""Keep only recent messages for a student"""
with self.conn.cursor() as cur:
cur.execute(
"""
DELETE FROM chat_history
WHERE student_id = %s
AND role != 'system'
AND id NOT IN (
SELECT id FROM chat_history
WHERE student_id = %s AND role != 'system'
ORDER BY created_at DESC
LIMIT %s
);
""",
(student_id, student_id, max_messages)
)
self.conn.commit()
conn = self.pool.getconn()
try:
with conn.cursor() as cur:
cur.execute(
"""
DELETE FROM chat_history
WHERE student_id = %s
AND role != 'system'
AND id NOT IN (
SELECT id FROM chat_history
WHERE student_id = %s AND role != 'system'
ORDER BY created_at DESC
LIMIT %s
);
""",
(student_id, student_id, max_messages)
)
conn.commit()
finally:
self.pool.putconn(conn)
def update_student_info(self, student_id: str, grade: Optional[int] = None,
language: Optional[bool] = None, nationality: Optional[str] = None):
......@@ -139,31 +170,39 @@ class ChatDatabaseService:
if updates:
params.append(student_id)
with self.conn.cursor() as cur:
cur.execute(
f"""
UPDATE students
SET {', '.join(updates)}
WHERE student_id = %s
""",
params
)
self.conn.commit()
conn = self.pool.getconn()
try:
with conn.cursor() as cur:
cur.execute(
f"""
UPDATE students
SET {', '.join(updates)}
WHERE student_id = %s
""",
params
)
conn.commit()
finally:
self.pool.putconn(conn)
def create_student(self, student_id: str, student_name: str, grade: int,
language: bool, nationality: str = 'EGYPTIAN'):
"""Create a new student record"""
with self.conn.cursor() as cur:
cur.execute(
"""
INSERT INTO students (student_id, student_name, grade, language, nationality)
VALUES (%s, %s, %s, %s, %s)
ON CONFLICT (student_id) DO NOTHING;
""",
(student_id, student_name, grade, language, nationality)
)
self.conn.commit()
conn = self.pool.getconn()
try:
with conn.cursor() as cur:
cur.execute(
"""
INSERT INTO students (student_id, student_name, grade, language, nationality)
VALUES (%s, %s, %s, %s, %s)
ON CONFLICT (student_id) DO NOTHING;
""",
(student_id, student_name, grade, language, nationality)
)
conn.commit()
finally:
self.pool.putconn(conn)
def close(self):
if self.conn:
self.conn.close()
\ No newline at end of file
def close_pool(self):
if self.pool:
self.pool.closeall()
\ No newline at end of file
self_hosted_env/voice_agent/services/pgvector_service.py
View file @ 12111380
import os
import psycopg2
from psycopg2.extras import RealDictCursor
from psycopg2.pool import ThreadedConnectionPool
from typing import List, Optional
# Import the pgvector adapter
from pgvector.psycopg2 import register_vector
class PGVectorService:
"""Service for managing embeddings with PostgreSQL pgvector"""
"""Service for managing embeddings with PostgreSQL pgvector using connection pooling"""
def __init__(self):
self.conn = psycopg2.connect(
self.pool = ThreadedConnectionPool(
minconn=1,
maxconn=20,
host=os.getenv("POSTGRES_HOST", "postgres"),
user=os.getenv("POSTGRES_USER"),
password=os.getenv("POSTGRES_PASSWORD"),
dbname=os.getenv("POSTGRES_DB"),
)
# Register the vector type with the connection
register_vector(self.conn)
# Test connection and register vector type to ensure the pool works
conn = self.pool.getconn()
try:
register_vector(conn)
finally:
self.pool.putconn(conn)
def _get_conn_with_vector(self):
"""Get a connection from the pool and register vector type"""
conn = self.pool.getconn()
register_vector(conn)
return conn
def insert_embedding(self, id: int, embedding: list):
"""Insert or update an embedding"""
with self.conn.cursor() as cur:
cur.execute(
"""
INSERT INTO embeddings_table (id, embedding)
VALUES (%s, %s)
ON CONFLICT (id) DO UPDATE SET embedding = EXCLUDED.embedding;
""",
(id, embedding),
)
self.conn.commit()
conn = self._get_conn_with_vector()
try:
with conn.cursor() as cur:
cur.execute(
"""
INSERT INTO embeddings_table (id, embedding)
VALUES (%s, %s)
ON CONFLICT (id) DO UPDATE SET embedding = EXCLUDED.embedding;
""",
(id, embedding),
)
conn.commit()
finally:
self.pool.putconn(conn)
def search_nearest(self, query_embedding: list, limit: int = 3):
"""Search nearest embeddings using cosine distance (<-> operator)"""
with self.conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute(
"""
SELECT id, embedding, embedding <-> %s AS distance
FROM embeddings_table
ORDER BY embedding <-> %s
LIMIT %s;
""",
(query_embedding, query_embedding, limit),
)
return cur.fetchall()
conn = self._get_conn_with_vector()
try:
with conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute(
"""
SELECT id, embedding, embedding <-> %s AS distance
FROM embeddings_table
ORDER BY embedding <-> %s
LIMIT %s;
""",
(query_embedding, query_embedding, limit),
)
return cur.fetchall()
finally:
self.pool.putconn(conn)
def search_filtered_nearest(
self,
......@@ -55,21 +76,25 @@ class PGVectorService:
limit: int = 3
):
"""Search nearest embeddings with filtering by grade, subject, and language"""
with self.conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute(
"""
SELECT id, grade, subject, unit, concept, lesson, chunk_text,
is_arabic, embedding <-> %s::vector AS distance
FROM educational_chunks
WHERE grade = %s
AND subject ILIKE %s
AND is_arabic = %s
ORDER BY embedding <-> %s::vector
LIMIT %s;
""",
(query_embedding, grade, f"%{subject}%", is_arabic, query_embedding, limit),
)
return cur.fetchall()
conn = self._get_conn_with_vector()
try:
with conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute(
"""
SELECT id, grade, subject, unit, concept, lesson, chunk_text,
is_arabic, embedding <-> %s::vector AS distance
FROM educational_chunks
WHERE grade = %s
AND subject ILIKE %s
AND is_arabic = %s
ORDER BY embedding <-> %s::vector
LIMIT %s;
""",
(query_embedding, grade, f"%{subject}%", is_arabic, query_embedding, limit),
)
return cur.fetchall()
finally:
self.pool.putconn(conn)
def search_flexible_filtered_nearest(
self,
......@@ -103,34 +128,42 @@ class PGVectorService:
params.append(query_embedding)
params.append(limit)
with self.conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute(
f"""
SELECT id, grade, subject, unit, concept, lesson, chunk_text,
is_arabic, embedding <-> %s::vector AS distance
FROM educational_chunks
{where_clause}
ORDER BY embedding <-> %s::vector
LIMIT %s;
""",
params
)
return cur.fetchall()
conn = self._get_conn_with_vector()
try:
with conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute(
f"""
SELECT id, grade, subject, unit, concept, lesson, chunk_text,
is_arabic, embedding <-> %s::vector AS distance
FROM educational_chunks
{where_clause}
ORDER BY embedding <-> %s::vector
LIMIT %s;
""",
params
)
return cur.fetchall()
finally:
self.pool.putconn(conn)
def get_subjects_by_grade_and_language(self, grade: int, is_arabic: bool) -> List[str]:
"""Get available subjects for a specific grade and language"""
with self.conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute(
"""
SELECT DISTINCT subject
FROM educational_chunks
WHERE grade = %s AND is_arabic = %s
ORDER BY subject;
""",
(grade, is_arabic)
)
return [row['subject'] for row in cur.fetchall()]
conn = self._get_conn_with_vector()
try:
with conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute(
"""
SELECT DISTINCT subject
FROM educational_chunks
WHERE grade = %s AND is_arabic = %s
ORDER BY subject;
""",
(grade, is_arabic)
)
return [row['subject'] for row in cur.fetchall()]
finally:
self.pool.putconn(conn)
def close(self):
if self.conn:
self.conn.close()
\ No newline at end of file
def close_pool(self):
if self.pool:
self.pool.closeall()
\ No newline at end of file
self_hosted_env/voice_agent/static/audio-recorder.html
View file @ 12111380
......@@ -185,10 +185,10 @@
</div>
<script>
// Configuration
// Configuration - Auto-detect current domain for CapRover
const Config = {
BACKEND_URL: "http://localhost:8000/chat",
AUDIO_RESPONSE_URL: "http://localhost:8000/get-audio-response"
BACKEND_URL: `${window.location.origin}/chat`,
AUDIO_RESPONSE_URL: `${window.location.origin}/get-audio-response`
};
// Enums
......@@ -206,19 +206,16 @@
PROCESSING: 'processing'
};
// Logger utility
class Logger {
static log(message, type = 'info') {
console.log(`[${type.toUpperCase()}] ${message}`);
}
}
// Base64 Decoder utility
class TextDecoder {
static decodeBase64Utf8(str) {
const bytes = Uint8Array.from(atob(str), c => c.charCodeAt(0));
const decoder = new window.TextDecoder('utf-8');
return decoder.decode(bytes);
try {
const bytes = Uint8Array.from(atob(str), c => c.charCodeAt(0));
const decoder = new window.TextDecoder('utf-8');
return decoder.decode(bytes);
} catch (error) {
return str; // Return original string if decode fails
}
}
}
......@@ -241,45 +238,68 @@
}
}
// API Client
// API Client with enhanced error handling
class APIClient {
async sendFormData(url, formData) {
const response = await fetch(url, {
method: 'POST',
body: formData
});
try {
const response = await fetch(url, {
method: 'POST',
body: formData,
mode: 'cors',
credentials: 'omit'
});
if (!response.ok) {
let errorData;
try {
errorData = await response.json();
} catch {
errorData = { detail: `HTTP ${response.status}: ${response.statusText}` };
}
throw new Error(errorData.detail || `Request failed with status ${response.status}`);
}
if (!response.ok) {
const errorData = await response.json();
throw new Error(errorData.detail || 'Request failed');
const responseData = await response.json();
return responseData;
} catch (error) {
throw error;
}
return response.json();
}
async fetchAudioResponse() {
const response = await fetch(Config.AUDIO_RESPONSE_URL);
if (response.ok) {
const encodedText = response.headers.get('X-Response-Text');
console.log('Encoded text from header:', encodedText);
try {
const response = await fetch(Config.AUDIO_RESPONSE_URL, {
method: 'GET',
mode: 'cors',
credentials: 'omit'
});
if (response.ok) {
const encodedText = response.headers.get('X-Response-Text');
let agentText = "لا يوجد رد متاح";
if (encodedText) {
try {
agentText = TextDecoder.decodeBase64Utf8(encodedText);
} catch (e) {
agentText = "خطأ في فك تشفير الرد";
}
}
let agentText = "لا يوجد رد متاح";
if (encodedText) {
const audioBlob = await response.blob();
return { agentText, audioBlob };
} else {
let errorData;
try {
agentText = TextDecoder.decodeBase64Utf8(encodedText);
console.log('Decoded agent text:', agentText);
} catch (e) {
console.log('Decoding error:', e);
agentText = "خطأ في فك تشفير الرد";
errorData = await response.json();
} catch {
errorData = { detail: `HTTP ${response.status}: ${response.statusText}` };
}
throw new Error(errorData.detail || 'Failed to get response');
}
const audioBlob = await response.blob();
return { agentText, audioBlob };
} else {
const errorData = await response.json();
throw new Error(errorData.detail || 'Failed to get response');
} catch (error) {
throw error;
}
}
}
......@@ -329,7 +349,6 @@
async startRecording() {
try {
Logger.log('طلب الوصول إلى الميكروفون...');
const stream = await navigator.mediaDevices.getUserMedia({ audio: true });
this.mediaRecorder = new MediaRecorder(stream, { mimeType: 'audio/webm;codecs=opus' });
......@@ -343,7 +362,6 @@
this.mediaRecorder.onstop = () => {
const recordedBlob = new Blob(this.audioChunks, { type: 'audio/webm;codecs=opus' });
Logger.log(`تم إيقاف التسجيل. الحجم: ${recordedBlob.size} بايت`);
stream.getTracks().forEach(track => track.stop());
this.onRecordingComplete(recordedBlob);
};
......@@ -353,7 +371,6 @@
this.uiManager.showStatus('التسجيل قيد التقدم...', StatusType.RECORDING);
} catch (error) {
Logger.log(`Error starting recording: ${error.message}`, 'error');
this.uiManager.showStatus(`خطأ في الوصول إلى الميكروفون: ${error.message}`, StatusType.ERROR);
this.stateMachine.setState(RecordingState.IDLE);
}
......@@ -478,15 +495,19 @@
// Auto-play audio if available
if (audioUrl) {
const audioPlayer = this.uiManager.chatContainer.lastChild.querySelector('audio');
if (audioPlayer) {
audioPlayer.play();
}
setTimeout(() => {
const audioPlayer = this.uiManager.chatContainer.lastChild.querySelector('audio');
if (audioPlayer) {
audioPlayer.play().catch(e => {
// Silent fail for auto-play
});
}
}, 100);
}
}
}
// Chat Service - Simplified for direct response handling
// Chat Service - Enhanced with better error handling
class ChatService {
constructor(apiClient, messageManager, uiManager) {
this.apiClient = apiClient;
......@@ -512,17 +533,16 @@
const formData = new FormData();
formData.append('text', text);
formData.append('student_id', studentId);
const response = await this.apiClient.sendFormData(Config.BACKEND_URL, formData);
if (response.status === 'success') {
// Response is ready immediately, get it
await this.getAgentResponse();
return true;
} else {
throw new Error(response.message || 'Unknown error');
}
} catch (error) {
Logger.log(`Error sending text chat: ${error.message}`, 'error');
this.uiManager.showStatus(`خطأ: ${error.message}`, StatusType.ERROR);
return false;
}
......@@ -538,17 +558,16 @@
const formData = new FormData();
formData.append('file', audioBlob, `voice_message_${Date.now()}.webm`);
formData.append('student_id', studentId);
const response = await this.apiClient.sendFormData(Config.BACKEND_URL, formData);
if (response.status === 'success') {
// Response is ready immediately, get it
await this.getAgentResponse();
return true;
} else {
throw new Error(response.message || 'Unknown error');
}
} catch (error) {
Logger.log(`Error sending voice chat: ${error.message}`, 'error');
this.uiManager.showStatus(`خطأ: ${error.message}`, StatusType.ERROR);
return false;
}
......@@ -557,22 +576,28 @@
async getAgentResponse() {
try {
this.uiManager.showStatus('جاري جلب رد المساعد...', StatusType.PROCESSING);
const { agentText, audioBlob } = await this.apiClient.fetchAudioResponse();
if (!agentText || agentText === "لا يوجد رد متاح") {
throw new Error('لم يتم استلام رد صالح من المساعد');
}
const audioUrl = URL.createObjectURL(audioBlob);
this.messageManager.addAgentMessage(agentText, audioUrl);
Logger.log('✓ Agent response received and played.', 'success');
this.uiManager.showStatus('✓ تم استلام الرد! جاهز للرسالة التالية.', StatusType.SUCCESS);
} catch (error) {
Logger.log(`Error fetching agent response: ${error.message}`, 'error');
this.uiManager.showStatus(`خطأ في الشبكة: ${error.message}`, StatusType.ERROR);
// Add fallback text message
this.messageManager.addAgentMessage('عذراً، حدث خطأ في استلام الرد الصوتي. يرجى المحاولة مرة أخرى.');
}
}
}
// Mediator Pattern - Simplified
// Mediator Pattern - Enhanced
class ChatMediator {
constructor() {
this.apiClient = new APIClient();
......@@ -612,8 +637,7 @@
if (!text) return;
this.uiManager.clearTextInput();
const success = await this.chatService.sendTextMessage(text, studentId);
// UI state is handled within the chat service
await this.chatService.sendTextMessage(text, studentId);
}
async handleStartRecording() {
......@@ -650,8 +674,12 @@
// Initialize application when DOM is ready
document.addEventListener('DOMContentLoaded', () => {
UnifiedChatApp.initialize();
console.log('Chat application with Student ID support initialized successfully!');
try {
UnifiedChatApp.initialize();
console.log('Chat application with Student ID support initialized successfully!');
} catch (error) {
console.error('Failed to initialize chat application:', error);
}
});
</script>
</body>
......
self_hosted_env/voice_agent/wait-for-postgres.sh
View file @ 12111380
#!/bin/bash
set -e
host="postgres"
# Use the environment variables set in CapRover for the host and user
host="${POSTGRES_HOST}"
user="${POSTGRES_USER}"
db="${POSTGRES_DB}"
password="${POSTGRES_PASSWORD}"
echo "Waiting for PostgreSQL database to be ready..."
echo "Waiting for PostgreSQL database at $host to be ready..."
# Wait for the database server to be available
until PGPASSWORD="$password" pg_isready -h "$host" -U "$user"; do
# Wait for the database server to be available. This is sufficient.
until PGPASSWORD="${POSTGRES_PASSWORD}" pg_isready -h "$host" -U "$user"; do
echo "PostgreSQL server is unavailable - sleeping"
sleep 1
done
# Wait for the specific database to be available
until PGPASSWORD="$password" psql -h "$host" -U "$user" -d "$db" -c '\q'; do
echo "Database '$db' is not yet available - sleeping"
sleep 1
done
echo "PostgreSQL database is up and ready - executing command"
exec "$@"
\ No newline at end of file
exec "$@"
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment