From 9c012b46a1ffd461106fd63331faebb346fc78ec Mon Sep 17 00:00:00 2001 From: timothycarambat Date: Wed, 18 Oct 2023 17:29:43 -0700 Subject: [PATCH] Add Hourly, Daily, Weekly, and Monthly cron function to workers --- backend/models/queue.js | 3 +- backend/models/ragTest.js | 6 + workers/functions/cron/ragTesting/index.js | 193 +++++++++++++++++++++ workers/functions/runRAGTest/index.js | 4 +- workers/index.js | 7 + 5 files changed, 210 insertions(+), 3 deletions(-) create mode 100644 workers/functions/cron/ragTesting/index.js diff --git a/backend/models/queue.js b/backend/models/queue.js index 1097a6c..2603498 100644 --- a/backend/models/queue.js +++ b/backend/models/queue.js @@ -83,8 +83,9 @@ const Queue = { }); }, - updateJob: async function (jobId, status, result) { + updateJob: async function (jobId = null, status, result) { try { + if (!jobId) return null; const updatedJob = await prisma.jobs.update({ where: { id: Number(jobId) }, data: { diff --git a/backend/models/ragTest.js b/backend/models/ragTest.js index 43c7689..69fd493 100644 --- a/backend/models/ragTest.js +++ b/backend/models/ragTest.js @@ -1,6 +1,12 @@ const prisma = require("../utils/prisma"); const RagTest = { + schedules: { + hourly: "hourly", + daily: "daily", + weekly: "weekly", + monthly: "monthly", + }, status: { running: "running", failed: "failed", diff --git a/workers/functions/cron/ragTesting/index.js b/workers/functions/cron/ragTesting/index.js new file mode 100644 index 0000000..1cda60d --- /dev/null +++ b/workers/functions/cron/ragTesting/index.js @@ -0,0 +1,193 @@ +const { RagTest } = require('../../../../backend/models/ragTest'); +const { InngestClient } = require('../../../utils/inngest'); + +const runHourlyRagTest = InngestClient.createFunction( + { + id: 'rag-testing-hourly-cron', + name: 'RAG Test for Workspace - Hourly Cron', + }, + { cron: 'TZ=America/Los_Angeles * * * * *' }, + async () => { + try { + console.log(`\n\x1b[33m[CRON: Running hourly RAG Test Cron]\x1b[0m`); + const eligibleTests = await RagTest.where( + { + enabled: true, + frequencyType: RagTest.schedules.hourly, + }, + null, + null, + { + id: true, + enabled: true, + organization: true, + } + ); + if (eligibleTests.length === 0) return; + + console.log( + `\x1b[34m[INFO]\x1b[0m ${eligibleTests.length} eligible tests to run. Fanning out workers.` + ); + eligibleTests.forEach(async (test) => { + await InngestClient.send({ + name: 'rag-test/run', + data: { + jobId: null, // not associated with a job. + organization: test.organization, + testId: test.id, + }, + }); + }); + + return; + } catch (e) { + console.error(e.message); + return; + } + } +); + +const runDailyRagTest = InngestClient.createFunction( + { + id: 'rag-testing-daily-cron', + name: 'RAG Test for Workspace - Daily Cron', + }, + { cron: 'TZ=America/Los_Angeles 0 18 * * *' }, + async () => { + try { + console.log(`\n\x1b[33m[CRON: Running daily RAG Test Cron]\x1b[0m`); + const eligibleTests = await RagTest.where( + { + enabled: true, + frequencyType: RagTest.schedules.daily, + }, + null, + null, + { + id: true, + enabled: true, + organization: true, + } + ); + if (eligibleTests.length === 0) return; + + console.log( + `\x1b[34m[INFO]\x1b[0m ${eligibleTests.length} eligible tests to run. Fanning out workers.` + ); + eligibleTests.forEach(async (test) => { + await InngestClient.send({ + name: 'rag-test/run', + data: { + jobId: null, // not associated with a job. + organization: test.organization, + testId: test.id, + }, + }); + }); + + return; + } catch (e) { + console.error(e.message); + return; + } + } +); + +const runWeeklyRagTest = InngestClient.createFunction( + { + id: 'rag-testing-weekly-cron', + name: 'RAG Test for Workspace - Weekly Cron', + }, + { cron: 'TZ=America/Los_Angeles 0 18 * * WED' }, + async () => { + try { + console.log(`\n\x1b[33m[CRON: Running weekly RAG Test Cron]\x1b[0m`); + const eligibleTests = await RagTest.where( + { + enabled: true, + frequencyType: RagTest.schedules.weekly, + }, + null, + null, + { + id: true, + enabled: true, + organization: true, + } + ); + if (eligibleTests.length === 0) return; + + console.log( + `\x1b[34m[INFO]\x1b[0m ${eligibleTests.length} eligible tests to run. Fanning out workers.` + ); + eligibleTests.forEach(async (test) => { + await InngestClient.send({ + name: 'rag-test/run', + data: { + jobId: null, // not associated with a job. + organization: test.organization, + testId: test.id, + }, + }); + }); + + return; + } catch (e) { + console.error(e.message); + return; + } + } +); + +const runMonthlyRagTest = InngestClient.createFunction( + { + id: 'rag-testing-monthly-cron', + name: 'RAG Test for Workspace - Monthly Cron', + }, + { cron: 'TZ=America/Los_Angeles 0 18 15 * *' }, + async () => { + try { + console.log(`\n\x1b[33m[CRON: Running monthly RAG Test Cron]\x1b[0m`); + const eligibleTests = await RagTest.where( + { + enabled: true, + frequencyType: RagTest.schedules.monthly, + }, + null, + null, + { + id: true, + enabled: true, + organization: true, + } + ); + if (eligibleTests.length === 0) return; + + console.log( + `\x1b[34m[INFO]\x1b[0m ${eligibleTests.length} eligible tests to run. Fanning out workers.` + ); + eligibleTests.forEach(async (test) => { + await InngestClient.send({ + name: 'rag-test/run', + data: { + jobId: null, // not associated with a job. + organization: test.organization, + testId: test.id, + }, + }); + }); + + return; + } catch (e) { + console.error(e.message); + return; + } + } +); + +module.exports = { + runHourlyRagTest, + runDailyRagTest, + runWeeklyRagTest, + runMonthlyRagTest, +}; diff --git a/workers/functions/runRAGTest/index.js b/workers/functions/runRAGTest/index.js index 57b6c01..e0e9d52 100644 --- a/workers/functions/runRAGTest/index.js +++ b/workers/functions/runRAGTest/index.js @@ -12,13 +12,13 @@ const { } = require('../../../backend/utils/vectordatabases/providers'); const { InngestClient } = require('../../utils/inngest'); -const DELTA_THRESHOLD = 0.3; +const DELTA_THRESHOLD = 0.25; const runRAGTest = InngestClient.createFunction( { name: 'RAG Test for Workspace' }, { event: 'rag-test/run' }, async ({ event, step: _step, logger }) => { var result = {}; - const { organization, testId, jobId } = event.data; + const { organization, testId, jobId = null } = event.data; const { run } = await RagTest.createRun(testId, RagTest.status.running); try { diff --git a/workers/index.js b/workers/index.js index 10fc6b0..1edf100 100644 --- a/workers/index.js +++ b/workers/index.js @@ -36,6 +36,7 @@ const { addWeaviateDocuments } = require("./functions/addWeaviateDocuments"); const { migrateOrganization } = require("./functions/migrateOrganization"); const { resetOrganization } = require("./functions/resetOrganization"); const { runRAGTest } = require("./functions/runRAGTest"); +const { runHourlyRagTest, runDailyRagTest, runWeeklyRagTest, runMonthlyRagTest } = require("./functions/cron/ragTesting"); const app = express(); app.use(cors({ origin: true })); @@ -104,6 +105,12 @@ app.use( // RAGTesting runRAGTest, + + // Cron Jobs + runHourlyRagTest, + runDailyRagTest, + runWeeklyRagTest, + runMonthlyRagTest, ], { landingPage: true }) );