API Reference

Protocols

`ChunkerProtocol`

Protocol for document chunking strategies.

Implementations split document text into smaller, overlapping or non-overlapping chunks suitable for embedding and retrieval.

chunk

def chunk(
    text: str,
    metadata: dict[str, Any] | None = None
) -> list[Any]

source

Split text into chunks.

Parameters

Parameter	Type	Description
`text`	str	The document text to chunk.
`metadata`	dict[str, Any] \| None	Optional metadata to attach to each chunk.

Returns

Type	Description
list[Any]	List of Chunk objects.

Classes

`Chunk`

source

A chunk of text with metadata.

Attributes: text: The chunk text content source: Source document identifier start_index: Starting character position in original document end_index: Ending character position in original document chunk_index: Sequential index of this chunk metadata: Optional metadata dictionary

`ChunkingConfig`

source

Configuration for chunking.

Example

config = ChunkingConfig(
strategy=ChunkingStrategy.FIXED_SIZE,
chunk_size=1000,
overlap=200
)

`Context`

source

Retrieved context for RAG generation.

Example

context = Context(
query="What is Lexigram?",
documents=[doc1, doc2],
metadata={"retrieval_time": 0.123}
)

`IngestionConfig`

source

Configuration for document ingestion stage.

`PipelineBuilder`

source

Builder for constructing RAG pipelines with fluent API.

The builder provides a convenient way to configure and build pipelines programmatically or from configuration files.

__init__

def __init__() -> None

source

Initialize the pipeline builder.

with_name

def with_name(name: str) -> PipelineBuilder

source

Set pipeline name.

Parameters

Parameter	Type	Description
`name`	str	Pipeline name

Returns

Type	Description
PipelineBuilder	Self for chaining

with_description

def with_description(description: str) -> PipelineBuilder

source

Set pipeline description.

Parameters

Parameter	Type	Description
`description`	str	Pipeline description

Returns

Type	Description
PipelineBuilder	Self for chaining

with_ingestion

def with_ingestion(**kwargs: Any) -> PipelineBuilder

source

Configure ingestion stage.

Parameters

Parameter	Type	Description

Returns

Type	Description
PipelineBuilder	Self for chaining

with_query_processing

def with_query_processing(**kwargs: Any) -> PipelineBuilder

source

Configure query processing stage.

Parameters

Parameter	Type	Description

Returns

Type	Description
PipelineBuilder	Self for chaining

with_retrieval

def with_retrieval(**kwargs: Any) -> PipelineBuilder

source

Configure retrieval stage.

Parameters

Parameter	Type	Description

Returns

Type	Description
PipelineBuilder	Self for chaining

with_context_optimization

def with_context_optimization(**kwargs: Any) -> PipelineBuilder

source

Configure context optimization stage.

Parameters

Parameter	Type	Description

Returns

Type	Description
PipelineBuilder	Self for chaining

with_synthesis

def with_synthesis(**kwargs: Any) -> PipelineBuilder

source

Configure synthesis stage.

Parameters

Parameter	Type	Description

Returns

Type	Description
PipelineBuilder	Self for chaining

with_quality_assurance

def with_quality_assurance(**kwargs: Any) -> PipelineBuilder

source

Configure quality assurance stage.

Parameters

Parameter	Type	Description

Returns

Type	Description
PipelineBuilder	Self for chaining

with_post_processing

def with_post_processing(**kwargs: Any) -> PipelineBuilder

source

Configure post-processing stage.

Parameters

Parameter	Type	Description

Returns

Type	Description
PipelineBuilder	Self for chaining

with_error_strategy

def with_error_strategy(
    strategy: ErrorStrategy,
    max_retries: int = 3,
    retry_delay: float = 1.0
) -> PipelineBuilder

source

Configure global error handling.

Parameters

Parameter	Type	Description
`strategy`	ErrorStrategy	Default error handling strategy
`max_retries`	int	Maximum number of retries
`retry_delay`	float	Initial delay between retries

Returns

Type	Description
PipelineBuilder	Self for chaining

with_stages

def with_stages(stages: list[PipelineStageType]) -> PipelineBuilder

source

Set the ordered list of pipeline stages.

Parameters

Parameter	Type	Description
`stages`	list[PipelineStageType]	List of stages

Returns

Type	Description
PipelineBuilder	Self for chaining

with_custom_stage

def with_custom_stage(stage: PipelineStageProtocol) -> PipelineBuilder

source

Add a custom pipeline stage.

Parameters

Parameter	Type	Description
`stage`	PipelineStageProtocol	Custom pipeline stage

Returns

Type	Description
PipelineBuilder	Self for chaining

from_dict

def from_dict(config_dict: dict[str, Any]) -> PipelineBuilder

source

Load configuration from dictionary.

Parameters

Parameter	Type	Description
`config_dict`	dict[str, Any]	Configuration dictionary

Returns

Type	Description
PipelineBuilder	Self for chaining

from_yaml

async def from_yaml(yaml_path: str | Path) -> PipelineBuilder

source

Load configuration from YAML file.

Parameters

Parameter	Type	Description
`yaml_path`	str \| Path	Path to YAML configuration file

Returns

Type	Description
PipelineBuilder	Self for chaining

retrieve

def retrieve(
    strategy: str = 'hybrid',
    top_k: int = 10,
    **kwargs: Any
) -> PipelineBuilder

source

Configure retrieval with high-level parameters.

Parameters

Parameter	Type	Description
`strategy`	str	Retrieval strategy name (``"hybrid"``, ``"dense"``, ``"sparse"``).
`top_k`	int	Number of chunks to retrieve. **kwargs: Additional retrieval parameters.

rerank

def rerank(
    strategy: str = 'cross-encoder',
    top_k: int = 5,
    **kwargs: Any
) -> PipelineBuilder

source

Configure context optimization / reranking.

Parameters

Parameter	Type	Description
`strategy`	str	Reranking strategy name.
`top_k`	int	Number of chunks to keep after reranking. **kwargs: Additional reranking parameters.

synthesize

def synthesize(
    strategy: str = 'abstractive',
    model: str | None = None,
    **kwargs: Any
) -> PipelineBuilder

source

Configure synthesis with high-level parameters.

Parameters

Parameter	Type	Description
`strategy`	str	Synthesis strategy (``"abstractive"``, ``"extractive"``).
`model`	str \| None	LLM model identifier for generation. **kwargs: Additional synthesis parameters.

with_citations

def with_citations(required: bool = True) -> PipelineBuilder

source

Enable citation tracking in the pipeline.

Parameters

Parameter	Type	Description
`required`	bool	Whether citations are required (pipeline fails without them if set to ``True``).

with_evaluation

def with_evaluation(
    evaluator: RAGEvaluatorProtocol | None = None,
    every_n: int = 1
) -> PipelineBuilder

source

Enable automatic evaluation of pipeline outputs.

Parameters

Parameter	Type	Description
`evaluator`	RAGEvaluatorProtocol \| None	Optional pre-built evaluator instance.
`every_n`	int	Evaluate every n-th request (default: every request).

with_timeout

def with_timeout(**stage_timeouts: float) -> PipelineBuilder

source

Set per-stage timeouts (in seconds).

Keyword arguments map stage names to their timeout values

builder.with_timeout(retrieval=10.0, synthesis=30.0)

builder.with_timeout(retrieval=10.0, synthesis=30.0)

Parameters

Parameter	Type	Description

with_working_memory

def with_working_memory(memory: WorkingMemoryProtocol) -> PipelineBuilder

source

Attach working memory for context enrichment.

Parameters

Parameter	Type	Description
`memory`	WorkingMemoryProtocol	Working memory instance.

build

def build() -> RAGPipeline

source

Build the RAG pipeline.

Returns

Type	Description
RAGPipeline	Configured RAG pipeline

Raises

Exception	Description
ValueError	If configuration is invalid

`PipelineConfig`

source

Complete pipeline configuration.

from_dict

def from_dict(
    cls,
    config_dict: dict[str, Any]
) -> PipelineConfig

source

Create configuration from dictionary.

to_dict

def to_dict() -> dict[str, Any]

source

Convert configuration to dictionary.

`RAGAnswerSynthesizedHook`

source

Payload fired after the synthesis stage produces a final answer.

Attributes: pipeline_name: Name or identifier of the pipeline that synthesised the answer.

`RAGConfig`

source

Configuration for RAG (Retrieval Augmented Generation) pipeline.

Example

config = RAGConfig(
vector_store_type="chroma",
collection_name="pet_knowledge",
top_k=5,
enable_citations=True
)

with_collection

def with_collection(name: str) -> RAGConfig

source

Return a copy of this config with a different collection_name.

Usage

tenant_config = base_config.with_collection("canon_t_tenant42")

tenant_config = base_config.with_collection("canon_t_tenant42")

`RAGDocumentsRetrievedHook`

source

Payload fired after the retrieval stage returns candidate chunks.

Attributes: chunk_count: Number of chunks returned by the retrieval step.

`RAGModule`

source

Retrieval-Augmented Generation (RAG) pipeline integration.

Call configure to register the RAG pipeline, strategy registries, and supporting services (knowledge graph, HyDE, compression, reasoning) for injection.

Usage

from lexigram.ai.rag.config import RAGConfig

@module(
    imports=[
        RAGModule.configure(RAGConfig(chunk_size=512))
    ]
)
class AppModule(Module):
    pass

from lexigram.ai.rag.config import RAGConfig

@module(
    imports=[
        RAGModule.configure(RAGConfig(chunk_size=512))
    ]
)
class AppModule(Module):
    pass

Error Handling

RAG pipeline failures surface as typed exceptions that can be caught
directly or handled via the Result pattern::

    from lexigram.ai.rag.exceptions import (
        RAGError,           # base — catch-all
        PreprocessingError, # document preprocessing failure
        RetrievalError,     # retrieval / vector-store failure
        SynthesisError,     # response synthesis failure
        ChunkingError,      # document chunking failure
    )

RAG pipeline failures surface as typed exceptions that can be caught
directly or handled via the Result pattern

from lexigram.ai.rag.exceptions import (
    RAGError,           # base — catch-all
    PreprocessingError, # document preprocessing failure
    RetrievalError,     # retrieval / vector-store failure
    SynthesisError,     # response synthesis failure
    ChunkingError,      # document chunking failure
)

    from lexigram.ai.rag.exceptions import (
        RAGError,           # base — catch-all
        PreprocessingError, # document preprocessing failure
        RetrievalError,     # retrieval / vector-store failure
        SynthesisError,     # response synthesis failure
        ChunkingError,      # document chunking failure
    )

Exports: RAGPipelineProtocol, RetrievalStrategyProtocol, RAGError, PreprocessingError, RetrievalError, SynthesisError, ChunkingError

configure

def configure(
    cls,
    config: RAGConfig | None = None
) -> DynamicModule

source

Create a RAGModule with explicit configuration.

Parameters

Parameter	Type	Description
`config`	RAGConfig \| None	RAGConfig or ``None`` to use defaults (reads from environment variables).

Returns

Type	Description
DynamicModule	A DynamicModule descriptor.

stub

def stub(
    cls,
    config: RAGConfig | None = None
) -> DynamicModule

source

Create a RAGModule suitable for unit and integration testing.

Uses in-memory or no-op implementations with minimal side effects.

Parameters

Parameter	Type	Description
`config`	RAGConfig \| None	Optional config override. Uses safe test defaults when None.

Returns

Type	Description
DynamicModule	A DynamicModule descriptor.

`RAGPipeline`

source

Main RAG pipeline that orchestrates all stages.

This class provides a simple interface for executing the complete RAG pipeline with configurable stages and error handling.

__init__

def __init__(
    config: PipelineConfig,
    stages: list[PipelineStageProtocol],
    evaluator: RAGEvaluatorProtocol | None = None,
    working_memory: WorkingMemoryProtocol | None = None
)

source

Initialize the RAG pipeline.

Parameters

Parameter	Type	Description
`config`	PipelineConfig	Pipeline configuration
`stages`	list[PipelineStageProtocol]	List of pipeline stages
`evaluator`	RAGEvaluatorProtocol \| None	Optional evaluator implementing RAGEvaluatorProtocol for automatic per-request quality evaluation. Evaluation frequency is controlled by auto_evaluate_every_n.
`working_memory`	WorkingMemoryProtocol \| None	Optional working memory for context enrichment.

run

async def run(
    query: str,
    documents: list[str] | None = None,
    document_paths: list[str] | None = None,
    metadata: dict[str, Any] | None = None
) -> PipelineContext

source

Execute the RAG pipeline.

Parameters

Parameter	Type	Description
`query`	str	User query
`documents`	list[str] \| None	Optional list of document content strings
`document_paths`	list[str] \| None	Optional list of document file paths
`metadata`	dict[str, Any] \| None	Optional custom metadata

Returns

Type	Description
PipelineContext	Pipeline context with results

execute

async def execute(context: RAGContext) -> Result[RAGResponse, RAGError]

source

Execute the RAG pipeline per the contract protocol.

Parameters

Parameter	Type	Description
`context`	RAGContext	Pipeline context with query and optional config/filters.

Returns

Type	Description
Result[RAGResponse, RAGError]	Ok(RAGResponse) on success, Err(RAGError) on failure.

run_parallel

async def run_parallel(
    query: str,
    stages: list[PipelineStageProtocol] | None = None,
    **kwargs: Any
) -> PipelineContext

source

Execute pipeline stages in parallel.

Parameters

Parameter	Type	Description
`query`	str	User query
`stages`	list[PipelineStageProtocol] \| None	Stages to execute in parallel (default: all stages) **kwargs: Additional context parameters

Returns

Type	Description
PipelineContext	Pipeline context with results

`RAGPipelineStartedHook`

source

Payload fired when a RAG pipeline begins processing a query.

Attributes: pipeline_name: Name or identifier of the pipeline that started.

`RAGProvider`

source

Registers RAG pipeline services and strategy registries with the DI container.

__init__

def __init__(config: RAGConfig | None = None) -> None

source

async def register(container: ContainerRegistrarProtocol) -> None

source

boot

async def boot(container: BootContainerProtocol) -> None

source

Boot RAG provider — wire optional integrations.

shutdown

async def shutdown() -> None

source

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Check RAG provider health — verifies embedding service and vector store.

Returns

Type	Description
HealthCheckResult	HealthCheckResult with status ``healthy`` when all configured dependencies are reachable, or ``degraded``/``unhealthy`` otherwise.

`RAGTenancyConfig`

source

Optional tenant-aware RAG pipeline configuration.

When enabled, the RAG provider wraps the RAGPipelineProtocol binding in a TenantScopedRAGPipeline that resolves the collection_name from the current tenant context at request time, with per-tenant pipeline instance caching.

Note

Requires lexigram-tenancy in the module graph when enabled is True — the provider resolves Context at boot.

`RerankResult`

source

Result of a reranking operation.

Attributes: documents: Reranked documents (most relevant first). scores: Relevance scores (parallel to documents). original_count: Number of documents passed to reranker. reranked_count: Number of documents returned (may be < original if top_k applied). model_name: Name of the reranking model used. metadata: Additional reranking metadata.

`RerankingStrategyRegistry`

source

Registry of reranking strategy handlers.

Reranking strategies reorder documents after initial retrieval using cross-encoders, LLM-based scoring, or fusion techniques.

Uses a handler-based dispatch pattern where handlers implement can_handle(strategy: str) and create_and_rerank() methods.

Usage

registry = RerankingStrategyRegistry()
registry.register(FlashRankStrategyHandler())
handler = registry.get("flashrank")
result = await handler.create_and_rerank(strategy="flashrank", ...)

registry = RerankingStrategyRegistry()
registry.register(FlashRankStrategyHandler())
handler = registry.get("flashrank")
result = await handler.create_and_rerank(strategy="flashrank", ...)

__init__

def __init__() -> None

source

Initialize an empty handler registry.

with_defaults

def with_defaults(cls) -> RerankingStrategyRegistry

source

Create a RerankingStrategyRegistry with default handlers.

Returns

Type	Description
RerankingStrategyRegistry	A RerankingStrategyRegistry with no default handlers. Handlers are registered conditionally by the provider.

def register(handler: object) -> None

source

Parameters

Parameter	Type	Description
`handler`	object	A handler instance with can_handle(strategy) method.

get

def get(strategy: str) -> object | None

source

Get a handler that can handle the given strategy.

Parameters

Parameter	Type	Description
`strategy`	str	Strategy name to look up.

Returns

Type	Description
object \| None	First handler where can_handle(strategy) is True, or None.

`RetrievalCompletedEvent`

source

Emitted when the retrieval stage of a RAG pipeline completes.

Consumed by: quality metrics, retrieval analytics, feedback loops.

`RetrievalConfig`

source

Configuration for retrieval stage.

`RetrievalStrategyRegistry`

source

Registry of retrieval strategy implementations.

Strategies take a query and a set of candidate documents and return an ordered subset ranked by relevance.

Usage

registry = RetrievalStrategyRegistry.with_defaults()
strategy = registry.instantiate("mmr", lambda_param=0.7)
results = await strategy.retrieve(query, candidates, top_k=5)

registry = RetrievalStrategyRegistry.with_defaults()
strategy = registry.instantiate("mmr", lambda_param=0.7)
results = await strategy.retrieve(query, candidates, top_k=5)

__init__

def __init__() -> None

source

with_defaults

def with_defaults(cls) -> RetrievalStrategyRegistry

source

Create a registry pre-loaded with built-in strategies.

Registered keys: - "vector": VectorRetrievalStrategy - "mmr": MMRRetrievalStrategy

Returns

Type	Description
RetrievalStrategyRegistry	A new registry instance with default strategies registered.

`SynthesisCompletedEvent`

source

Emitted when the synthesis stage of a RAG pipeline completes.

Consumed by: quality metrics, answer analytics, audit.

`SynthesisConfig`

source

Configuration for response synthesis.

Attributes: strategy: Synthesis strategy to use max_context_length: Maximum context length in tokens max_response_length: Maximum response length in tokens include_citations: Whether to include citations output_format: Desired output format quality_check: Whether to run quality checks min_confidence: Minimum confidence threshold metadata: Additional configuration metadata

`TenantScopedRAGPipeline`

source

Resolves per-tenant RAG pipelines at request time.

Caches pipeline instances per tenant with LRU eviction. When no tenant context is available, delegates to a default pipeline built from the base config.

The factory callable receives a tenant-scoped RAGConfig (with collection_name already resolved) and should return a fully constructed RAGPipelineProtocol.

Example usage

factory = TenantScopedRAGPipeline(
    base_config=RAGConfig(collection_name="canon"),
    resolver=TemplatedTenantCollectionResolver(),
    ctx=context,
    pipeline_factory=build_rag_pipeline,
)
result = await factory.execute(RAGContext(query="..."))

factory = TenantScopedRAGPipeline(
    base_config=RAGConfig(collection_name="canon"),
    resolver=TemplatedTenantCollectionResolver(),
    ctx=context,
    pipeline_factory=build_rag_pipeline,
)
result = await factory.execute(RAGContext(query="..."))

__init__

def __init__(
    base_config: RAGConfig,
    resolver: TenantCollectionResolver,
    ctx: Context,
    pipeline_factory: Any,
    cache_size: int = 100
) -> None

source

execute

async def execute(context: RAGContext) -> Result[RAGResponse, RAGError]

source

Execute the RAG pipeline with tenant-aware collection resolution.

When a tenant ID is present in the current Context, the collection_name from base_config is resolved via the TenantCollectionResolver before delegating to the tenant’s cached pipeline instance.

Parameters

Parameter	Type	Description
`context`	RAGContext	The RAG execution context.

Returns

Type	Description
Result[RAGResponse, RAGError]	The pipeline result or an error.

query

async def query(
    question: str,
    **kwargs: Any
) -> RAGResponse

source

Convenience: execute a query string and return the response.

Builds a RAGContext from question and extra keyword arguments, then delegates to execute. Raises on error.

Parameters

Parameter	Type	Description
`question`	str	The user query. **kwargs: Additional ``RAGContext`` fields.

Returns

Type	Description
RAGResponse	The RAG response.

Raises

Exception	Description
RAGError	If the pipeline execution fails.

Functions

`create_chunker`

create_chunker

def create_chunker(
    strategy: ChunkingStrategy = ChunkingStrategy.FIXED_SIZE,
    config: ChunkingConfig | None = None,
    **kwargs: Any
) -> AbstractChunker

source

Create a chunker instance for the given strategy.

Convenience wrapper around ChunkingStrategyRegistry.

Parameters

Parameter	Type	Description
`strategy`	ChunkingStrategy	Which chunking strategy to use.
`config`	ChunkingConfig \| None	Optional chunking configuration. **kwargs: Additional keyword arguments forwarded to the chunker constructor (override config defaults).

Returns

Type	Description
AbstractChunker	A configured Chunker instance.

Raises

Exception	Description
ValueError	If no chunker is registered for the given strategy.

Exceptions

`RAGError`

source

Base exception for RAG errors.