API Reference

Protocols

ChunkerProtocol

source

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 Oridecon?",
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. Fluent configuration methods live in PipelineConfigWiring; this class owns pipeline assembly.

__init__

def __init__() -> None

source

Initialize the pipeline builder.

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 oridecon.ai.rag.config import RAGConfig

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

from oridecon.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 oridecon.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 oridecon.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 oridecon.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

register

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

register

def register(handler: object) -> None

source

Register a handler instance.

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

default_strategies

def default_strategies(cls) -> dict[str, type]

source

Declare the built-in retrieval strategies.

Returns

Type	Description
dict[str, type]	Mapping of strategy key → class: ``"vector"`` and ``"mmr"``.

---

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.

---