Full-stack application combining LlamaIndex vector search with Neo4j knowledge graph (GraphRAG) for answering queries about Netflix marketing materials. Flask/Hypercorn backend with custom ReAct agent, React frontend. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
168 lines
No EOL
6.8 KiB
Python
168 lines
No EOL
6.8 KiB
Python
# netflix_chatbot/main.py
|
|
|
|
import asyncio
|
|
import os
|
|
import sys
|
|
from flask import Flask
|
|
from flask_cors import CORS
|
|
|
|
# Ensure the project directory is in the Python path
|
|
current_dir = os.path.dirname(os.path.abspath(__file__))
|
|
if current_dir not in sys.path:
|
|
sys.path.insert(0, current_dir)
|
|
|
|
# Import necessary components from our modules
|
|
from config import (
|
|
APPLICATION_ROOT, MAX_CONTENT_LENGTH,
|
|
CORS_ALLOWED_ORIGINS, CORS_SUPPORTS_CREDENTIALS,
|
|
SERVER_HOST, SERVER_PORT, USE_RELOADER, LOG_LEVEL,
|
|
KEEP_ALIVE_TIMEOUT, READ_TIMEOUT, WRITE_TIMEOUT
|
|
)
|
|
from utils import logger, log_structured
|
|
from json_utils import CustomJSONProvider
|
|
from ai_core import initialize_global_index # Import the initialization function
|
|
from shared_state import global_workflow_agent, is_agent_available # Import shared state
|
|
from routes import register_routes
|
|
from init_mongodb import init_mongodb # Your MongoDB initialization script
|
|
|
|
# --- Flask App Initialization ---
|
|
app = Flask(__name__)
|
|
|
|
# Apply custom JSON provider for handling special types (LlamaIndex objects, etc.)
|
|
app.json_provider_class = CustomJSONProvider
|
|
app.json = CustomJSONProvider(app)
|
|
|
|
# Configuration
|
|
app.config['MAX_CONTENT_LENGTH'] = MAX_CONTENT_LENGTH
|
|
if APPLICATION_ROOT:
|
|
app.config['APPLICATION_ROOT'] = APPLICATION_ROOT
|
|
# If using APPLICATION_ROOT, you might need to adjust route prefixes
|
|
# or use a Blueprint with url_prefix=APPLICATION_ROOT
|
|
log_structured('info', f"Flask Application Root set to: {APPLICATION_ROOT}")
|
|
|
|
# CORS Configuration
|
|
CORS(app,
|
|
resources={r"/*": {"origins": CORS_ALLOWED_ORIGINS}},
|
|
supports_credentials=CORS_SUPPORTS_CREDENTIALS,
|
|
# Expose custom headers if needed by the frontend
|
|
# expose_headers=["Content-Disposition"] # Example for downloads
|
|
)
|
|
log_structured('info', f"CORS configured for origins: {CORS_ALLOWED_ORIGINS}")
|
|
|
|
|
|
# --- Register Routes ---
|
|
# Pass the app object to the function in routes.py
|
|
register_routes(app)
|
|
log_structured('info', "Flask routes registered.")
|
|
|
|
|
|
# --- Startup Function ---
|
|
async def startup_event() -> bool:
|
|
"""Tasks to run when the application starts.
|
|
|
|
Returns:
|
|
bool: True if all startup tasks completed successfully, False otherwise
|
|
"""
|
|
log_structured('info', "Application startup sequence initiated.")
|
|
all_success = True
|
|
|
|
# 1. Initialize MongoDB Connection & Schema (using your script)
|
|
log_structured('info', "Initializing MongoDB connection...")
|
|
mongo_success = False
|
|
try:
|
|
if init_mongodb():
|
|
log_structured('info', "MongoDB initialized successfully.")
|
|
mongo_success = True
|
|
else:
|
|
log_structured('warning', "MongoDB initialization script finished, but reported issues.")
|
|
all_success = False
|
|
except Exception as db_err:
|
|
log_structured('critical', "FATAL: MongoDB initialization failed.", {'error': str(db_err)})
|
|
all_success = False
|
|
# We'll continue in a degraded state
|
|
|
|
# 2. Initialize Global AI Index and Agent
|
|
log_structured('info', "Initializing global AI index and agent...")
|
|
index_success = await initialize_global_index()
|
|
|
|
# Explicitly check the status after initialization
|
|
if not is_agent_available():
|
|
log_structured('critical', "After initialize_global_index, global_workflow_agent is still unavailable, even though function may have reported success")
|
|
all_success = False
|
|
elif not index_success:
|
|
log_structured('warning', "AI initialization reported failure, but will continue in degraded state")
|
|
all_success = False
|
|
else:
|
|
log_structured('info', "AI initialization successful, global_workflow_agent is available")
|
|
|
|
log_structured('info', f"Application startup sequence complete. Overall success: {all_success}")
|
|
return all_success
|
|
|
|
|
|
# --- Shutdown Function (Optional) ---
|
|
async def shutdown_event():
|
|
"""Tasks to run when the application stops."""
|
|
log_structured('info', "Application shutdown sequence initiated.")
|
|
# Add any cleanup tasks here (e.g., closing connections if not handled elsewhere)
|
|
# Note: Hypercorn might not always guarantee graceful shutdown execution.
|
|
log_structured('info', "Application shutdown sequence complete.")
|
|
|
|
|
|
# --- Main Execution Block ---
|
|
if __name__ == '__main__':
|
|
from hypercorn.config import Config as HypercornConfig
|
|
from hypercorn.asyncio import serve as hypercorn_serve
|
|
|
|
# Create Hypercorn config object
|
|
config = HypercornConfig()
|
|
|
|
# Basic settings
|
|
config.bind = [f"{SERVER_HOST}:{SERVER_PORT}"]
|
|
config.use_reloader = USE_RELOADER
|
|
config.accesslog = '-' # Log to stdout/stderr
|
|
config.errorlog = '-' # Log to stdout/stderr
|
|
config.loglevel = LOG_LEVEL.upper()
|
|
config.worker_class = 'asyncio'
|
|
|
|
# Timeouts (ensure these are floats or ints)
|
|
config.keep_alive_timeout = float(KEEP_ALIVE_TIMEOUT)
|
|
config.read_timeout = float(READ_TIMEOUT)
|
|
config.write_timeout = float(WRITE_TIMEOUT)
|
|
|
|
# Request size limits (check Hypercorn docs for exact names, might vary slightly)
|
|
# These might apply to HTTP/1.1 or HTTP/2 differently.
|
|
# config.h11_max_incomplete_size = MAX_CONTENT_LENGTH # Example for HTTP/1.1
|
|
# config.h2_max_concurrent_streams = 100 # Example for HTTP/2
|
|
# config.max_app_buffer_size = MAX_CONTENT_LENGTH # Another potential setting
|
|
# It's safer to configure these via a reverse proxy (like Nginx) in production.
|
|
# Hypercorn's defaults are usually reasonable. Let's comment these out for now.
|
|
|
|
# Assign startup and shutdown handlers
|
|
config.startup_hooks = [startup_event]
|
|
config.shutdown_hooks = [shutdown_event]
|
|
|
|
log_structured('info', f"Starting Hypercorn server on {SERVER_HOST}:{SERVER_PORT}")
|
|
log_structured('info', f"Reload mode: {'Enabled' if USE_RELOADER else 'Disabled'}")
|
|
|
|
# Execute startup task before running the server
|
|
log_structured('info', "Manually executing startup sequence before server start")
|
|
startup_success = asyncio.run(startup_event())
|
|
|
|
# Double-check that the agent is initialized
|
|
if not is_agent_available():
|
|
log_structured('critical', "After startup, global_workflow_agent is still unavailable. Forcing re-initialization...")
|
|
# Try once more to initialize
|
|
index_success = asyncio.run(initialize_global_index())
|
|
if not index_success or not is_agent_available():
|
|
log_structured('critical', "Emergency initialization also failed. Server will run but chat functionality will be impaired.")
|
|
else:
|
|
log_structured('info', "Emergency initialization succeeded.")
|
|
|
|
# Run the server
|
|
try:
|
|
asyncio.run(hypercorn_serve(app, config))
|
|
except KeyboardInterrupt:
|
|
log_structured('info', "Server stopped manually (KeyboardInterrupt).")
|
|
except Exception as run_err:
|
|
log_structured('critical', "Hypercorn server failed to run.", {'error': str(run_err)})
|
|
sys.exit(1) |