Skip to content

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

anan character v0 and perfecly long lasting deployment

parent 12111380
Show whitespace changes
Inline Side-by-side
Showing with 581 additions and 265 deletions
self_hosted_env/docker-compose.yml
View file @ ca894367
version: "3.8"
services:
postgres:
# Use the new custom image from Docker Hub
image: salmamohammedhamedmustafa/postgres:latest
build: ./postgres
environment:
POSTGRES_USER: ${POSTGRES_USER}
POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
......@@ -35,7 +34,7 @@ services:
retries: 3
voice-agent:
image: salmamohammedhamedmustafa/voice-agent:latest
build: ./voice_agent
ports:
- "8000:8000"
environment:
......@@ -48,6 +47,8 @@ services:
POSTGRES_USER: "${POSTGRES_USER}"
POSTGRES_PASSWORD: "${POSTGRES_PASSWORD}"
POSTGRES_DB: "${POSTGRES_DB}"
DB_PORT: "${DB_PORT}"
DB_HOST: "${DB_HOST}"
depends_on:
- minio
- postgres
......
self_hosted_env/minio/Dockerfile 0 → 100644
View file @ ca894367
# Start from the official MinIO image
FROM minio/minio:latest
# Set the command to run MinIO with the console address
CMD ["server", "/data", "--console-address", ":9001"]
\ No newline at end of file
self_hosted_env/minio/captain-definition 0 → 100644
View file @ ca894367
{
"schemaVersion": 2,
"dockerfilePath": "./Dockerfile",
"containerHttpPort": "9001",
"ports": [
"9000:9000",
"9001:9001"
],
"volumes": [
"/data"
]
}
self_hosted_env/postgres/captain-definition 0 → 100644
View file @ ca894367
{
"schemaVersion": 2,
"dockerfilePath": "./Dockerfile",
"containerHttpPort": "5432",
"ports": [
"5432:5432"
],
"volumes": [
"/var/lib/postgresql/data"
]
}
self_hosted_env/postgres/init.sql
View file @ ca894367
-- Create the main database
DO
$$
BEGIN
IF NOT EXISTS (
SELECT FROM pg_database WHERE datname = 'embeddings_db'
) THEN
PERFORM dblink_exec('dbname=' || current_database(), 'CREATE DATABASE embeddings_db');
END IF;
END
$$;
-- Connect to the newly created database
\c embeddings_db
CREATE EXTENSION IF NOT EXISTS vector;
\ No newline at end of file
self_hosted_env/voice_agent/main.py
View file @ ca894367
......@@ -14,7 +14,7 @@ from repositories import StorageRepository, MinIOStorageRepository
from handlers import AudioMessageHandler, TextMessageHandler
from services import (
AudioService, ChatService, HealthService, ResponseService,
ResponseManager, OpenAIService, AgentService
ResponseManager, OpenAIService, AgentService, ConnectionPool, PGVectorService, ChatDatabaseService
)
class DIContainer:
......@@ -25,7 +25,16 @@ class DIContainer:
# Initialize OpenAI and Agent services
self.openai_service = OpenAIService()
self.agent_service = AgentService()
self.pool_handler = ConnectionPool(
dbname=os.getenv("POSTGRES_DB"),
user=os.getenv("POSTGRES_USER"),
password=os.getenv("POSTGRES_PASSWORD"),
host=os.getenv("DB_HOST"), # This is the crucial part
port=int(os.getenv("DB_PORT"))
)
print(os.getenv("DB_HOST"), os.getenv("POSTGRES_DB"), os.getenv("POSTGRES_USER"))
self.agent_service = AgentService(pool_handler=self.pool_handler)
# Initialize services
self.audio_service = AudioService(self.storage_repo, self.config.minio_bucket)
......
self_hosted_env/voice_agent/services/__init__.py
View file @ ca894367
......@@ -6,3 +6,6 @@ from .response_manager import ResponseManager
from .openai_service import OpenAIService
from .agent_service import AgentService
from .pgvector_service import PGVectorService
from .chat_database_service import ChatDatabaseService
from .connection_pool import ConnectionPool
from .pedagogy_service import PedagogyService
self_hosted_env/voice_agent/services/agent_service.py
View file @ ca894367
......@@ -10,13 +10,26 @@ 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
from services.connection_pool import ConnectionPool
logger = logging.getLogger(__name__)
SYSTEM_PROMPTS: Dict[StudentNationality, str] = {
StudentNationality.EGYPTIAN: """
إنت مُدرّس لطفل في ابتدائي اسمه {student_name} في الصف {grade}.
رَدّ باللهجة المصريّة الطبيعيّة كأنّك بتكَلِّم {student_name} قصادك.
فقط لو الطِّفل سأل عن هويتك بصراحة وواضح (مثل "إنت مين؟"، "عرّفني بنفسك"، "إنت بتعمل إيه هنا؟")،
رُد بالنصّ الثابت ده:
"أنا عَنان مؤسِّس شارع العلوم، وإنت هنا على مَنَصّة Science Street Lab،
وأنا هنا عشان أساعدك تتعلَّم أي حاجة عايز تتعلَّمها في العلوم."
لو سأل أسئلة عامة زي "نت عارف انا مين؟" أو "إزيك؟" أو "شكرا"، رد بطريقة طبيعية ودودة باستخدام اسمه {student_name}.
أما لو سأل عن أي حاجة في العلوم أو المنهج، اشرح له بالطريقة التعليمية المناسبة.
ملاحظة مُلزِمة: كلمة "منصّة" لازم تكتبها دايمًا كده بالظبط: **مَنَصّة** (بالفتحة على الميم والنون)،
عشان الـTTS يِنطِقها صح.
في باقي الردود، رُد باللهجة المصريّة الطبيعيّة كأنّك بتكَلِّم {student_name} قصادك.
خَلّي الكلام بسيط، واضح، وقَريب من وُدنه.
الجُمَل قُصَيَّرة ومُترابطة، مِش مَقطَّعة.
اشرح كأنّك بتحكي له حِكاية أو بتورّيه حاجَة من الحَياة حَوالينا، مِش بتِقرا من كتاب.
......@@ -32,12 +45,25 @@ SYSTEM_PROMPTS: Dict[StudentNationality, str] = {
لَمّا تِذكُر الصف {grade}، قُله بالطريقة الطبيعيّة زي ما الأطفال بيقولوها: الصف 4 = سنة رابعة ابتدائي، الصف 5 = سنة خامسة ابتدائي، وهكذا.
الهَدف: رَد قُصَيَّر يِعلِّم ويُوصَّل المَعلومة، ويِبان إن فيه مُعلِّم بيِشرَح للطفل مِش كتاب بيتقري.
الهَدف: رَد قُصَيَّر يِعلِّم ويُوصَّل المَعلومة، ويِبان إن "عَنان" بيِشرَح للطفل جوّه مَنَصّة "شارِع العُلوم"، مِش كتاب بيتقري.
""",
StudentNationality.SAUDI: """
إنت معلّم لطفل في ابتدائي اسمه {student_name} في الصف {grade}.
رَد باللهجة السعوديّة الطبيعيّة، كأنك تشرح له قدّامك.
إنت مُعلّم لطفل في ابتدائي اسمه {student_name} في الصف {grade}.
فقط لو الطفل سأل عن هويتك بصراحة وواضح (مثل "إنت مين؟"، "عرّفني بنفسك"، "إنت وش تسوي هنا؟")،
رُد بالنص الثابت هذا:
"أنا عَنان مؤسِّس شارع العلوم، وإنت هنا على مَنَصّة Science Street Lab،
وأنا هنا عشان أساعدك تتعلَّم أي حاجة عايز تتعلَّمها في العلوم."
لو سأل أسئلة عامة مثل "نت عارف انا مين؟" أو "كيفك؟" أو "شكرا"، رد بطريقة طبيعية ودودة باستخدام اسمه {student_name}.
أما لو سأل عن أي شيء في العلوم أو المنهج، اشرح له بالطريقة التعليمية المناسبة.
ملاحظة مُلزِمة: كلمة "منصّة" لازم تكتبها دايمًا كده بالظبط: **مَنَصّة** (بالفتحة على الميم والنون)،
عشان الـTTS يِنطِقها صح.
في باقي الردود، رَد باللهجة السعوديّة الطبيعيّة، كأنك تشرح له قدّامك.
خل الشرح واضح وسهل، لكن لا يكون ناشف.
اشرح كأنك تسولف معه وتشبّه بأشياء من حياته اليومية.
......@@ -54,17 +80,16 @@ SYSTEM_PROMPTS: Dict[StudentNationality, str] = {
لما تذكر الصف {grade}، قُلها بالطريقة اللي الطفل متعود يسمعها: الصف 4 = رابع ابتدائي، الصف 5 = خامس ابتدائي، وهكذا.
الهدف: رد مبسّط، قريب، ويبيّن إن المعلّم يشرح للطفل، مو يقرأ من كتاب.
الهدف: رد مبسّط، قريب، ويبيّن إن "عَنان" يشرح للطفل جوّه مَنَصّة "شارع العلوم"، مو يقرأ من كتاب.
"""
}
class AgentService:
"""Service class for handling AI agent conversations using database memory"""
def __init__(self, use_pgvector: bool = True):
def __init__(self, use_pgvector: bool = True, pool_handler: Optional[ConnectionPool] = None):
self.openai_service = OpenAIService()
if not self.openai_service.is_available():
logger.warning("Warning: OPENAI_API_KEY not found. Agent service will be disabled.")
......@@ -72,32 +97,55 @@ class AgentService:
else:
self.client = self.openai_service.client
# Use database for conversation memory
self.db_service = ChatDatabaseService()
self.pgvector = PGVectorService() if use_pgvector else None
self.pool_handler = pool_handler
if self.pool_handler is None:
self.pool_handler = ConnectionPool(
minconn=1,
maxconn=20,
dbname=os.getenv("DB_NAME"),
user=os.getenv("DB_USER"),
password=os.getenv("DB_PASSWORD"),
host=os.getenv("DB_HOST"),
port=os.getenv("DB_PORT")
)
self.use_pgvector = use_pgvector
# Pass the same pool handler to both services
self.db_service = ChatDatabaseService(self.pool_handler)
if self.use_pgvector:
self.pgvector = PGVectorService(self.pool_handler)
else:
self.pgvector = None
# Initialize pedagogy service for Socratic questioning
self.pedagogy_service = PedagogyService()
logger.info("AgentService initialized with PedagogyService for Socratic questioning")
self.student_info = {}
def is_available(self) -> bool:
return self.client is not None
def get_conversation_history(self, student_id: str) -> List[Dict[str, str]]:
"""Get conversation history from database"""
try:
return self.db_service.get_chat_history(student_id)
except Exception as e:
logger.error(f"Error getting conversation history for {student_id}: {e}")
return []
def add_message_to_history(self, student_id: str, message: str, role: str = "user"):
"""Add message to database"""
try:
self.db_service.add_message(student_id, role, message)
# Limit history to prevent growth
self.db_service.limit_history(student_id, max_messages=38)
except Exception as e:
logger.error(f"Error adding message to history for {student_id}: {e}")
def get_available_subjects(self, student_id: str) -> List[str]:
"""Get available subjects for the student based on their grade and language"""
if not self.pgvector:
return []
try:
student_info = self.db_service.get_student_info(student_id)
if not student_info:
return []
......@@ -106,6 +154,9 @@ class AgentService:
student_info['grade'],
student_info['is_arabic']
)
except Exception as e:
logger.error(f"Error getting available subjects for {student_id}: {e}")
return []
def generate_response(
self,
......@@ -113,53 +164,36 @@ class AgentService:
student_id: str,
subject: str = "Science",
model: str = Models.chat,
temperature: float = 1.0,
temperature: float = 0.3,
top_k: int = 3
) -> str:
"""Generate AI response using database memory with subject filtering"""
"""Generate AI response using database memory with optional retrieval based on question type"""
if not self.is_available():
raise HTTPException(status_code=500, detail="Agent service not available")
try:
# Get complete student information from database
# Get student info
student_info = self.db_service.get_student_info(student_id)
if not student_info:
raise HTTPException(status_code=404, detail=f"Student with ID {student_id} not found")
# Extract student first name only
# Extract first name
full_name = student_info.get('student_name', 'الطالب')
student_name = full_name.split()[0] if full_name else "الطالب"
# Print student information
print("----------------- Student Info Retrieved -----------------")
print(f"Student ID: {student_id}")
for key, value in student_info.items():
print(f"{key.capitalize()}: {value}")
print("---------------------------------------------------------")
logger.info(f"Retrieved student info from DB: {student_info} for student: {student_id}")
# Convert nationality string to StudentNationality enum
# Map nationality
nationality_lower = student_info['nationality'].lower().strip()
nationality_mapping = {
'egyptian': StudentNationality.EGYPTIAN,
'saudi': StudentNationality.SAUDI
}
nationality = nationality_mapping.get(nationality_lower, StudentNationality.EGYPTIAN)
if nationality_lower in nationality_mapping:
nationality = nationality_mapping[nationality_lower]
logger.info(f"Successfully mapped '{student_info['nationality']}' to {nationality}")
else:
logger.warning(f"Unknown nationality '{student_info['nationality']}' for student {student_id}, defaulting to EGYPTIAN")
nationality = StudentNationality.EGYPTIAN
# Add user message to database
# Add user message to DB
self.add_message_to_history(student_id, user_message, "user")
# Get conversation history from database
conversation_history = self.get_conversation_history(student_id)
# Create subject-specific system prompt with first name only
# Format system prompt
base_system_prompt = SYSTEM_PROMPTS.get(nationality, SYSTEM_PROMPTS[StudentNationality.EGYPTIAN])
formatted_base_prompt = base_system_prompt.format(
student_name=student_name,
......@@ -167,38 +201,57 @@ class AgentService:
)
subject_specific_prompt = f"{formatted_base_prompt}\n\nإنت بتدرّس مادة {subject} للطفل {student_name}."
# Add Socratic questioning instructions if applicable (grades 4-6)
# Add Socratic instructions if any
socratic_instructions = self.pedagogy_service.get_socratic_instructions(
student_info['grade'],
student_info['nationality']
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
messages = []
# Check if system message exists
has_system_message = conversation_history and conversation_history[0].get("role") == "system"
if not has_system_message:
messages.append({"role": "system", "content": subject_specific_prompt})
# Add system message to database
self.add_message_to_history(student_id, subject_specific_prompt, "system")
# Add conversation history
messages.extend(conversation_history)
# Enhanced pgvector enrichment with filtering
if self.pgvector:
# ----------------- DYNAMIC RETRIEVAL DECISION -----------------
# Ask model to classify if retrieval needed
try:
query_embedding = self.openai_service.generate_embedding(user_message)
classification_prompt = f"""
صنف السؤال التالي: هل يحتاج لإجابة تعتمد على المنهج الدراسي أو محتوى المادة العلمية المتخصصة؟
رد فقط بـ "YES" لو يحتاج retrieval من المحتوى التعليمي، و "NO" لو مجرد سؤال عام أو عن الشخصية أو محادثة عادية.
أمثلة على أسئلة تحتاج "YES":
- ما هو التمثيل الضوئي؟
- اشرح لي الجهاز الهضمي
- كيف تتكون الأمطار؟
أمثلة على أسئلة تحتاج "NO":
- إنت مين؟
- إزيك؟
- نت عارف انا مين؟
- شكرا ليك
السؤال: "{user_message}"
"""
classification_response = self.client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": classification_prompt}],
temperature=0
)
classification_answer = classification_response.choices[0].message.content.strip().upper()
need_retrieval = classification_answer == "YES"
logger.info(f"Classification for '{user_message}': {classification_answer} (need_retrieval: {need_retrieval})")
except Exception as e:
logger.warning(f"Error in classification, defaulting to no retrieval: {e}")
need_retrieval = False
# ----------------- RETRIEVAL (if needed) -----------------
if self.pgvector and need_retrieval:
try:
query_embedding = self.openai_service.generate_embedding(user_message)
neighbors = self.pgvector.search_filtered_nearest(
query_embedding=query_embedding,
grade=student_info['grade'],
......@@ -207,54 +260,40 @@ class AgentService:
limit=top_k
)
if neighbors:
print("\n----------------- Retrieval Results -----------------")
context_message = f"معلومات من المنهج لمادة {subject} للصف {student_info['grade']} للطالب {student_name}:\n"
for i, n in enumerate(neighbors, 1):
unit_info = f" - الوحدة: {n['unit']}" if n['unit'] else ""
concept_info = f" - المفهوم: {n['concept']}" if n['concept'] else ""
lesson_info = f" - الدرس: {n['lesson']}" if n['lesson'] else ""
context_message += f"\n{i}. {unit_info}{concept_info}{lesson_info}\n"
context_message += f"المحتوى: {n['chunk_text'][:200]}...\n"
context_message += f"(درجة التشابه: {n['distance']:.3f})\n"
print(f"Result {i}:")
print(f" Unit: {n['unit']}")
print(f" Concept: {n['concept']}")
print(f" Lesson: {n['lesson']}")
print(f" Chunk Text: {n['chunk_text']}...")
print(f" Distance: {n['distance']:.3f}")
print("-" * 20)
print("-----------------------------------------------------")
relevant_neighbors = [n for n in neighbors if n['distance'] < 1.3] if neighbors else []
if relevant_neighbors:
context_message = f"معلومات من المنهج لمادة {subject} للصف {student_info['grade']}:\n\n"
for n in relevant_neighbors:
unit_info = f"الوحدة: {n['unit']}" if n.get('unit') else ""
concept_info = f"المفهوم: {n['concept']}" if n.get('concept') else ""
lesson_info = f"الدرس: {n['lesson']}" if n.get('lesson') else ""
context_header = " - ".join(filter(None, [unit_info, concept_info, lesson_info]))
if context_header:
context_message += f"**{context_header}**\n"
context_message += f"{n['chunk_text']}\n\n---\n\n"
context_message += "استخدم هذه المعلومات لتقديم شرح دقيق ومناسب للطفل."
messages.append({"role": "system", "content": context_message})
logger.info(f"Added {len(neighbors)} filtered knowledge base results for subject: {subject}")
else:
print("\n----------------- Retrieval Results -----------------")
print(f"No relevant content found for subject: {subject}, grade: {student_info['grade']}, Arabic: {student_info['is_arabic']}")
print("-----------------------------------------------------")
logger.info(f"No relevant content found for subject: {subject}, grade: {student_info['grade']}, Arabic: {student_info['is_arabic']}")
except Exception as e:
logger.warning(f"Error using pgvector with filtering: {e}")
logger.warning(f"Error using pgvector: {e}")
# Generate AI response
# ----------------- GENERATE RESPONSE -----------------
response = self.client.chat.completions.create(
model=model,
messages=messages,
temperature=temperature
)
ai_response = response.choices[0].message.content.strip()
if not ai_response:
raise ValueError("Empty response from AI model")
# Add AI response to database
# Save AI response
self.add_message_to_history(student_id, ai_response, "assistant")
return ai_response
except HTTPException:
raise
except Exception as e:
logger.error(f"Error generating AI response: {e}")
raise HTTPException(status_code=500, detail=f"AI response generation failed: {str(e)}")
......@@ -265,6 +304,7 @@ class AgentService:
if not self.pgvector:
raise HTTPException(status_code=400, detail="PGVector service not enabled")
try:
student_info = self.db_service.get_student_info(student_id)
if not student_info:
raise HTTPException(status_code=404, detail=f"Student with ID {student_id} not found")
......@@ -276,6 +316,11 @@ class AgentService:
is_arabic=student_info['is_arabic'],
limit=top_k
)
except HTTPException:
raise
except Exception as e:
logger.error(f"Error in search_similar: {e}")
raise HTTPException(status_code=500, detail=f"Search failed: {str(e)}")
def update_student_subject_context(self, student_id: str, subject: str):
"""Update the system message for a new subject"""
......@@ -285,7 +330,8 @@ class AgentService:
return False
# Extract student name
student_name = student_info.get('student_name', 'الطالب')
full_name = student_info.get('student_name', 'الطالب')
student_name = full_name.split()[0] if full_name else "الطالب"
# Clear existing history to reset context
self.db_service.clear_history(student_id)
......@@ -327,10 +373,15 @@ class AgentService:
def export_conversation(self, student_id: str) -> List[Dict[str, str]]:
"""Export conversation history for a student"""
try:
return self.get_conversation_history(student_id)
except Exception as e:
logger.error(f"Error exporting conversation for {student_id}: {e}")
return []
def import_conversation(self, messages: List[Dict[str, str]], student_id: str):
"""Import conversation history for a student"""
try:
# Clear existing history first
self.db_service.clear_history(student_id)
......@@ -340,6 +391,9 @@ class AgentService:
content = message.get("content", "")
if content:
self.add_message_to_history(student_id, content, role)
except Exception as e:
logger.error(f"Error importing conversation for {student_id}: {e}")
raise
def clear_conversation(self, student_id: str) -> Dict[str, str]:
"""Clear conversation history for a student"""
......@@ -394,41 +448,69 @@ class AgentService:
# Return a default system prompt - this could be made more sophisticated
return "Default system prompt for educational AI assistant"
def close(self):
"""Close database connection pools"""
if self.db_service:
self.db_service.close_pool()
if self.pgvector:
self.pgvector.close_pool()
def debug_retrieval_pipeline(self, student_id: str, query: str):
"""Debug function to trace the retrieval pipeline"""
print("=== RETRIEVAL DEBUG PIPELINE ===")
try:
# 1. Check student info
student_info = self.db_service.get_student_info(student_id)
print(f"1. Student Info: {student_info}")
# ----------------- Test -----------------
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
if not student_info:
print("❌ No student info found!")
return
agent = AgentService(use_pgvector=True)
# 2. Test embedding generation
print(f"2. Testing embedding generation for query: '{query}'")
query_embedding = self.openai_service.generate_embedding(query)
print(f"✅ Generated embedding (length: {len(query_embedding)})")
if agent.is_available():
try:
# Test with chemistry (default)
reply = agent.generate_response(
"هو يعني إيه ذَرّة؟",
student_id="student_001",
subject="chemistry"
# 3. Test vector search
print(f"3. Testing vector search...")
if self.pgvector:
neighbors = self.pgvector.search_filtered_nearest(
query_embedding=query_embedding,
grade=student_info['grade'],
subject="Science",
is_arabic=student_info['is_arabic'],
limit=3
)
print("AI (Chemistry):", reply)
# Test with math
reply = agent.generate_response(
"إيه هو الجمع؟",
student_id="student_001",
subject="math"
print(f"✅ Found {len(neighbors)} neighbors:")
for i, neighbor in enumerate(neighbors):
print(f" {i+1}. Distance: {neighbor['distance']:.3f}")
print(f" Unit: {neighbor.get('unit', 'N/A')}")
print(f" Concept: {neighbor.get('concept', 'N/A')}")
print(f" Text preview: {neighbor['chunk_text'][:100]}...")
print()
else:
print("❌ PGVector not available")
# 4. Test full response generation
print(f"4. Testing full response generation...")
response = self.generate_response(
user_message=query,
student_id=student_id,
subject="Science"
)
print("AI (Math):", reply)
print(f"✅ Generated response (first 200 chars): {response[:200]}...")
print("=== DEBUG COMPLETE ===")
except Exception as e:
print(f"Test failed: {e}")
finally:
agent.close()
else:
print("Agent service not available. Check OPENAI_API_KEY.")
\ No newline at end of file
print(f"❌ Debug pipeline failed at step: {e}")
import traceback
traceback.print_exc()
def close(self):
"""Close database connection pools"""
if self.pool_handler:
try:
self.pool_handler.close_all()
except Exception as e:
logger.error(f"Error closing connection pools: {e}")
def __del__(self):
"""Destructor to ensure connection pools are closed"""
self.close()
\ No newline at end of file
self_hosted_env/voice_agent/services/chat_database_service.py
View file @ ca894367
import os
import psycopg2
from psycopg2.extras import RealDictCursor
from psycopg2.pool import ThreadedConnectionPool
from typing import List, Dict, Optional, Tuple
import logging
from services.connection_pool import ConnectionPool
logger = logging.getLogger(__name__)
class ChatDatabaseService:
"""Simple service for managing chat history in PostgreSQL with connection pooling"""
def __init__(self):
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"),
)
"""Service for managing chat history using a shared, robust connection pool"""
def __init__(self, pool_handler: 'ConnectionPoolHandler'):
self.pool_handler = pool_handler
def get_student_nationality(self, student_id: str) -> Optional[str]:
"""Get student nationality from database"""
conn = self.pool.getconn()
try:
with self.pool_handler.get_connection() as conn:
with conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute(
"SELECT nationality FROM students WHERE student_id = %s",
......@@ -32,13 +25,10 @@ class ChatDatabaseService:
)
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"""
conn = self.pool.getconn()
try:
with self.pool_handler.get_connection() as conn:
with conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute(
"""
......@@ -53,18 +43,15 @@ class ChatDatabaseService:
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
'grade': result['grade'],
'is_arabic': result['language'],
'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"""
conn = self.pool.getconn()
try:
with self.pool_handler.get_connection() as conn:
with conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute(
"SELECT grade, language FROM students WHERE student_id = %s",
......@@ -74,13 +61,10 @@ class ChatDatabaseService:
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"""
conn = self.pool.getconn()
try:
with self.pool_handler.get_connection() as conn:
with conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute(
"""
......@@ -93,15 +77,11 @@ class ChatDatabaseService:
(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"""
conn = self.pool.getconn()
try:
with self.pool_handler.get_connection() as conn:
with conn.cursor() as cur:
cur.execute(
"""
......@@ -111,26 +91,20 @@ class ChatDatabaseService:
(student_id, role, content)
)
conn.commit()
finally:
self.pool.putconn(conn)
def clear_history(self, student_id: str):
"""Clear chat history for a student"""
conn = self.pool.getconn()
try:
with self.pool_handler.get_connection() as conn:
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"""
conn = self.pool.getconn()
try:
with self.pool_handler.get_connection() as conn:
with conn.cursor() as cur:
cur.execute(
"""
......@@ -147,8 +121,6 @@ class ChatDatabaseService:
(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):
......@@ -170,8 +142,7 @@ class ChatDatabaseService:
if updates:
params.append(student_id)
conn = self.pool.getconn()
try:
with self.pool_handler.get_connection() as conn:
with conn.cursor() as cur:
cur.execute(
f"""
......@@ -182,14 +153,11 @@ class ChatDatabaseService:
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"""
conn = self.pool.getconn()
try:
with self.pool_handler.get_connection() as conn:
with conn.cursor() as cur:
cur.execute(
"""
......@@ -200,9 +168,3 @@ class ChatDatabaseService:
(student_id, student_name, grade, language, nationality)
)
conn.commit()
finally:
self.pool.putconn(conn)
def close_pool(self):
if self.pool:
self.pool.closeall()
\ No newline at end of file
self_hosted_env/voice_agent/services/connection_pool.py 0 → 100644
View file @ ca894367
import os
import psycopg2
from psycopg2.pool import ThreadedConnectionPool
import logging
import time
import threading
from contextlib import contextmanager
logger = logging.getLogger(__name__)
class ConnectionPool:
"""connection pool with health monitoring and automatic recovery"""
def __init__(self, minconn=2, maxconn=20, **db_params):
self.db_params = db_params
self.minconn = minconn
self.maxconn = maxconn
self.pool = None
self._pool_lock = threading.RLock()
self._last_health_check = 0
self._health_check_interval = 300 # 5 minutes
self._connection_timeout = 30
self._idle_timeout = 7200 # 2 hours
self._initialize_pool()
def _initialize_pool(self):
"""Initialize the connection pool with proper parameters"""
try:
# Add connection parameters to handle idle connections
pool_params = {
**self.db_params,
'connect_timeout': self._connection_timeout,
# These parameters help with connection management
'keepalives_idle': 600, # Start keepalives after 10 min idle
'keepalives_interval': 30, # Send keepalive every 30 seconds
'keepalives_count': 3, # Close connection after 3 failed keepalives
}
self.pool = ThreadedConnectionPool(
minconn=self.minconn,
maxconn=self.maxconn,
**pool_params
)
logger.info(f"Connection pool initialized with {self.minconn}-{self.maxconn} connections")
except Exception as e:
logger.error(f"Failed to initialize connection pool: {e}")
raise
def _recreate_pool(self):
"""Recreate the connection pool in case of catastrophic failure"""
with self._pool_lock:
if self.pool:
try:
self.pool.closeall()
except:
pass
logger.warning("Recreating connection pool...")
self._initialize_pool()
def _validate_connection(self, conn):
"""Validate a connection with comprehensive checks"""
try:
# Check if connection is alive
with conn.cursor() as cur:
cur.execute("SELECT 1")
cur.fetchone()
# Check connection status
if conn.closed != 0:
return False
# Check for any pending transactions
if conn.info.transaction_status != psycopg2.extensions.TRANSACTION_STATUS_IDLE:
try:
conn.rollback()
except:
return False
return True
except (psycopg2.OperationalError, psycopg2.InterfaceError,
psycopg2.DatabaseError) as e:
logger.debug(f"Connection validation failed: {e}")
return False
except Exception as e:
logger.warning(f"Unexpected error during connection validation: {e}")
return False
def _health_check(self):
"""Perform periodic health check on the pool"""
current_time = time.time()
if current_time - self._last_health_check < self._health_check_interval:
return
try:
with self._pool_lock:
if self.pool:
# Try to get a connection to test pool health
test_conn = self.pool.getconn()
if test_conn and self._validate_connection(test_conn):
self.pool.putconn(test_conn)
self._last_health_check = current_time
return
else:
# Connection is bad, try to close it
if test_conn:
try:
test_conn.close()
except:
pass
# Pool seems unhealthy, recreate it
logger.warning("Pool health check failed, recreating pool")
self._recreate_pool()
self._last_health_check = current_time
except Exception as e:
logger.error(f"Health check failed: {e}")
try:
self._recreate_pool()
except Exception as recreate_error:
logger.error(f"Failed to recreate pool during health check: {recreate_error}")
@contextmanager
def get_connection(self, max_retries=3):
"""Get a validated connection with automatic retry and recovery"""
self._health_check()
conn = None
for attempt in range(max_retries):
try:
with self._pool_lock:
if not self.pool:
self._initialize_pool()
conn = self.pool.getconn()
if conn and self._validate_connection(conn):
try:
yield conn
return
finally:
if conn:
try:
# Ensure connection is in a clean state
if conn.info.transaction_status != psycopg2.extensions.TRANSACTION_STATUS_IDLE:
conn.rollback()
self.pool.putconn(conn)
except Exception as e:
logger.warning(f"Error returning connection to pool: {e}")
try:
conn.close()
except:
pass
else:
# Bad connection, close it
if conn:
try:
conn.close()
except:
pass
conn = None
except Exception as e:
logger.warning(f"Connection attempt {attempt + 1} failed: {e}")
if conn:
try:
conn.close()
except:
pass
conn = None
if attempt == max_retries - 1:
# Last attempt, try to recreate pool
try:
self._recreate_pool()
except Exception as recreate_error:
logger.error(f"Failed to recreate pool: {recreate_error}")
raise ConnectionError(f"Failed to get valid connection after {max_retries} attempts")
# Wait before retry with exponential backoff
time.sleep(min(2 ** attempt, 10))
def get_valid_conn(self):
"""Legacy method for backward compatibility - get a validated connection"""
max_retries = 3
for attempt in range(max_retries):
try:
with self._pool_lock:
if not self.pool:
self._initialize_pool()
conn = self.pool.getconn()
if conn and self._validate_connection(conn):
return conn
else:
# Bad connection, close it
if conn:
try:
conn.close()
except:
pass
conn = None
except Exception as e:
logger.warning(f"Connection attempt {attempt + 1} failed: {e}")
if conn:
try:
conn.close()
except:
pass
conn = None
if attempt == max_retries - 1:
# Last attempt, try to recreate pool
try:
self._recreate_pool()
except Exception as recreate_error:
logger.error(f"Failed to recreate pool: {recreate_error}")
raise ConnectionError(f"Failed to get valid connection after {max_retries} attempts")
# Wait before retry with exponential backoff
time.sleep(min(2 ** attempt, 10))
def put_conn(self, conn):
"""Return connection to pool - legacy method for backward compatibility"""
try:
if conn:
# Ensure connection is in a clean state
if conn.info.transaction_status != psycopg2.extensions.TRANSACTION_STATUS_IDLE:
conn.rollback()
self.pool.putconn(conn)
except Exception as e:
logger.warning(f"Error returning connection to pool: {e}")
try:
conn.close()
except:
pass
def close_all(self):
"""Close all connections in the pool"""
with self._pool_lock:
if self.pool:
try:
self.pool.closeall()
finally:
self.pool = None
\ No newline at end of file
self_hosted_env/voice_agent/services/pgvector_service.py
View file @ ca894367
import os
import psycopg2
from psycopg2.extras import RealDictCursor
from psycopg2.pool import ThreadedConnectionPool
from typing import List, Optional
# Import the pgvector adapter
import logging
from pgvector.psycopg2 import register_vector
from services.connection_pool import ConnectionPool
logger = logging.getLogger(__name__)
class PGVectorService:
"""Service for managing embeddings with PostgreSQL pgvector using connection pooling"""
def __init__(self):
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"),
)
# Test connection and register vector type to ensure the pool works
conn = self.pool.getconn()
try:
register_vector(conn)
finally:
self.pool.putconn(conn)
class PGVectorService:
"""Service for managing embeddings with PostgreSQL pgvector using a shared, robust connection pool"""
def _get_conn_with_vector(self):
"""Get a connection from the pool and register vector type"""
conn = self.pool.getconn()
def __init__(self, pool_handler: 'ConnectionPool'):
self.pool_handler = pool_handler
# Test connection and register vector type
with self.pool_handler.get_connection() as conn:
register_vector(conn)
return conn
def insert_embedding(self, id: int, embedding: list):
"""Insert or update an embedding"""
conn = self._get_conn_with_vector()
try:
with self.pool_handler.get_connection() as conn:
with conn.cursor() as cur:
cur.execute(
"""
......@@ -46,13 +31,10 @@ class PGVectorService:
(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)"""
conn = self._get_conn_with_vector()
try:
with self.pool_handler.get_connection() as conn:
with conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute(
"""
......@@ -64,8 +46,6 @@ class PGVectorService:
(query_embedding, query_embedding, limit),
)
return cur.fetchall()
finally:
self.pool.putconn(conn)
def search_filtered_nearest(
self,
......@@ -76,8 +56,7 @@ class PGVectorService:
limit: int = 3
):
"""Search nearest embeddings with filtering by grade, subject, and language"""
conn = self._get_conn_with_vector()
try:
with self.pool_handler.get_connection() as conn:
with conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute(
"""
......@@ -93,8 +72,6 @@ class PGVectorService:
(query_embedding, grade, f"%{subject}%", is_arabic, query_embedding, limit),
)
return cur.fetchall()
finally:
self.pool.putconn(conn)
def search_flexible_filtered_nearest(
self,
......@@ -128,8 +105,7 @@ class PGVectorService:
params.append(query_embedding)
params.append(limit)
conn = self._get_conn_with_vector()
try:
with self.pool_handler.get_connection() as conn:
with conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute(
f"""
......@@ -143,13 +119,10 @@ class PGVectorService:
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"""
conn = self._get_conn_with_vector()
try:
with self.pool_handler.get_connection() as conn:
with conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute(
"""
......@@ -161,9 +134,3 @@ class PGVectorService:
(grade, is_arabic)
)
return [row['subject'] for row in cur.fetchall()]
finally:
self.pool.putconn(conn)
def close_pool(self):
if self.pool:
self.pool.closeall()
\ No newline at end of file
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