version 1.1

.env.example

@@ -23,4 +23,11 @@ WEBSOCKET_PATH=/notify
 # Application settings
 ENVIRONMENT=production
 DEBUG=false
-PORT=7860 
+PORT=7860 
+
+# Cache Configuration
+CACHE_TTL_SECONDS=300
+CACHE_CLEANUP_INTERVAL=60
+CACHE_MAX_SIZE=1000
+HISTORY_QUEUE_SIZE=10
+HISTORY_CACHE_TTL=3600 

.gitattributes

@@ -1,29 +0,0 @@
-# Auto detect text files and perform LF normalization
-* text=auto eol=lf
-
-# Documents
-*.md text
-*.txt text
-*.ini text
-*.yaml text
-*.yml text
-*.json text
-*.py text
-*.env.example text
-
-# Binary files
-*.png binary
-*.jpg binary
-*.jpeg binary
-*.gif binary
-*.ico binary
-*.db binary
-
-# Git related files
-.gitignore text
-.gitattributes text
-
-# Docker related files
-Dockerfile text
-docker-compose.yml text
-.dockerignore text 

.gitignore

@@ -60,6 +60,8 @@ tests/
 
 Admin_bot/
 
+Pix-Agent/
+
 # Hugging Face Spaces
 .gitattributes
 
@@ -77,3 +79,5 @@ Thumbs.db
 *.log
 .env
 main.py
+
+test/

README.md

@@ -358,4 +358,62 @@ You can customize the retrieval parameters when making API requests:
 
 ## Implementation Details
 
-The system is implemented as a custom retriever class `ThresholdRetriever` that integrates with LangChain's retrieval infrastructure while providing enhanced functionality. 
+The system is implemented as a custom retriever class `ThresholdRetriever` that integrates with LangChain's retrieval infrastructure while providing enhanced functionality.
+
+## In-Memory Cache
+
+Dự án bao gồm một hệ thống cache trong bộ nhớ để giảm thiểu truy cập đến cơ sở dữ liệu PostgreSQL và MongoDB.
+
+### Cấu hình Cache
+
+Cache được cấu hình thông qua các biến môi trường:
+
+```
+# Cache Configuration
+CACHE_TTL_SECONDS=300           # Thời gian tồn tại của cache item (giây)
+CACHE_CLEANUP_INTERVAL=60       # Chu kỳ xóa cache hết hạn (giây)
+CACHE_MAX_SIZE=1000             # Số lượng item tối đa trong cache
+HISTORY_QUEUE_SIZE=10           # Số lượng item tối đa trong queue lịch sử người dùng
+HISTORY_CACHE_TTL=3600          # Thời gian tồn tại của lịch sử người dùng (giây)
+```
+
+### Cơ chế Cache
+
+Hệ thống cache kết hợp hai cơ chế hết hạn:
+
+1. **Lazy Expiration**: Kiểm tra thời hạn khi truy cập cache item. Nếu item đã hết hạn, nó sẽ bị xóa và trả về kết quả là không tìm thấy.
+
+2. **Active Expiration**: Một background thread định kỳ quét và xóa các item đã hết hạn. Điều này giúp tránh tình trạng cache quá lớn với các item không còn được sử dụng.
+
+### Các loại dữ liệu được cache
+
+- **Dữ liệu PostgreSQL**: Thông tin từ các bảng FAQ, Emergency Contacts, và Events.
+- **Lịch sử người dùng từ MongoDB**: Lịch sử hội thoại người dùng được lưu trong queue với thời gian sống tính theo lần truy cập cuối cùng.
+
+### API Cache
+
+Dự án cung cấp các API endpoints để quản lý cache:
+
+- `GET /cache/stats`: Xem thống kê về cache (tổng số item, bộ nhớ sử dụng, v.v.)
+- `DELETE /cache/clear`: Xóa toàn bộ cache
+- `GET /debug/cache`: (Chỉ trong chế độ debug) Xem thông tin chi tiết về cache, bao gồm các keys và cấu hình
+
+### Cách hoạt động
+
+1. Khi một request đến, hệ thống sẽ kiểm tra dữ liệu trong cache trước.
+2. Nếu dữ liệu tồn tại và còn hạn, trả về từ cache.
+3. Nếu dữ liệu không tồn tại hoặc đã hết hạn, truy vấn từ database và lưu kết quả vào cache.
+4. Khi dữ liệu được cập nhật hoặc xóa, cache liên quan sẽ tự động được xóa.
+
+### Lịch sử người dùng
+
+Lịch sử hội thoại người dùng được lưu trong queue riêng với cơ chế đặc biệt:
+
+- Mỗi người dùng có một queue riêng với kích thước giới hạn (`HISTORY_QUEUE_SIZE`).
+- Thời gian sống của queue được làm mới mỗi khi có tương tác mới.
+- Khi queue đầy, các item cũ nhất sẽ bị loại bỏ.
+- Queue tự động bị xóa sau một thời gian không hoạt động.
+
+## Tác giả
+
+- **PIX Project Team** 

api_documentation.txt

@@ -1,318 +0,0 @@
-# Frontend Integration Guide for PixAgent API
-
-This guide provides instructions for integrating with the optimized PostgreSQL-based API endpoints for Event, FAQ, and Emergency data.
-
-## API Endpoints
-
-### Events
-
-| Endpoint | Method | Description |
-|----------|--------|-------------|
-| /postgres/events/ | GET | Fetch all events (with optional filtering) |
-| /postgres/events/{event_id} | GET | Fetch a specific event by ID |
-| /postgres/events/featured | GET | Fetch featured events |
-| /postgres/events/ | POST | Create a new event |
-| /postgres/events/{event_id} | PUT | Update an existing event |
-| /postgres/events/{event_id} | DELETE | Delete an event |
-
-### FAQs
-
-| Endpoint | Method | Description |
-|----------|--------|-------------|
-| /postgres/faqs/ | GET | Fetch all FAQs |
-| /postgres/faqs/{faq_id} | GET | Fetch a specific FAQ by ID |
-| /postgres/faqs/ | POST | Create a new FAQ |
-| /postgres/faqs/{faq_id} | PUT | Update an existing FAQ |
-| /postgres/faqs/{faq_id} | DELETE | Delete a FAQ |
-
-### Emergency Contacts
-
-| Endpoint | Method | Description |
-|----------|--------|-------------|
-| /postgres/emergencies/ | GET | Fetch all emergency contacts |
-| /postgres/emergencies/{emergency_id} | GET | Fetch a specific emergency contact by ID |
-| /postgres/emergencies/ | POST | Create a new emergency contact |
-| /postgres/emergencies/{emergency_id} | PUT | Update an existing emergency contact |
-| /postgres/emergencies/{emergency_id} | DELETE | Delete an emergency contact |
-
-## Response Models
-
-### Event Response Model
-
-interface EventResponse {
-  id: number;
-  name: string;
-  description: string;
-  date_start: string;  // ISO format date
-  date_end: string;    // ISO format date
-  location: string;
-  image_url: string;
-  price: {
-    currency: string;
-    amount: string;
-  };
-  featured: boolean;
-  is_active: boolean;
-  created_at: string;  // ISO format date
-  updated_at: string;  // ISO format date
-}
-
-### FAQ Response Model
-
-interface FaqResponse {
-  id: number;
-  question: string;
-  answer: string;
-  is_active: boolean;
-  created_at: string;  // ISO format date
-  updated_at: string;  // ISO format date
-}
-
-### Emergency Response Model
-
-interface EmergencyResponse {
-  id: number;
-  name: string;
-  phone_number: string;
-  description: string;
-  address: string;
-  priority: number;
-  is_active: boolean;
-  created_at: string;  // ISO format date
-  updated_at: string;  // ISO format date
-}
-
-## Example Usage (React)
-
-### Fetching Events
-
-import { useState, useEffect } from 'react';
-import axios from 'axios';
-
-const API_BASE_URL = 'http://localhost:8000';
-
-function EventList() {
-  const [events, setEvents] = useState([]);
-  const [loading, setLoading] = useState(true);
-  const [error, setError] = useState(null);
-
-  useEffect(() => {
-    const fetchEvents = async () => {
-      try {
-        setLoading(true);
-        const response = await axios.get(`${API_BASE_URL}/postgres/events/`);
-        setEvents(response.data);
-        setLoading(false);
-      } catch (err) {
-        setError('Failed to fetch events');
-        setLoading(false);
-        console.error('Error fetching events:', err);
-      }
-    };
-
-    fetchEvents();
-  }, []);
-
-  if (loading) return <p>Loading events...</p>;
-  if (error) return <p>{error}</p>;
-
-  return (
-    <div>
-      <h1>Events</h1>
-      <div className="event-list">
-        {events.map(event => (
-          <div key={event.id} className="event-card">
-            <h2>{event.name}</h2>
-            <p>{event.description}</p>
-            <p>
-              <strong>When:</strong> {new Date(event.date_start).toLocaleDateString()} - {new Date(event.date_end).toLocaleDateString()}
-            </p>
-            <p><strong>Where:</strong> {event.location}</p>
-            <p><strong>Price:</strong> {event.price.amount} {event.price.currency}</p>
-            {event.featured && <span className="featured-badge">Featured</span>}
-          </div>
-        ))}
-      </div>
-    </div>
-  );
-}
-
-### Creating an Event
-
-import { useState } from 'react';
-import axios from 'axios';
-
-function CreateEvent() {
-  const [eventData, setEventData] = useState({
-    name: '',
-    description: '',
-    date_start: '',
-    date_end: '',
-    location: '',
-    image_url: '',
-    price: {
-      currency: 'USD',
-      amount: '0'
-    },
-    featured: false,
-    is_active: true
-  });
-  const [loading, setLoading] = useState(false);
-  const [error, setError] = useState(null);
-  const [success, setSuccess] = useState(false);
-
-  const handleChange = (e) => {
-    const { name, value, type, checked } = e.target;
-    
-    if (name === 'price_amount') {
-      setEventData(prev => ({
-        ...prev,
-        price: {
-          ...prev.price,
-          amount: value
-        }
-      }));
-    } else if (name === 'price_currency') {
-      setEventData(prev => ({
-        ...prev,
-        price: {
-          ...prev.price,
-          currency: value
-        }
-      }));
-    } else {
-      setEventData(prev => ({
-        ...prev,
-        [name]: type === 'checkbox' ? checked : value
-      }));
-    }
-  };
-
-  const handleSubmit = async (e) => {
-    e.preventDefault();
-    try {
-      setLoading(true);
-      setError(null);
-      setSuccess(false);
-      
-      const response = await axios.post(`${API_BASE_URL}/postgres/events/`, eventData);
-      setSuccess(true);
-      setEventData({
-        name: '',
-        description: '',
-        date_start: '',
-        date_end: '',
-        location: '',
-        image_url: '',
-        price: {
-          currency: 'USD',
-          amount: '0'
-        },
-        featured: false,
-        is_active: true
-      });
-      setLoading(false);
-    } catch (err) {
-      setError('Failed to create event');
-      setLoading(false);
-      console.error('Error creating event:', err);
-    }
-  };
-
-  return (
-    <div>
-      <h1>Create New Event</h1>
-      {success && <div className="success-message">Event created successfully!</div>}
-      {error && <div className="error-message">{error}</div>}
-      <form onSubmit={handleSubmit}>
-        {/* Form fields would go here */}
-        <button type="submit" disabled={loading}>
-          {loading ? 'Creating...' : 'Create Event'}
-        </button>
-      </form>
-    </div>
-  );
-}
-
-## Performance Optimizations
-
-The API now includes several performance optimizations:
-
-### Caching
-
-The server implements caching for read operations, which significantly improves response times for repeated requests. The average cache improvement is over 70%.
-
-Frontend considerations:
-No need to implement client-side caching for data that doesn't change frequently
-For real-time data, consider adding a refresh button in the UI
-If data might be updated by other users, consider adding a polling mechanism or websocket for updates
-
-
-### Error Handling
-
-The API returns standardized error responses. Example:
-
-async function fetchData(url) {
-  try {
-    const response = await fetch(url);
-    if (!response.ok) {
-      const errorData = await response.json();
-      throw new Error(errorData.detail || 'An error occurred');
-    }
-    return await response.json();
-  } catch (error) {
-    console.error('API request failed:', error);
-    // Handle error in UI
-    return null;
-  }
-}
-
-### Price Field Handling
-
-The price field of events is a JSON object with currency and amount properties. When creating or updating events, ensure this is properly formatted:
-
-// Correct format for price field
-const eventData = {
-  // other fields...
-  price: {
-    currency: 'USD',
-    amount: '10.99'
-  }
-};
-
-// When displaying price
-function formatPrice(price) {
-  if (!price) return 'Free';
-  if (typeof price === 'string') {
-    try {
-      price = JSON.parse(price);
-    } catch {
-      return price;
-    }
-  }
-  return `${price.amount} ${price.currency}`;
-}
-
-## CORS Configuration
-
-The API has CORS enabled for frontend applications. If you're experiencing CORS issues, ensure your frontend domain is allowed in the server configuration.
-
-For local development, the following origins are typically allowed:
-- http://localhost:3000
-- http://localhost:5000
-- http://localhost:8080
-
-## Status Codes
-
-| Status Code | Description |
-|-------------|-------------|
-| 200 | Success - The request was successful |
-| 201 | Created - A new resource was successfully created |
-| 400 | Bad Request - The request could not be understood or was missing required parameters |
-| 404 | Not Found - Resource not found |
-| 422 | Validation Error - Request data failed validation |
-| 500 | Internal Server Error - An error occurred on the server |
-
-## Questions?
-
-For further inquiries about the API, please contact the development team.

app.py

@@ -64,11 +64,9 @@ async def lifespan(app: FastAPI):
     # Startup: kiểm tra kết nối các database
     logger.info("Starting application...")
     db_status = check_database_connections()
-    if all(db_status.values()):
-        logger.info("All database connections are working")
     
     # Khởi tạo bảng trong cơ sở dữ liệu (nếu chưa tồn tại)
-    if DEBUG:  # Chỉ khởi tạo bảng trong chế độ debug
+    if DEBUG and all(db_status.values()):  # Chỉ khởi tạo bảng trong chế độ debug và khi tất cả kết nối DB thành công
         from app.database.postgresql import create_tables
         if create_tables():
             logger.info("Database tables created or already exist")
@@ -84,6 +82,7 @@ try:
     from app.api.postgresql_routes import router as postgresql_router
     from app.api.rag_routes import router as rag_router
     from app.api.websocket_routes import router as websocket_router
+    from app.api.pdf_routes import router as pdf_router
     
     # Import middlewares
     from app.utils.middleware import RequestLoggingMiddleware, ErrorHandlingMiddleware, DatabaseCheckMiddleware
@@ -91,6 +90,9 @@ try:
     # Import debug utilities
     from app.utils.debug_utils import debug_view, DebugInfo, error_tracker, performance_monitor
     
+    # Import cache
+    from app.utils.cache import get_cache
+    
 except ImportError as e:
     logger.error(f"Error importing routes or middlewares: {e}")
     raise
@@ -126,6 +128,7 @@ app.include_router(mongodb_router)
 app.include_router(postgresql_router)
 app.include_router(rag_router)
 app.include_router(websocket_router)
+app.include_router(pdf_router)
 
 # Root endpoint
 @app.get("/")
@@ -149,6 +152,25 @@ def health_check():
         "databases": db_status
     }
 
+@app.get("/api/ping")
+async def ping():
+    return {"status": "pong"}
+
+# Cache stats endpoint
+@app.get("/cache/stats")
+def cache_stats():
+    """Trả về thống kê về cache"""
+    cache = get_cache()
+    return cache.stats()
+
+# Cache clear endpoint
+@app.delete("/cache/clear")
+def cache_clear():
+    """Xóa tất cả dữ liệu trong cache"""
+    cache = get_cache()
+    cache.clear()
+    return {"message": "Cache cleared successfully"}
+
 # Debug endpoints (chỉ có trong chế độ debug)
 if DEBUG:
     @app.get("/debug/config")
@@ -190,6 +212,29 @@ if DEBUG:
     def debug_full_report(request: Request):
         """Hiển thị báo cáo debug đầy đủ (chỉ trong chế độ debug)"""
         return debug_view(request)
+    
+    @app.get("/debug/cache")
+    def debug_cache():
+        """Hiển thị thông tin chi tiết về cache (chỉ trong chế độ debug)"""
+        cache = get_cache()
+        cache_stats = cache.stats()
+        
+        # Thêm thông tin chi tiết về các key trong cache
+        cache_keys = list(cache.cache.keys())
+        history_users = list(cache.user_history_queues.keys())
+        
+        return {
+            "stats": cache_stats,
+            "keys": cache_keys,
+            "history_users": history_users,
+            "config": {
+                "ttl": cache.ttl,
+                "cleanup_interval": cache.cleanup_interval,
+                "max_size": cache.max_size,
+                "history_queue_size": os.getenv("HISTORY_QUEUE_SIZE", "10"),
+                "history_cache_ttl": os.getenv("HISTORY_CACHE_TTL", "3600"),
+            }
+        }
 
 # Run the app with uvicorn when executed directly
 if __name__ == "__main__":

app/__init__.py

@@ -11,7 +11,9 @@ import os
 sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
 try:
-    from app.py import app
+    # Sửa lại cách import đúng - 'app.py' không phải là module hợp lệ
+    # 'app' là tên module, '.py' là phần mở rộng tệp
+    from app import app
 except ImportError:
     # Thử cách khác nếu import trực tiếp không hoạt động
     import importlib.util

app/api/mongodb_routes.py

@@ -8,7 +8,7 @@ import asyncio
 
 from app.database.mongodb import (
     save_session, 
-    get_user_history,
+    get_chat_history,
     update_session_response,
     check_db_connection,
     session_collection
@@ -178,7 +178,7 @@ async def get_history(user_id: str, n: int = Query(3, ge=1, le=10)):
             )
         
         # Get user history from MongoDB
-        history_data = get_user_history(user_id=user_id, n=n)
+        history_data = get_chat_history(user_id=user_id, n=n)
         
         # Convert to response model
         return HistoryResponse(history=history_data)

app/api/pdf_routes.py

@@ -0,0 +1,233 @@
+import os
+import shutil
+import uuid
+from fastapi import APIRouter, UploadFile, File, Form, HTTPException, BackgroundTasks
+from fastapi.responses import JSONResponse
+from typing import Optional, List, Dict, Any
+
+from app.utils.pdf_processor import PDFProcessor
+from app.models.pdf_models import PDFResponse, DeleteDocumentRequest, DocumentsListResponse
+from app.api.pdf_websocket import (
+    send_pdf_upload_started, 
+    send_pdf_upload_progress, 
+    send_pdf_upload_completed,
+    send_pdf_upload_failed,
+    send_pdf_delete_started,
+    send_pdf_delete_completed,
+    send_pdf_delete_failed
+)
+
+# Khởi tạo router
+router = APIRouter(
+    prefix="/pdf",
+    tags=["PDF Processing"],
+)
+
+# Thư mục lưu file tạm - sử dụng /tmp để tránh lỗi quyền truy cập
+TEMP_UPLOAD_DIR = "/tmp/uploads/temp"
+STORAGE_DIR = "/tmp/uploads/pdfs"
+
+# Đảm bảo thư mục upload tồn tại
+os.makedirs(TEMP_UPLOAD_DIR, exist_ok=True)
+os.makedirs(STORAGE_DIR, exist_ok=True)
+
+# Endpoint upload và xử lý PDF
+@router.post("/upload", response_model=PDFResponse)
+async def upload_pdf(
+    file: UploadFile = File(...),
+    namespace: str = Form("Default"),
+    index_name: str = Form("testbot768"),
+    title: Optional[str] = Form(None),
+    description: Optional[str] = Form(None),
+    user_id: Optional[str] = Form(None),
+    background_tasks: BackgroundTasks = None
+):
+    """
+    Upload và xử lý file PDF để tạo embeddings và lưu vào Pinecone
+    
+    - **file**: File PDF cần xử lý
+    - **namespace**: Namespace trong Pinecone để lưu embeddings (mặc định: "Default")
+    - **index_name**: Tên index Pinecone (mặc định: "testbot768")
+    - **title**: Tiêu đề của tài liệu (tùy chọn)
+    - **description**: Mô tả về tài liệu (tùy chọn)
+    - **user_id**: ID của người dùng để cập nhật trạng thái qua WebSocket
+    """
+    try:
+        # Kiểm tra file có phải PDF không
+        if not file.filename.lower().endswith('.pdf'):
+            raise HTTPException(status_code=400, detail="Chỉ chấp nhận file PDF")
+        
+        # Tạo file_id và lưu file tạm
+        file_id = str(uuid.uuid4())
+        temp_file_path = os.path.join(TEMP_UPLOAD_DIR, f"{file_id}.pdf")
+        
+        # Gửi thông báo bắt đầu xử lý qua WebSocket nếu có user_id
+        if user_id:
+            await send_pdf_upload_started(user_id, file.filename, file_id)
+        
+        # Lưu file
+        with open(temp_file_path, "wb") as buffer:
+            shutil.copyfileobj(file.file, buffer)
+            
+        # Tạo metadata
+        metadata = {
+            "filename": file.filename,
+            "content_type": file.content_type
+        }
+        
+        if title:
+            metadata["title"] = title
+        if description:
+            metadata["description"] = description
+        
+        # Gửi thông báo tiến độ qua WebSocket
+        if user_id:
+            await send_pdf_upload_progress(
+                user_id, 
+                file_id, 
+                "file_preparation", 
+                0.2, 
+                "File saved, preparing for processing"
+            )
+            
+        # Khởi tạo PDF processor
+        processor = PDFProcessor(index_name=index_name, namespace=namespace)
+        
+        # Gửi thông báo bắt đầu embedding qua WebSocket
+        if user_id:
+            await send_pdf_upload_progress(
+                user_id, 
+                file_id, 
+                "embedding_start", 
+                0.4, 
+                "Starting to process PDF and create embeddings"
+            )
+        
+        # Xử lý PDF và tạo embeddings
+        # Tạo callback function để xử lý cập nhật tiến độ
+        async def progress_callback_wrapper(step, progress, message):
+            if user_id:
+                await send_progress_update(user_id, file_id, step, progress, message)
+        
+        # Xử lý PDF và tạo embeddings với callback đã được xử lý đúng cách
+        result = await processor.process_pdf(
+            file_path=temp_file_path,
+            document_id=file_id,
+            metadata=metadata,
+            progress_callback=progress_callback_wrapper
+        )
+        
+        # Nếu thành công, chuyển file vào storage
+        if result.get('success'):
+            storage_path = os.path.join(STORAGE_DIR, f"{file_id}.pdf")
+            shutil.move(temp_file_path, storage_path)
+            
+            # Gửi thông báo hoàn thành qua WebSocket
+            if user_id:
+                await send_pdf_upload_completed(
+                    user_id,
+                    file_id,
+                    file.filename,
+                    result.get('chunks_processed', 0)
+                )
+        else:
+            # Gửi thông báo lỗi qua WebSocket
+            if user_id:
+                await send_pdf_upload_failed(
+                    user_id,
+                    file_id,
+                    file.filename,
+                    result.get('error', 'Unknown error')
+                )
+            
+        # Dọn dẹp: xóa file tạm nếu vẫn còn
+        if os.path.exists(temp_file_path):
+            os.remove(temp_file_path)
+            
+        return result
+    except Exception as e:
+        # Dọn dẹp nếu có lỗi
+        if 'temp_file_path' in locals() and os.path.exists(temp_file_path):
+            os.remove(temp_file_path)
+            
+        # Gửi thông báo lỗi qua WebSocket
+        if 'user_id' in locals() and user_id and 'file_id' in locals():
+            await send_pdf_upload_failed(
+                user_id,
+                file_id,
+                file.filename,
+                str(e)
+            )
+            
+        return PDFResponse(
+            success=False,
+            error=str(e)
+        )
+
+# Function để gửi cập nhật tiến độ - được sử dụng trong callback
+async def send_progress_update(user_id, document_id, step, progress, message):
+    if user_id:
+        await send_pdf_upload_progress(user_id, document_id, step, progress, message)
+
+# Endpoint xóa tài liệu
+@router.delete("/namespace", response_model=PDFResponse)
+async def delete_namespace(
+    namespace: str = "Default",
+    index_name: str = "testbot768",
+    user_id: Optional[str] = None
+):
+    """
+    Xóa toàn bộ embeddings trong một namespace từ Pinecone (tương ứng xoá namespace)
+
+    - **namespace**: Namespace trong Pinecone (mặc định: "Default")
+    - **index_name**: Tên index Pinecone (mặc định: "testbot768")
+    - **user_id**: ID của người dùng để cập nhật trạng thái qua WebSocket
+    """
+    try:
+        # Gửi thông báo bắt đầu xóa qua WebSocket
+        if user_id:
+            await send_pdf_delete_started(user_id, namespace)
+            
+        processor = PDFProcessor(index_name=index_name, namespace=namespace)
+        result = await processor.delete_namespace()
+        
+        # Gửi thông báo kết quả qua WebSocket
+        if user_id:
+            if result.get('success'):
+                await send_pdf_delete_completed(user_id, namespace)
+            else:
+                await send_pdf_delete_failed(user_id, namespace, result.get('error', 'Unknown error'))
+                
+        return result
+    except Exception as e:
+        # Gửi thông báo lỗi qua WebSocket
+        if user_id:
+            await send_pdf_delete_failed(user_id, namespace, str(e))
+            
+        return PDFResponse(
+            success=False,
+            error=str(e)
+        )
+
+# Endpoint lấy danh sách tài liệu
+@router.get("/documents", response_model=DocumentsListResponse)
+async def get_documents(namespace: str = "Default", index_name: str = "testbot768"):
+    """
+    Lấy thông tin về tất cả tài liệu đã được embed
+    
+    - **namespace**: Namespace trong Pinecone (mặc định: "Default")
+    - **index_name**: Tên index Pinecone (mặc định: "testbot768")
+    """
+    try:
+        # Khởi tạo PDF processor
+        processor = PDFProcessor(index_name=index_name, namespace=namespace)
+        
+        # Lấy danh sách documents
+        result = await processor.list_documents()
+        
+        return result
+    except Exception as e:
+        return DocumentsListResponse(
+            success=False,
+            error=str(e)
+        ) 

app/api/pdf_websocket.py

@@ -0,0 +1,263 @@
+import logging
+from typing import Dict, List, Optional, Any
+from fastapi import WebSocket, WebSocketDisconnect, APIRouter
+from pydantic import BaseModel
+import json
+import time
+
+# Cấu hình logging
+logger = logging.getLogger(__name__)
+
+# Models cho Swagger documentation
+class ConnectionStatus(BaseModel):
+    user_id: str
+    active: bool
+    connection_count: int
+    last_activity: Optional[float] = None
+
+class UserConnection(BaseModel):
+    user_id: str
+    connection_count: int
+
+class AllConnectionsStatus(BaseModel):
+    total_users: int
+    total_connections: int
+    users: List[UserConnection]
+
+# Khởi tạo router
+router = APIRouter(
+    prefix="/ws",
+    tags=["WebSockets"],
+)
+
+class ConnectionManager:
+    """Quản lý các kết nối WebSocket"""
+    
+    def __init__(self):
+        # Lưu trữ các kết nối theo user_id
+        self.active_connections: Dict[str, List[WebSocket]] = {}
+        
+    async def connect(self, websocket: WebSocket, user_id: str):
+        """Kết nối một WebSocket mới"""
+        await websocket.accept()
+        if user_id not in self.active_connections:
+            self.active_connections[user_id] = []
+        self.active_connections[user_id].append(websocket)
+        logger.info(f"New WebSocket connection for user {user_id}. Total connections: {len(self.active_connections[user_id])}")
+        
+    def disconnect(self, websocket: WebSocket, user_id: str):
+        """Ngắt kết nối WebSocket"""
+        if user_id in self.active_connections:
+            if websocket in self.active_connections[user_id]:
+                self.active_connections[user_id].remove(websocket)
+            # Xóa user_id khỏi dict nếu không còn kết nối nào
+            if not self.active_connections[user_id]:
+                del self.active_connections[user_id]
+        logger.info(f"WebSocket disconnected for user {user_id}")
+        
+    async def send_message(self, message: Dict[str, Any], user_id: str):
+        """Gửi tin nhắn tới tất cả kết nối của một user"""
+        if user_id in self.active_connections:
+            disconnected_websockets = []
+            for websocket in self.active_connections[user_id]:
+                try:
+                    await websocket.send_text(json.dumps(message))
+                except Exception as e:
+                    logger.error(f"Error sending message to WebSocket: {str(e)}")
+                    disconnected_websockets.append(websocket)
+            
+            # Xóa các kết nối bị ngắt
+            for websocket in disconnected_websockets:
+                self.disconnect(websocket, user_id)
+    
+    def get_connection_status(self, user_id: str = None) -> Dict[str, Any]:
+        """Lấy thông tin về trạng thái kết nối WebSocket"""
+        if user_id:
+            # Trả về thông tin kết nối cho user cụ thể
+            if user_id in self.active_connections:
+                return {
+                    "user_id": user_id,
+                    "active": True,
+                    "connection_count": len(self.active_connections[user_id]),
+                    "last_activity": time.time()
+                }
+            else:
+                return {
+                    "user_id": user_id,
+                    "active": False,
+                    "connection_count": 0,
+                    "last_activity": None
+                }
+        else:
+            # Trả về thông tin tất cả kết nối
+            result = {
+                "total_users": len(self.active_connections),
+                "total_connections": sum(len(connections) for connections in self.active_connections.values()),
+                "users": []
+            }
+            
+            for uid, connections in self.active_connections.items():
+                result["users"].append({
+                    "user_id": uid,
+                    "connection_count": len(connections)
+                })
+            
+            return result
+
+
+# Tạo instance của ConnectionManager
+manager = ConnectionManager()
+
+@router.websocket("/pdf/{user_id}")
+async def websocket_endpoint(websocket: WebSocket, user_id: str):
+    """Endpoint WebSocket để cập nhật tiến trình xử lý PDF"""
+    await manager.connect(websocket, user_id)
+    try:
+        while True:
+            # Đợi tin nhắn từ client (chỉ để giữ kết nối)
+            await websocket.receive_text()
+    except WebSocketDisconnect:
+        manager.disconnect(websocket, user_id)
+    except Exception as e:
+        logger.error(f"WebSocket error: {str(e)}")
+        manager.disconnect(websocket, user_id)
+
+# API endpoints để kiểm tra trạng thái WebSocket
+@router.get("/status", response_model=AllConnectionsStatus, responses={
+    200: {
+        "description": "Successful response",
+        "content": {
+            "application/json": {
+                "example": {
+                    "total_users": 2,
+                    "total_connections": 3,
+                    "users": [
+                        {"user_id": "user1", "connection_count": 2},
+                        {"user_id": "user2", "connection_count": 1}
+                    ]
+                }
+            }
+        }
+    }
+})
+async def get_all_websocket_connections():
+    """
+    Lấy thông tin về tất cả kết nối WebSocket hiện tại.
+    
+    Endpoint này trả về:
+    - Tổng số người dùng đang kết nối
+    - Tổng số kết nối WebSocket
+    - Danh sách người dùng kèm theo số lượng kết nối của mỗi người
+    """
+    return manager.get_connection_status()
+
+@router.get("/status/{user_id}", response_model=ConnectionStatus, responses={
+    200: {
+        "description": "Successful response for active connection",
+        "content": {
+            "application/json": {
+                "examples": {
+                    "active_connection": {
+                        "summary": "Active connection",
+                        "value": {
+                            "user_id": "user123",
+                            "active": True,
+                            "connection_count": 2,
+                            "last_activity": 1634567890.123
+                        }
+                    },
+                    "no_connection": {
+                        "summary": "No active connection",
+                        "value": {
+                            "user_id": "user456",
+                            "active": False,
+                            "connection_count": 0,
+                            "last_activity": None
+                        }
+                    }
+                }
+            }
+        }
+    }
+})
+async def get_user_websocket_status(user_id: str):
+    """
+    Lấy thông tin về kết nối WebSocket của một người dùng cụ thể.
+    
+    Parameters:
+    - **user_id**: ID của người dùng cần kiểm tra
+    
+    Returns:
+    - Thông tin về trạng thái kết nối, bao gồm:
+      - active: Có đang kết nối hay không
+      - connection_count: Số lượng kết nối hiện tại
+      - last_activity: Thời gian hoạt động gần nhất
+    """
+    return manager.get_connection_status(user_id)
+
+# Các hàm gửi thông báo cập nhật trạng thái
+
+async def send_pdf_upload_started(user_id: str, filename: str, document_id: str):
+    """Gửi thông báo bắt đầu upload PDF"""
+    await manager.send_message({
+        "type": "pdf_upload_started",
+        "document_id": document_id,
+        "filename": filename,
+        "timestamp": int(time.time())
+    }, user_id)
+
+async def send_pdf_upload_progress(user_id: str, document_id: str, step: str, progress: float, message: str):
+    """Gửi thông báo tiến độ upload PDF"""
+    await manager.send_message({
+        "type": "pdf_upload_progress",
+        "document_id": document_id,
+        "step": step,
+        "progress": progress,
+        "message": message,
+        "timestamp": int(time.time())
+    }, user_id)
+
+async def send_pdf_upload_completed(user_id: str, document_id: str, filename: str, chunks: int):
+    """Gửi thông báo hoàn thành upload PDF"""
+    await manager.send_message({
+        "type": "pdf_upload_completed",
+        "document_id": document_id,
+        "filename": filename,
+        "chunks": chunks,
+        "timestamp": int(time.time())
+    }, user_id)
+
+async def send_pdf_upload_failed(user_id: str, document_id: str, filename: str, error: str):
+    """Gửi thông báo lỗi upload PDF"""
+    await manager.send_message({
+        "type": "pdf_upload_failed",
+        "document_id": document_id,
+        "filename": filename,
+        "error": error,
+        "timestamp": int(time.time())
+    }, user_id)
+
+async def send_pdf_delete_started(user_id: str, namespace: str):
+    """Gửi thông báo bắt đầu xóa PDF"""
+    await manager.send_message({
+        "type": "pdf_delete_started",
+        "namespace": namespace,
+        "timestamp": int(time.time())
+    }, user_id)
+
+async def send_pdf_delete_completed(user_id: str, namespace: str):
+    """Gửi thông báo hoàn thành xóa PDF"""
+    await manager.send_message({
+        "type": "pdf_delete_completed",
+        "namespace": namespace,
+        "timestamp": int(time.time())
+    }, user_id)
+
+async def send_pdf_delete_failed(user_id: str, namespace: str, error: str):
+    """Gửi thông báo lỗi xóa PDF"""
+    await manager.send_message({
+        "type": "pdf_delete_failed",
+        "namespace": namespace,
+        "error": error,
+        "timestamp": int(time.time())
+    }, user_id)

app/api/rag_routes.py

@@ -11,9 +11,9 @@ import google.generativeai as genai
 from datetime import datetime
 from langchain.prompts import PromptTemplate
 from langchain_google_genai import GoogleGenerativeAIEmbeddings
-from app.utils.utils import cache, timer_decorator
+from app.utils.utils import timer_decorator
 
-from app.database.mongodb import get_user_history, get_chat_history, get_request_history, save_session, session_collection
+from app.database.mongodb import get_chat_history, get_request_history, session_collection
 from app.database.pinecone import (
     search_vectors, 
     get_chain, 
@@ -33,32 +33,6 @@ from app.models.rag_models import (
     UserMessageModel
 )
 
-# Sử dụng bộ nhớ đệm thay vì Redis
-class SimpleCache:
-    def __init__(self):
-        self.cache = {}
-        self.expiration = {}
-    
-    async def get(self, key):
-        if key in self.cache:
-            # Kiểm tra xem cache đã hết hạn chưa
-            if key in self.expiration and self.expiration[key] > time.time():
-                return self.cache[key]
-            else:
-                # Xóa cache đã hết hạn
-                if key in self.cache:
-                    del self.cache[key]
-                if key in self.expiration:
-                    del self.expiration[key]
-        return None
-    
-    async def set(self, key, value, ex=300):  # Mặc định 5 phút
-        self.cache[key] = value
-        self.expiration[key] = time.time() + ex
-
-# Khởi tạo SimpleCache
-redis_client = SimpleCache()
-
 # Configure logging
 logger = logging.getLogger(__name__)
 
@@ -72,6 +46,29 @@ router = APIRouter(
     tags=["RAG"],
 )
 
+fix_request = PromptTemplate(
+    template = """Goal:
+Your task is fixing user'srequest to get all information of history chat.
+You will received a conversation history and current request of user.
+Generate a new request that make sense if current request related to history conversation.
+
+Return Format:
+Only return the fully users' request with all the important keywords.
+If the current message is NOT related to the conversation history or there is no chat history: Return user's current request.
+If the current message IS related to the conversation history: Return new request based on information from the conversation history and the current request.
+
+Warning:
+Only use history chat if current request is truly relevant to the previous conversation.
+
+Conversation History:
+{chat_history}
+
+User current message:
+{question}
+""",
+    input_variables = ["chat_history", "question"],
+)
+
 # Create a prompt template with conversation history
 prompt = PromptTemplate(
     template = """Goal:
@@ -87,7 +84,7 @@ Warning:
 Let's support users like a real tour guide, not a bot. The information in core knowledge is your own knowledge.
 Your knowledge is provided in the Core Knowledge. All of information in Core Knowledge is about Da Nang, Vietnam.
 You just care about current time that user mention when user ask about Solana event.
-If you do not have enough information to answer user's question, please reply with "I don't know. I don't have information about that".
+Only use core knowledge to answer. If you do not have enough information to answer user's question, please reply with "I'm sorry. I don't have information about that" and Give users some more options to ask.
 
 Core knowledge:
 {context}
@@ -162,102 +159,18 @@ async def chat(request: ChatRequest, background_tasks: BackgroundTasks):
     """
     start_time = time.time()
     try:
-        # Create cache key for request
-        cache_key = f"rag_chat:{request.user_id}:{request.question}:{request.include_history}:{request.use_rag}:{request.similarity_top_k}:{request.limit_k}:{request.similarity_metric}:{request.similarity_threshold}"
-        
-        # Check cache using redis_client instead of cache
-        cached_response = await redis_client.get(cache_key)
-        if cached_response is not None:
-            logger.info(f"Cache hit for RAG chat request from user {request.user_id}")
-            try:
-                # If cached_response is string (JSON), parse it
-                if isinstance(cached_response, str):
-                    cached_data = json.loads(cached_response)
-                    return ChatResponse(
-                        answer=cached_data.get("answer", ""),
-                        processing_time=cached_data.get("processing_time", 0.0)
-                    )
-                # If cached_response is object with sources, extract answer and processing_time
-                elif hasattr(cached_response, 'sources'):
-                    return ChatResponse(
-                        answer=cached_response.answer,
-                        processing_time=cached_response.processing_time
-                    )
-                # Otherwise, return cached response as is
-                return cached_response
-            except Exception as e:
-                logger.error(f"Error parsing cached response: {e}")
-                # Continue processing if cache parsing fails
-        
         # Save user message first (so it's available for user history)
         session_id = request.session_id or f"{request.user_id}_{datetime.now().strftime('%Y-%m-%d_%H:%M:%S')}"
-        logger.info(f"Processing chat request for user {request.user_id}, session {session_id}")
-        
-        # First, save the user's message so it's available for history lookups
-        try:
-            # Save user's question
-            save_session(
-                session_id=session_id,
-                factor="user",
-                action="asking_freely",
-                first_name=getattr(request, 'first_name', "User"),
-                last_name=getattr(request, 'last_name', ""),
-                message=request.question,
-                user_id=request.user_id,
-                username=getattr(request, 'username', ""),
-                response=None  # No response yet
-            )
-            logger.info(f"User message saved for session {session_id}")
-        except Exception as e:
-            logger.error(f"Error saving user message to session: {e}")
-            # Continue processing even if saving fails
-        
-        # Use the RAG pipeline
-        if request.use_rag:
-            # Get the retriever with custom parameters
-            retriever = get_chain(
-                top_k=request.similarity_top_k,
-                limit_k=request.limit_k,
-                similarity_metric=request.similarity_metric,
-                similarity_threshold=request.similarity_threshold
-            )
-            if not retriever:
-                raise HTTPException(status_code=500, detail="Failed to initialize retriever")
-            
-            # Get request history for context
-            context_query = get_request_history(request.user_id) if request.include_history else request.question
-            logger.info(f"Using context query for retrieval: {context_query[:100]}...")
-            
-            # Retrieve relevant documents
-            retrieved_docs = retriever.invoke(context_query)
-            context = "\n".join([doc.page_content for doc in retrieved_docs])
-            
-            # Prepare sources
-            sources = []
-            for doc in retrieved_docs:
-                source = None
-                metadata = {}
-                
-                if hasattr(doc, 'metadata'):
-                    source = doc.metadata.get('source', None)
-                    # Extract score information
-                    score = doc.metadata.get('score', None)
-                    normalized_score = doc.metadata.get('normalized_score', None)
-                    # Remove score info from metadata to avoid duplication
-                    metadata = {k: v for k, v in doc.metadata.items() 
-                               if k not in ['text', 'source', 'score', 'normalized_score']}
-                
-                sources.append(SourceDocument(
-                    text=doc.page_content,
-                    source=source,
-                    score=score,
-                    normalized_score=normalized_score,
-                    metadata=metadata
-                ))
-        else:
-            # No RAG
-            context = ""
-            sources = None
+        # logger.info(f"Processing chat request for user {request.user_id}, session {session_id}")
+
+        retriever = get_chain(
+            top_k=request.similarity_top_k,
+            limit_k=request.limit_k,
+            similarity_metric=request.similarity_metric,
+            similarity_threshold=request.similarity_threshold
+        )
+        if not retriever:
+            raise HTTPException(status_code=500, detail="Failed to initialize retriever")
         
         # Get chat history
         chat_history = get_chat_history(request.user_id) if request.include_history else ""
@@ -295,11 +208,50 @@ async def chat(request: ChatRequest, background_tasks: BackgroundTasks):
             generation_config=generation_config,
             safety_settings=safety_settings
         )
+
+        prompt_request = fix_request.format(
+            question=request.question,
+            chat_history=chat_history
+        )
+        
+        # Log thời gian bắt đầu final_request
+        final_request_start_time = time.time()
+        final_request = model.generate_content(prompt_request)
+        # Log thời gian hoàn thành final_request
+        logger.info(f"Fixed Request: {final_request.text}")
+        logger.info(f"Final request generation time: {time.time() - final_request_start_time:.2f} seconds")
+        # print(final_request.text)
+
+        retrieved_docs = retriever.invoke(final_request.text)
+        logger.info(f"Retrieve: {retrieved_docs}")
+        context = "\n".join([doc.page_content for doc in retrieved_docs])
+
+        sources = []
+        for doc in retrieved_docs:
+            source = None
+            metadata = {}
+            
+            if hasattr(doc, 'metadata'):
+                source = doc.metadata.get('source', None)
+                # Extract score information
+                score = doc.metadata.get('score', None)
+                normalized_score = doc.metadata.get('normalized_score', None)
+                # Remove score info from metadata to avoid duplication
+                metadata = {k: v for k, v in doc.metadata.items() 
+                            if k not in ['text', 'source', 'score', 'normalized_score']}
+            
+            sources.append(SourceDocument(
+                text=doc.page_content,
+                source=source,
+                score=score,
+                normalized_score=normalized_score,
+                metadata=metadata
+            ))
         
         # Generate the prompt using template
         prompt_text = prompt.format(
             context=context,
-            question=request.question,
+            question=final_request.text,
             chat_history=chat_history
         )
         logger.info(f"Full prompt with history and context: {prompt_text}")
@@ -308,59 +260,11 @@ async def chat(request: ChatRequest, background_tasks: BackgroundTasks):
         response = model.generate_content(prompt_text)
         answer = response.text
         
-        # Save the RAG response
-        try:
-            # Now save the RAG response with the same session_id
-            save_session(
-                session_id=session_id,
-                factor="rag",
-                action="RAG_response",
-                first_name=getattr(request, 'first_name', "User"),
-                last_name=getattr(request, 'last_name', ""),
-                message=request.question,
-                user_id=request.user_id,
-                username=getattr(request, 'username', ""),
-                response=answer
-            )
-            logger.info(f"RAG response saved for session {session_id}")
-            
-            # Check if the response starts with "I don't know" and trigger notification
-            if answer.strip().lower().startswith("i don't know"):
-                from app.api.websocket_routes import send_notification
-                notification_data = {
-                    "session_id": session_id,
-                    "factor": "rag",
-                    "action": "RAG_response",
-                    "message": request.question,
-                    "user_id": request.user_id,
-                    "username": getattr(request, 'username', ""),
-                    "first_name": getattr(request, 'first_name', "User"),
-                    "last_name": getattr(request, 'last_name', ""),
-                    "response": answer,
-                    "created_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
-                }
-                background_tasks.add_task(send_notification, notification_data)
-                logger.info(f"Notification queued for session {session_id} - response starts with 'I don't know'")
-        except Exception as e:
-            logger.error(f"Error saving RAG response to session: {e}")
-            # Continue processing even if saving fails
-        
         # Calculate processing time
         processing_time = time.time() - start_time
         
-        # Create internal response object with sources for logging
-        internal_response = ChatResponseInternal(
-            answer=answer,
-            sources=sources,
-            processing_time=processing_time
-        )
-        
         # Log full response with sources
-        logger.info(f"Generated response for user {request.user_id}: {answer}")
-        if sources:
-            logger.info(f"Sources used: {len(sources)} documents")
-            for i, source in enumerate(sources):
-                logger.info(f"Source {i+1}: {source.source or 'Unknown'} (score: {source.score})")
+        # logger.info(f"Generated response for user {request.user_id}: {answer}")
         
         # Create response object for API (without sources)
         chat_response = ChatResponse(
@@ -368,18 +272,6 @@ async def chat(request: ChatRequest, background_tasks: BackgroundTasks):
             processing_time=processing_time
         )
         
-        # Cache result using redis_client instead of cache
-        try:
-            # Convert to JSON to ensure it can be cached
-            cache_data = {
-                "answer": answer,
-                "processing_time": processing_time
-            }
-            await redis_client.set(cache_key, json.dumps(cache_data), ex=300)
-        except Exception as e:
-            logger.error(f"Error caching response: {e}")
-            # Continue even if caching fails
-        
         # Return response
         return chat_response
     except Exception as e:
@@ -443,96 +335,4 @@ async def health_check():
         "services": services,
         "retrieval_config": retrieval_config,
         "timestamp": datetime.now().isoformat()
-    }
-
-@router.post("/rag")
-async def process_rag(request: Request, user_data: UserMessageModel, background_tasks: BackgroundTasks):
-    """
-    Process a user message through the RAG pipeline and return a response.
-    
-    Parameters:
-    - **user_id**: User ID from the client application
-    - **session_id**: Session ID for tracking the conversation
-    - **message**: User's message/question
-    - **similarity_top_k**: (Optional) Number of top similar documents to return after filtering
-    - **limit_k**: (Optional) Maximum number of documents to retrieve from vector store
-    - **similarity_metric**: (Optional) Similarity metric to use (cosine, dotproduct, euclidean)
-    - **similarity_threshold**: (Optional) Threshold for vector similarity (0-1)
-    """
-    try:
-        # Extract request data
-        user_id = user_data.user_id
-        session_id = user_data.session_id
-        message = user_data.message
-        
-        # Extract retrieval parameters (use defaults if not provided)
-        top_k = user_data.similarity_top_k or DEFAULT_TOP_K
-        limit_k = user_data.limit_k or DEFAULT_LIMIT_K
-        similarity_metric = user_data.similarity_metric or DEFAULT_SIMILARITY_METRIC
-        similarity_threshold = user_data.similarity_threshold or DEFAULT_SIMILARITY_THRESHOLD
-        
-        logger.info(f"RAG request received for user_id={user_id}, session_id={session_id}")
-        logger.info(f"Message: {message[:100]}..." if len(message) > 100 else f"Message: {message}")
-        logger.info(f"Retrieval parameters: top_k={top_k}, limit_k={limit_k}, metric={similarity_metric}, threshold={similarity_threshold}")
-        
-        # Create a cache key for this request to avoid reprocessing identical questions
-        cache_key = f"rag_{user_id}_{session_id}_{hashlib.md5(message.encode()).hexdigest()}_{top_k}_{limit_k}_{similarity_metric}_{similarity_threshold}"
-        
-        # Check if we have this response cached
-        cached_result = await redis_client.get(cache_key)
-        if cached_result:
-            logger.info(f"Cache hit for key: {cache_key}")
-            if isinstance(cached_result, str):  # If stored as JSON string
-                return json.loads(cached_result)
-            return cached_result
-        
-        # Save user message to MongoDB
-        try:
-            # Save user's question
-            save_session(
-                session_id=session_id,
-                factor="user",
-                action="asking_freely",
-                first_name="User",  # You can update this with actual data if available
-                last_name="",
-                message=message,
-                user_id=user_id,
-                username="",
-                response=None  # No response yet
-            )
-            logger.info(f"User message saved to MongoDB with session_id: {session_id}")
-        except Exception as e:
-            logger.error(f"Error saving user message: {e}")
-            # Continue anyway to try to get a response
-        
-        # Create a ChatRequest object to reuse the existing chat endpoint
-        chat_request = ChatRequest(
-            user_id=user_id,
-            question=message,
-            include_history=True,
-            use_rag=True,
-            similarity_top_k=top_k,
-            limit_k=limit_k,
-            similarity_metric=similarity_metric,
-            similarity_threshold=similarity_threshold,
-            session_id=session_id
-        )
-        
-        # Process through the chat endpoint
-        response = await chat(chat_request, background_tasks)
-        
-        # Cache the response
-        try:
-            await redis_client.set(cache_key, json.dumps({
-                "answer": response.answer,
-                "processing_time": response.processing_time
-            }))
-            logger.info(f"Cached response for key: {cache_key}")
-        except Exception as e:
-            logger.error(f"Failed to cache response: {e}")
-        
-        return response
-    except Exception as e:
-        logger.error(f"Error processing RAG request: {e}")
-        logger.error(traceback.format_exc())
-        raise HTTPException(status_code=500, detail=f"Error processing request: {str(e)}") 
+    }

app/database/models.py

@@ -25,6 +25,8 @@ class EmergencyItem(Base):
     location = Column(String, nullable=True)  # Will be converted to/from PostGIS POINT type
     priority = Column(Integer, default=0)
     is_active = Column(Boolean, default=True)
+    section = Column(String, nullable=True)  # Section field (16.1, 16.2.1, 16.2.2, 16.3)
+    section_id = Column(Integer, nullable=True)  # Numeric identifier for section
     created_at = Column(DateTime, server_default=func.now())
     updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now())
 
@@ -39,11 +41,36 @@ class EventItem(Base):
     date_start = Column(DateTime, nullable=False)
     date_end = Column(DateTime, nullable=True)
     price = Column(JSON, nullable=True)
+    url = Column(String, nullable=True)
     is_active = Column(Boolean, default=True)
     featured = Column(Boolean, default=False)
     created_at = Column(DateTime, server_default=func.now())
     updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now())
 
+class AboutPixity(Base):
+    __tablename__ = "about_pixity"
+    
+    id = Column(Integer, primary_key=True, index=True)
+    content = Column(Text, nullable=False)
+    created_at = Column(DateTime, server_default=func.now())
+    updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now())
+
+class SolanaSummit(Base):
+    __tablename__ = "solana_summit"
+    
+    id = Column(Integer, primary_key=True, index=True)
+    content = Column(Text, nullable=False)
+    created_at = Column(DateTime, server_default=func.now())
+    updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now())
+
+class DaNangBucketList(Base):
+    __tablename__ = "danang_bucket_list"
+    
+    id = Column(Integer, primary_key=True, index=True)
+    content = Column(Text, nullable=False)
+    created_at = Column(DateTime, server_default=func.now())
+    updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now())
+
 class VectorDatabase(Base):
     __tablename__ = "vector_database"
     

app/database/mongodb.py

@@ -20,6 +20,10 @@ COLLECTION_NAME = os.getenv("COLLECTION_NAME", "session_chat")
 # Set timeout for MongoDB connection
 MONGODB_TIMEOUT = int(os.getenv("MONGODB_TIMEOUT", "5000"))  # 5 seconds by default
 
+# Legacy cache settings - now only used for configuration purposes
+HISTORY_CACHE_TTL = int(os.getenv("HISTORY_CACHE_TTL", "3600"))  # 1 hour by default
+HISTORY_QUEUE_SIZE = int(os.getenv("HISTORY_QUEUE_SIZE", "10"))  # 10 items by default
+
 # Create MongoDB connection with timeout
 try:
     client = MongoClient(MONGODB_URL, serverSelectionTimeoutMS=MONGODB_TIMEOUT)
@@ -82,6 +86,7 @@ def save_session(session_id, factor, action, first_name, last_name, message, use
         }
         result = session_collection.insert_one(session_data)
         logger.info(f"Session saved with ID: {result.inserted_id}")
+        
         return {
             "acknowledged": result.acknowledged,
             "inserted_id": str(result.inserted_id),
@@ -94,15 +99,18 @@ def save_session(session_id, factor, action, first_name, last_name, message, use
 def update_session_response(session_id, response):
     """Update a session with response"""
     try:
+        # Lấy session hiện có
+        existing_session = session_collection.find_one({"session_id": session_id})
+        
+        if not existing_session:
+            logger.warning(f"No session found with ID: {session_id}")
+            return False
+        
         result = session_collection.update_one(
             {"session_id": session_id},
             {"$set": {"response": response}}
         )
         
-        if result.matched_count == 0:
-            logger.warning(f"No session found with ID: {session_id}")
-            return False
-            
         logger.info(f"Session {session_id} updated with response")
         return True
     except Exception as e:
@@ -112,80 +120,61 @@ def update_session_response(session_id, response):
 def get_recent_sessions(user_id, action, n=3):
     """Get n most recent sessions for a specific user and action"""
     try:
-        return list(
+        # Truy vấn trực tiếp từ MongoDB
+        result = list(
             session_collection.find(
                 {"user_id": user_id, "action": action},
                 {"_id": 0, "message": 1, "response": 1}
             ).sort("created_at_datetime", -1).limit(n)
         )
-    except Exception as e:
-        logger.error(f"Error getting recent sessions: {e}")
-        return []
-
-def get_user_history(user_id, n=3):
-    """Get user history for a specific user"""
-    try:
-        # Find all messages of this user
-        user_messages = list(
-            session_collection.find(
-                {
-                    "user_id": user_id, 
-                    "message": {"$exists": True, "$ne": None},
-                    # Include all user messages regardless of action type
-                }
-            ).sort("created_at_datetime", -1).limit(n * 2)  # Get more to ensure we have enough pairs
-        )
-        
-        # Group messages by session_id to find pairs
-        session_dict = {}
-        for msg in user_messages:
-            session_id = msg.get("session_id")
-            if session_id not in session_dict:
-                session_dict[session_id] = {}
-                
-            if msg.get("factor", "").lower() == "user":
-                session_dict[session_id]["question"] = msg.get("message", "")
-                session_dict[session_id]["timestamp"] = msg.get("created_at_datetime")
-            elif msg.get("factor", "").lower() == "rag":
-                session_dict[session_id]["answer"] = msg.get("response", "")
-                
-        # Build history from complete pairs only (with both question and answer)
-        history = []
-        for session_id, data in session_dict.items():
-            if "question" in data and "answer" in data and data.get("answer"):
-                history.append({
-                    "question": data["question"],
-                    "answer": data["answer"]
-                })
-        
-        # Sort by timestamp and limit to n
-        history = sorted(history, key=lambda x: x.get("timestamp", 0), reverse=True)[:n]
         
-        logger.info(f"Retrieved {len(history)} history items for user {user_id}")
-        return history
+        logger.debug(f"Retrieved {len(result)} recent sessions for user {user_id}, action {action}")
+        return result
     except Exception as e:
-        logger.error(f"Error getting user history: {e}")
+        logger.error(f"Error getting recent sessions: {e}")
         return []
 
-# Functions from chatbot.py
-def get_chat_history(user_id, n=5):
-    """Get conversation history for a specific user from MongoDB in format suitable for LLM prompt"""
+def get_chat_history(user_id, n = 5) -> str:
+    """
+    Lấy lịch sử chat cho user_id từ MongoDB và ghép thành chuỗi theo định dạng:
+    
+    User: ...
+    Bot: ...
+    User: ...
+    Bot: ...
+    """
     try:
-        history = get_user_history(user_id, n)
+        # Truy vấn các document có user_id, sắp xếp theo created_at tăng dần
+        # Get the 4 most recent documents first, then sort them in ascending order
+        docs = list(session_collection.find({"user_id": str(user_id)}).sort("created_at", -1).limit(n))
+        # Reverse the list to get chronological order (oldest to newest)
+        docs.reverse()
+        if not docs:
+            logger.info(f"Không tìm thấy dữ liệu cho user_id: {user_id}")
+            return ""
         
-        # Format history for prompt context
-        formatted_history = ""
-        for item in history:
-            formatted_history += f"User: {item['question']}\nAssistant: {item['answer']}\n\n"
+        conversation_lines = []
+        # Xử lý từng document theo cấu trúc mới
+        for doc in docs:
+            factor = doc.get("factor", "").lower()
+            action = doc.get("action", "").lower()
+            message = doc.get("message", "")
+            response = doc.get("response", "")
             
-        return formatted_history
+            if factor == "user" and action == "asking_freely":
+                conversation_lines.append(f"User: {message}")
+                conversation_lines.append(f"Bot: {response}")
+        
+        # Ghép các dòng thành chuỗi
+        return "\n".join(conversation_lines)
     except Exception as e:
-        logger.error(f"Error getting chat history for prompt: {e}")
+        logger.error(f"Lỗi khi lấy lịch sử chat cho user_id {user_id}: {e}")
         return ""
 
 def get_request_history(user_id, n=3):
     """Get the most recent user requests to use as context for retrieval"""
     try:
+        # Lấy lịch sử trực tiếp từ MongoDB (thông qua get_user_history đã sửa đổi)
         history = get_user_history(user_id, n)
         
         # Just extract the questions for context

app/database/pinecone.py

@@ -6,7 +6,6 @@ from typing import Optional, List, Dict, Any, Union, Tuple
 import time
 from langchain_google_genai import GoogleGenerativeAIEmbeddings
 import google.generativeai as genai
-from app.utils.utils import cache
 from langchain_core.retrievers import BaseRetriever
 from langchain.callbacks.manager import Callbacks
 from langchain_core.documents import Document
@@ -73,23 +72,39 @@ def init_pinecone():
         if pc is None:
             logger.info(f"Initializing Pinecone connection to index {PINECONE_INDEX_NAME}...")
             
+            # Check if API key and index name are set
+            if not PINECONE_API_KEY:
+                logger.error("PINECONE_API_KEY is not set in environment variables")
+                return None
+                
+            if not PINECONE_INDEX_NAME:
+                logger.error("PINECONE_INDEX_NAME is not set in environment variables")
+                return None
+            
             # Initialize Pinecone client using the new API
             pc = Pinecone(api_key=PINECONE_API_KEY)
             
-            # Check if index exists
-            index_list = pc.list_indexes()
-            
-            if not hasattr(index_list, 'names') or PINECONE_INDEX_NAME not in index_list.names():
-                logger.error(f"Index {PINECONE_INDEX_NAME} does not exist in Pinecone")
+            try:
+                # Check if index exists
+                index_list = pc.list_indexes()
+                
+                if not hasattr(index_list, 'names') or PINECONE_INDEX_NAME not in index_list.names():
+                    logger.error(f"Index {PINECONE_INDEX_NAME} does not exist in Pinecone")
+                    return None
+                
+                # Get existing index
+                index = pc.Index(PINECONE_INDEX_NAME)
+                logger.info(f"Pinecone connection established to index {PINECONE_INDEX_NAME}")
+            except Exception as connection_error:
+                logger.error(f"Error connecting to Pinecone index: {connection_error}")
                 return None
             
-            # Get existing index
-            index = pc.Index(PINECONE_INDEX_NAME)
-            logger.info(f"Pinecone connection established to index {PINECONE_INDEX_NAME}")
-            
         return index
+    except ImportError as e:
+        logger.error(f"Required package for Pinecone is missing: {e}")
+        return None
     except Exception as e:
-        logger.error(f"Error initializing Pinecone: {e}")
+        logger.error(f"Unexpected error initializing Pinecone: {e}")
         return None
 
 # Get Pinecone index singleton
@@ -184,7 +199,7 @@ async def search_vectors(
     limit_k: int = DEFAULT_LIMIT_K,
     similarity_metric: str = DEFAULT_SIMILARITY_METRIC,
     similarity_threshold: float = DEFAULT_SIMILARITY_THRESHOLD,
-    namespace: str = "", 
+    namespace: str = "Default", 
     filter: Optional[Dict] = None
 ) -> Dict:
     """
@@ -211,23 +226,13 @@ async def search_vectors(
         if limit_k < top_k:
             logger.warning(f"limit_k ({limit_k}) must be greater than or equal to top_k ({top_k}). Setting limit_k to {top_k}")
             limit_k = top_k
-            
-        # Create cache key from parameters
-        vector_hash = hash(str(query_vector))
-        cache_key = f"pinecone_search:{vector_hash}:{limit_k}:{similarity_metric}:{similarity_threshold}:{namespace}:{filter}"
-        
-        # Check cache first
-        cached_result = cache.get(cache_key)
-        if cached_result is not None:
-            logger.info("Returning cached Pinecone search results")
-            return cached_result
         
-        # If not in cache, perform search
+        # Perform search directly without cache
         pinecone_index = get_pinecone_index()
         if pinecone_index is None:
             logger.error("Failed to get Pinecone index for search")
             return None
-        
+            
         # Query Pinecone with the provided metric and higher limit_k to allow for threshold filtering
         results = pinecone_index.query(
             vector=query_vector,
@@ -250,10 +255,7 @@ async def search_vectors(
         
         # Log search result metrics
         match_count = len(filtered_matches)
-        logger.info(f"Pinecone search returned {match_count} matches after threshold filtering (metric: {similarity_metric}, threshold: {similarity_threshold})")
-        
-        # Store result in cache with 5 minute TTL
-        cache.set(cache_key, results, ttl=300)
+        logger.info(f"Pinecone search returned {match_count} matches after threshold filtering (metric: {similarity_metric}, threshold: {similarity_threshold}, namespace: {namespace})")
         
         return results
     except Exception as e:
@@ -261,7 +263,7 @@ async def search_vectors(
         return None
 
 # Upsert vectors to Pinecone
-async def upsert_vectors(vectors, namespace=""):
+async def upsert_vectors(vectors, namespace="Default"):
     """Upsert vectors to Pinecone index"""
     try:
         pinecone_index = get_pinecone_index()
@@ -284,7 +286,7 @@ async def upsert_vectors(vectors, namespace=""):
         return None
 
 # Delete vectors from Pinecone
-async def delete_vectors(ids, namespace=""):
+async def delete_vectors(ids, namespace="Default"):
     """Delete vectors from Pinecone index"""
     try:
         pinecone_index = get_pinecone_index()
@@ -304,7 +306,7 @@ async def delete_vectors(ids, namespace=""):
         return False
 
 # Fetch vector metadata from Pinecone
-async def fetch_metadata(ids, namespace=""):
+async def fetch_metadata(ids, namespace="Default"):
     """Fetch metadata for specific vector IDs"""
     try:
         pinecone_index = get_pinecone_index()
@@ -336,7 +338,8 @@ class ThresholdRetriever(BaseRetriever):
     limit_k: int = Field(default=DEFAULT_LIMIT_K, description="Maximum number of results to retrieve from Pinecone")
     similarity_metric: str = Field(default=DEFAULT_SIMILARITY_METRIC, description="Similarity metric to use")
     similarity_threshold: float = Field(default=DEFAULT_SIMILARITY_THRESHOLD, description="Threshold for similarity")
-    
+    namespace: str = "Default"
+
     class Config:
         """Configuration for this pydantic object."""
         arbitrary_types_allowed = True
@@ -347,7 +350,7 @@ class ThresholdRetriever(BaseRetriever):
         limit_k: int = DEFAULT_LIMIT_K,
         similarity_metric: str = DEFAULT_SIMILARITY_METRIC,
         similarity_threshold: float = DEFAULT_SIMILARITY_THRESHOLD,
-        namespace: str = "", 
+        namespace: str = "Default", 
         filter: Optional[Dict] = None
     ) -> Dict:
         """Synchronous wrapper for search_vectors"""
@@ -440,8 +443,8 @@ class ThresholdRetriever(BaseRetriever):
                     limit_k=self.limit_k,
                     similarity_metric=self.similarity_metric,
                     similarity_threshold=self.similarity_threshold,
-                    namespace=getattr(self.vectorstore, "namespace", ""),
-                    filter=self.search_kwargs.get("filter", None)
+                    namespace=self.namespace,
+                    # filter=self.search_kwargs.get("filter", None)
                 ))
             
             # Run the async function in a thread
@@ -455,8 +458,8 @@ class ThresholdRetriever(BaseRetriever):
                 limit_k=self.limit_k,
                 similarity_metric=self.similarity_metric,
                 similarity_threshold=self.similarity_threshold,
-                namespace=getattr(self.vectorstore, "namespace", ""),
-                filter=self.search_kwargs.get("filter", None)
+                namespace=self.namespace,
+                # filter=self.search_kwargs.get("filter", None)
             ))
         
         # Convert to documents
@@ -517,14 +520,6 @@ def get_chain(
         if _retriever_instance is not None:
             return _retriever_instance
             
-        # Check if chain has been cached
-        cache_key = f"pinecone_retriever:{index_name}:{namespace}:{top_k}:{limit_k}:{similarity_metric}:{similarity_threshold}"
-        cached_retriever = cache.get(cache_key)
-        if cached_retriever is not None:
-            _retriever_instance = cached_retriever
-            logger.info("Retrieved cached Pinecone retriever")
-            return _retriever_instance
-            
         start_time = time.time()
         logger.info("Initializing new retriever chain with threshold-based filtering")
         
@@ -572,9 +567,6 @@ def get_chain(
         
         logger.info(f"Pinecone retriever initialized in {time.time() - start_time:.2f} seconds")
         
-        # Cache the retriever with longer TTL (1 hour) since it rarely changes
-        cache.set(cache_key, _retriever_instance, ttl=3600)
-        
         return _retriever_instance
     except Exception as e:
         logger.error(f"Error creating retrieval chain: {e}")

app/database/postgresql.py

@@ -6,7 +6,7 @@ from sqlalchemy.exc import SQLAlchemyError, OperationalError
 from dotenv import load_dotenv
 import logging
 
-# Cấu hình logging
+# Configure logging
 logger = logging.getLogger(__name__)
 
 # Load environment variables
@@ -24,66 +24,76 @@ else:
 
 if not DATABASE_URL:
     logger.error("No database URL configured. Please set AIVEN_DB_URL environment variable.")
-    DATABASE_URL = "postgresql://localhost/test"  # Fallback để không crash khi khởi động
+    DATABASE_URL = "postgresql://localhost/test"  # Fallback to avoid crash on startup
 
-# Create SQLAlchemy engine
+# Create SQLAlchemy engine with optimized settings
 try:
     engine = create_engine(
         DATABASE_URL,
-        pool_pre_ping=True,
-        pool_recycle=300,  # Recycle connections every 5 minutes
-        pool_size=10,      # Tăng kích thước pool từ 5 lên 10
-        max_overflow=20,   # Tăng số lượng kết nối tối đa từ 10 lên 20
+        pool_pre_ping=True,         # Enable connection health checks
+        pool_recycle=300,           # Recycle connections every 5 minutes
+        pool_size=20,               # Increase pool size for more concurrent connections
+        max_overflow=30,            # Allow more overflow connections
+        pool_timeout=30,            # Timeout for getting connection from pool
         connect_args={
-            "connect_timeout": 3,  # Giảm timeout từ 5 xuống 3 giây
-            "keepalives": 1,      # Bật keepalive
-            "keepalives_idle": 30, # Thời gian idle trước khi gửi keepalive
-            "keepalives_interval": 10, # Khoảng thời gian giữa các gói keepalive
-            "keepalives_count": 5  # Số lần thử lại trước khi đóng kết nối
+            "connect_timeout": 5,   # Connection timeout in seconds
+            "keepalives": 1,        # Enable TCP keepalives
+            "keepalives_idle": 30,  # Time before sending keepalives
+            "keepalives_interval": 10, # Time between keepalives
+            "keepalives_count": 5,  # Number of keepalive probes
+            "application_name": "pixagent_api" # Identify app in PostgreSQL logs
         },
-        # Thêm các tùy chọn hiệu suất
-        isolation_level="READ COMMITTED",  # Mức cô lập thấp hơn READ COMMITTED
-        echo=False,  # Tắt echo SQL để giảm overhead logging
-        echo_pool=False  # Tắt echo pool để giảm overhead logging
+        # Performance optimizations
+        isolation_level="READ COMMITTED",  # Lower isolation level for better performance
+        echo=False,                 # Disable SQL echo to reduce overhead
+        echo_pool=False,            # Disable pool logging
+        future=True,                # Use SQLAlchemy 2.0 features
+        # Execution options for common queries
+        execution_options={
+            "compiled_cache": {},   # Use an empty dict for compiled query caching
+            "logging_token": "SQL", # Tag for query logging
+        }
     )
-    logger.info("PostgreSQL engine initialized")
+    logger.info("PostgreSQL engine initialized with optimized settings")
 except Exception as e:
     logger.error(f"Failed to initialize PostgreSQL engine: {e}")
-    # Không raise exception để tránh crash khi khởi động, các xử lý lỗi sẽ được thực hiện ở các function
+    # Don't raise exception to avoid crash on startup
 
-# Create session factory with optimized settings
+# Create optimized session factory
 SessionLocal = sessionmaker(
     autocommit=False, 
     autoflush=False, 
     bind=engine,
-    expire_on_commit=False  # Tránh truy vấn lại DB sau khi commit
+    expire_on_commit=False  # Prevent automatic reloading after commit
 )
 
 # Base class for declarative models - use sqlalchemy.orm for SQLAlchemy 2.0 compatibility
 from sqlalchemy.orm import declarative_base
 Base = declarative_base()
 
-# Kiểm tra kết nối PostgreSQL
+# Check PostgreSQL connection
 def check_db_connection():
-    """Kiểm tra kết nối PostgreSQL"""
+    """Check PostgreSQL connection status"""
     try:
-        # Thực hiện một truy vấn đơn giản để kiểm tra kết nối
+        # Simple query to verify connection
         with engine.connect() as connection:
-            connection.execute(text("SELECT 1"))
-        logger.info("PostgreSQL connection is working")
+            connection.execute(text("SELECT 1")).fetchone()
+        logger.info("PostgreSQL connection successful")
         return True
     except OperationalError as e:
         logger.error(f"PostgreSQL connection failed: {e}")
         return False
     except Exception as e:
-        logger.error(f"Unknown error when checking PostgreSQL connection: {e}")
+        logger.error(f"Unknown error checking PostgreSQL connection: {e}")
         return False
 
-# Dependency to get DB session
+# Dependency to get DB session with improved error handling
 def get_db():
     """Get database session dependency for FastAPI endpoints"""
     db = SessionLocal()
     try:
+        # Test connection is valid before returning
+        db.execute(text("SELECT 1")).fetchone()
         yield db
     except SQLAlchemyError as e:
         logger.error(f"Database session error: {e}")
@@ -92,13 +102,92 @@ def get_db():
     finally:
         db.close()
 
-# Tạo các bảng trong cơ sở dữ liệu nếu chưa tồn tại
+# Create tables in database if they don't exist
 def create_tables():
-    """Tạo các bảng trong cơ sở dữ liệu"""
+    """Create tables in database"""
     try:
         Base.metadata.create_all(bind=engine)
         logger.info("Database tables created or already exist")
         return True
     except SQLAlchemyError as e:
-        logger.error(f"Failed to create database tables: {e}")
+        logger.error(f"Failed to create database tables (SQLAlchemy error): {e}")
+        return False
+    except Exception as e:
+        logger.error(f"Failed to create database tables (unexpected error): {e}")
+        return False
+
+# Function to create indexes for better performance
+def create_indexes():
+    """Create indexes for better query performance"""
+    try:
+        with engine.connect() as conn:
+            try:
+                # Index for featured events - use try-except to handle if index already exists
+                conn.execute(text("""
+                    CREATE INDEX idx_event_featured 
+                    ON event_item(featured)
+                """))
+            except SQLAlchemyError:
+                logger.info("Index idx_event_featured already exists")
+            
+            try:
+                # Index for active events
+                conn.execute(text("""
+                    CREATE INDEX idx_event_active 
+                    ON event_item(is_active)
+                """))
+            except SQLAlchemyError:
+                logger.info("Index idx_event_active already exists")
+            
+            try:
+                # Index for date filtering
+                conn.execute(text("""
+                    CREATE INDEX idx_event_date_start 
+                    ON event_item(date_start)
+                """))
+            except SQLAlchemyError:
+                logger.info("Index idx_event_date_start already exists")
+            
+            try:
+                # Composite index for combined filtering
+                conn.execute(text("""
+                    CREATE INDEX idx_event_featured_active 
+                    ON event_item(featured, is_active)
+                """))
+            except SQLAlchemyError:
+                logger.info("Index idx_event_featured_active already exists")
+                
+            # Indexes for FAQ and Emergency tables
+            try:
+                # FAQ active flag index
+                conn.execute(text("""
+                    CREATE INDEX idx_faq_active 
+                    ON faq_item(is_active)
+                """))
+            except SQLAlchemyError:
+                logger.info("Index idx_faq_active already exists")
+            
+            try:
+                # Emergency contact active flag and priority indexes
+                conn.execute(text("""
+                    CREATE INDEX idx_emergency_active 
+                    ON emergency_item(is_active)
+                """))
+            except SQLAlchemyError:
+                logger.info("Index idx_emergency_active already exists")
+                
+            try:
+                conn.execute(text("""
+                    CREATE INDEX idx_emergency_priority 
+                    ON emergency_item(priority)
+                """))
+            except SQLAlchemyError:
+                logger.info("Index idx_emergency_priority already exists")
+            
+            conn.commit()
+            
+        logger.info("Database indexes created or verified")
+        return True
+    except SQLAlchemyError as e:
+        logger.error(f"Failed to create indexes: {e}")
         return False 

app/models/pdf_models.py

@@ -0,0 +1,51 @@
+from pydantic import BaseModel, Field
+from typing import Optional, List, Dict, Any
+
+class PDFUploadRequest(BaseModel):
+    """Request model cho upload PDF"""
+    namespace: Optional[str] = Field("Default", description="Namespace trong Pinecone")
+    index_name: Optional[str] = Field("testbot768", description="Tên index trong Pinecone")
+    title: Optional[str] = Field(None, description="Tiêu đề của tài liệu")
+    description: Optional[str] = Field(None, description="Mô tả về tài liệu")
+
+class PDFResponse(BaseModel):
+    """Response model cho xử lý PDF"""
+    success: bool = Field(..., description="Trạng thái xử lý thành công hay không")
+    document_id: Optional[str] = Field(None, description="ID của tài liệu")
+    chunks_processed: Optional[int] = Field(None, description="Số lượng chunks đã xử lý")
+    total_text_length: Optional[int] = Field(None, description="Tổng độ dài văn bản")
+    error: Optional[str] = Field(None, description="Thông báo lỗi nếu có")
+    
+    class Config:
+        schema_extra = {
+            "example": {
+                "success": True,
+                "document_id": "550e8400-e29b-41d4-a716-446655440000",
+                "chunks_processed": 25,
+                "total_text_length": 50000
+            }
+        }
+
+class DeleteDocumentRequest(BaseModel):
+    """Request model cho xóa document"""
+    document_id: str = Field(..., description="ID của tài liệu cần xóa")
+    namespace: Optional[str] = Field("Default", description="Namespace trong Pinecone")
+    index_name: Optional[str] = Field("testbot768", description="Tên index trong Pinecone")
+
+class DocumentsListResponse(BaseModel):
+    """Response model cho lấy danh sách tài liệu"""
+    success: bool = Field(..., description="Trạng thái xử lý thành công hay không")
+    total_vectors: Optional[int] = Field(None, description="Tổng số vectors trong index")
+    namespace: Optional[str] = Field(None, description="Namespace đang sử dụng")
+    index_name: Optional[str] = Field(None, description="Tên index đang sử dụng")
+    error: Optional[str] = Field(None, description="Thông báo lỗi nếu có")
+    
+    class Config:
+        schema_extra = {
+            "example": {
+                "success": True,
+                "total_vectors": 5000,
+                "namespace": "Default",
+                "index_name": "testbot768"
+            }
+        } 

app/utils/cache.py

@@ -0,0 +1,271 @@
+import os
+import time
+import threading
+import logging
+from typing import Dict, Any, Optional, Tuple, List, Callable, Generic, TypeVar, Union
+from datetime import datetime
+from dotenv import load_dotenv
+import json
+
+# Thiết lập logging
+logger = logging.getLogger(__name__)
+
+# Load biến môi trường
+load_dotenv()
+
+# Cấu hình cache từ biến môi trường
+DEFAULT_CACHE_TTL = int(os.getenv("CACHE_TTL_SECONDS", "300"))  # Mặc định 5 phút
+DEFAULT_CACHE_CLEANUP_INTERVAL = int(os.getenv("CACHE_CLEANUP_INTERVAL", "60"))  # Mặc định 1 phút
+DEFAULT_CACHE_MAX_SIZE = int(os.getenv("CACHE_MAX_SIZE", "1000"))  # Mặc định 1000 phần tử
+DEFAULT_HISTORY_QUEUE_SIZE = int(os.getenv("HISTORY_QUEUE_SIZE", "10"))  # Mặc định queue size là 10
+DEFAULT_HISTORY_CACHE_TTL = int(os.getenv("HISTORY_CACHE_TTL", "3600"))  # Mặc định 1 giờ
+
+# Generic type để có thể sử dụng cho nhiều loại giá trị khác nhau
+T = TypeVar('T')
+
+# Cấu trúc cho một phần tử trong cache
+class CacheItem(Generic[T]):
+    def __init__(self, value: T, ttl: int = DEFAULT_CACHE_TTL):
+        self.value = value
+        self.expire_at = time.time() + ttl
+        self.last_accessed = time.time()
+    
+    def is_expired(self) -> bool:
+        """Kiểm tra xem item có hết hạn chưa"""
+        return time.time() > self.expire_at
+    
+    def touch(self) -> None:
+        """Cập nhật thời gian truy cập lần cuối"""
+        self.last_accessed = time.time()
+    
+    def extend(self, ttl: int = DEFAULT_CACHE_TTL) -> None:
+        """Gia hạn thời gian sống của item"""
+        self.expire_at = time.time() + ttl
+
+
+# Lớp HistoryQueue để lưu trữ lịch sử người dùng
+class HistoryQueue:
+    def __init__(self, max_size: int = DEFAULT_HISTORY_QUEUE_SIZE, ttl: int = DEFAULT_HISTORY_CACHE_TTL):
+        self.items: List[Dict[str, Any]] = []
+        self.max_size = max_size
+        self.ttl = ttl
+        self.expire_at = time.time() + ttl
+    
+    def add(self, item: Dict[str, Any]) -> None:
+        """Thêm một item vào queue, nếu đã đầy thì loại bỏ item cũ nhất"""
+        if len(self.items) >= self.max_size:
+            self.items.pop(0)
+        self.items.append(item)
+        # Mỗi khi thêm item mới, cập nhật thời gian hết hạn
+        self.refresh_expiry()
+    
+    def get_all(self) -> List[Dict[str, Any]]:
+        """Lấy tất cả items trong queue"""
+        return self.items
+    
+    def is_expired(self) -> bool:
+        """Kiểm tra xem queue có hết hạn chưa"""
+        return time.time() > self.expire_at
+    
+    def refresh_expiry(self) -> None:
+        """Làm mới thời gian hết hạn"""
+        self.expire_at = time.time() + self.ttl
+
+
+# Lớp cache chính
+class InMemoryCache:
+    def __init__(
+        self, 
+        ttl: int = DEFAULT_CACHE_TTL,
+        cleanup_interval: int = DEFAULT_CACHE_CLEANUP_INTERVAL,
+        max_size: int = DEFAULT_CACHE_MAX_SIZE
+    ):
+        self.cache: Dict[str, CacheItem] = {}
+        self.ttl = ttl
+        self.cleanup_interval = cleanup_interval
+        self.max_size = max_size
+        self.user_history_queues: Dict[str, HistoryQueue] = {}
+        self.lock = threading.RLock()  # Sử dụng RLock để tránh deadlock
+        
+        # Khởi động thread dọn dẹp cache định kỳ (active expiration)
+        self.cleanup_thread = threading.Thread(target=self._cleanup_task, daemon=True)
+        self.cleanup_thread.start()
+    
+    def set(self, key: str, value: Any, ttl: Optional[int] = None) -> None:
+        """Lưu một giá trị vào cache"""
+        with self.lock:
+            ttl_value = ttl if ttl is not None else self.ttl
+            
+            # Nếu cache đã đầy, xóa bớt các item ít được truy cập nhất
+            if len(self.cache) >= self.max_size and key not in self.cache:
+                self._evict_lru_items()
+                
+            self.cache[key] = CacheItem(value, ttl_value)
+            logger.debug(f"Cache set: {key} (expires in {ttl_value}s)")
+    
+    def get(self, key: str, default: Any = None) -> Any:
+        """
+        Lấy giá trị từ cache. Nếu key không tồn tại hoặc đã hết hạn, trả về giá trị mặc định.
+        Áp dụng lazy expiration: kiểm tra và xóa các item hết hạn khi truy cập.
+        """
+        with self.lock:
+            item = self.cache.get(key)
+            
+            # Nếu không tìm thấy key hoặc item đã hết hạn
+            if item is None or item.is_expired():
+                # Nếu item tồn tại nhưng đã hết hạn, xóa nó (lazy expiration)
+                if item is not None:
+                    logger.debug(f"Cache miss (expired): {key}")
+                    del self.cache[key]
+                else:
+                    logger.debug(f"Cache miss (not found): {key}")
+                return default
+            
+            # Cập nhật thời gian truy cập
+            item.touch()
+            logger.debug(f"Cache hit: {key}")
+            return item.value
+    
+    def delete(self, key: str) -> bool:
+        """Xóa một key khỏi cache"""
+        with self.lock:
+            if key in self.cache:
+                del self.cache[key]
+                logger.debug(f"Cache delete: {key}")
+                return True
+            return False
+    
+    def clear(self) -> None:
+        """Xóa tất cả dữ liệu trong cache"""
+        with self.lock:
+            self.cache.clear()
+            logger.debug("Cache cleared")
+    
+    def get_or_set(self, key: str, callback: Callable[[], T], ttl: Optional[int] = None) -> T:
+        """
+        Lấy giá trị từ cache nếu tồn tại, nếu không thì gọi callback để lấy giá trị
+        và lưu vào cache trước khi trả về.
+        """
+        with self.lock:
+            value = self.get(key)
+            if value is None:
+                value = callback()
+                self.set(key, value, ttl)
+            return value
+    
+    def _cleanup_task(self) -> None:
+        """Thread để dọn dẹp các item đã hết hạn (active expiration)"""
+        while True:
+            time.sleep(self.cleanup_interval)
+            try:
+                self._remove_expired_items()
+            except Exception as e:
+                logger.error(f"Error in cache cleanup task: {e}")
+    
+    def _remove_expired_items(self) -> None:
+        """Xóa tất cả các item đã hết hạn trong cache"""
+        with self.lock:
+            now = time.time()
+            expired_keys = [k for k, v in self.cache.items() if v.is_expired()]
+            for key in expired_keys:
+                del self.cache[key]
+            
+            # Xóa các user history queue đã hết hạn
+            expired_user_ids = [uid for uid, queue in self.user_history_queues.items() if queue.is_expired()]
+            for user_id in expired_user_ids:
+                del self.user_history_queues[user_id]
+                
+            if expired_keys or expired_user_ids:
+                logger.debug(f"Cleaned up {len(expired_keys)} expired cache items and {len(expired_user_ids)} expired history queues")
+    
+    def _evict_lru_items(self, count: int = 1) -> None:
+        """Xóa bỏ các item ít được truy cập nhất khi cache đầy"""
+        items = sorted(self.cache.items(), key=lambda x: x[1].last_accessed)
+        for i in range(min(count, len(items))):
+            del self.cache[items[i][0]]
+        logger.debug(f"Evicted {min(count, len(items))} least recently used items from cache")
+    
+    def stats(self) -> Dict[str, Any]:
+        """Trả về thống kê về cache"""
+        with self.lock:
+            now = time.time()
+            total_items = len(self.cache)
+            expired_items = sum(1 for item in self.cache.values() if item.is_expired())
+            memory_usage = self._estimate_memory_usage()
+            return {
+                "total_items": total_items,
+                "expired_items": expired_items,
+                "active_items": total_items - expired_items,
+                "memory_usage_bytes": memory_usage,
+                "memory_usage_mb": memory_usage / (1024 * 1024),
+                "max_size": self.max_size,
+                "history_queues": len(self.user_history_queues)
+            }
+    
+    def _estimate_memory_usage(self) -> int:
+        """Ước tính dung lượng bộ nhớ của cache (gần đúng)"""
+        # Ước tính dựa trên kích thước của các key và giá trị
+        cache_size = sum(len(k) for k in self.cache.keys())
+        for item in self.cache.values():
+            try:
+                # Ước tính kích thước của value (gần đúng)
+                if isinstance(item.value, (str, bytes)):
+                    cache_size += len(item.value)
+                elif isinstance(item.value, (dict, list)):
+                    cache_size += len(json.dumps(item.value))
+                else:
+                    # Giá trị mặc định cho các loại dữ liệu khác
+                    cache_size += 100
+            except:
+                cache_size += 100
+        
+        # Ước tính kích thước của user history queues
+        for queue in self.user_history_queues.values():
+            try:
+                cache_size += len(json.dumps(queue.items)) + 100  # 100 bytes cho metadata
+            except:
+                cache_size += 100
+        
+        return cache_size
+    
+    # Các phương thức chuyên biệt cho việc quản lý lịch sử người dùng
+    def add_user_history(self, user_id: str, item: Dict[str, Any], queue_size: Optional[int] = None, ttl: Optional[int] = None) -> None:
+        """Thêm một item vào history queue của người dùng"""
+        with self.lock:
+            # Tạo queue nếu chưa tồn tại
+            if user_id not in self.user_history_queues:
+                queue_size_value = queue_size if queue_size is not None else DEFAULT_HISTORY_QUEUE_SIZE
+                ttl_value = ttl if ttl is not None else DEFAULT_HISTORY_CACHE_TTL
+                self.user_history_queues[user_id] = HistoryQueue(max_size=queue_size_value, ttl=ttl_value)
+            
+            # Thêm item vào queue
+            self.user_history_queues[user_id].add(item)
+            logger.debug(f"Added history item for user {user_id}")
+    
+    def get_user_history(self, user_id: str, default: Any = None) -> List[Dict[str, Any]]:
+        """Lấy lịch sử của người dùng từ cache"""
+        with self.lock:
+            queue = self.user_history_queues.get(user_id)
+            
+            # Nếu không tìm thấy queue hoặc queue đã hết hạn
+            if queue is None or queue.is_expired():
+                if queue is not None and queue.is_expired():
+                    del self.user_history_queues[user_id]
+                    logger.debug(f"User history queue expired: {user_id}")
+                return default if default is not None else []
+            
+            # Làm mới thời gian hết hạn
+            queue.refresh_expiry()
+            logger.debug(f"Retrieved history for user {user_id}: {len(queue.items)} items")
+            return queue.get_all()
+
+
+# Singleton instance
+_cache_instance = None
+
+def get_cache() -> InMemoryCache:
+    """Trả về instance singleton của InMemoryCache"""
+    global _cache_instance
+    if _cache_instance is None:
+        _cache_instance = InMemoryCache()
+    return _cache_instance 

app/utils/pdf_processor.py

@@ -0,0 +1,211 @@
+import os
+import time
+import uuid
+from langchain.text_splitter import RecursiveCharacterTextSplitter
+from langchain_community.document_loaders import PyPDFLoader
+from langchain_google_genai import GoogleGenerativeAIEmbeddings
+import logging
+
+from app.database.pinecone import get_pinecone_index, init_pinecone
+
+# Cấu hình logging
+logger = logging.getLogger(__name__)
+
+# Khởi tạo embeddings model
+embeddings_model = GoogleGenerativeAIEmbeddings(model="models/embedding-001")
+
+class PDFProcessor:
+    """Lớp xử lý file PDF và tạo embeddings"""
+    
+    def __init__(self, index_name="testbot768", namespace="Default"):
+        """Khởi tạo với tên index và namespace Pinecone mặc định"""
+        self.index_name = index_name
+        self.namespace = namespace
+        self.pinecone_index = None
+        
+    def _init_pinecone_connection(self):
+        """Khởi tạo kết nối đến Pinecone"""
+        try:
+            # Sử dụng singleton pattern từ module database.pinecone
+            self.pinecone_index = get_pinecone_index()
+            if not self.pinecone_index:
+                logger.error("Không thể kết nối đến Pinecone")
+                return False
+            return True
+        except Exception as e:
+            logger.error(f"Lỗi khi kết nối Pinecone: {str(e)}")
+            return False
+            
+    async def process_pdf(self, file_path, document_id=None, metadata=None, progress_callback=None):
+        """
+        Xử lý file PDF, chia thành chunks và tạo embeddings
+        
+        Args:
+            file_path (str): Đường dẫn tới file PDF
+            document_id (str, optional): ID của tài liệu, nếu không cung cấp sẽ tạo ID mới
+            metadata (dict, optional): Metadata bổ sung cho tài liệu
+            progress_callback (callable, optional): Callback function để cập nhật tiến độ
+            
+        Returns:
+            dict: Thông tin kết quả xử lý gồm document_id và số chunks đã xử lý
+        """
+        try:
+            # Khởi tạo kết nối Pinecone nếu chưa có
+            if not self.pinecone_index:
+                if not self._init_pinecone_connection():
+                    return {"success": False, "error": "Không thể kết nối đến Pinecone"}
+            
+            # Tạo document_id nếu không có
+            if not document_id:
+                document_id = str(uuid.uuid4())
+            
+            # Đọc file PDF bằng PyPDFLoader
+            logger.info(f"Đang đọc file PDF: {file_path}")
+            if progress_callback:
+                await progress_callback("pdf_loading", 0.5, "Loading PDF file")
+                
+            loader = PyPDFLoader(file_path)
+            pages = loader.load()
+            
+            # Trích xuất và nối text từ tất cả các trang
+            all_text = ""
+            for page in pages:
+                all_text += page.page_content + "\n"
+            
+            if progress_callback:
+                await progress_callback("text_extraction", 0.6, "Extracted text from PDF")
+                
+            # Chia văn bản thành các chunk
+            text_splitter = RecursiveCharacterTextSplitter(chunk_size=800, chunk_overlap=300)
+            chunks = text_splitter.split_text(all_text)
+            
+            logger.info(f"Đã chia file PDF thành {len(chunks)} chunks")
+            if progress_callback:
+                await progress_callback("chunking", 0.7, f"Split document into {len(chunks)} chunks")
+            
+            # Xử lý embedding cho từng chunk và upsert lên Pinecone
+            vectors = []
+            for i, chunk in enumerate(chunks):
+                # Cập nhật tiến độ embedding
+                if progress_callback and i % 5 == 0:  # Cập nhật sau mỗi 5 chunks để tránh quá nhiều thông báo
+                    embedding_progress = 0.7 + (0.3 * (i / len(chunks)))
+                    await progress_callback("embedding", embedding_progress, f"Processing chunk {i+1}/{len(chunks)}")
+                
+                # Tạo vector embedding cho từng chunk
+                vector = embeddings_model.embed_query(chunk)
+                
+                # Chuẩn bị metadata cho vector
+                vector_metadata = {
+                    "document_id": document_id,
+                    "chunk_index": i,
+                    "text": chunk
+                }
+                
+                # Thêm metadata bổ sung nếu có
+                if metadata:
+                    for key, value in metadata.items():
+                        if key not in vector_metadata:
+                            vector_metadata[key] = value
+                
+                # Thêm vector vào danh sách để upsert
+                vectors.append({
+                    "id": f"{document_id}_{i}",
+                    "values": vector,
+                    "metadata": vector_metadata
+                })
+                
+                # Upsert mỗi 100 vectors để tránh quá lớn
+                if len(vectors) >= 100:
+                    await self._upsert_vectors(vectors)
+                    vectors = []
+            
+            # Upsert các vectors còn lại
+            if vectors:
+                await self._upsert_vectors(vectors)
+            
+            logger.info(f"Đã embedding và lưu {len(chunks)} chunks từ PDF với document_id: {document_id}")
+            
+            # Final progress update
+            if progress_callback:
+                await progress_callback("completed", 1.0, "PDF processing complete")
+            
+            return {
+                "success": True,
+                "document_id": document_id,
+                "chunks_processed": len(chunks),
+                "total_text_length": len(all_text)
+            }
+            
+        except Exception as e:
+            logger.error(f"Lỗi khi xử lý PDF: {str(e)}")
+            if progress_callback:
+                await progress_callback("error", 0, f"Error processing PDF: {str(e)}")
+            return {
+                "success": False,
+                "error": str(e)
+            }
+    
+    async def _upsert_vectors(self, vectors):
+        """Upsert vectors vào Pinecone"""
+        try:
+            if not vectors:
+                return
+                
+            result = self.pinecone_index.upsert(
+                vectors=vectors,
+                namespace=self.namespace
+            )
+            
+            logger.info(f"Đã upsert {len(vectors)} vectors vào Pinecone")
+            return result
+        except Exception as e:
+            logger.error(f"Lỗi khi upsert vectors: {str(e)}")
+            raise
+    
+    async def delete_namespace(self):
+        """
+        Xóa toàn bộ vectors trong namespace hiện tại (tương đương xoá namespace).
+        """
+        # Khởi tạo kết nối nếu cần
+        if not self.pinecone_index and not self._init_pinecone_connection():
+            return {"success": False, "error": "Không thể kết nối đến Pinecone"}
+
+        try:
+            # delete_all=True sẽ xóa toàn bộ vectors trong namespace
+            result = self.pinecone_index.delete(
+                delete_all=True,
+                namespace=self.namespace
+            )
+            logger.info(f"Đã xóa namespace '{self.namespace}' (tất cả vectors).")
+            return {"success": True, "detail": result}
+        except Exception as e:
+            logger.error(f"Lỗi khi xóa namespace '{self.namespace}': {e}")
+            return {"success": False, "error": str(e)}
+    
+    async def list_documents(self):
+        """Lấy danh sách tất cả document_id từ Pinecone"""
+        try:
+            # Khởi tạo kết nối Pinecone nếu chưa có
+            if not self.pinecone_index:
+                if not self._init_pinecone_connection():
+                    return {"success": False, "error": "Không thể kết nối đến Pinecone"}
+            
+            # Lấy thông tin index
+            stats = self.pinecone_index.describe_index_stats()
+            
+            # Thực hiện truy vấn để lấy danh sách tất cả document_id duy nhất
+            # Phương pháp này có thể không hiệu quả với dataset lớn, nhưng là cách đơn giản nhất
+            # Trong thực tế, nên lưu danh sách document_id trong một database riêng
+            
+            return {
+                "success": True,
+                "total_vectors": stats.get('total_vector_count', 0),
+                "namespace": self.namespace,
+                "index_name": self.index_name
+            }
+        except Exception as e:
+            logger.error(f"Lỗi khi lấy danh sách documents: {str(e)}")
+            return {
+                "success": False,
+                "error": str(e)
+            } 

app/utils/utils.py

@@ -2,10 +2,13 @@ import logging
 import time
 import uuid
 import threading
+import os
 from functools import wraps
 from datetime import datetime, timedelta
 import pytz
-from typing import Callable, Any, Dict, Optional
+from typing import Callable, Any, Dict, Optional, List, Tuple, Set
+import gc
+import heapq
 
 # Configure logging
 logging.basicConfig(
@@ -70,46 +73,395 @@ def truncate_text(text, max_length=100):
         return text
     return text[:max_length] + "..."
 
-# Simple in-memory cache implementation (replaces Redis dependency)
-class SimpleCache:
-    def __init__(self):
-        self._cache = {}
-        self._expiry = {}
+class CacheStrategy:
+    """Cache loading strategy enumeration"""
+    LAZY = "lazy"  # Only load items into cache when requested
+    EAGER = "eager"  # Preload items into cache at initialization
+    MIXED = "mixed"  # Preload high-priority items, lazy load others
+
+class CacheItem:
+    """Represents an item in the cache with metadata"""
+    def __init__(self, key: str, value: Any, ttl: int = 300, priority: int = 1):
+        self.key = key
+        self.value = value
+        self.expiry = datetime.now() + timedelta(seconds=ttl)
+        self.priority = priority  # Higher number = higher priority
+        self.access_count = 0     # Track number of accesses
+        self.last_accessed = datetime.now()
+        
+    def is_expired(self) -> bool:
+        """Check if the item is expired"""
+        return datetime.now() > self.expiry
     
-    def get(self, key: str) -> Optional[Any]:
+    def touch(self):
+        """Update last accessed time and access count"""
+        self.last_accessed = datetime.now()
+        self.access_count += 1
+        
+    def __lt__(self, other):
+        """For heap comparisons - lower priority items are evicted first"""
+        # First compare priority
+        if self.priority != other.priority:
+            return self.priority < other.priority
+        # Then compare access frequency (less frequently accessed items are evicted first)
+        if self.access_count != other.access_count:
+            return self.access_count < other.access_count
+        # Finally compare last access time (oldest accessed first)
+        return self.last_accessed < other.last_accessed
+
+    def get_size(self) -> int:
+        """Approximate memory size of the cache item in bytes"""
+        try:
+            import sys
+            return sys.getsizeof(self.value) + sys.getsizeof(self.key) + 64  # Additional overhead
+        except:
+            # Default estimate if we can't get the size
+            return 1024
+
+# Enhanced in-memory cache implementation
+class EnhancedCache:
+    def __init__(self, 
+                 strategy: str = "lazy", 
+                 max_items: int = 10000, 
+                 max_size_mb: int = 100,
+                 cleanup_interval: int = 60,
+                 stats_enabled: bool = True):
+        """
+        Initialize enhanced cache with configurable strategy.
+        
+        Args:
+            strategy: Cache loading strategy (lazy, eager, mixed)
+            max_items: Maximum number of items to store in cache
+            max_size_mb: Maximum size of cache in MB
+            cleanup_interval: Interval in seconds to run cleanup
+            stats_enabled: Whether to collect cache statistics
+        """
+        self._cache: Dict[str, CacheItem] = {}
+        self._namespace_cache: Dict[str, Set[str]] = {}  # Tracking keys by namespace
+        self._strategy = strategy
+        self._max_items = max_items
+        self._max_size_bytes = max_size_mb * 1024 * 1024
+        self._current_size_bytes = 0
+        self._stats_enabled = stats_enabled
+        
+        # Statistics
+        self._hits = 0
+        self._misses = 0
+        self._evictions = 0
+        self._total_get_time = 0
+        self._total_set_time = 0
+        
+        # Setup cleanup thread
+        self._last_cleanup = datetime.now()
+        self._cleanup_interval = cleanup_interval
+        self._lock = threading.RLock()
+        
+        if cleanup_interval > 0:
+            self._start_cleanup_thread(cleanup_interval)
+        
+        logger.info(f"Enhanced cache initialized with strategy={strategy}, max_items={max_items}, max_size={max_size_mb}MB")
+    
+    def _start_cleanup_thread(self, interval: int):
+        """Start background thread for periodic cleanup"""
+        def cleanup_worker():
+            while True:
+                time.sleep(interval)
+                try:
+                    self.cleanup()
+                except Exception as e:
+                    logger.error(f"Error in cache cleanup: {e}")
+        
+        thread = threading.Thread(target=cleanup_worker, daemon=True)
+        thread.start()
+        logger.info(f"Cache cleanup thread started with interval {interval}s")
+    
+    def get(self, key: str, namespace: str = None) -> Optional[Any]:
         """Get value from cache if it exists and hasn't expired"""
+        if self._stats_enabled:
+            start_time = time.time()
+            
+        # Use namespaced key if namespace is provided
+        cache_key = f"{namespace}:{key}" if namespace else key
+        
+        with self._lock:
+            cache_item = self._cache.get(cache_key)
+            
+            if cache_item:
+                if cache_item.is_expired():
+                    # Clean up expired key
+                    self._remove_item(cache_key, namespace)
+                    if self._stats_enabled:
+                        self._misses += 1
+                    value = None
+                else:
+                    # Update access metadata
+                    cache_item.touch()
+                    if self._stats_enabled:
+                        self._hits += 1
+                    value = cache_item.value
+            else:
+                if self._stats_enabled:
+                    self._misses += 1
+                value = None
+                
+            if self._stats_enabled:
+                self._total_get_time += time.time() - start_time
+                
+            return value
+    
+    def set(self, key: str, value: Any, ttl: int = 300, priority: int = 1, namespace: str = None) -> None:
+        """Set a value in the cache with TTL in seconds"""
+        if self._stats_enabled:
+            start_time = time.time()
+            
+        # Use namespaced key if namespace is provided
+        cache_key = f"{namespace}:{key}" if namespace else key
+            
+        with self._lock:
+            # Create cache item
+            cache_item = CacheItem(cache_key, value, ttl, priority)
+            item_size = cache_item.get_size()
+            
+            # Check if we need to make room
+            if (len(self._cache) >= self._max_items or 
+                self._current_size_bytes + item_size > self._max_size_bytes):
+                self._evict_items(item_size)
+            
+            # Update size tracking
+            if cache_key in self._cache:
+                # If replacing, subtract old size first
+                self._current_size_bytes -= self._cache[cache_key].get_size()
+            self._current_size_bytes += item_size
+            
+            # Store the item
+            self._cache[cache_key] = cache_item
+            
+            # Update namespace tracking
+            if namespace:
+                if namespace not in self._namespace_cache:
+                    self._namespace_cache[namespace] = set()
+                self._namespace_cache[namespace].add(cache_key)
+                
+            if self._stats_enabled:
+                self._total_set_time += time.time() - start_time
+    
+    def delete(self, key: str, namespace: str = None) -> None:
+        """Delete a key from the cache"""
+        # Use namespaced key if namespace is provided
+        cache_key = f"{namespace}:{key}" if namespace else key
+        
+        with self._lock:
+            self._remove_item(cache_key, namespace)
+    
+    def _remove_item(self, key: str, namespace: str = None):
+        """Internal method to remove an item and update tracking"""
         if key in self._cache:
-            # Check if the key has expired
-            if key in self._expiry and self._expiry[key] > datetime.now():
-                return self._cache[key]
+            # Update size tracking
+            self._current_size_bytes -= self._cache[key].get_size()
+            # Remove from cache
+            del self._cache[key]
+            
+            # Update namespace tracking
+            if namespace and namespace in self._namespace_cache:
+                if key in self._namespace_cache[namespace]:
+                    self._namespace_cache[namespace].remove(key)
+                # Cleanup empty sets
+                if not self._namespace_cache[namespace]:
+                    del self._namespace_cache[namespace]
+    
+    def _evict_items(self, needed_space: int = 0) -> None:
+        """Evict items to make room in the cache"""
+        if not self._cache:
+            return
+            
+        with self._lock:
+            # Convert cache items to a list for sorting
+            items = list(self._cache.values())
+            
+            # Sort by priority, access count, and last accessed time
+            items.sort()  # Uses the __lt__ method of CacheItem
+            
+            # Evict items until we have enough space
+            space_freed = 0
+            evicted_count = 0
+            
+            for item in items:
+                # Stop if we've made enough room
+                if (len(self._cache) - evicted_count <= self._max_items * 0.9 and
+                    (space_freed >= needed_space or 
+                     self._current_size_bytes - space_freed <= self._max_size_bytes * 0.9)):
+                    break
+                    
+                # Skip high priority items unless absolutely necessary
+                if item.priority > 9 and evicted_count < len(items) // 2:
+                    continue
+                    
+                # Evict this item
+                item_size = item.get_size()
+                namespace = item.key.split(':', 1)[0] if ':' in item.key else None
+                self._remove_item(item.key, namespace)
+                
+                space_freed += item_size
+                evicted_count += 1
+                if self._stats_enabled:
+                    self._evictions += 1
+            
+            logger.info(f"Cache eviction: removed {evicted_count} items, freed {space_freed / 1024:.2f}KB")
+    
+    def clear(self, namespace: str = None) -> None:
+        """
+        Clear the cache or a specific namespace
+        """
+        with self._lock:
+            if namespace:
+                # Clear only keys in the specified namespace
+                if namespace in self._namespace_cache:
+                    keys_to_remove = list(self._namespace_cache[namespace])
+                    for key in keys_to_remove:
+                        self._remove_item(key, namespace)
+                    # The namespace should be auto-cleaned in _remove_item
             else:
-                # Clean up expired keys
-                if key in self._cache:
-                    del self._cache[key]
-                if key in self._expiry:
-                    del self._expiry[key]
-        return None
+                # Clear the entire cache
+                self._cache.clear()
+                self._namespace_cache.clear()
+                self._current_size_bytes = 0
+                
+            logger.info(f"Cache cleared{' for namespace ' + namespace if namespace else ''}")
+    
+    def cleanup(self) -> None:
+        """Remove expired items and run garbage collection if needed"""
+        with self._lock:
+            now = datetime.now()
+            # Only run if it's been at least cleanup_interval since last cleanup
+            if (now - self._last_cleanup).total_seconds() < self._cleanup_interval:
+                return
+                
+            # Find expired items
+            expired_keys = []
+            for key, item in self._cache.items():
+                if item.is_expired():
+                    expired_keys.append((key, key.split(':', 1)[0] if ':' in key else None))
+            
+            # Remove expired items
+            for key, namespace in expired_keys:
+                self._remove_item(key, namespace)
+            
+            # Update last cleanup time
+            self._last_cleanup = now
+            
+            # Run garbage collection if we removed several items
+            if len(expired_keys) > 100:
+                gc.collect()
+                
+            logger.info(f"Cache cleanup: removed {len(expired_keys)} expired items")
+    
+    def get_stats(self) -> Dict:
+        """Get cache statistics"""
+        with self._lock:
+            if not self._stats_enabled:
+                return {"stats_enabled": False}
+                
+            # Calculate hit rate
+            total_requests = self._hits + self._misses
+            hit_rate = (self._hits / total_requests) * 100 if total_requests > 0 else 0
+            
+            # Calculate average times
+            avg_get_time = (self._total_get_time / total_requests) * 1000 if total_requests > 0 else 0
+            avg_set_time = (self._total_set_time / self._evictions) * 1000 if self._evictions > 0 else 0
+            
+            return {
+                "stats_enabled": True,
+                "item_count": len(self._cache),
+                "max_items": self._max_items,
+                "size_bytes": self._current_size_bytes,
+                "max_size_bytes": self._max_size_bytes,
+                "hits": self._hits,
+                "misses": self._misses,
+                "hit_rate_percent": round(hit_rate, 2),
+                "evictions": self._evictions,
+                "avg_get_time_ms": round(avg_get_time, 3),
+                "avg_set_time_ms": round(avg_set_time, 3),
+                "namespace_count": len(self._namespace_cache),
+                "namespaces": list(self._namespace_cache.keys())
+            }
+    
+    def preload(self, items: List[Tuple[str, Any, int, int]], namespace: str = None) -> None:
+        """
+        Preload a list of items into the cache
+        
+        Args:
+            items: List of (key, value, ttl, priority) tuples
+            namespace: Optional namespace for all items
+        """
+        for key, value, ttl, priority in items:
+            self.set(key, value, ttl, priority, namespace)
+        
+        logger.info(f"Preloaded {len(items)} items into cache{' namespace ' + namespace if namespace else ''}")
+    
+    def get_or_load(self, key: str, loader_func: Callable[[], Any], 
+                   ttl: int = 300, priority: int = 1, namespace: str = None) -> Any:
+        """
+        Get from cache or load using the provided function
+        
+        Args:
+            key: Cache key
+            loader_func: Function to call if cache miss occurs
+            ttl: TTL in seconds
+            priority: Item priority
+            namespace: Optional namespace
+            
+        Returns:
+            Cached or freshly loaded value
+        """
+        # Try to get from cache first
+        value = self.get(key, namespace)
+        
+        # If not in cache, load it
+        if value is None:
+            value = loader_func()
+            # Only cache if we got a valid value
+            if value is not None:
+                self.set(key, value, ttl, priority, namespace)
+                
+        return value
+
+# Load cache configuration from environment variables
+CACHE_STRATEGY = os.getenv("CACHE_STRATEGY", "mixed")
+CACHE_MAX_ITEMS = int(os.getenv("CACHE_MAX_ITEMS", "10000"))
+CACHE_MAX_SIZE_MB = int(os.getenv("CACHE_MAX_SIZE_MB", "100"))
+CACHE_CLEANUP_INTERVAL = int(os.getenv("CACHE_CLEANUP_INTERVAL", "60"))
+CACHE_STATS_ENABLED = os.getenv("CACHE_STATS_ENABLED", "true").lower() in ("true", "1", "yes")
+
+# Initialize the enhanced cache
+cache = EnhancedCache(
+    strategy=CACHE_STRATEGY,
+    max_items=CACHE_MAX_ITEMS,
+    max_size_mb=CACHE_MAX_SIZE_MB,
+    cleanup_interval=CACHE_CLEANUP_INTERVAL,
+    stats_enabled=CACHE_STATS_ENABLED
+)
+
+# Backward compatibility for SimpleCache - for a transition period
+class SimpleCache:
+    def __init__(self):
+        """Legacy SimpleCache implementation that uses EnhancedCache underneath"""
+        logger.warning("SimpleCache is deprecated, please use EnhancedCache directly")
+        
+    def get(self, key: str) -> Optional[Any]:
+        """Get value from cache if it exists and hasn't expired"""
+        return cache.get(key)
     
     def set(self, key: str, value: Any, ttl: int = 300) -> None:
         """Set a value in the cache with TTL in seconds"""
-        self._cache[key] = value
-        # Set expiry time
-        self._expiry[key] = datetime.now() + timedelta(seconds=ttl)
+        cache.set(key, value, ttl)
     
     def delete(self, key: str) -> None:
         """Delete a key from the cache"""
-        if key in self._cache:
-            del self._cache[key]
-        if key in self._expiry:
-            del self._expiry[key]
+        cache.delete(key)
     
     def clear(self) -> None:
         """Clear the entire cache"""
-        self._cache.clear()
-        self._expiry.clear()
-
-# Initialize cache
-cache = SimpleCache()
+        cache.clear()
 
 def get_host_url(request) -> str:
     """

docs/api_documentation.md

@@ -0,0 +1,581 @@
+# API Documentation
+
+## Frontend Setup
+
+```javascript
+// Basic Axios setup
+import axios from 'axios';
+
+const api = axios.create({
+  baseURL: 'https://api.your-domain.com',
+  timeout: 10000,
+  headers: {
+    'Content-Type': 'application/json',
+    'Accept': 'application/json'
+  }
+});
+
+// Error handling
+api.interceptors.response.use(
+  response => response.data,
+  error => {
+    const errorMessage = error.response?.data?.detail || 'An error occurred';
+    console.error('API Error:', errorMessage);
+    return Promise.reject(errorMessage);
+  }
+);
+```
+
+## Caching System
+
+- All GET endpoints support `use_cache=true` parameter (default)
+- Cache TTL: 300 seconds (5 minutes)
+- Cache is automatically invalidated on data changes
+
+## Authentication
+
+Currently no authentication is required. If implemented in the future, use JWT Bearer tokens:
+
+```javascript
+const api = axios.create({
+  // ...other config
+  headers: {
+    // ...other headers
+    'Authorization': `Bearer ${token}`
+  }
+});
+```
+
+## Error Codes
+
+| Code | Description |
+|------|-------------|
+| 400 | Bad Request |
+| 404 | Not Found |
+| 500 | Internal Server Error |
+| 503 | Service Unavailable |
+
+## PostgreSQL Endpoints
+
+### FAQ Endpoints
+
+#### Get FAQs List
+```
+GET /postgres/faq
+```
+
+Parameters:
+- `skip`: Number of items to skip (default: 0)
+- `limit`: Maximum items to return (default: 100)
+- `active_only`: Return only active items (default: false)
+- `use_cache`: Use cached data if available (default: true)
+
+Response:
+```json
+[
+  {
+    "question": "How do I book a hotel?",
+    "answer": "You can book a hotel through our app or website.",
+    "is_active": true,
+    "id": 1,
+    "created_at": "2023-01-01T00:00:00",
+    "updated_at": "2023-01-01T00:00:00"
+  }
+]
+```
+
+Example:
+```javascript
+async function getFAQs() {
+  try {
+    const data = await api.get('/postgres/faq', {
+      params: { active_only: true, limit: 20 }
+    });
+    return data;
+  } catch (error) {
+    console.error('Error fetching FAQs:', error);
+    throw error;
+  }
+}
+```
+
+#### Create FAQ
+```
+POST /postgres/faq
+```
+
+Request Body:
+```json
+{
+  "question": "How do I book a hotel?",
+  "answer": "You can book a hotel through our app or website.",
+  "is_active": true
+}
+```
+
+Response: Created FAQ object
+
+#### Get FAQ Detail
+```
+GET /postgres/faq/{faq_id}
+```
+
+Parameters:
+- `faq_id`: ID of FAQ (required)
+- `use_cache`: Use cached data if available (default: true)
+
+Response: FAQ object
+
+#### Update FAQ
+```
+PUT /postgres/faq/{faq_id}
+```
+
+Parameters:
+- `faq_id`: ID of FAQ to update (required)
+
+Request Body: Partial or complete FAQ object
+Response: Updated FAQ object
+
+#### Delete FAQ
+```
+DELETE /postgres/faq/{faq_id}
+```
+
+Parameters:
+- `faq_id`: ID of FAQ to delete (required)
+
+Response:
+```json
+{
+  "status": "success",
+  "message": "FAQ item 1 deleted"
+}
+```
+
+#### Batch Operations
+
+Create multiple FAQs:
+```
+POST /postgres/faqs/batch
+```
+
+Update status of multiple FAQs:
+```
+PUT /postgres/faqs/batch-update-status
+```
+
+Delete multiple FAQs:
+```
+DELETE /postgres/faqs/batch
+```
+
+### Emergency Contact Endpoints
+
+#### Get Emergency Contacts
+```
+GET /postgres/emergency
+```
+
+Parameters:
+- `skip`: Number of items to skip (default: 0)
+- `limit`: Maximum items to return (default: 100)
+- `active_only`: Return only active items (default: false)
+- `use_cache`: Use cached data if available (default: true)
+
+Response: Array of Emergency Contact objects
+
+#### Create Emergency Contact
+```
+POST /postgres/emergency
+```
+
+Request Body:
+```json
+{
+  "name": "Fire Department",
+  "phone_number": "114",
+  "description": "Fire rescue services",
+  "address": "Da Nang",
+  "location": "16.0544, 108.2022",
+  "priority": 1,
+  "is_active": true
+}
+```
+
+Response: Created Emergency Contact object
+
+#### Get Emergency Contact
+```
+GET /postgres/emergency/{emergency_id}
+```
+
+#### Update Emergency Contact
+```
+PUT /postgres/emergency/{emergency_id}
+```
+
+#### Delete Emergency Contact
+```
+DELETE /postgres/emergency/{emergency_id}
+```
+
+#### Batch Operations
+
+Create multiple Emergency Contacts:
+```
+POST /postgres/emergency/batch
+```
+
+Update status of multiple Emergency Contacts:
+```
+PUT /postgres/emergency/batch-update-status
+```
+
+Delete multiple Emergency Contacts:
+```
+DELETE /postgres/emergency/batch
+```
+
+### Event Endpoints
+
+#### Get Events
+```
+GET /postgres/events
+```
+
+Parameters:
+- `skip`: Number of items to skip (default: 0)
+- `limit`: Maximum items to return (default: 100)
+- `active_only`: Return only active items (default: false)
+- `featured_only`: Return only featured items (default: false)
+- `use_cache`: Use cached data if available (default: true)
+
+Response: Array of Event objects
+
+#### Create Event
+```
+POST /postgres/events
+```
+
+Request Body:
+```json
+{
+  "name": "Da Nang Fireworks Festival",
+  "description": "International Fireworks Festival Da Nang 2023",
+  "address": "Dragon Bridge, Da Nang",
+  "location": "16.0610, 108.2277",
+  "date_start": "2023-06-01T19:00:00",
+  "date_end": "2023-06-01T22:00:00",
+  "price": [
+    {"type": "VIP", "amount": 500000},
+    {"type": "Standard", "amount": 300000}
+  ],
+  "url": "https://danangfireworks.com",
+  "is_active": true,
+  "featured": true
+}
+```
+
+Response: Created Event object
+
+#### Get Event
+```
+GET /postgres/events/{event_id}
+```
+
+#### Update Event
+```
+PUT /postgres/events/{event_id}
+```
+
+#### Delete Event
+```
+DELETE /postgres/events/{event_id}
+```
+
+#### Batch Operations
+
+Create multiple Events:
+```
+POST /postgres/events/batch
+```
+
+Update status of multiple Events:
+```
+PUT /postgres/events/batch-update-status
+```
+
+Delete multiple Events:
+```
+DELETE /postgres/events/batch
+```
+
+### About Pixity Endpoints
+
+#### Get About Pixity
+```
+GET /postgres/about-pixity
+```
+
+Response:
+```json
+{
+  "content": "PiXity is your smart, AI-powered local companion...",
+  "id": 1,
+  "created_at": "2023-01-01T00:00:00",
+  "updated_at": "2023-01-01T00:00:00"
+}
+```
+
+#### Update About Pixity
+```
+PUT /postgres/about-pixity
+```
+
+Request Body:
+```json
+{
+  "content": "PiXity is your smart, AI-powered local companion..."
+}
+```
+
+Response: Updated About Pixity object
+
+### Da Nang Bucket List Endpoints
+
+#### Get Da Nang Bucket List
+```
+GET /postgres/danang-bucket-list
+```
+
+Response: Bucket List object with JSON content string
+
+#### Update Da Nang Bucket List
+```
+PUT /postgres/danang-bucket-list
+```
+
+### Solana Summit Endpoints
+
+#### Get Solana Summit
+```
+GET /postgres/solana-summit
+```
+
+Response: Solana Summit object with JSON content string
+
+#### Update Solana Summit
+```
+PUT /postgres/solana-summit
+```
+
+### Health Check
+```
+GET /postgres/health
+```
+
+Response:
+```json
+{
+  "status": "healthy",
+  "message": "PostgreSQL connection is working",
+  "timestamp": "2023-01-01T00:00:00"
+}
+```
+
+## MongoDB Endpoints
+
+### Session Endpoints
+
+#### Create Session
+```
+POST /session
+```
+
+Request Body:
+```json
+{
+  "user_id": "user123",
+  "query": "How do I book a room?",
+  "timestamp": "2023-01-01T00:00:00",
+  "metadata": {
+    "client_info": "web",
+    "location": "Da Nang"
+  }
+}
+```
+
+Response: Created Session object with session_id
+
+#### Update Session with Response
+```
+PUT /session/{session_id}/response
+```
+
+Request Body:
+```json
+{
+  "response": "You can book a room through our app or website.",
+  "response_timestamp": "2023-01-01T00:00:05",
+  "metadata": {
+    "response_time_ms": 234,
+    "model_version": "gpt-4"
+  }
+}
+```
+
+Response: Updated Session object
+
+#### Get Session
+```
+GET /session/{session_id}
+```
+
+Response: Session object
+
+#### Get User History
+```
+GET /history
+```
+
+Parameters:
+- `user_id`: User ID (required)
+- `limit`: Maximum sessions to return (default: 10)
+- `skip`: Number of sessions to skip (default: 0)
+
+Response:
+```json
+{
+  "user_id": "user123",
+  "sessions": [
+    {
+      "session_id": "60f7a8b9c1d2e3f4a5b6c7d8",
+      "query": "How do I book a room?",
+      "timestamp": "2023-01-01T00:00:00",
+      "response": "You can book a room through our app or website.",
+      "response_timestamp": "2023-01-01T00:00:05"
+    }
+  ],
+  "total_count": 1
+}
+```
+
+#### Health Check
+```
+GET /health
+```
+
+## RAG Endpoints
+
+### Create Embedding
+```
+POST /embedding
+```
+
+Request Body:
+```json
+{
+  "text": "Text to embed"
+}
+```
+
+Response:
+```json
+{
+  "embedding": [0.1, 0.2, 0.3, ...],
+  "dimensions": 1536
+}
+```
+
+### Process Chat Request
+```
+POST /chat
+```
+
+Request Body:
+```json
+{
+  "query": "Can you tell me about Pixity?",
+  "chat_history": [
+    {"role": "user", "content": "Hello"},
+    {"role": "assistant", "content": "Hello! How can I help you?"}
+  ]
+}
+```
+
+Response:
+```json
+{
+  "answer": "Pixity is a platform...",
+  "sources": [
+    {
+      "document_id": "doc123",
+      "chunk_id": "chunk456",
+      "chunk_text": "Pixity was founded in...",
+      "relevance_score": 0.92
+    }
+  ]
+}
+```
+
+### Direct RAG Query
+```
+POST /rag
+```
+
+Request Body:
+```json
+{
+  "query": "Can you tell me about Pixity?",
+  "namespace": "about_pixity",
+  "top_k": 3
+}
+```
+
+Response: Query results with relevance scores
+
+### Health Check
+```
+GET /health
+```
+
+## PDF Processing Endpoints
+
+### Upload and Process PDF
+```
+POST /pdf/upload
+```
+
+Form Data:
+- `file`: PDF file (required)
+- `namespace`: Vector database namespace (default: "Default")
+- `index_name`: Vector database index name (default: "testbot768")
+- `title`: Document title (optional)
+- `description`: Document description (optional)
+- `user_id`: User ID for WebSocket updates (optional)
+
+Response: Processing results with document_id
+
+### Delete Documents in Namespace
+```
+DELETE /pdf/namespace
+```
+
+Parameters:
+- `namespace`: Vector database namespace (default: "Default")
+- `index_name`: Vector database index name (default: "testbot768")
+- `user_id`: User ID for WebSocket updates (optional)
+
+Response: Deletion results
+
+### Get Documents List
+```
+GET /pdf/documents
+```
+
+Parameters:
+- `namespace`: Vector database namespace (default: "Default")
+- `index_name`: Vector database index name (default: "testbot768")
+
+Response: List of documents in the namespace 

requirements.txt

@@ -40,4 +40,8 @@ watchfiles==0.21.0
 
 # Core dependencies
 starlette==0.27.0
-psutil==5.9.6 
+psutil==5.9.6 
+
+# Upload PDF
+pypdf==3.17.4
+