@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.

InstallCopy Anchor Link

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

Wire itCopy Anchor Link

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, size } metadata (size is the real byte length, the fingerprint the reconciler compares); deleting the record removes its chunks. Add your own keys (e.g. an ACL audience) with metadata: (event) => ({ … }) — merged over the defaults.

APICopy Anchor Link

maxBytes defaults to 25 MB. MediaAttachEvent.size is a declared value and can only reject early; the real byte length is re-checked after the read. Wire statFile: (disk, path) => Promise<number> (S3 HeadObject, fs.stat) so an oversized object is never downloaded — reconcileMediaRag uses it too, keeping its size fingerprint comparison real-vs-real.

Owner-scoped retrievalCopy Anchor Link

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 extractorsCopy Anchor Link

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)Copy Anchor Link

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 syncCopy Anchor Link

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. The result also lists reingested ids. It uses VectorStore.listDocuments(filter?) to enumerate what's indexed for the owner.

At-least-once ingestionCopy Anchor Link

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 }); // throws → the queue retries
}

If the enqueue call itself throws, the service logs it and publishes media.failed.

ObservabilityCopy Anchor Link

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 (with error and, when a phase ran, kind). Payloads carry mediaId, ownerType and ownerId; outcomes of an attach also carry collection, source, size and mimeType. The Telescope bridge captures them automatically.

Chat attachmentsCopy Anchor Link

Files users attach in the chat through AgentMediaAttachmentsModule (@dudousxd/nestjs-agent/media) are media records too (ownerType: 'agent-actor', ownerId = the actor's id, collection agent-attachments by default). With indexForRag: true that module publishes each ready attachment on aviary:media:attach (and its removal on aviary:media:delete), so this package indexes it like any other upload. Scope retrieval by owner (FilteredRetriever on ownerType / ownerId) and list the attachment collection in collections. See File uploads.

RelatedCopy Anchor Link

• @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