Implementation

Technology stack, code structure, and key components

Technology Stack

Backend

FastAPI with async support

Database

MongoDB with Motor async driver

AI/ML

Google Gemini 2.5 Flash

Classification

Ollama (Qwen2 7B)

Full Technology List

Category	Technology	Purpose
Backend Framework	FastAPI	High-performance async API
Database	MongoDB + Motor	Flexible schema, async driver
AI/ML	Google Gemini 2.5 Flash	Chat generation, image parsing
AI/ML	Ollama (Qwen2 7B)	Local intent classification
String Matching	TheFuzz	Fuzzy keyword matching
WhatsApp	WPPConnect	WhatsApp Web automation
Scheduler	APScheduler	Data pipeline cron jobs

---

Key Components

Intent Classifier

Location: backend/app/llm_engine/classifier.py

Combines Strategy and Factory patterns:

CLASSIFIER_MAP = {
    "ollama": _create_ollama_classifier,
    "fuzzy": _create_fuzzy_classifier,
    "hybrid": _create_hybrid_classifier,
}

def get_classifier() -> IntentClassifier:
    method = settings.CLASSIFICATION_METHOD
    factory_fn = CLASSIFIER_MAP.get(method)
    return factory_fn()

Configure via .env: CLASSIFICATION_METHOD=hybrid

Context Strategies

Location: backend/app/strategies/context_strategies.py

Each strategy encapsulates data fetching + LLM formatting:

class DiningStrategy(ContextStrategy):
    async def fetch(self, query: str) -> str:
        raw_menus = await fetch_dining_data_raw()
        return self._format_for_llm(raw_menus)

Strategy	Data Source
DiningStrategy	MongoDB menus collection
AnnouncementStrategy	MongoDB announcements
GeneralStrategy	No database (fallback)

MongoDB Singleton

Location: backend/app/db/mongo.py

Thread-safe singleton with lazy initialization:

class MongoDB:
    def __init__(self):
        self._client = None
        self._initialized = False
    
    @property
    def client(self):
        if self._client is None:
            raise RuntimeError("Not connected")
        return self._client

db = MongoDB()  # Global singleton

Always call connect_to_mongo() in app startup event!

Data Pipeline

Location: data-pipeline/

Job	Schedule	Action
Menu Sync	Daily 8:00	Scrape → Gemini Vision → MongoDB
Announcement Sync	Hourly	Scrape → Parse → MongoDB

# Menu extraction with Gemini Vision
image_bytes = download_menu_image()
json_menu = await gemini_client.parse_image(image_bytes)
await db.menus.insert_one(json_menu)

---

Configuration

Environment Variables

# Database
MONGO_CONNECTION_STRING=mongodb://localhost:27017
MONGO_DB_NAME=chatbot

# AI Services
GEMINI_API_KEY=your_api_key
OLLAMA_BASE_URL=http://localhost:11434
OLLAMA_MODEL=qwen2:7b

# Classification
CLASSIFICATION_METHOD=hybrid
FUZZY_THRESHOLD=80

Pydantic Settings

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    MONGO_CONNECTION_STRING: str
    GEMINI_API_KEY: str
    CLASSIFICATION_METHOD: Literal["ollama", "fuzzy", "hybrid"] = "hybrid"
    
    model_config = SettingsConfigDict(env_file=".env")

settings = Settings()

---

Running the Project

1. Clone & Setup

   git clone https://github.com/Akdeniz-CSE-Students/csebot_documentation/.git
   cd ChatbotForCse
   cp .env.example .env  # Configure your API keys
   
2. Start Backend

   cd backend
   pip install -r requirements.txt
   uvicorn main:app --reload
   
3. Start WhatsApp Client

   cd whatsapp
   npm install
   npm start  # First run requires QR scan
   
4. Start Data Pipeline (Optional)

   cd data-pipeline
   pip install -r requirements.txt
   python jobs/run_scheduler.py
   

---

Testing

Running Tests

# Run classifier tests
cd backend
pytest test_classifier.py -v

Mock injection example:

def test_hybrid_fallback():
    reset_classifier()  # Reset singleton
    
    classifier = get_classifier()
    result = await classifier.classify("test message")
    
    assert result in ["dining", "announcement", "general"]

The reset_classifier() function allows injecting mock classifiers for testing.