A modern, scalable backend for an AI-powered teaching assistant that automates grading, provides intelligent feedback, and enables Socratic teaching methods.
The backend is built with FastAPI and follows modern Python async patterns with a clean, layered architecture:
βββ app/ # Main application code
β βββ api/ # API route handlers
β β βββ analytics.py # Analytics & reporting endpoints
β β βββ assignments.py # Assignment management API
β β βββ auth.py # Authentication endpoints
β β βββ files.py # File upload/download API
β β βββ grading.py # AI grading operations
β β βββ health.py # Health check endpoints
β β βββ notifications.py # Notification management
β β βββ socratic.py # Socratic teaching API
β β βββ submissions.py # Submission handling
β β βββ users.py # User management
β β βββ websocket.py # Real-time WebSocket API
β βββ core/ # Core functionality & utilities
β β βββ auth.py # Authentication & authorization logic
β β βββ config.py # Application configuration
β β βββ events.py # Application lifecycle events
β β βββ exceptions.py # Custom exception classes
β β βββ logging.py # Structured logging setup
β β βββ middleware.py # HTTP middleware (CORS, logging, etc.)
β β βββ monitoring.py # Prometheus metrics & monitoring
β β βββ security.py # Security utilities (password hashing, JWT)
β βββ db/ # Database layer
β β βββ models/ # SQLAlchemy ORM models
β β β βββ assignment.py # Assignment model
β β β βββ base.py # Base model class
β β β βββ notification.py # Notification model
β β β βββ socratic.py # Socratic session models
β β β βββ submission.py # Submission model
β β β βββ user.py # User model with roles
β β βββ migrations/ # Alembic database migrations
β β βββ repositories/ # Data access layer (Repository pattern)
β β β βββ assignment.py # Assignment repository
β β β βββ base.py # Base repository class
β β β βββ notification.py # Notification repository
β β β βββ socratic.py # Socratic repository
β β β βββ user.py # User repository
β β βββ init_db.py # Database initialization & seed data
β β βββ session.py # Database session management
β βββ services/ # Business logic services
β β βββ document_parser.py # PDF/document parsing
β β βββ grading_agent.py # AI grading service using LangChain
β β βββ socratic.py # Socratic teaching logic
β β βββ vector_store.py # Vector embeddings & similarity search
β β βββ websocket.py # WebSocket connection manager
β βββ tasks/ # Background tasks (Celery)
β β βββ analytics.py # Analytics data processing
β β βββ celery.py # Celery configuration
β β βββ file_processing.py # File processing tasks
β β βββ grading.py # Async grading tasks
β β βββ monitoring.py # System monitoring tasks
β β βββ notifications.py # Email/notification tasks
β βββ main.py # FastAPI application entry point
βββ tests/ # Test suite
βββ scripts/ # Utility scripts
βββ data/ # Data storage (assignments, submissions, etc.)
βββ requirements-*.txt # Dependencies organized by environment
- Python 3.10+
- PostgreSQL 14+ (or use Docker)
- Redis 6+ (or use Docker)
- Elasticsearch 8+ (optional, for analytics)
# Clone the repository
git clone <repository-url>
cd ai-ta-grading-agent/backend
# Create virtual environment
python -m venv .venv
source .venv/bin/activate # On Windows: venv\Scripts\activate
# Install dependencies
pip install -r requirements-dev.txt # For development
# OR
pip install -r requirements-core.txt # For minimal setup# Copy environment template
cp .env.example .env
# Edit configuration (required settings)
# - DATABASE_URL: PostgreSQL connection string
# - SECRET_KEY: JWT secret key
# - OPENAI_API_KEY: OpenAI API key for AI grading
# - REDIS_URL: Redis connection string# Initialize database with sample data
python -c "
import asyncio
from app.db.init_db import init_db
asyncio.run(init_db())
"
# Or run migrations manually
alembic upgrade head# Development server (with auto-reload)
uvicorn app.main:app --reload --host 0.0.0.0 --port 8000
# Production server
gunicorn app.main:app -w 4 -k uvicorn.workers.UvicornWorkerVisit these URLs to confirm everything is working:
- API Documentation: http://localhost:8000/docs
- Health Check: http://localhost:8000/api/v1/health
- Metrics: http://localhost:8000/metrics
We use a modular requirements structure for different environments:
| File | Purpose | Usage |
|---|---|---|
requirements-core.txt |
28 essential packages | pip install -r requirements-core.txt |
requirements-dev.txt |
Core + development tools | pip install -r requirements-dev.txt |
requirements-prod.txt |
Core + production extras | pip install -r requirements-prod.txt |
requirements-categorized.txt |
Detailed breakdown | Reference documentation |
- π Web Framework: FastAPI, Uvicorn, Starlette, Pydantic
- ποΈ Database: SQLAlchemy, Alembic, Redis, Elasticsearch
- π€ AI/ML: LangChain, OpenAI, Hugging Face, NumPy, Pandas
- π Storage: MinIO, AioFiles
- π Security: Passlib, Python-JOSE
- π Monitoring: Prometheus, Structlog, Psutil
- βοΈ Communication: AioSMTPLib, Aio-Pika
app/main.py- Entry point that creates the FastAPI appcreate_application()- Configures the app with:- Middleware setup (CORS, logging, monitoring)
- API router registration
- Event handlers (startup/shutdown)
- Exception handlers
- Startup Events - Initialize connections to:
- Database (PostgreSQL)
- Cache (Redis)
- Search (Elasticsearch)
- Message Queue (RabbitMQ)
- API Routers - Register all endpoint handlers from
app/api/
main.py β create_application() β middleware β routers β services β repositories β models
# Run all tests
pytest
# Run with coverage
pytest --cov=app tests/
# Run specific test categories
pytest tests/test_db.py # Database tests
pytest tests/test_auth.py # Authentication tests
pytest tests/test_health.py # Health check testsFor local testing, you can comment/uncomment these in your .env:
# Disable external services for local testing
# ELASTICSEARCH_URL=http://localhost:9200 # Comment out to disable analytics
# RABBITMQ_HOST=localhost # Comment out to disable task queue
# MINIO_URL=localhost:9000 # Comment out to use local file storage
# Use simpler backends for development
DATABASE_URL=sqlite:///./test.db # Use SQLite instead of PostgreSQL
REDIS_URL=redis://localhost:6379/0 # Use local Redis| Endpoint | Purpose | Usage |
|---|---|---|
/api/v1/health |
Application health | Basic liveness check |
/api/v1/health/detailed |
Detailed health | Database, Redis, external services |
/metrics |
Prometheus metrics | System metrics for monitoring |
/docs |
API Documentation | Interactive Swagger UI |
/redoc |
Alternative docs | ReDoc interface |
{
"status": "healthy",
"timestamp": "2024-01-20T10:30:00Z",
"version": "0.1.0",
"services": {
"database": "healthy",
"redis": "healthy",
"elasticsearch": "healthy"
}
}The application exposes these metrics:
- Request counts by endpoint and status code
- Response times (latency histograms)
- Active connections count
- Database connection pool metrics
- Custom business metrics (grading operations, user activity)
- π Admin: Full system access
- π¨βπ« Instructor: Manage assignments and grade submissions
- π¨βπ TA: Grade submissions and assist students
- π€ Student: Submit assignments and view feedback
# Login to get JWT token
curl -X POST "/api/v1/auth/login" \
-d "username=user@example.com&password=password"
# Use token in requests
curl -H "Authorization: Bearer <token>" "/api/v1/assignments"# Build image
docker build -t ai-ta-backend .
# Run with docker-compose (includes PostgreSQL, Redis)
docker-compose up -d-
Database Connection Error
# Check PostgreSQL is running pg_isready -h localhost -p 5432 # Verify connection string in .env DATABASE_URL=postgresql://user:password@localhost:5432/dbname
-
OpenAI API Errors
# Verify API key is set echo $OPENAI_API_KEY # Test API key curl -H "Authorization: Bearer $OPENAI_API_KEY" \ https://api.openai.com/v1/models
-
Import Errors
# Ensure virtual environment is activated which python # Should show venv path # Reinstall dependencies pip install -r requirements-dev.txt
# Enable debug logging
export LOG_LEVEL=DEBUG
# Run with reload and debug
uvicorn app.main:app --reload --log-level debug- Async I/O: All database and external service calls are async
- Connection Pooling: PostgreSQL and Redis connection pools
- Caching: Multi-level caching strategy with Redis
- Background Tasks: Celery for heavy operations (grading, analytics)
- Horizontal: Add more FastAPI workers/containers
- Database: PostgreSQL read replicas
- Cache: Redis cluster
- Storage: S3/MinIO for file storage
- Search: Elasticsearch cluster for analytics
- Follow PEP 8 code style
- Add type hints to all functions
- Write tests for new features
- Update documentation for API changes
- Use conventional commit messages
MIT License - see LICENSE file for details.