llm4s-docs
API
Generated with
llm4s-docs/org.llm4s/org.llm4s.rag/RAG

RAG

org.llm4s.rag.RAG
See theRAG companion object
final class RAG extends Closeable

High-level RAG (Retrieval-Augmented Generation) pipeline.

Provides a unified interface for:

  • Document ingestion (from files, directories, or raw text)
  • Semantic search with hybrid fusion
  • Answer generation with retrieved context

Attributes

Example 
// Create pipeline
val rag = RAG.builder()
 .withEmbeddings("openai")
 .withChunking(ChunkerFactory.Strategy.Sentence, 800, 150)
 .build()
 .toOption.get
// Ingest documents
rag.ingest("./docs")
// Search
val results = rag.query("What is X?")
// With answer generation (requires LLM client)
val ragWithLLM = RAG.builder()
 .withEmbeddings("openai")
 .withLLM(llmClient)
 .build()
 .toOption.get
val answer = ragWithLLM.queryWithAnswer("What is X?")
Companion
object
Graph
Supertypes
trait Closeable
trait AutoCloseable
class Object
trait Matchable
class Any

Members list

Value members

Concrete methods

def chunkCount: Int

Number of chunks this instance has indexed and not since deleted: a re-ingest replaces a document's count.

Number of chunks this instance has indexed and not since deleted: a re-ingest replaces a document's count.

Attributes

def clear(): Result[Unit]

Clear all indexed data.

Clear all indexed data.

Attributes

override def close(): Unit

Close resources.

Close resources.

Attributes

Definition Classes
Closeable -> AutoCloseable
def deleteDocument(docId: String): Result[Unit]

Delete a specific document and its chunks.

Delete a specific document and its chunks.

Limitation: a document's chunks are named <docId>-chunk-<n> and the stores delete by prefix, so deleting a also deletes the chunks of a document whose id itself looks like a-chunk-<n> (for example a-chunk-1, whose chunks are a-chunk-1-chunk-0, ...). Avoid document ids of that shape. Re-ingesting a document is not affected: it replaces chunks by their exact ids.

Value parameters

docId

Document ID to delete

Attributes

Returns

Unit on success

def deleteFromCollection(collectionPath: CollectionPath, documentId: String): Result[Long]

Delete a document from a collection.

Delete a document from a collection.

Value parameters

collectionPath

The collection containing the document

documentId

The document identifier

Attributes

Returns

Number of chunks deleted

def documentCount: Int

Number of documents this instance has indexed and not since deleted. A re-ingested document counts once; one that was re-ingested with no chunks does not count.

Number of documents this instance has indexed and not since deleted. A re-ingested document counts once; one that was re-ingested with no chunks does not count.

Attributes

def hasGraphRAG: Boolean

Check if GraphRAG is available for this pipeline.

Check if GraphRAG is available for this pipeline.

Attributes

def hasPermissions: Boolean

Check if permission-based search is available

Check if permission-based search is available

Attributes

def ingest(path: String, metadata: Map[String, String]): Result[Int]

Ingest a document from a file path.

Ingest a document from a file path.

Supports: .txt, .md, .pdf, .docx and other text-like formats.

Value parameters

metadata

Additional metadata to attach to all chunks

path

Path to file or directory

Attributes

Returns

Number of chunks created

def ingest(path: Path): Result[Int]

Ingest a document from a Path.

Ingest a document from a Path.

Attributes

def ingest(loader: DocumentLoader): Result[LoadStats]

Ingest documents from a DocumentLoader.

Ingest documents from a DocumentLoader.

A document that fails to load or ingest is counted in the returned LoadStats (or, with failFast, returned as the Left). A LoadResult.ListingFailure - the loader could not enumerate its documents at all - is always returned as the Left.

Value parameters

loader

The document loader to ingest from

Attributes

Returns

Loading statistics with success/failure counts

def ingestAsync(loader: DocumentLoader)(implicit ec: ExecutionContext): Future[Result[LoadStats]]

Async version of ingest with parallel document processing.

Async version of ingest with parallel document processing.

Processes documents in batches with configurable parallelism. Uses the parallelism and batchSize settings from LoadingConfig.

As with ingest, a LoadResult.ListingFailure is returned as the Left; as the results are collected first, nothing is ingested in that case.

Value parameters

ec

Execution context for async operations

loader

The document loader to ingest from

Attributes

Returns

Future with loading statistics

def ingestBytes(content: Array[Byte], filename: String, documentId: String, metadata: Map[String, String]): Result[Int]

Ingest a document from raw bytes.

Ingest a document from raw bytes.

This enables source-agnostic document ingestion - the same extraction logic works for documents from S3, HTTP responses, databases, etc.

Supported formats: PDF, DOCX, plain text, and anything Apache Tika can handle.

Value parameters

content

Raw document bytes

documentId

Unique identifier for this document

filename

Filename for format detection (e.g., "report.pdf")

metadata

Additional metadata

Attributes

Returns

Number of chunks created

def ingestBytesMultiple(documents: Iterator[(Array[Byte], String, String, Map[String, String])]): Result[LoadStats]

Ingest multiple documents from raw bytes.

Ingest multiple documents from raw bytes.

Value parameters

documents

Iterator of (content, filename, documentId, metadata) tuples

Attributes

Returns

Loading statistics with success/failure counts

def ingestChunks(documentId: String, chunks: Seq[String], metadata: Map[String, String]): Result[Int]

Ingest pre-chunked content (for advanced use cases).

Ingest pre-chunked content (for advanced use cases).

Attributes

def ingestPath(path: Path, metadata: Map[String, String]): Result[Int]

Ingest a document from a Path with metadata.

Ingest a document from a Path with metadata.

Attributes

def ingestText(content: String, documentId: String, metadata: Map[String, String]): Result[Int]

Ingest raw text content.

Ingest raw text content.

Ingesting an id that is already indexed replaces what is indexed under it. Replacing one document is not atomic across the two stores, so concurrent re-ingests of the SAME document id are not supported: serialise them in the caller. Different documents may be ingested concurrently.

Value parameters

content

The text content to ingest

documentId

Unique identifier for this document

metadata

Additional metadata

Attributes

Returns

Number of chunks created

def ingestWithPermissions(collectionPath: CollectionPath, documentId: String, content: String, metadata: Map[String, String], readableBy: Set[PrincipalId]): Result[Int]

Ingest a document into a specific collection with permission control.

Ingest a document into a specific collection with permission control.

Requires a SearchIndex to be configured via .withSearchIndex().

Value parameters

collectionPath

Target collection (must be a leaf collection)

content

Document text content

documentId

Unique document identifier

metadata

Additional document metadata

readableBy

Principal IDs that can read this document (empty = inherit from collection)

Attributes

Returns

Number of chunks indexed

def needsUpdate(doc: Document): Result[Boolean]

Check if a document needs updating based on version.

Check if a document needs updating based on version.

Value parameters

doc

Document to check

Attributes

Returns

true if document is new or changed

def query(query: String, topK: Option[Int]): Result[Seq[RAGSearchResult]]

Search for relevant chunks.

Search for relevant chunks.

If query transformers are configured, they are applied sequentially to the query before embedding. The transformed query is used for both vector search and keyword search.

Value parameters

query

The search query

topK

Override default topK (optional)

Attributes

Returns

Ranked search results

def queryWithAnswer(question: String, topK: Option[Int]): Result[RAGAnswerResult]

Search and generate an answer using LLM.

Search and generate an answer using LLM.

Requires an LLM client to be configured.

Value parameters

question

The question to answer

topK

Override default topK (optional)

Attributes

Returns

Answer with supporting contexts

def queryWithGraphRAG(question: String, topK: Option[Int]): Result[GraphRAGAnswer]

Query through GraphRAG using router-selected mode.

Query through GraphRAG using router-selected mode.

The router decides Global/Local/Hybrid first. Vector retrieval is only executed when Hybrid is selected.

Attributes

def queryWithGraphRAGMode(question: String, mode: GraphRAGMode, topK: Option[Int]): Result[GraphRAGAnswer]

Query GraphRAG in a specific mode.

Query GraphRAG in a specific mode.

Attributes

def queryWithPermissions(auth: UserAuthorization, collectionPattern: CollectionPattern, queryText: String, topK: Option[Int]): Result[Seq[RAGSearchResult]]

Search with permission filtering.

Search with permission filtering.

Requires a SearchIndex to be configured via .withSearchIndex(). Results are filtered based on user authorization:

  • Collection-level: Only search collections the user can access
  • Document-level: Only return documents the user can read

Value parameters

auth

User authorization context with principal IDs

collectionPattern

Pattern to filter collections (e.g., "confluence" for all descendants)

queryText

The search query

topK

Maximum number of results (defaults to config.topK)

Attributes

Returns

Permission-filtered search results

def queryWithPermissionsAndAnswer(auth: UserAuthorization, collectionPattern: CollectionPattern, question: String, topK: Option[Int]): Result[RAGAnswerResult]

Search with permissions and generate an answer using LLM.

Search with permissions and generate an answer using LLM.

Requires both a SearchIndex and LLM client to be configured.

Value parameters

auth

User authorization context

collectionPattern

Pattern to filter collections

question

The question to answer

topK

Maximum number of results

Attributes

Returns

Answer with permission-filtered supporting contexts

def refresh(loader: DocumentLoader): Result[SyncStats]

Full refresh - re-process all documents.

Full refresh - re-process all documents.

Clears the registry and the index, then re-ingests every document from the loader. Use this when you want to ensure a clean slate.

'''The index is cleared only after the loader has been read in full.''' If the loader reports a LoadResult.ListingFailure (at any point, including a later S3 page), or - with failFast - any LoadResult.Failure, refresh returns that error as the Left and leaves the existing index and registry untouched. Without failFast, a document that fails to read is left out of the rebuilt index, as a clean slate implies, and logged at WARN; use sync to keep a document's indexed version while its source is unreadable.

'''Memory:''' to give that guarantee, refresh holds every loaded Document - its extracted text and metadata, not the raw bytes - in memory before it clears anything. For a source too large for that, use sync, which streams one document at a time.

The guarantee covers reading the loader, not ingesting: an embedding or store error after the index is cleared counts that document as failed (or, with failFast, is returned as the Left), and the index then holds what was ingested before it.

Value parameters

loader

The document loader to refresh from

Attributes

Returns

Sync statistics (all as "added"), or the loader's error with the index untouched

def refreshAsync(loader: DocumentLoader)(implicit ec: ExecutionContext): Future[Result[SyncStats]]

Async full refresh: refresh, run on ec.

Async full refresh: refresh, run on ec.

It gives the same guarantee: the loader is read in full before the index is cleared, so a LoadResult.ListingFailure (or, with failFast, any read failure) is returned as the Left with the index untouched - and it holds every loaded document in memory to do so.

Value parameters

ec

Execution context for async operations

loader

The document loader to refresh from

Attributes

Returns

Future with sync statistics

def searchIndex: Option[SearchIndex]

Access the permission-based search index (if configured)

Access the permission-based search index (if configured)

Attributes

def stats: Result[RAGStats]

Get store statistics

Get store statistics

Attributes

def sync(loader: DocumentLoader): Result[SyncStats]

Sync with a loader - only process changed documents.

Sync with a loader - only process changed documents.

Compares document versions to detect:

  • New documents (added)
  • Changed documents (updated - old chunks removed, new chunks added)
  • Deleted documents (removed from source)
  • Unchanged documents (skipped)

Sync streams the loader, holding one document at a time plus the ids it has seen. Its deletion pass removes only documents the source no longer lists, never ones it could not read:

  • If the loader reports a LoadResult.ListingFailure - it could not enumerate its documents, e.g. an S3 listing with no credentials or a missing bucket - sync stops there and returns that error, and deletes nothing: an unlistable source is not an empty one. Documents synced before the failure (earlier S3 pages) stay synced.
  • A document that is listed but whose read fails - a LoadResult.Failure with a documentId, e.g. a transient S3 GetObject error or a failed extraction - keeps its indexed version. It is not deleted, and not counted in the returned SyncStats: it was neither verified unchanged nor updated. The next sync that reads it picks up any change.
  • A LoadResult.Failure with no documentId could be any document, so sync cannot tell which unlisted documents are really gone: it skips the deletion pass for that run, and logs a warning. Adds and updates still apply.

Read failures are logged at WARN.

Value parameters

loader

The document loader to sync with

Attributes

Returns

Sync statistics, or the listing error

def syncAsync(loader: DocumentLoader)(implicit ec: ExecutionContext): Future[Result[SyncStats]]

Async sync with parallel change detection.

Async sync with parallel change detection.

Performs change detection in parallel, but applies updates sequentially to avoid conflicts in the vector store.

As with sync, a LoadResult.ListingFailure is returned as the Left; as the results are collected first, nothing is added, updated or deleted in that case. A document whose read fails keeps its indexed version, and a read failure that does not name its document skips the deletion pass, as in sync. Unlike sync, it holds every loaded document in memory.

Value parameters

ec

Execution context for async operations

loader

The document loader to sync with

Attributes

Returns

Future with sync statistics

Concrete fields

Generated with