From 7b602dbde9748634e7442f05123b68f89dadd897 Mon Sep 17 00:00:00 2001 From: Degentle12 Date: Mon, 24 Aug 2026 11:17:49 +0000 Subject: [PATCH] feat(backend): add RAG pipeline for the AGI tutor over course content Builds a retrieval-augmented generation pipeline so the AGI tutor answers from indexed course material instead of a static surface: - backend/src/services/tutor/: chunking + deterministic embeddings, a Qdrant vector store (with in-memory fallback for tests), content provider, answer generators (extractive by default, OpenAI-compatible when OPENAI_API_KEY is set) and grounding/faithfulness metrics. - backend/src/workers/indexingJob.ts: content indexing worker, started on server boot, that embeds and upserts course chunks idempotently. - Routes: POST /api/agi-tutor/rag/ask returns grounded answers with inline citations and metrics; POST /api/agi-tutor/rag/index triggers indexing; GET /api/agi-tutor/rag/status reports store state. - Answers fall back to a safe "I don't know" when retrieval or faithfulness confidence is below the configured threshold. - docker-compose.yml: adds a Qdrant vector store container. Closes #406 --- backend/.env.example | 17 ++ backend/src/__tests__/ragPipeline.test.ts | 257 ++++++++++++++++++ backend/src/controllers/agiTutorController.ts | 73 +++++ backend/src/index.ts | 8 + backend/src/routes/agiTutorRoutes.ts | 56 ++++ backend/src/services/tutor/contentProvider.ts | 212 +++++++++++++++ backend/src/services/tutor/embeddings.ts | 104 +++++++ backend/src/services/tutor/generators.ts | 209 ++++++++++++++ .../src/services/tutor/groundingMetrics.ts | 140 ++++++++++ backend/src/services/tutor/index.ts | 44 +++ backend/src/services/tutor/ragPipeline.ts | 257 ++++++++++++++++++ backend/src/services/tutor/types.ts | 90 ++++++ backend/src/services/tutor/vectorStore.ts | 250 +++++++++++++++++ backend/src/workers/indexingJob.ts | 53 ++++ docker-compose.yml | 27 ++ 15 files changed, 1797 insertions(+) create mode 100644 backend/src/__tests__/ragPipeline.test.ts create mode 100644 backend/src/services/tutor/contentProvider.ts create mode 100644 backend/src/services/tutor/embeddings.ts create mode 100644 backend/src/services/tutor/generators.ts create mode 100644 backend/src/services/tutor/groundingMetrics.ts create mode 100644 backend/src/services/tutor/index.ts create mode 100644 backend/src/services/tutor/ragPipeline.ts create mode 100644 backend/src/services/tutor/types.ts create mode 100644 backend/src/services/tutor/vectorStore.ts create mode 100644 backend/src/workers/indexingJob.ts diff --git a/backend/.env.example b/backend/.env.example index b3176fba..dcb3ec6d 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -168,3 +168,20 @@ GLOSSARY_TECH_ID=your-tech-glossary-id GLOSSARY_SUBTITLE_ID=your-subtitle-glossary-id TRANSLATION_CACHE_TTL=2592000 TRANSLATION_QUALITY_THRESHOLD=0.85 + +# ============================================================================= +# AGI Tutor RAG Pipeline Configuration (Issue #406) +# ============================================================================= +# Vector store mode: "qdrant" (default) or "memory" (tests / no-Docker dev) +RAG_VECTOR_STORE=qdrant +QDRANT_URL=http://localhost:6333 +QDRANT_COLLECTION=aethermint_course_content +RAG_EMBEDDING_DIM=768 +RAG_TOP_K=5 +RAG_CONFIDENCE_THRESHOLD=0.55 +# Content indexing worker poll interval (ms) +RAG_INDEX_INTERVAL_MS=300000 +# Optional LLM for answer generation. When unset, a deterministic +# extractive generator grounded in the retrieved passages is used. +OPENAI_API_KEY= +RAG_LLM_MODEL=gpt-4o-mini diff --git a/backend/src/__tests__/ragPipeline.test.ts b/backend/src/__tests__/ragPipeline.test.ts new file mode 100644 index 00000000..82c34c2c --- /dev/null +++ b/backend/src/__tests__/ragPipeline.test.ts @@ -0,0 +1,257 @@ +/** + * Tests for the AGI tutor RAG pipeline (Issue #406). + * + * All tests run against the in-memory vector store so they are + * deterministic and require no external services. + */ + +import { + LocalHashEmbedder, + cosineSimilarity, + MemoryVectorStore, + SeedCourseContentProvider, + RagPipeline, + faithfulnessScore, + citationCoverage, + computeGroundingMetrics, + OpenAiAnswerGenerator, +} from '../services/tutor'; + +describe('LocalHashEmbedder', () => { + const embedder = new LocalHashEmbedder(256); + + it('produces fixed-dimension, unit-norm vectors', () => { + const vector = embedder.embed('smart contracts run on a blockchain'); + expect(vector).toHaveLength(256); + const norm = Math.sqrt(vector.reduce((sum, v) => sum + v * v, 0)); + expect(norm).toBeCloseTo(1, 5); + }); + + it('is deterministic for the same input', () => { + expect(embedder.embed('what is a smart contract?')).toEqual( + embedder.embed('what is a smart contract?') + ); + }); + + it('ranks related text above unrelated text', () => { + const query = embedder.embed('what is a smart contract?'); + const related = embedder.embed( + 'A smart contract is a program that runs on a blockchain.' + ); + const unrelated = embedder.embed('the quick brown fox jumps over the lazy dog'); + + expect(cosineSimilarity(query, related)).toBeGreaterThan( + cosineSimilarity(query, unrelated) + ); + }); +}); + +describe('MemoryVectorStore', () => { + const store = new MemoryVectorStore({ collection: 'test' }); + const embedder = new LocalHashEmbedder(256); + + beforeEach(async () => { + await store.clear(); + }); + + it('upserts, counts, and searches by similarity', async () => { + const chunks = [ + { + id: 'c1', + courseId: 'course_a', + courseTitle: 'Course A', + moduleId: 'm1', + moduleTitle: 'Module 1', + lessonId: 'l1', + lessonTitle: 'Smart Contracts', + title: 'Smart Contracts', + content: 'A smart contract is a program that runs on a blockchain.', + contentType: 'lesson' as const, + }, + { + id: 'c2', + courseId: 'course_a', + courseTitle: 'Course A', + moduleId: 'm2', + moduleTitle: 'Module 2', + lessonId: 'l2', + lessonTitle: 'Graph Theory', + title: 'Graph Theory', + content: 'A graph is a collection of vertices connected by edges.', + contentType: 'lesson' as const, + }, + ]; + const vectors = chunks.map((c) => embedder.embed(c.content)); + + await store.upsert(chunks, vectors); + expect(await store.count()).toBe(2); + + const results = await store.search(embedder.embed('smart contracts'), 1); + expect(results).toHaveLength(1); + expect(results[0].chunk.id).toBe('c1'); + expect(results[0].score).toBeGreaterThan(0); + expect(results[0].excerpt.length).toBeGreaterThan(0); + }); + + it('replaces existing chunks on re-upsert', async () => { + const chunk = { + id: 'c1', + courseId: 'course_a', + courseTitle: 'Course A', + moduleId: 'm1', + moduleTitle: 'Module 1', + title: 'Lesson', + content: 'version one', + contentType: 'lesson' as const, + }; + await store.upsert([chunk], [embedder.embed(chunk.content)]); + await store.upsert([{ ...chunk, content: 'version two' }], [embedder.embed('version two')]); + expect(await store.count()).toBe(1); + }); +}); + +describe('Grounding metrics', () => { + const passage = (content: string) => ({ + chunk: { + id: 'p1', + courseId: 'course_a', + courseTitle: 'Course A', + moduleId: 'm1', + moduleTitle: 'Module 1', + title: 'Lesson', + content, + contentType: 'lesson' as const, + }, + score: 0.8, + excerpt: content.slice(0, 100), + }); + + it('scores grounded answers as faithful', () => { + const source = + 'Consensus is the process by which participants in a blockchain network agree on the state of the ledger. Proof of Work requires miners to solve a computationally expensive puzzle.'; + const answer = + 'Consensus is the process by which participants agree on the state of the ledger [1]. Proof of Work requires solving a computationally expensive puzzle [1].'; + const metrics = computeGroundingMetrics('what is consensus?', answer, [ + passage(source), + ]); + expect(metrics.faithfulnessScore).toBeGreaterThan(0.9); + expect(metrics.citationCoverage).toBe(1); + expect(metrics.grounded).toBe(true); + }); + + it('scores answers unrelated to the context as unfaithful', () => { + const source = + 'Consensus is the process by which participants in a blockchain network agree on the state of the ledger.'; + const answer = 'The capital of France is Paris, a city famous for its art.'; + const metrics = computeGroundingMetrics('what is the capital?', answer, [ + passage(source), + ]); + expect(metrics.faithfulnessScore).toBeLessThan(0.5); + expect(metrics.grounded).toBe(false); + }); + + it('citation coverage reflects inline markers', () => { + expect(citationCoverage('First claim [1]. Second claim [2].')).toBe(1); + expect(citationCoverage('First claim. Second claim.')).toBe(0); + }); + + it('faithfulness handles empty inputs', () => { + expect(faithfulnessScore('', [passage('some content')])).toBe(0); + }); +}); + +describe('RagPipeline (in-memory)', () => { + const buildPipeline = () => + new RagPipeline({ + store: new MemoryVectorStore({ collection: 'test' }), + provider: new SeedCourseContentProvider(), + embedder: new LocalHashEmbedder(768), + }); + + it('indexes seed course content into the store', async () => { + const pipeline = buildPipeline(); + const result = await pipeline.indexContent(); + + expect(result.indexed).toBeGreaterThan(0); + expect(result.chunkCount).toBe(result.indexed); + + const status = await pipeline.getStatus(); + expect(status.chunkCount).toBe(result.indexed); + expect(status.lastIndexedAt).not.toBeNull(); + }); + + it('skips re-indexing when content is unchanged', async () => { + const pipeline = buildPipeline(); + await pipeline.indexContent(); + const second = await pipeline.indexContent(); + + expect(second.indexed).toBe(0); + expect(second.skipped).toBeGreaterThan(0); + }); + + it('answers a question grounded in course material with citations', async () => { + const pipeline = buildPipeline(); + await pipeline.indexContent(); + + const answer = await pipeline.answer('what is a smart contract?'); + + expect(answer.fallback).toBe(false); + expect(answer.grounded).toBe(true); + expect(answer.answer).not.toContain("don't know"); + expect(answer.citations.length).toBeGreaterThan(0); + expect(answer.citations[0]).toMatchObject({ + courseId: 'course_blockchain_fundamentals', + }); + expect(answer.sources.length).toBeGreaterThan(0); + expect(answer.metrics.confidence).toBeGreaterThanOrEqual(answer.metrics.threshold); + expect(answer.metrics.faithfulnessScore).toBeGreaterThan(0.5); + expect(answer.metrics.citationCoverage).toBeGreaterThan(0); + }); + + it('falls back to a safe "I don\'t know" when confidence is low', async () => { + const pipeline = buildPipeline(); + await pipeline.indexContent(); + + const answer = await pipeline.answer( + 'what is the population of the city of Atlantis?' + ); + + expect(answer.fallback).toBe(true); + expect(answer.grounded).toBe(false); + expect(answer.answer).toContain("don't know"); + expect(answer.citations).toHaveLength(0); + expect(answer.sources).toHaveLength(0); + }); + + it('answers without indexing (empty store) via fallback', async () => { + const pipeline = buildPipeline(); + const answer = await pipeline.answer('what is a smart contract?'); + + expect(answer.fallback).toBe(true); + expect(answer.answer).toContain("don't know"); + }); +}); + +describe('OpenAiAnswerGenerator', () => { + it('is unavailable when no API key is configured', () => { + const previous = process.env.OPENAI_API_KEY; + delete process.env.OPENAI_API_KEY; + const generator = new OpenAiAnswerGenerator(); + expect(generator.available).toBe(false); + if (previous !== undefined) { + process.env.OPENAI_API_KEY = previous; + } + }); + + it('throws when asked to generate without an API key', async () => { + const previous = process.env.OPENAI_API_KEY; + delete process.env.OPENAI_API_KEY; + const generator = new OpenAiAnswerGenerator(); + await expect(generator.generate('question', [])).rejects.toThrow( + 'OPENAI_API_KEY' + ); + if (previous !== undefined) { + process.env.OPENAI_API_KEY = previous; + } + }); +}); diff --git a/backend/src/controllers/agiTutorController.ts b/backend/src/controllers/agiTutorController.ts index dcd1c31d..a5b347ff 100644 --- a/backend/src/controllers/agiTutorController.ts +++ b/backend/src/controllers/agiTutorController.ts @@ -4,6 +4,7 @@ import { UniversalKnowledgeService } from '../services/universalKnowledgeService import { StudentAdaptationService } from '../services/studentAdaptationService'; import { EmotionalIntelligenceService } from '../services/emotionalIntelligenceService'; import { CrossDomainIntegrationService } from '../services/crossDomainIntegrationService'; +import { ragPipeline } from '../services/tutor'; import logger from '../utils/logger'; export class AGITutorController { @@ -289,4 +290,76 @@ export class AGITutorController { }); } } + + /** + * Answer a question using the RAG pipeline over indexed course content + */ + async askRagQuestion(req: Request, res: Response) { + try { + const { question, topK } = req.body; + + if (!question || typeof question !== 'string' || question.trim() === '') { + res.status(400).json({ + success: false, + error: 'question is required' + }); + return; + } + + const answer = await ragPipeline.answer(question, { + topK: typeof topK === 'number' && topK > 0 ? topK : undefined + }); + + res.json({ + success: true, + data: answer + }); + } catch (error) { + logger.error('Error answering RAG question:', error); + res.status(500).json({ + success: false, + error: 'Failed to answer question' + }); + } + } + + /** + * Trigger indexing of course content into the vector store + */ + async triggerRagIndexing(req: Request, res: Response) { + try { + const result = await ragPipeline.indexContent(); + + res.json({ + success: true, + data: result + }); + } catch (error) { + logger.error('Error indexing course content:', error); + res.status(500).json({ + success: false, + error: 'Failed to index course content' + }); + } + } + + /** + * Get RAG pipeline indexing status + */ + async getRagStatus(req: Request, res: Response) { + try { + const status = await ragPipeline.getStatus(); + + res.json({ + success: true, + data: status + }); + } catch (error) { + logger.error('Error getting RAG status:', error); + res.status(500).json({ + success: false, + error: 'Failed to get RAG status' + }); + } + } } diff --git a/backend/src/index.ts b/backend/src/index.ts index ad04c5bf..d4ae0527 100644 --- a/backend/src/index.ts +++ b/backend/src/index.ts @@ -43,6 +43,8 @@ import * as transactionEvents from './events/transactionEvents'; // Bridge relayer monitor watch job — Issue #423 import { bridgeMonitorJob } from './workers/bridgeMonitorJob'; import { processQuestionGenerationJob } from './workers/questionGenJob'; +// Course content indexing worker — Issue #406 (AGI tutor RAG pipeline) +import { startIndexingJob, stopIndexingJob } from './workers/indexingJob'; import questionGeneratorService from './services/questionGen/questionGenerator'; // Background job queue — Issue #258 @@ -543,6 +545,11 @@ async function startServer() { await bridgeMonitorJob.start(); } + // Start the course content indexing worker for the AGI tutor RAG + // pipeline (Issue #406). Fails open when the vector store is + // unreachable so single-node deployments keep working. + startIndexingJob(); + // Initialise the presence & availability system (Issue #405). Fails open // when Redis is unreachable so single-node deployments keep working. await presenceService.initialize(); @@ -665,6 +672,7 @@ if (require.main === module) { { name: 'transaction-processor', run: () => typeof (transactionProcessor as any).stop === 'function' && (transactionProcessor as any).stop() }, { name: 'transaction-events', run: () => typeof (transactionEvents as any).stopListening === 'function' && (transactionEvents as any).stopListening() }, { name: 'presence', run: () => presenceService.destroy() }, + { name: 'indexing-job', run: () => stopIndexingJob() }, { name: 'job-queue', run: async () => { try { const jq = getJobQueue(); await jq.destroy(); } catch { /* queue may not be initialised */ } } }, { name: 'redis', diff --git a/backend/src/routes/agiTutorRoutes.ts b/backend/src/routes/agiTutorRoutes.ts index 88aa5497..64ad9e49 100644 --- a/backend/src/routes/agiTutorRoutes.ts +++ b/backend/src/routes/agiTutorRoutes.ts @@ -123,4 +123,60 @@ router.post('/emotional-support', async (req, res) => { await agiTutorController.provideEmotionalSupport(req, res); }); +/** + * @openapi + * /api/agi-tutor/rag/ask: + * post: + * tags: [AGITutor] + * summary: Answer a question with retrieval-augmented generation over course content + * requestBody: + * required: true + * content: + * application/json: + * schema: + * type: object + * required: [question] + * properties: + * question: + * type: string + * topK: + * type: number + * responses: + * '200': + * description: Grounded answer with citations and grounding metrics + * '400': + * description: Missing question + */ +router.post('/rag/ask', async (req, res) => { + await agiTutorController.askRagQuestion(req, res); +}); + +/** + * @openapi + * /api/agi-tutor/rag/index: + * post: + * tags: [AGITutor] + * summary: Trigger indexing of course content into the vector store + * responses: + * '200': + * description: Indexing completed + */ +router.post('/rag/index', async (req, res) => { + await agiTutorController.triggerRagIndexing(req, res); +}); + +/** + * @openapi + * /api/agi-tutor/rag/status: + * get: + * tags: [AGITutor] + * summary: Get RAG pipeline indexing status + * responses: + * '200': + * description: Indexing status + */ +router.get('/rag/status', async (req, res) => { + await agiTutorController.getRagStatus(req, res); +}); + export default router; diff --git a/backend/src/services/tutor/contentProvider.ts b/backend/src/services/tutor/contentProvider.ts new file mode 100644 index 00000000..4135e921 --- /dev/null +++ b/backend/src/services/tutor/contentProvider.ts @@ -0,0 +1,212 @@ +/** + * Course Content Provider for the AGI Tutor RAG pipeline. + * + * Supplies course material to be chunked and embedded into the vector + * store. The default implementation ships with a curated seed dataset so + * the pipeline works out of the box and tests stay deterministic. The + * interface is deliberately small so a database-backed provider (e.g. + * reading published `Content`/`Course` records) can be swapped in without + * touching the rest of the pipeline. + */ + +import logger from '../../utils/logger'; +import { DocumentChunk } from './types'; + +export interface CourseContentProvider { + getCourseContent(): Promise; +} + +interface SeedLesson { + id: string; + title: string; + content: string; +} + +interface SeedModule { + id: string; + title: string; + lessons: SeedLesson[]; +} + +interface SeedCourse { + id: string; + title: string; + modules: SeedModule[]; +} + +const MAX_CHUNK_LENGTH = 1000; +const CHUNK_OVERLAP = 150; + +const SEED_COURSES: SeedCourse[] = [ + { + id: 'course_blockchain_fundamentals', + title: 'Introduction to Blockchain', + modules: [ + { + id: 'mod_what_is_blockchain', + title: 'What is a Blockchain?', + lessons: [ + { + id: 'lesson_distributed_ledger', + title: 'Distributed Ledgers', + content: + 'A blockchain is a distributed ledger that records transactions across many computers so that the record cannot be altered retroactively without the alteration of all subsequent blocks and the consensus of the network. Each participant in the network maintains a copy of the ledger, which removes the need for a central authority. Because every node holds the same history, tampering with a single copy is immediately detectable. Blocks are linked using cryptographic hashes: each block contains the hash of the previous block, forming an unbroken chain back to the genesis block.', + }, + { + id: 'lesson_consensus', + title: 'Consensus Mechanisms', + content: + 'Consensus is the process by which participants in a blockchain network agree on the state of the ledger. Proof of Work requires miners to solve a computationally expensive puzzle before proposing a new block, which makes attacks costly. Proof of Stake instead selects validators based on the amount of cryptocurrency they lock up as collateral, which is far more energy efficient. Consensus ensures that all honest nodes converge on the same canonical chain even when some nodes behave maliciously or the network is partitioned.', + }, + ], + }, + { + id: 'mod_smart_contracts', + title: 'Smart Contracts', + lessons: [ + { + id: 'lesson_smart_contract_basics', + title: 'Smart Contract Basics', + content: + 'A smart contract is a program that runs on a blockchain and executes automatically when predetermined conditions are met. Smart contracts enable trustless agreements because the code, not any single party, enforces the terms. They are deterministic: given the same input, every node that executes the contract reaches the same result, which is why contracts are typically written in restricted languages without nondeterministic operations. Common use cases include token transfers, escrow services, decentralized finance, and automated payments.', + }, + { + id: 'lesson_gas_and_fees', + title: 'Gas and Transaction Fees', + content: + 'Executing a smart contract consumes computational resources, and users pay for those resources in the network currency. On Ethereum this cost is called gas, and the total fee is the gas used multiplied by the gas price. Estimating gas correctly matters: setting the limit too low causes the transaction to fail and revert, while setting it too high wastes funds. Fee markets prioritise transactions that offer higher prices, which is why fees spike during periods of network congestion.', + }, + ], + }, + ], + }, + { + id: 'course_decentralized_finance', + title: 'Decentralized Finance (DeFi)', + modules: [ + { + id: 'mod_defi_primitives', + title: 'DeFi Primitives', + lessons: [ + { + id: 'lesson_lending', + title: 'Lending and Borrowing Protocols', + content: + 'Decentralized lending protocols allow users to deposit assets into a pool and earn interest, or borrow against their collateral without a bank. Borrowers must over-collateralize their positions because the protocol cannot perform credit checks. Interest rates are often determined algorithmically based on the utilization of the pool: when demand for borrowing is high, rates rise to attract more suppliers. If the value of collateral falls below the required ratio, the position can be liquidated to protect lenders.', + }, + { + id: 'lesson_amm', + title: 'Automated Market Makers', + content: + 'An automated market maker (AMM) is a smart contract that provides liquidity and sets prices using a mathematical formula rather than an order book. The constant product formula, where the product of the quantities of two assets in a pool stays constant, is the most famous example. Trades move the price along the curve, so large trades relative to pool size cause significant slippage. Liquidity providers earn a share of trading fees but also take on impermanent loss when the relative price of the pooled assets changes.', + }, + ], + }, + ], + }, + { + id: 'course_math_for_cs', + title: 'Mathematics for Computer Science', + modules: [ + { + id: 'mod_discrete_math', + title: 'Discrete Mathematics', + lessons: [ + { + id: 'lesson_graph_theory', + title: 'Graph Theory', + content: + 'A graph is a collection of vertices connected by edges, and it is one of the most important structures in computer science. Graphs model networks of all kinds: social connections, roads, the web, and dependency relationships. A tree is a connected graph with no cycles. Breadth-first search explores a graph level by level and finds the shortest path in unweighted graphs, while depth-first search explores as far as possible along each branch before backtracking and is useful for detecting cycles and topological sorting.', + }, + { + id: 'lesson_logic', + title: 'Propositional Logic', + content: + 'Propositional logic studies statements that are either true or false and the connectives that combine them: conjunction, disjunction, negation, implication, and biconditional. A truth table lists the truth value of a compound statement for every combination of its atomic propositions. Logical equivalence means two statements have identical truth tables. Proof techniques such as modus ponens, proof by contrapositive, and proof by contradiction are fundamental to reasoning about the correctness of algorithms and programs.', + }, + ], + }, + ], + }, +]; + +/** + * Split long lesson text into overlapping windows so each stored chunk stays + * within the embedding model's practical context size. + */ +function chunkText(text: string, maxLength: number, overlap: number): string[] { + const cleaned = text.replace(/\s+/g, ' ').trim(); + if (cleaned.length <= maxLength) { + return [cleaned]; + } + const chunks: string[] = []; + let start = 0; + while (start < cleaned.length) { + let end = Math.min(start + maxLength, cleaned.length); + if (end < cleaned.length) { + // Break on a word boundary when possible. + const boundary = cleaned.lastIndexOf(' ', end); + if (boundary > start + maxLength / 2) { + end = boundary; + } + } + chunks.push(cleaned.slice(start, end).trim()); + if (end >= cleaned.length) { + break; + } + start = Math.max(end - overlap, start + 1); + } + return chunks; +} + +export class SeedCourseContentProvider implements CourseContentProvider { + async getCourseContent(): Promise { + const chunks: DocumentChunk[] = []; + + for (const course of SEED_COURSES) { + for (const module of course.modules) { + chunks.push({ + id: `chunk_${course.id}_${module.id}`, + courseId: course.id, + courseTitle: course.title, + moduleId: module.id, + moduleTitle: module.title, + title: `${course.title} — ${module.title}`, + content: module.title, + contentType: 'module', + metadata: { source: 'seed' }, + }); + + for (const lesson of module.lessons) { + const parts = chunkText( + lesson.content, + MAX_CHUNK_LENGTH, + CHUNK_OVERLAP + ); + parts.forEach((part, index) => { + chunks.push({ + id: `chunk_${course.id}_${module.id}_${lesson.id}_${index}`, + courseId: course.id, + courseTitle: course.title, + moduleId: module.id, + moduleTitle: module.title, + lessonId: lesson.id, + lessonTitle: lesson.title, + title: lesson.title, + content: part, + contentType: 'lesson', + metadata: { source: 'seed', part: parts.length > 1 ? index + 1 : undefined }, + }); + }); + } + } + } + + logger.info(`Seed content provider produced ${chunks.length} chunks`); + return chunks; + } +} + +export function getCourseContentProvider(): CourseContentProvider { + return new SeedCourseContentProvider(); +} diff --git a/backend/src/services/tutor/embeddings.ts b/backend/src/services/tutor/embeddings.ts new file mode 100644 index 00000000..96bd503d --- /dev/null +++ b/backend/src/services/tutor/embeddings.ts @@ -0,0 +1,104 @@ +/** + * Embedding Service for the AGI Tutor RAG pipeline. + * + * Produces deterministic, keyless embeddings using feature hashing: + * character n-grams and word tokens are hashed into a fixed-dimension + * vector with signed weights, then L2-normalised. Cosine similarity over + * these vectors gives a stable, offline-computable notion of topical + * closeness, which keeps the pipeline fully self-contained (no external + * embedding API required) while remaining deterministic for tests. + */ + +export interface Embedder { + readonly dimension: number; + embed(text: string): number[]; +} + +export class LocalHashEmbedder implements Embedder { + readonly dimension: number; + + constructor(dimension: number = 768) { + this.dimension = dimension; + } + + embed(text: string): number[] { + const vector = new Float64Array(this.dimension); + const normalized = text.toLowerCase().replace(/\s+/g, ' ').trim(); + if (normalized.length === 0) { + return Array.from(vector); + } + + for (const feature of this.extractFeatures(normalized)) { + const hash = this.hashFeature(feature); + const index = Math.abs(hash) % this.dimension; + vector[index] += hash > 0 ? 1 : -1; + } + + return this.normalize(vector); + } + + private extractFeatures(text: string): string[] { + const features: string[] = []; + + // Word unigrams (length > 1 to skip noise tokens). + for (const token of text.split(/[^a-z0-9]+/).filter(Boolean)) { + if (token.length > 1) { + features.push(`w:${token}`); + } + } + + // Character n-grams (2-4) capture morphology and near-miss spellings. + const compact = text.replace(/[^a-z0-9]/g, ''); + for (let n = 2; n <= 4; n++) { + for (let i = 0; i + n <= compact.length; i++) { + features.push(`c${n}:${compact.slice(i, i + n)}`); + } + } + + return features; + } + + /** FNV-1a 32-bit hash returned as a signed 32-bit integer. */ + private hashFeature(feature: string): number { + let hash = 0x811c9dc5; + for (let i = 0; i < feature.length; i++) { + hash ^= feature.charCodeAt(i); + hash = Math.imul(hash, 0x01000193); + } + return hash | 0; + } + + private normalize(vector: Float64Array): number[] { + let norm = 0; + for (let i = 0; i < vector.length; i++) { + norm += vector[i] * vector[i]; + } + norm = Math.sqrt(norm); + if (norm === 0) { + return Array.from(vector); + } + return Array.from(vector, (v) => v / norm); + } +} + +export function cosineSimilarity(a: number[], b: number[]): number { + const length = Math.min(a.length, b.length); + let dot = 0; + let normA = 0; + let normB = 0; + for (let i = 0; i < length; i++) { + dot += a[i] * b[i]; + normA += a[i] * a[i]; + normB += b[i] * b[i]; + } + if (normA === 0 || normB === 0) { + return 0; + } + return dot / (Math.sqrt(normA) * Math.sqrt(normB)); +} + +export function getEmbedder(dimension?: number): Embedder { + const resolved = + dimension ?? parseInt(process.env.RAG_EMBEDDING_DIM ?? '768', 10); + return new LocalHashEmbedder(resolved); +} diff --git a/backend/src/services/tutor/generators.ts b/backend/src/services/tutor/generators.ts new file mode 100644 index 00000000..4d36f18e --- /dev/null +++ b/backend/src/services/tutor/generators.ts @@ -0,0 +1,209 @@ +/** + * Answer Generators for the AGI Tutor RAG pipeline. + * + * A generator turns retrieved passages into a grounded answer with inline + * citation markers. Two implementations exist: + * - ExtractiveAnswerGenerator: deterministic, offline, keyless. It + * assembles the most relevant sentences from the retrieved passages and + * tags each with its citation. Used by default and as a fallback. + * - OpenAiAnswerGenerator: used when OPENAI_API_KEY is configured. The + * model is instructed to answer strictly from the provided context and + * to mark sources inline as [1], [2], ... + */ + +import axios from 'axios'; +import { Citation, RetrievedPassage } from './types'; + +export interface GeneratedAnswer { + answer: string; + citations: Citation[]; + model: string; +} + +export interface AnswerGenerator { + readonly name: string; + generate(question: string, passages: RetrievedPassage[]): Promise; +} + +const SIGNIFICANT_WORD_MIN_LENGTH = 3; +const SIGNIFICANT_WORD_BLACKLIST = new Set([ + 'the', 'and', 'for', 'are', 'was', 'with', 'from', 'that', 'this', + 'what', 'how', 'why', 'does', 'can', 'you', 'your', 'about', 'which', +]); + +export class ExtractiveAnswerGenerator implements AnswerGenerator { + readonly name = 'extractive'; + + async generate( + question: string, + passages: RetrievedPassage[] + ): Promise { + if (passages.length === 0) { + return { answer: '', citations: [], model: this.name }; + } + + const keywords = this.significantWords(question); + const selected: { passage: RetrievedPassage; sentences: string[] }[] = []; + + for (const passage of passages.slice(0, 3)) { + const sentences = this.splitSentences(passage.chunk.content); + const relevant = sentences.filter((sentence) => + this.containsAny(sentence, keywords) + ); + if (relevant.length === 0) { + continue; + } + selected.push({ passage, sentences: relevant.slice(0, 2) }); + if (selected.length >= 2) { + break; + } + } + + // No retrieved passage answers the question — signal the pipeline to + // fall back to the safe "I don't know" response rather than quoting + // unrelated content. + if (selected.length === 0) { + return { answer: '', citations: [], model: this.name }; + } + + const citations: Citation[] = []; + const parts: string[] = []; + + selected.forEach(({ passage, sentences }, passageIndex) => { + const index = passageIndex + 1; + citations.push({ + index, + sourceId: passage.chunk.id, + courseId: passage.chunk.courseId, + courseTitle: passage.chunk.courseTitle, + moduleTitle: passage.chunk.moduleTitle, + lessonTitle: passage.chunk.lessonTitle, + title: passage.chunk.title, + excerpt: passage.excerpt, + sourceUrl: passage.chunk.sourceUrl, + }); + for (const sentence of sentences) { + parts.push(`${sentence} [${index}]`); + } + }); + + const answer = `Based on the course material, here is what the retrieved lessons say:\n\n${parts.join(' ')}`; + return { answer, citations, model: this.name }; + } + + private splitSentences(text: string): string[] { + return text + .split(/(?<=[.!?])\s+/) + .map((sentence) => sentence.trim()) + .filter((sentence) => sentence.length > 0); + } + + private significantWords(question: string): string[] { + return question + .toLowerCase() + .split(/[^a-z0-9]+/) + .filter( + (word) => + word.length >= SIGNIFICANT_WORD_MIN_LENGTH && + !SIGNIFICANT_WORD_BLACKLIST.has(word) + ); + } + + private containsAny(sentence: string, words: string[]): boolean { + const normalized = sentence.toLowerCase(); + return words.some((word) => normalized.includes(word)); + } +} + +export class OpenAiAnswerGenerator implements AnswerGenerator { + readonly name = 'openai'; + private readonly apiKey: string; + private readonly model: string; + + constructor() { + this.apiKey = process.env.OPENAI_API_KEY ?? ''; + this.model = process.env.RAG_LLM_MODEL ?? 'gpt-4o-mini'; + } + + get available(): boolean { + return this.apiKey.length > 0; + } + + async generate( + question: string, + passages: RetrievedPassage[] + ): Promise { + if (!this.available) { + throw new Error('OPENAI_API_KEY is not configured'); + } + + const context = passages + .map((passage, i) => `[${i + 1}] ${passage.chunk.content}`) + .join('\n\n'); + + const { data } = await axios.post( + 'https://api.openai.com/v1/chat/completions', + { + model: this.model, + temperature: 0.2, + messages: [ + { + role: 'system', + content: + 'You are a tutor that answers strictly from the provided course material. ' + + 'Answer only using the context passages. Cite the source of each claim inline ' + + 'with a bracket number such as [1] or [2]. If the context does not contain ' + + 'enough information to answer, reply exactly with: I don\'t know.', + }, + { + role: 'user', + content: `Question: ${question}\n\nContext:\n${context}`, + }, + ], + }, + { + headers: { Authorization: `Bearer ${this.apiKey}` }, + timeout: 30000, + } + ); + + const content: string = data?.choices?.[0]?.message?.content ?? ''; + return { + answer: content, + citations: this.extractCitations(content, passages), + model: this.model, + }; + } + + private extractCitations( + answer: string, + passages: RetrievedPassage[] + ): Citation[] { + const used = new Set(); + const pattern = /\[(\d+)\]/g; + let match: RegExpExecArray | null; + while ((match = pattern.exec(answer)) !== null) { + const index = parseInt(match[1], 10); + if (index >= 1 && index <= passages.length) { + used.add(index); + } + } + + return Array.from(used) + .sort((a, b) => a - b) + .map((index) => { + const passage = passages[index - 1]; + return { + index, + sourceId: passage.chunk.id, + courseId: passage.chunk.courseId, + courseTitle: passage.chunk.courseTitle, + moduleTitle: passage.chunk.moduleTitle, + lessonTitle: passage.chunk.lessonTitle, + title: passage.chunk.title, + excerpt: passage.excerpt, + sourceUrl: passage.chunk.sourceUrl, + }; + }); + } +} diff --git a/backend/src/services/tutor/groundingMetrics.ts b/backend/src/services/tutor/groundingMetrics.ts new file mode 100644 index 00000000..d8e99141 --- /dev/null +++ b/backend/src/services/tutor/groundingMetrics.ts @@ -0,0 +1,140 @@ +/** + * Grounding & Faithfulness Metrics for the AGI Tutor RAG pipeline. + * + * Computes, per answer: + * - retrievalConfidence: max similarity of the retrieved passages; + * - contextRelevance: average similarity of the retrieved passages; + * - faithfulnessScore: how much of the answer text is supported by the + * retrieved passages, estimated via character n-gram containment; + * - citationCoverage: share of answer sentences carrying a citation marker. + * + * The overall confidence is a weighted blend of retrieval confidence and + * faithfulness, compared against a threshold to decide whether the answer + * is safe to surface or should fall back to "I don't know". + */ + +import { GroundingMetrics, RetrievedPassage } from './types'; + +export interface GroundingMetricsOptions { + retrievalWeight?: number; + faithfulnessWeight?: number; + threshold?: number; +} + +const NGRAM_SIZE = 3; +const MIN_SENTENCE_LENGTH = 8; + +function normalize(text: string): string { + return text.toLowerCase().replace(/[^a-z0-9]+/g, ' '); +} + +function ngrams(text: string, size: number): Set { + const normalized = normalize(text).replace(/\s+/g, ''); + const grams = new Set(); + for (let i = 0; i + size <= normalized.length; i++) { + grams.add(normalized.slice(i, i + size)); + } + return grams; +} + +function stripCitationMarkers(text: string): string { + return text.replace(/\[\d+\]/g, ' ').replace(/\s+/g, ' ').trim(); +} + +function splitSentences(text: string): string[] { + return text + .split(/(?<=[.!?])\s+/) + .map((sentence) => sentence.trim()) + .filter((sentence) => sentence.length >= MIN_SENTENCE_LENGTH); +} + +/** + * Faithfulness proxy: the mean, over answer sentences, of the fraction of + * the sentence's character n-grams that appear in the retrieved context. + */ +export function faithfulnessScore( + answer: string, + passages: RetrievedPassage[] +): number { + const cleaned = stripCitationMarkers(answer); + const sentences = splitSentences(cleaned); + if (sentences.length === 0) { + return 0; + } + + const sourceText = passages + .map((passage) => passage.chunk.content) + .join(' '); + const sourceGrams = ngrams(sourceText, NGRAM_SIZE); + if (sourceGrams.size === 0) { + return 0; + } + + let total = 0; + let counted = 0; + for (const sentence of sentences) { + const sentenceGrams = ngrams(sentence, NGRAM_SIZE); + if (sentenceGrams.size === 0) { + continue; + } + let contained = 0; + for (const gram of sentenceGrams) { + if (sourceGrams.has(gram)) { + contained++; + } + } + total += contained / sentenceGrams.size; + counted++; + } + + return counted === 0 ? 0 : total / counted; +} + +export function citationCoverage(answer: string): number { + const sentences = splitSentences(answer); + if (sentences.length === 0) { + return 0; + } + let cited = 0; + for (const sentence of sentences) { + if (/\[\d+\]/.test(sentence)) { + cited++; + } + } + return cited / sentences.length; +} + +export function computeGroundingMetrics( + question: string, + answer: string, + passages: RetrievedPassage[], + options: GroundingMetricsOptions = {} +): GroundingMetrics { + const retrievalWeight = options.retrievalWeight ?? 0.6; + const faithfulnessWeight = options.faithfulnessWeight ?? 0.4; + const threshold = + options.threshold ?? parseFloat(process.env.RAG_CONFIDENCE_THRESHOLD ?? '0.55'); + + const retrievalConfidence = + passages.length > 0 ? Math.max(...passages.map((p) => p.score)) : 0; + const contextRelevance = + passages.length > 0 + ? passages.reduce((sum, p) => sum + p.score, 0) / passages.length + : 0; + const faithfulness = faithfulnessScore(answer, passages); + const coverage = citationCoverage(answer); + + const confidence = + retrievalWeight * retrievalConfidence + + faithfulnessWeight * faithfulness; + + return { + retrievalConfidence, + faithfulnessScore: faithfulness, + citationCoverage: coverage, + contextRelevance, + confidence, + grounded: confidence >= threshold, + threshold, + }; +} diff --git a/backend/src/services/tutor/index.ts b/backend/src/services/tutor/index.ts new file mode 100644 index 00000000..0f113316 --- /dev/null +++ b/backend/src/services/tutor/index.ts @@ -0,0 +1,44 @@ +/** + * AGI Tutor RAG pipeline — public surface. + */ + +export { RagPipeline, ragPipeline } from './ragPipeline'; +export type { RagPipelineOptions } from './ragPipeline'; +export { + LocalHashEmbedder, + cosineSimilarity, + getEmbedder, +} from './embeddings'; +export type { Embedder } from './embeddings'; +export { + QdrantVectorStore, + MemoryVectorStore, + getVectorStore, + resetVectorStore, +} from './vectorStore'; +export type { VectorStore, VectorStoreOptions } from './vectorStore'; +export { + SeedCourseContentProvider, + getCourseContentProvider, +} from './contentProvider'; +export type { CourseContentProvider } from './contentProvider'; +export { + ExtractiveAnswerGenerator, + OpenAiAnswerGenerator, +} from './generators'; +export type { AnswerGenerator, GeneratedAnswer } from './generators'; +export { + computeGroundingMetrics, + faithfulnessScore, + citationCoverage, +} from './groundingMetrics'; +export type { GroundingMetricsOptions } from './groundingMetrics'; +export type { + DocumentChunk, + RetrievedPassage, + Citation, + GroundingMetrics, + RagAnswer, + IndexingStatus, + IndexingResult, +} from './types'; diff --git a/backend/src/services/tutor/ragPipeline.ts b/backend/src/services/tutor/ragPipeline.ts new file mode 100644 index 00000000..11a51fde --- /dev/null +++ b/backend/src/services/tutor/ragPipeline.ts @@ -0,0 +1,257 @@ +/** + * AGI Tutor RAG Pipeline + * + * Orchestrates the retrieval-augmented generation flow over course content: + * 1. Course material is chunked and embedded into a vector store + * (see contentProvider.ts and vectorStore.ts). + * 2. For each question, relevant passages are retrieved by embedding the + * question and searching the store. + * 3. An answer generator composes a response grounded in those passages + * with inline citations. + * 4. Grounding/faithfulness metrics decide whether the answer is safe to + * surface; below the confidence threshold the pipeline falls back to a + * safe "I don't know" response. + */ + +import { createHash } from 'crypto'; +import logger from '../../utils/logger'; +import { Embedder, getEmbedder } from './embeddings'; +import { VectorStore, getVectorStore } from './vectorStore'; +import { + CourseContentProvider, + getCourseContentProvider, +} from './contentProvider'; +import { + AnswerGenerator, + ExtractiveAnswerGenerator, + OpenAiAnswerGenerator, +} from './generators'; +import { computeGroundingMetrics } from './groundingMetrics'; +import { + IndexingResult, + IndexingStatus, + RagAnswer, + RetrievedPassage, +} from './types'; + +export interface RagPipelineOptions { + store?: VectorStore; + embedder?: Embedder; + provider?: CourseContentProvider; + generator?: AnswerGenerator; + topK?: number; + threshold?: number; +} + +const FALLBACK_MESSAGE = + "I don't know — the course material doesn't contain enough information to answer this question confidently."; + +export class RagPipeline { + private readonly store: VectorStore; + private readonly embedder: Embedder; + private readonly provider: CourseContentProvider; + private readonly generator: AnswerGenerator; + private readonly topK: number; + private readonly threshold: number; + + private isIndexing = false; + private lastContentHash: string | null = null; + private status: IndexingStatus = { + store: 'unknown', + collection: 'unknown', + chunkCount: 0, + isIndexing: false, + lastIndexedAt: null, + lastIndexDurationMs: null, + lastError: null, + totalIndexed: 0, + }; + + constructor(options: RagPipelineOptions = {}) { + this.store = + options.store ?? + getVectorStore({ collection: process.env.QDRANT_COLLECTION }); + this.embedder = options.embedder ?? getEmbedder(); + this.provider = options.provider ?? getCourseContentProvider(); + this.generator = + options.generator ?? this.resolveDefaultGenerator(); + this.topK = + options.topK ?? parseInt(process.env.RAG_TOP_K ?? '5', 10); + this.threshold = + options.threshold ?? + parseFloat(process.env.RAG_CONFIDENCE_THRESHOLD ?? '0.55'); + + this.status.store = this.store.name; + this.status.collection = this.store.collection; + } + + /** + * Answer a question against the indexed course material. Returns a + * grounded answer with citations and metrics, or a safe fallback when + * confidence is low. + */ + async answer( + question: string, + options: { topK?: number } = {} + ): Promise { + const vector = this.embedder.embed(question); + const topK = options.topK ?? this.topK; + + let passages: RetrievedPassage[]; + try { + passages = await this.store.search(vector, topK); + } catch (err) { + logger.warn( + 'RAG vector store search failed; answering with empty context', + err + ); + passages = []; + } + + let generated; + try { + generated = await this.generator.generate(question, passages); + } catch (err) { + logger.warn( + `RAG generator '${this.generator.name}' failed; using extractive fallback`, + err + ); + generated = await new ExtractiveAnswerGenerator().generate( + question, + passages + ); + } + + const metrics = computeGroundingMetrics(question, generated.answer, passages, { + threshold: this.threshold, + }); + + const safeToSurface = metrics.grounded && passages.length > 0; + const fallback = !safeToSurface; + + return { + question, + answer: safeToSurface ? generated.answer : FALLBACK_MESSAGE, + citations: safeToSurface ? generated.citations : [], + metrics, + sources: safeToSurface ? passages : [], + grounded: metrics.grounded, + fallback, + fallbackMessage: fallback ? FALLBACK_MESSAGE : undefined, + model: safeToSurface ? generated.model : 'none', + generatedAt: new Date().toISOString(), + }; + } + + /** + * Index (or re-index) course content into the vector store. Skips the + * work when the content corpus is unchanged since the last successful run. + */ + async indexContent(): Promise { + if (this.isIndexing) { + return { + indexed: 0, + skipped: 0, + durationMs: 0, + store: this.store.name, + collection: this.store.collection, + chunkCount: this.status.chunkCount, + }; + } + + this.isIndexing = true; + this.status.isIndexing = true; + const startedAt = Date.now(); + + try { + const chunks = await this.provider.getCourseContent(); + const contentHash = this.hashChunks(chunks); + + if (this.lastContentHash === contentHash && this.status.chunkCount > 0) { + const durationMs = Date.now() - startedAt; + logger.info( + `Course content unchanged; skipping re-index (${chunks.length} chunks)` + ); + return { + indexed: 0, + skipped: chunks.length, + durationMs, + store: this.store.name, + collection: this.store.collection, + chunkCount: this.status.chunkCount, + }; + } + + const vectors = chunks.map((chunk) => this.embedder.embed(chunk.content)); + await this.store.ensureCollection(); + await this.store.upsert(chunks, vectors); + + const chunkCount = await this.store.count(); + const durationMs = Date.now() - startedAt; + + this.lastContentHash = contentHash; + this.status = { + store: this.store.name, + collection: this.store.collection, + chunkCount, + isIndexing: false, + lastIndexedAt: new Date().toISOString(), + lastIndexDurationMs: durationMs, + lastError: null, + totalIndexed: this.status.totalIndexed + chunks.length, + }; + + logger.info( + `Indexed ${chunks.length} course chunks into ${this.store.name} in ${durationMs}ms` + ); + return { + indexed: chunks.length, + skipped: 0, + durationMs, + store: this.store.name, + collection: this.store.collection, + chunkCount, + }; + } catch (err) { + this.status.lastError = + err instanceof Error ? err.message : 'Unknown indexing error'; + this.status.lastIndexedAt = null; + logger.error('Failed to index course content', err); + throw err; + } finally { + this.isIndexing = false; + this.status.isIndexing = false; + } + } + + async getStatus(): Promise { + try { + this.status.chunkCount = await this.store.count(); + } catch { + // Keep the cached count if the store is unreachable. + } + return { ...this.status }; + } + + async health(): Promise<{ ok: boolean; detail?: string }> { + return this.store.health(); + } + + private resolveDefaultGenerator(): AnswerGenerator { + const openAi = new OpenAiAnswerGenerator(); + return openAi.available ? openAi : new ExtractiveAnswerGenerator(); + } + + private hashChunks(chunks: { id: string; content: string }[]): string { + const hash = createHash('sha256'); + for (const chunk of chunks) { + hash.update(chunk.id); + hash.update('\x00'); + hash.update(chunk.content); + hash.update('\x00'); + } + return hash.digest('hex'); + } +} + +export const ragPipeline = new RagPipeline(); diff --git a/backend/src/services/tutor/types.ts b/backend/src/services/tutor/types.ts new file mode 100644 index 00000000..939c1e2a --- /dev/null +++ b/backend/src/services/tutor/types.ts @@ -0,0 +1,90 @@ +/** + * AGI Tutor RAG Pipeline Types + * + * Types shared across the retrieval-augmented generation pipeline: + * content chunking, vector retrieval, answer generation, citation + * tracking and grounding/faithfulness metrics. + */ + +export interface DocumentChunk { + id: string; + courseId: string; + courseTitle: string; + moduleId: string; + moduleTitle: string; + lessonId?: string; + lessonTitle?: string; + title: string; + content: string; + sourceUrl?: string; + contentType: 'course' | 'module' | 'lesson' | 'resource'; + metadata?: Record; +} + +export interface RetrievedPassage { + chunk: DocumentChunk; + score: number; + excerpt: string; +} + +export interface Citation { + index: number; + sourceId: string; + courseId: string; + courseTitle: string; + moduleTitle: string; + lessonTitle?: string; + title: string; + excerpt: string; + sourceUrl?: string; +} + +export interface GroundingMetrics { + /** Max similarity of the retrieved passages (how confident retrieval was). */ + retrievalConfidence: number; + /** Fraction of answer content supported by the retrieved passages (0-1). */ + faithfulnessScore: number; + /** Fraction of answer sentences carrying an inline citation marker (0-1). */ + citationCoverage: number; + /** Average similarity of the retrieved passages (0-1). */ + contextRelevance: number; + /** Overall confidence, a weighted blend of retrieval and faithfulness. */ + confidence: number; + /** Whether the answer meets the grounding threshold. */ + grounded: boolean; + /** The confidence threshold the answer was evaluated against. */ + threshold: number; +} + +export interface RagAnswer { + question: string; + answer: string; + citations: Citation[]; + metrics: GroundingMetrics; + sources: RetrievedPassage[]; + grounded: boolean; + fallback: boolean; + fallbackMessage?: string; + model: string; + generatedAt: string; +} + +export interface IndexingStatus { + store: string; + collection: string; + chunkCount: number; + isIndexing: boolean; + lastIndexedAt: string | null; + lastIndexDurationMs: number | null; + lastError: string | null; + totalIndexed: number; +} + +export interface IndexingResult { + indexed: number; + skipped: number; + durationMs: number; + store: string; + collection: string; + chunkCount: number; +} diff --git a/backend/src/services/tutor/vectorStore.ts b/backend/src/services/tutor/vectorStore.ts new file mode 100644 index 00000000..ba08b728 --- /dev/null +++ b/backend/src/services/tutor/vectorStore.ts @@ -0,0 +1,250 @@ +/** + * Vector Store for the AGI Tutor RAG pipeline. + * + * Two implementations back the same interface: + * - QdrantVectorStore: the production store (see the `qdrant` service in + * docker-compose.yml), accessed through Qdrant's REST API. + * - MemoryVectorStore: an in-memory brute-force store used in tests and as + * a failsafe when Qdrant is unreachable, so the pipeline degrades + * gracefully instead of erroring. + */ + +import axios from 'axios'; +import { createHash } from 'crypto'; +import logger from '../../utils/logger'; +import { DocumentChunk, RetrievedPassage } from './types'; +import { cosineSimilarity } from './embeddings'; + +export const DEFAULT_COLLECTION = 'aethermint_course_content'; + +export interface VectorStore { + readonly name: string; + readonly collection: string; + ensureCollection(): Promise; + upsert(chunks: DocumentChunk[], vectors: number[][]): Promise; + search(vector: number[], topK: number): Promise; + count(): Promise; + clear(): Promise; + health(): Promise<{ ok: boolean; detail?: string }>; +} + +export interface VectorStoreOptions { + url?: string; + collection?: string; +} + +export class QdrantVectorStore implements VectorStore { + readonly name = 'qdrant'; + readonly collection: string; + private readonly baseUrl: string; + private readonly dimension: number; + + constructor(options: VectorStoreOptions = {}) { + this.collection = + options.collection ?? process.env.QDRANT_COLLECTION ?? DEFAULT_COLLECTION; + this.baseUrl = ( + options.url ?? process.env.QDRANT_URL ?? 'http://localhost:6333' + ).replace(/\/+$/, ''); + this.dimension = parseInt( + process.env.RAG_EMBEDDING_DIM ?? '768', + 10 + ); + } + + async ensureCollection(): Promise { + const url = `${this.baseUrl}/collections/${this.collection}`; + try { + await axios.get(url, { timeout: 5000 }); + } catch (err: unknown) { + const status = (err as { response?: { status?: number } })?.response?.status; + if (status === 404) { + await axios.put( + url, + { + vectors: { size: this.dimension, distance: 'Cosine' }, + }, + { timeout: 10000 } + ); + logger.info(`Created Qdrant collection '${this.collection}'`); + } else { + throw err; + } + } + } + + async upsert(chunks: DocumentChunk[], vectors: number[][]): Promise { + if (chunks.length === 0) { + return; + } + await this.ensureCollection(); + + const points = chunks.map((chunk, i) => ({ + id: this.pointId(chunk.id), + vector: vectors[i], + payload: { ...chunk }, + })); + + // Upsert in batches to stay well under Qdrant's per-request limits. + const batchSize = 64; + for (let i = 0; i < points.length; i += batchSize) { + const batch = points.slice(i, i + batchSize); + await axios.put( + `${this.baseUrl}/collections/${this.collection}/points?wait=true`, + { points: batch }, + { timeout: 30000 } + ); + } + } + + async search(vector: number[], topK: number): Promise { + if (topK <= 0) { + return []; + } + await this.ensureCollection(); + const { data } = await axios.post( + `${this.baseUrl}/collections/${this.collection}/points/search`, + { vector, limit: topK, with_payload: true }, + { timeout: 10000 } + ); + return (data?.result ?? []).map((hit: Record) => + this.toPassage(hit) + ); + } + + async count(): Promise { + try { + const { data } = await axios.get( + `${this.baseUrl}/collections/${this.collection}`, + { timeout: 5000 } + ); + const pointsCount = data?.result?.points_count as number | null; + return typeof pointsCount === 'number' ? pointsCount : 0; + } catch { + return 0; + } + } + + async clear(): Promise { + await axios.post( + `${this.baseUrl}/collections/${this.collection}/points/delete`, + { filter: { must: [] } }, + { timeout: 30000 } + ); + } + + async health(): Promise<{ ok: boolean; detail?: string }> { + try { + const { data } = await axios.get(`${this.baseUrl}/healthz`, { + timeout: 2000, + }); + return { ok: data?.status === 'ok', detail: 'qdrant reachable' }; + } catch (err) { + return { + ok: false, + detail: err instanceof Error ? err.message : 'qdrant unreachable', + }; + } + } + + /** Stable 64-bit (safe-integer) point id derived from a chunk id. */ + private pointId(chunkId: string): number { + const hex = createHash('sha1').update(chunkId).digest('hex').slice(0, 16); + const big = BigInt(`0x${hex}`); + return Number(big % BigInt(Number.MAX_SAFE_INTEGER)); + } + + private toPassage(hit: Record): RetrievedPassage { + const payload = (hit.payload ?? {}) as DocumentChunk; + const content = payload.content ?? ''; + return { + chunk: payload, + score: typeof hit.score === 'number' ? hit.score : 0, + excerpt: content.slice(0, 500), + }; + } +} + +export class MemoryVectorStore implements VectorStore { + readonly name = 'memory'; + readonly collection: string; + private items: { chunk: DocumentChunk; vector: number[] }[] = []; + + constructor(options: VectorStoreOptions = {}) { + this.collection = + options.collection ?? process.env.QDRANT_COLLECTION ?? DEFAULT_COLLECTION; + } + + async ensureCollection(): Promise { + // In-memory store has no collection lifecycle. + } + + async upsert(chunks: DocumentChunk[], vectors: number[][]): Promise { + for (let i = 0; i < chunks.length; i++) { + const existing = this.items.findIndex( + (item) => item.chunk.id === chunks[i].id + ); + const entry = { chunk: chunks[i], vector: vectors[i] }; + if (existing >= 0) { + this.items[existing] = entry; + } else { + this.items.push(entry); + } + } + } + + async search(vector: number[], topK: number): Promise { + if (topK <= 0) { + return []; + } + return this.items + .map((item) => ({ + chunk: item.chunk, + score: cosineSimilarity(vector, item.vector), + })) + .sort((a, b) => b.score - a.score) + .slice(0, topK) + .map((item) => ({ + chunk: item.chunk, + score: item.score, + excerpt: item.chunk.content.slice(0, 500), + })); + } + + async count(): Promise { + return this.items.length; + } + + async clear(): Promise { + this.items = []; + } + + async health(): Promise<{ ok: boolean; detail?: string }> { + return { ok: true, detail: 'in-memory store' }; + } +} + +let cachedStore: VectorStore | null = null; + +/** + * Resolve the configured vector store. Defaults to Qdrant; set + * RAG_VECTOR_STORE=memory (or NODE_ENV=test) for an in-memory store. + */ +export function getVectorStore(options?: VectorStoreOptions): VectorStore { + if (cachedStore) { + return cachedStore; + } + const mode = + process.env.NODE_ENV === 'test' + ? 'memory' + : (process.env.RAG_VECTOR_STORE ?? 'qdrant').toLowerCase(); + cachedStore = + mode === 'memory' + ? new MemoryVectorStore(options) + : new QdrantVectorStore(options); + return cachedStore; +} + +/** Reset the cached store (useful in tests). */ +export function resetVectorStore(): void { + cachedStore = null; +} diff --git a/backend/src/workers/indexingJob.ts b/backend/src/workers/indexingJob.ts new file mode 100644 index 00000000..0c7081aa --- /dev/null +++ b/backend/src/workers/indexingJob.ts @@ -0,0 +1,53 @@ +/** + * Course Content Indexing Worker (Issue #406) + * + * Periodically indexes course material into the vector store so the AGI + * tutor RAG pipeline can retrieve relevant context. The job is idempotent: + * unchanged content is skipped (see RagPipeline.indexContent) and failures + * are logged without crashing the process, so the worker fails open when the + * vector store is unreachable (e.g. local development without Docker). + */ + +import { ragPipeline } from '../services/tutor'; +import logger from '../utils/logger'; + +let intervalId: NodeJS.Timeout | null = null; +let isRunning = false; + +export const runIndexingJob = async (): Promise => { + if (isRunning) { + return; + } + isRunning = true; + try { + await ragPipeline.indexContent(); + } catch (err) { + logger.error('Course content indexing job failed', err); + } finally { + isRunning = false; + } +}; + +export const startIndexingJob = (intervalMs?: number): void => { + if (intervalId) { + return; + } + const resolvedInterval = + intervalMs ?? parseInt(process.env.RAG_INDEX_INTERVAL_MS ?? '300000', 10); + + // Index once at startup, then poll for content changes. + runIndexingJob(); + intervalId = setInterval(runIndexingJob, resolvedInterval); + logger.info( + `Course content indexing worker started (interval ${resolvedInterval}ms)` + ); +}; + +export const stopIndexingJob = (): void => { + if (intervalId) { + clearInterval(intervalId); + intervalId = null; + } + isRunning = false; + logger.info('Course content indexing worker stopped'); +}; diff --git a/docker-compose.yml b/docker-compose.yml index 19aede2d..61cd28ef 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -78,6 +78,8 @@ services: - REDIS_PORT=6379 - REDIS_PASSWORD=${REDIS_PASSWORD:-} - MONGODB_URI=mongodb://mongodb:27017/aethermint + - QDRANT_URL=http://qdrant:6333 + - QDRANT_COLLECTION=aethermint_course_content depends_on: postgres: condition: service_healthy @@ -163,6 +165,29 @@ services: retries: 5 start_period: 10s + # -------------------------------------------------------------------------- + # Qdrant - Vector store for the AGI tutor RAG pipeline (Issue #406) + # -------------------------------------------------------------------------- + qdrant: + image: qdrant/qdrant:v1.9.7 + container_name: aethermint-qdrant + ports: + - "6333:6333" + - "6334:6334" + environment: + - QDRANT__SERVICE__GRPC_PORT=6334 + volumes: + - qdrant_data:/qdrant/storage + networks: + - aethermint-network + restart: unless-stopped + healthcheck: + test: ["CMD-SHELL", "bash -c 'exec 3<>/dev/tcp/127.0.0.1/6333 && echo ok' || exit 1"] + interval: 10s + timeout: 5s + retries: 5 + start_period: 10s + # -------------------------------------------------------------------------- # Backup - on-demand PostgreSQL backup runner # -------------------------------------------------------------------------- @@ -219,3 +244,5 @@ volumes: name: aethermint-redis-data mongo_data: name: aethermint-mongo-data + qdrant_data: + name: aethermint-qdrant-data