@dudousxd/nestjs-agent-rag-media

Attach a file through @dudousxd/nestjs-media and it becomes searchable by the agent automatically: extract text, chunk, embed, index — scoped to its owner, with delete kept in sync.

It couples to the media library through the diagnostics channel — the same aviary:media:* seam Telescope rides — so it has no hard dependency on @dudousxd/nestjs-media. You supply a one-line readFile; the integration listens for attach and delete events.

Install

pnpm add @dudousxd/nestjs-agent-rag-media @dudousxd/nestjs-agent-rag @dudousxd/nestjs-agent-core

Wire it

import { AgentMediaIngestionModule } from '@dudousxd/nestjs-agent-rag-media';

@Module({
  imports: [
    AgentMediaIngestionModule.forRoot({
      store,                           // any VectorStore — Pg, Redis, Memory
      embedder,                        // any EmbeddingProvider
      collections: ['knowledge-base'], // omit = every collection
      readFile: (disk, path) => media.disk(disk).get(path), // the only glue
    }),
  ],
})
export class AppModule {}

Every file attached to knowledge-base is now chunked, embedded, and indexed with { mediaId, ownerType, ownerId, collection } metadata; deleting the record removes its chunks.

API

Owner-scoped retrieval

Ingestion stamps the owner onto every chunk, so a per-user / per-tenant retriever is one wrapper — FilteredRetriever from -rag:

import { EmbeddingRetriever, FilteredRetriever } from '@dudousxd/nestjs-agent-rag';

const base = new EmbeddingRetriever(embedder, store);
const forThisUser = new FilteredRetriever(base, { ownerId: actor.id });

Pass forThisUser to createRetrievalTool(...) or forRoot({ retrieval }) and the agent only ever retrieves that owner's documents. The fixed filter takes precedence, so a crafted query can't widen the scope.

Custom extractors

defaultTextExtractor() decodes text/* and JSON, strips text/html, and skips anything else (unsupported types are dropped, not indexed as binary garbage). Register a parser to add a format:

import { defaultTextExtractor } from '@dudousxd/nestjs-agent-rag-media';
import pdfParse from 'pdf-parse';

const extractor = defaultTextExtractor().register('application/pdf', async (bytes) =>
  (await pdfParse(bytes)).text,
);

AgentMediaIngestionModule.forRoot({ store, embedder, readFile, extractor });

Ingesting conversions (PDF, OCR)

For binary formats, the better move is to let the media library's own conversion pipeline produce the text and ingest that — no PDF/OCR parser on the RAG side. Point conversions at the conversion names you want; resolve maps the (owner-less) conversion event back to an ingestable descriptor, reusing the media record's id so the derived text shares the original's document id.

AgentMediaIngestionModule.forRoot({
  store,
  embedder,
  readFile,
  conversions: {
    names: ['text'], // media conversion names to ingest; others ignored
    resolve: async ({ id, path }) => {
      const record = await mediaLibrary.find(id);
      if (!record) return null;
      return {
        id, // same document id as the original
        ownerType: record.ownerType,
        ownerId: record.ownerId,
        collection: record.collection,
        disk: record.conversions.text.disk,
        path,
        mimeType: 'text/plain',
        size: record.conversions.text.size,
      };
    },
  },
});

Keeping the index in sync

Ingestion rides the diagnostics channel, so it's eventual and best-effort: the vectors update just after the media event fires. That covers the normal MediaLibrary.attach() / delete() path — but an event can be missed if the process was down when it fired, or if a record was deleted straight in the database, bypassing the library. reconcileMediaRag repairs that drift by diffing the source of truth against the index:

import { reconcileMediaRag } from '@dudousxd/nestjs-agent-rag-media';

const { ingested, removed } = await reconcileMediaRag(
  { ownerType: 'user', ownerId: actor.id, collection: 'knowledge-base' },
  {
    store,
    embedder,
    readFile,
    // your source of truth — the media records currently attached
    source: {
      listMedia: ({ ownerType, ownerId, collection }) =>
        mediaLibrary.list(ownerType, ownerId, collection),
    },
  },
);

It ingests media that isn't indexed yet, re-ingests media whose content changed (the indexed size fingerprint no longer matches — catching a file replaced while nobody was listening), and removes indexed documents whose media record is gone — touching only the difference, so it's cheap to run on a schedule or at boot. It leans on VectorStore.listDocuments(filter?) (new in -rag) to enumerate what's indexed for the owner.

At-least-once ingestion

AgentMediaIngestionModule ingests inline. For large files or an at-least-once guarantee, pass an enqueue hook — attach/delete are then handed to your durable queue instead of ingested inline, and a worker replays them with applyMediaIngestJob. This keeps durability opt-in: the package pulls in no durable dependency of its own.

import { applyMediaIngestJob } from '@dudousxd/nestjs-agent-rag-media';

AgentMediaIngestionModule.forRoot({
  store,
  embedder,
  readFile,
  enqueue: (job) => durableQueue.add('media-rag', job), // { type: 'ingest' | 'remove', event }
});

// in the durable worker — pass the same config object you gave the module:
async function handle(job) {
  await applyMediaIngestJob(job, { store, embedder, readFile });
}

Observability

Each file emits an aviary:rag:media.* diagnostics event — media.ingested (with chunk count), media.removed, media.skipped (unsupported-type / too-large / empty-text), media.failed. The Telescope bridge captures them automatically.

Related

• @dudousxd/nestjs-agent-rag — the RAG package this builds on
• RAG guide — modes, citations, hybrid + reranking, owner scoping
• @dudousxd/nestjs-media — the upload library it ingests from