harnest.extensions.rag

All Python modules · Import and compatibility guidance

PostgreSQL configuration

Option	Default	Purpose
`dsn`	None	asyncpg DSN; required unless `pool` is supplied
`table`	`harnest_rag_chunks`	Safe extension-owned table name
`pool`	None	Borrow an existing asyncpg-compatible pool without closing it
`setup_schema`	`True`	Create the schema, or validate an operator-managed schema when false
`pool_options`	None	Additional options for `asyncpg.create_pool`
`embedder`	None	Application-owned embedding adapter
`chunker`	`FixedSizeChunker()`	Application chunking policy
`namespace`	`default`	Fixed or per-operation tenant namespace
`batch_size`	100	Chunk texts sent to the embedder per request

Public API and bounds

API	Role
`rag.postgres(...)`	Create the stock-PostgreSQL service
`rag.memory(...)`	Create the bounded in-memory service
`rag.service(...)`	Wrap a custom `RAGBackend`
`RAGService`	Lifecycle-owned ingestion, deletion, fetch, and search API
`RAGDocument`, `RAGChunk`, `RAGQuery`, `RAGHit`	Immutable provider-neutral data contracts
`SearchMode`	Keyword, semantic, or hybrid selection
`Embedder`, `Chunker`, `RAGBackend`	Replaceable provider protocols
`FixedSizeChunker`, `PostgresBackend`, `MemoryBackend`	Built-in implementations

The shared boundary permits at most 1,000 documents, 50 million source characters, and 10,000 generated chunks per ingestion request. Documents may contain up to 10 million characters, individual chunks up to 1 million, queries up to 16,384, and embeddings up to 65,536 dimensions. Metadata and filters allow 64 fields, with up to 100 scalar values in one membership filter. Empty text, duplicate identities, nested metadata, boolean vector values, and non-finite numbers fail before model or datastore I/O.

Exports

Name	Kind
`Chunker`	Class
`Embedder`	Class
`FilterValue`	Value or type alias
`FixedSizeChunker`	Class
`MemoryBackend`	Class
`PostgresBackend`	Class
`RAGBackend`	Class
`RAGChunk`	Class
`RAGDocument`	Class
`RAGExtension`	Class
`RAGHit`	Class
`RAGQuery`	Class
`RAGService`	Class
`Scalar`	Value or type alias
`SearchMode`	Class
`extension`	Namespace
`rag`	Namespace

Chunker

Split one document into ordered text passages.

Chunker(*args, **kwargs)

Example

from harnest.extensions.rag import Chunker

class MyChunker(Chunker):
    ...

Chunker.split

Chunker.split(document: RAGDocument) -> Sequence[str]

Return non-empty chunks in source order.

Parameters and example

Parameter	Type	Default
`document`	`RAGDocument`	Required

from harnest.extensions.rag import Chunker

chunker: Chunker
result = chunker.split(document=...)

Embedder

Turn text into vectors without coupling RAG to a model provider.

Embedder(*args, **kwargs)

Example

from harnest.extensions.rag import Embedder

class MyEmbedder(Embedder):
    ...

Embedder.embed

async Embedder.embed(texts: Sequence[str]) -> Sequence[Sequence[float]]

Return one finite vector for every supplied text.

Parameters and example

Parameter	Type	Default
`texts`	`Sequence[str]`	Required

from harnest.extensions.rag import Embedder

embedder: Embedder
result = await embedder.embed(texts=...)

FilterValue

FilterValue = str | int | float | bool | tuple[str | int | float | bool, ...]

Example

from harnest.extensions.rag import FilterValue

FixedSizeChunker

Split text on nearby whitespace with deterministic overlap.

FixedSizeChunker(*, size: int = 2000, overlap: int = 200) -> None

Parameters and example

Parameter	Type	Default
`size`	`int`	`2000`
`overlap`	`int`	`200`

from harnest.extensions.rag import FixedSizeChunker

chunker = FixedSizeChunker()

FixedSizeChunker.split

FixedSizeChunker.split(document: RAGDocument) -> tuple[str, ...]

Return stable non-empty passages without dropping source text.

Parameters and example

Parameter	Type	Default
`document`	`RAGDocument`	Required

from harnest.extensions.rag import FixedSizeChunker

chunker: FixedSizeChunker
result = chunker.split(document=...)

MemoryBackend

Store a bounded corpus in process behind the standard backend contract.

MemoryBackend(*, max_chunks: int = 10000) -> None

Parameters and example

Parameter	Type	Default
`max_chunks`	`int`	`10000`

from harnest.extensions.rag import MemoryBackend

backend = MemoryBackend()

MemoryBackend.start

async MemoryBackend.start() -> None

Activate the backend without allocating external resources.

Example

from harnest.extensions.rag import MemoryBackend

backend: MemoryBackend
await backend.start()

MemoryBackend.close

async MemoryBackend.close() -> None

Deactivate the backend while retaining its development corpus.

Example

from harnest.extensions.rag import MemoryBackend

backend: MemoryBackend
await backend.close()

MemoryBackend.replace

async MemoryBackend.replace(namespace: str, document_ids: Sequence[str], chunks: Sequence[RAGChunk]) -> None

Replace complete documents without exceeding configured capacity.

Parameters and example

Parameter	Type	Default
`namespace`	`str`	Required
`document_ids`	`Sequence[str]`	Required
`chunks`	`Sequence[RAGChunk]`	Required

from harnest.extensions.rag import MemoryBackend

backend: MemoryBackend
await backend.replace(namespace=..., document_ids=..., chunks=...)

MemoryBackend.delete

async MemoryBackend.delete(namespace: str, document_ids: Sequence[str]) -> None

Delete matching document chunks inside one namespace.

Parameters and example

Parameter	Type	Default
`namespace`	`str`	Required
`document_ids`	`Sequence[str]`	Required

from harnest.extensions.rag import MemoryBackend

backend: MemoryBackend
await backend.delete(namespace=..., document_ids=...)

MemoryBackend.fetch

async MemoryBackend.fetch(namespace: str, chunk_ids: Sequence[str]) -> tuple[RAGChunk, ...]

Return exact chunks in caller order and omit missing identities.

Parameters and example

Parameter	Type	Default
`namespace`	`str`	Required
`chunk_ids`	`Sequence[str]`	Required

from harnest.extensions.rag import MemoryBackend

backend: MemoryBackend
result = await backend.fetch(namespace=..., chunk_ids=...)

MemoryBackend.search

async MemoryBackend.search(query: RAGQuery) -> tuple[RAGHit, ...]

Filter, score, sort, and limit the development corpus in memory.

Parameters and example

Parameter	Type	Default
`query`	`RAGQuery`	Required

from harnest.extensions.rag import MemoryBackend

backend: MemoryBackend
result = await backend.search(query=...)

PostgresBackend

Persist chunks in PostgreSQL with full-text and array-vector ranking.

PostgresBackend(
    dsn: str | None = None,
    *,
    table: str = 'harnest_rag_chunks',
    pool: Any | None = None,
    setup_schema: bool = True,
    pool_options: Mapping[str, Any] | None = None,
) -> None

Parameters and example

Parameter	Type	Default
`dsn`	`str \| None`	`None`
`table`	`str`	`'harnest_rag_chunks'`
`pool`	`Any \| None`	`None`
`setup_schema`	`bool`	`True`
`pool_options`	`Mapping[str, Any] \| None`	`None`

from harnest.extensions.rag import PostgresBackend

backend = PostgresBackend()

PostgresBackend.start

async PostgresBackend.start() -> None

Open an owned pool and ensure the isolated RAG schema is usable.

Example

from harnest.extensions.rag import PostgresBackend

backend: PostgresBackend
await backend.start()

PostgresBackend.close

async PostgresBackend.close() -> None

Close only a pool created by this backend.

Example

from harnest.extensions.rag import PostgresBackend

backend: PostgresBackend
await backend.close()

PostgresBackend.replace

async PostgresBackend.replace(namespace: str, document_ids: Sequence[str], chunks: Sequence[RAGChunk]) -> None

Replace complete documents atomically so obsolete chunks cannot survive.

Parameters and example

Parameter	Type	Default
`namespace`	`str`	Required
`document_ids`	`Sequence[str]`	Required
`chunks`	`Sequence[RAGChunk]`	Required

from harnest.extensions.rag import PostgresBackend

backend: PostgresBackend
await backend.replace(namespace=..., document_ids=..., chunks=...)

PostgresBackend.delete

async PostgresBackend.delete(namespace: str, document_ids: Sequence[str]) -> None

Delete complete documents inside one namespace.

Parameters and example

Parameter	Type	Default
`namespace`	`str`	Required
`document_ids`	`Sequence[str]`	Required

from harnest.extensions.rag import PostgresBackend

backend: PostgresBackend
await backend.delete(namespace=..., document_ids=...)

PostgresBackend.fetch

async PostgresBackend.fetch(namespace: str, chunk_ids: Sequence[str]) -> tuple[RAGChunk, ...]

Fetch exact chunks in caller order with ordering performed by PostgreSQL.

Parameters and example

Parameter	Type	Default
`namespace`	`str`	Required
`chunk_ids`	`Sequence[str]`	Required

from harnest.extensions.rag import PostgresBackend

backend: PostgresBackend
result = await backend.fetch(namespace=..., chunk_ids=...)

PostgresBackend.search

async PostgresBackend.search(query: RAGQuery) -> tuple[RAGHit, ...]

Filter, rank, order, and limit one query entirely in PostgreSQL.

Parameters and example

Parameter	Type	Default
`query`	`RAGQuery`	Required

from harnest.extensions.rag import PostgresBackend

backend: PostgresBackend
result = await backend.search(query=...)

RAGBackend

Storage adapter required by the provider-neutral RAG service.

RAGBackend(*args, **kwargs)

Example

from harnest.extensions.rag import RAGBackend

class MyRAGBackend(RAGBackend):
    ...

RAGBackend.start

async RAGBackend.start() -> None

Open backend resources and validate required indexes.

Example

from harnest.extensions.rag import RAGBackend

backend: RAGBackend
await backend.start()

RAGBackend.close

async RAGBackend.close() -> None

Release only resources owned by this backend.

Example

from harnest.extensions.rag import RAGBackend

backend: RAGBackend
await backend.close()

RAGBackend.replace

async RAGBackend.replace(namespace: str, document_ids: Sequence[str], chunks: Sequence[RAGChunk]) -> None

Atomically replace complete documents and remove obsolete chunks.

Parameters and example

Parameter	Type	Default
`namespace`	`str`	Required
`document_ids`	`Sequence[str]`	Required
`chunks`	`Sequence[RAGChunk]`	Required

from harnest.extensions.rag import RAGBackend

backend: RAGBackend
await backend.replace(namespace=..., document_ids=..., chunks=...)

RAGBackend.delete

async RAGBackend.delete(namespace: str, document_ids: Sequence[str]) -> None

Delete documents only inside the supplied namespace.

Parameters and example

Parameter	Type	Default
`namespace`	`str`	Required
`document_ids`	`Sequence[str]`	Required

from harnest.extensions.rag import RAGBackend

backend: RAGBackend
await backend.delete(namespace=..., document_ids=...)

RAGBackend.fetch

async RAGBackend.fetch(namespace: str, chunk_ids: Sequence[str]) -> Sequence[RAGChunk]

Fetch exact chunks without performing relevance ranking.

Parameters and example

Parameter	Type	Default
`namespace`	`str`	Required
`chunk_ids`	`Sequence[str]`	Required

from harnest.extensions.rag import RAGBackend

backend: RAGBackend
result = await backend.fetch(namespace=..., chunk_ids=...)

RAGBackend.search

async RAGBackend.search(query: RAGQuery) -> Sequence[RAGHit]

Apply filtering, ranking, ordering, and limiting in the backend.

Parameters and example

Parameter	Type	Default
`query`	`RAGQuery`	Required

from harnest.extensions.rag import RAGBackend

backend: RAGBackend
result = await backend.search(query=...)

RAGChunk

One independently retrievable, namespace-scoped document passage.

RAGChunk(
    id: str,
    document_id: str,
    namespace: str,
    text: str,
    title: str | None = None,
    uri: str | None = None,
    metadata: Mapping[str, Scalar] = ...,
    embedding: tuple[float, ...] | None = None,
) -> None

Fields (— means no declared default):

Field	Type	Default
`id`	`str`	`—`
`document_id`	`str`	`—`
`namespace`	`str`	`—`
`text`	`str`	`—`
`title`	`str \| None`	`None`
`uri`	`str \| None`	`None`
`metadata`	`Mapping[str, Scalar]`	`dict()`
`embedding`	`tuple[float, ...] \| None`	`None`

Parameters and example

Parameter	Type	Default
`id`	`str`	Required
`document_id`	`str`	Required
`namespace`	`str`	Required
`text`	`str`	Required
`title`	`str \| None`	`None`
`uri`	`str \| None`	`None`
`metadata`	`Mapping[str, Scalar]`	`...`
`embedding`	`tuple[float, ...] \| None`	`None`

from harnest.extensions.rag import RAGChunk

chunk = RAGChunk(id=..., document_id=..., namespace=..., text=...)

RAGDocument

One source document supplied to the chunking and indexing pipeline.

RAGDocument(
    id: str,
    text: str,
    title: str | None = None,
    uri: str | None = None,
    metadata: Mapping[str, Scalar] = ...,
) -> None

Fields (— means no declared default):

Field	Type	Default
`id`	`str`	`—`
`text`	`str`	`—`
`title`	`str \| None`	`None`
`uri`	`str \| None`	`None`
`metadata`	`Mapping[str, Scalar]`	`dict()`

Parameters and example

Parameter	Type	Default
`id`	`str`	Required
`text`	`str`	Required
`title`	`str \| None`	`None`
`uri`	`str \| None`	`None`
`metadata`	`Mapping[str, Scalar]`	`...`

from harnest.extensions.rag import RAGDocument

document = RAGDocument(id=..., text=...)

RAGExtension

Create lifecycle-owned RAG services without choosing a storage provider.

RAGExtension(*args: object, **kwargs: object) -> 'Extension[ContextT]'

Parameters and example

Parameter	Type	Default
`*args`	`object`	Optional
`**kwargs`	`object`	Optional

from harnest.extensions.rag import RAGExtension

extension = RAGExtension()

RAGExtension.start

async RAGExtension.start(context: 'ExtensionStartContext') -> None

Initialize application-scoped resources after dependencies start.

Parameters and example

Parameter	Type	Default
`context`	`'ExtensionStartContext'`	Required

from harnest.extensions.rag import RAGExtension

extension: RAGExtension
await extension.start(context=...)

RAGExtension.stop

async RAGExtension.stop() -> None

Release application-scoped resources before dependencies stop.

Example

from harnest.extensions.rag import RAGExtension

extension: RAGExtension
await extension.stop()

RAGExtension.create_context

RAGExtension.create_context(context: ExtensionContext) -> ContextT

Adapt the managed invocation base to an extension-specific context.

Parameters and example

Parameter	Type	Default
`context`	`ExtensionContext`	Required

from harnest.extensions.rag import RAGExtension

extension: RAGExtension
result = extension.create_context(context=...)

RAGExtension.context

Return the task-local extension context for the active invocation.

Example

from harnest.extensions.rag import RAGExtension

extension: RAGExtension
value = extension.context

RAGExtension.service

RAGExtension.service(
    backend: RAGBackend,
    *,
    embedder: Embedder | None = None,
    chunker: Chunker | None = None,
    namespace: NamespaceSource = 'default',
    batch_size: int = 100,
) -> RAGService

Wrap any compatible backend in the shared ingestion and retrieval API.

Parameters and example

Parameter	Type	Default
`backend`	`RAGBackend`	Required
`embedder`	`Embedder \| None`	`None`
`chunker`	`Chunker \| None`	`None`
`namespace`	`NamespaceSource`	`'default'`
`batch_size`	`int`	`100`

from harnest.extensions.rag import RAGExtension

extension: RAGExtension
service = extension.service(backend=...)

RAGExtension.memory

RAGExtension.memory(
    *,
    embedder: Embedder | None = None,
    chunker: Chunker | None = None,
    namespace: NamespaceSource = 'default',
    batch_size: int = 100,
    max_chunks: int = 10000,
) -> RAGService

Create a bounded in-memory service for development and unit tests.

Parameters and example

Parameter	Type	Default
`embedder`	`Embedder \| None`	`None`
`chunker`	`Chunker \| None`	`None`
`namespace`	`NamespaceSource`	`'default'`
`batch_size`	`int`	`100`
`max_chunks`	`int`	`10000`

from harnest.extensions.rag import RAGExtension

extension: RAGExtension
service = extension.memory()

RAGExtension.postgres

RAGExtension.postgres(
    dsn: str | None = None,
    *,
    table: str = 'harnest_rag_chunks',
    pool: Any | None = None,
    setup_schema: bool = True,
    pool_options: Mapping[str, Any] | None = None,
    embedder: Embedder | None = None,
    chunker: Chunker | None = None,
    namespace: NamespaceSource = 'default',
    batch_size: int = 100,
) -> RAGService

Create a PostgreSQL-backed lexical, semantic, or hybrid service.

Parameters and example

Parameter	Type	Default
`dsn`	`str \| None`	`None`
`table`	`str`	`'harnest_rag_chunks'`
`pool`	`Any \| None`	`None`
`setup_schema`	`bool`	`True`
`pool_options`	`Mapping[str, Any] \| None`	`None`
`embedder`	`Embedder \| None`	`None`
`chunker`	`Chunker \| None`	`None`
`namespace`	`NamespaceSource`	`'default'`
`batch_size`	`int`	`100`

from harnest.extensions.rag import RAGExtension

extension: RAGExtension
service = extension.postgres()

RAGHit

One ranked passage returned from a RAG backend.

RAGHit(chunk: RAGChunk, score: float) -> None

Fields (— means no declared default):

Field	Type	Default
`chunk`	`RAGChunk`	`—`
`score`	`float`	`—`

Parameters and example

Parameter	Type	Default
`chunk`	`RAGChunk`	Required
`score`	`float`	Required

from harnest.extensions.rag import RAGHit

hit = RAGHit(chunk=..., score=...)

RAGQuery

A bounded retrieval request understood by every backend.

RAGQuery(
    text: str,
    namespace: str,
    mode: SearchMode = <SearchMode.HYBRID: 'hybrid'>,
    limit: int = 8,
    filters: Mapping[str, FilterValue] = ...,
    vector: tuple[float, ...] | None = None,
    min_score: float | None = None,
) -> None

Fields (— means no declared default):

Field	Type	Default
`text`	`str`	`—`
`namespace`	`str`	`—`
`mode`	`SearchMode`	`<SearchMode.HYBRID: 'hybrid'>`
`limit`	`int`	`8`
`filters`	`Mapping[str, FilterValue]`	`dict()`
`vector`	`tuple[float, ...] \| None`	`None`
`min_score`	`float \| None`	`None`

Parameters and example

Parameter	Type	Default
`text`	`str`	Required
`namespace`	`str`	Required
`mode`	`SearchMode`	`<SearchMode.HYBRID: 'hybrid'>`
`limit`	`int`	`8`
`filters`	`Mapping[str, FilterValue]`	`...`
`vector`	`tuple[float, ...] \| None`	`None`
`min_score`	`float \| None`	`None`

from harnest.extensions.rag import RAGQuery

query = RAGQuery(text=..., namespace=...)

RAGService

Own one backend and expose the same RAG API to Tools and Tasks.

RAGService(
    backend: RAGBackend,
    *,
    embedder: Embedder | None = None,
    chunker: Chunker | None = None,
    namespace: NamespaceSource = 'default',
    batch_size: int = 100,
) -> None

Parameters and example

Parameter	Type	Default
`backend`	`RAGBackend`	Required
`embedder`	`Embedder \| None`	`None`
`chunker`	`Chunker \| None`	`None`
`namespace`	`NamespaceSource`	`'default'`
`batch_size`	`int`	`100`

from harnest.extensions.rag import RAGService

service = RAGService(backend=...)

RAGService.backend

Expose the configured adapter for provider-specific administration.

Example

from harnest.extensions.rag import RAGService

service: RAGService
value = service.backend

RAGService.start

async RAGService.start() -> None

Open the backend once under application lifecycle ownership.

Example

from harnest.extensions.rag import RAGService

service: RAGService
await service.start()

RAGService.close

async RAGService.close() -> None

Close the backend once and make later calls fail closed.

Example

from harnest.extensions.rag import RAGService

service: RAGService
await service.close()

RAGService.aenter

async RAGService.__aenter__() -> 'RAGService'

Allow direct use as a Harnest lifecycle resource.

Example

from harnest.extensions.rag import RAGService

service: RAGService
async with service:
    ...

RAGService.aexit

async RAGService.__aexit__() -> None

Release the backend when its lifecycle resource exits.

Example

from harnest.extensions.rag import RAGService

service: RAGService
async with service:
    ...

RAGService.ingest

async RAGService.ingest(documents: Sequence[RAGDocument], *, trigger: MutationTrigger = 'user') -> int

Chunk, optionally embed, and atomically audit one ingestion request.

Parameters and example

Parameter	Type	Default
`documents`	`Sequence[RAGDocument]`	Required
`trigger`	`MutationTrigger`	`'user'`

from harnest.extensions.rag import RAGService

service: RAGService
result = await service.ingest(documents=...)

RAGService.delete

async RAGService.delete(document_ids: Sequence[str], *, trigger: MutationTrigger = 'user') -> None

Delete namespaced documents through the audited mutation boundary.

Parameters and example

Parameter	Type	Default
`document_ids`	`Sequence[str]`	Required
`trigger`	`MutationTrigger`	`'user'`

from harnest.extensions.rag import RAGService

service: RAGService
await service.delete(document_ids=...)

RAGService.fetch

async RAGService.fetch(chunk_ids: Sequence[str]) -> tuple[RAGChunk, ...]

Fetch exact namespaced chunks in backend-defined source order.

Parameters and example

Parameter	Type	Default
`chunk_ids`	`Sequence[str]`	Required

from harnest.extensions.rag import RAGService

service: RAGService
result = await service.fetch(chunk_ids=...)

RAGService.search

async RAGService.search(
    text: str,
    *,
    mode: SearchMode = <SearchMode.HYBRID: 'hybrid'>,
    limit: int = 8,
    filters: Mapping[str, FilterValue] | None = None,
    min_score: float | None = None,
) -> tuple[RAGHit, ...]

Embed when required and delegate bounded ranking to the backend.

Parameters and example

Parameter	Type	Default
`text`	`str`	Required
`mode`	`SearchMode`	`<SearchMode.HYBRID: 'hybrid'>`
`limit`	`int`	`8`
`filters`	`Mapping[str, FilterValue] \| None`	`None`
`min_score`	`float \| None`	`None`

from harnest.extensions.rag import RAGService

service: RAGService
result = await service.search(text=...)

Scalar

Scalar = str | int | float | bool

Example

from harnest.extensions.rag import Scalar

SearchMode

Choose lexical, vector, or rank-fused retrieval.

SearchMode(*values)

Member	Value
`KEYWORD`	`'keyword'`
`SEMANTIC`	`'semantic'`
`HYBRID`	`'hybrid'`

Example

from harnest.extensions.rag import SearchMode

value = SearchMode.KEYWORD

extension

Create lifecycle-owned RAG services without choosing a storage provider.

Example

from harnest.extensions.rag import extension

extension.start

async extension.start(context: 'ExtensionStartContext') -> None

Initialize application-scoped resources after dependencies start.

Parameters and example

Parameter	Type	Default
`context`	`'ExtensionStartContext'`	Required

from harnest.extensions.rag import extension

await extension.start(context=...)

extension.stop

async extension.stop() -> None

Release application-scoped resources before dependencies stop.

Example

from harnest.extensions.rag import extension

await extension.stop()

extension.create_context

extension.create_context(context: ExtensionContext) -> ContextT

Adapt the managed invocation base to an extension-specific context.

Parameters and example

Parameter	Type	Default
`context`	`ExtensionContext`	Required

from harnest.extensions.rag import extension

result = extension.create_context(context=...)

extension.context

Return the task-local extension context for the active invocation.

Example

from harnest.extensions.rag import extension

value = extension.context

extension.service

extension.service(
    backend: RAGBackend,
    *,
    embedder: Embedder | None = None,
    chunker: Chunker | None = None,
    namespace: NamespaceSource = 'default',
    batch_size: int = 100,
) -> RAGService

Wrap any compatible backend in the shared ingestion and retrieval API.

Parameters and example

Parameter	Type	Default
`backend`	`RAGBackend`	Required
`embedder`	`Embedder \| None`	`None`
`chunker`	`Chunker \| None`	`None`
`namespace`	`NamespaceSource`	`'default'`
`batch_size`	`int`	`100`

from harnest.extensions.rag import extension

service = extension.service(backend=...)

extension.memory

extension.memory(
    *,
    embedder: Embedder | None = None,
    chunker: Chunker | None = None,
    namespace: NamespaceSource = 'default',
    batch_size: int = 100,
    max_chunks: int = 10000,
) -> RAGService

Create a bounded in-memory service for development and unit tests.

Parameters and example

Parameter	Type	Default
`embedder`	`Embedder \| None`	`None`
`chunker`	`Chunker \| None`	`None`
`namespace`	`NamespaceSource`	`'default'`
`batch_size`	`int`	`100`
`max_chunks`	`int`	`10000`

from harnest.extensions.rag import extension

service = extension.memory()

extension.postgres

extension.postgres(
    dsn: str | None = None,
    *,
    table: str = 'harnest_rag_chunks',
    pool: Any | None = None,
    setup_schema: bool = True,
    pool_options: Mapping[str, Any] | None = None,
    embedder: Embedder | None = None,
    chunker: Chunker | None = None,
    namespace: NamespaceSource = 'default',
    batch_size: int = 100,
) -> RAGService

Create a PostgreSQL-backed lexical, semantic, or hybrid service.

Parameters and example

Parameter	Type	Default
`dsn`	`str \| None`	`None`
`table`	`str`	`'harnest_rag_chunks'`
`pool`	`Any \| None`	`None`
`setup_schema`	`bool`	`True`
`pool_options`	`Mapping[str, Any] \| None`	`None`
`embedder`	`Embedder \| None`	`None`
`chunker`	`Chunker \| None`	`None`
`namespace`	`NamespaceSource`	`'default'`
`batch_size`	`int`	`100`

from harnest.extensions.rag import extension

service = extension.postgres()

rag

Create lifecycle-owned RAG services without choosing a storage provider.

Example

from harnest.extensions.rag import rag

rag.start

async rag.start(context: 'ExtensionStartContext') -> None

Initialize application-scoped resources after dependencies start.

Parameters and example

Parameter	Type	Default
`context`	`'ExtensionStartContext'`	Required

from harnest.extensions.rag import rag

await rag.start(context=...)

rag.stop

async rag.stop() -> None

Release application-scoped resources before dependencies stop.

Example

from harnest.extensions.rag import rag

await rag.stop()

rag.create_context

rag.create_context(context: ExtensionContext) -> ContextT

Adapt the managed invocation base to an extension-specific context.

Parameters and example

Parameter	Type	Default
`context`	`ExtensionContext`	Required

from harnest.extensions.rag import rag

result = rag.create_context(context=...)

rag.context

Return the task-local extension context for the active invocation.

Example

from harnest.extensions.rag import rag

value = rag.context

rag.service

rag.service(
    backend: RAGBackend,
    *,
    embedder: Embedder | None = None,
    chunker: Chunker | None = None,
    namespace: NamespaceSource = 'default',
    batch_size: int = 100,
) -> RAGService

Wrap any compatible backend in the shared ingestion and retrieval API.

Parameters and example

Parameter	Type	Default
`backend`	`RAGBackend`	Required
`embedder`	`Embedder \| None`	`None`
`chunker`	`Chunker \| None`	`None`
`namespace`	`NamespaceSource`	`'default'`
`batch_size`	`int`	`100`

from harnest.extensions.rag import rag

service = rag.service(backend=...)

rag.memory

rag.memory(
    *,
    embedder: Embedder | None = None,
    chunker: Chunker | None = None,
    namespace: NamespaceSource = 'default',
    batch_size: int = 100,
    max_chunks: int = 10000,
) -> RAGService

Create a bounded in-memory service for development and unit tests.

Parameters and example

Parameter	Type	Default
`embedder`	`Embedder \| None`	`None`
`chunker`	`Chunker \| None`	`None`
`namespace`	`NamespaceSource`	`'default'`
`batch_size`	`int`	`100`
`max_chunks`	`int`	`10000`

from harnest.extensions.rag import rag

service = rag.memory()

rag.postgres

rag.postgres(
    dsn: str | None = None,
    *,
    table: str = 'harnest_rag_chunks',
    pool: Any | None = None,
    setup_schema: bool = True,
    pool_options: Mapping[str, Any] | None = None,
    embedder: Embedder | None = None,
    chunker: Chunker | None = None,
    namespace: NamespaceSource = 'default',
    batch_size: int = 100,
) -> RAGService

Create a PostgreSQL-backed lexical, semantic, or hybrid service.

Parameters and example

Parameter	Type	Default
`dsn`	`str \| None`	`None`
`table`	`str`	`'harnest_rag_chunks'`
`pool`	`Any \| None`	`None`
`setup_schema`	`bool`	`True`
`pool_options`	`Mapping[str, Any] \| None`	`None`
`embedder`	`Embedder \| None`	`None`
`chunker`	`Chunker \| None`	`None`
`namespace`	`NamespaceSource`	`'default'`
`batch_size`	`int`	`100`

from harnest.extensions.rag import rag

service = rag.postgres()

​PostgreSQL configuration

​Public API and bounds

​Exports

​Chunker

​Chunker.split

​Embedder

​Embedder.embed

​FilterValue

​FixedSizeChunker

​FixedSizeChunker.split

​MemoryBackend

​MemoryBackend.start

​MemoryBackend.close

​MemoryBackend.replace

​MemoryBackend.delete

​MemoryBackend.fetch

​MemoryBackend.search

​PostgresBackend

​PostgresBackend.start

​PostgresBackend.close

​PostgresBackend.replace

​PostgresBackend.delete

​PostgresBackend.fetch

​PostgresBackend.search

​RAGBackend

​RAGBackend.start

​RAGBackend.close

​RAGBackend.replace

​RAGBackend.delete

​RAGBackend.fetch

​RAGBackend.search

​RAGChunk

​RAGDocument

​RAGExtension

​RAGExtension.start

​RAGExtension.stop

​RAGExtension.create_context

​RAGExtension.context

​RAGExtension.service

​RAGExtension.memory

​RAGExtension.postgres

​RAGHit

​RAGQuery

​RAGService

​RAGService.backend

​RAGService.start

​RAGService.close

​RAGService.aenter

​RAGService.aexit

​RAGService.ingest

​RAGService.delete

​RAGService.fetch

​RAGService.search

​Scalar

​SearchMode

​extension

​extension.start

​extension.stop

​extension.create_context

​extension.context

​extension.service

​extension.memory

​extension.postgres

​rag

​rag.start

​rag.stop

​rag.create_context

​rag.context

​rag.service

​rag.memory

​rag.postgres

PostgreSQL configuration

Public API and bounds

Exports

Chunker

Chunker.split

Embedder

Embedder.embed

FilterValue

FixedSizeChunker

FixedSizeChunker.split

MemoryBackend

MemoryBackend.start

MemoryBackend.close

MemoryBackend.replace

MemoryBackend.delete

MemoryBackend.fetch

MemoryBackend.search

PostgresBackend

PostgresBackend.start

PostgresBackend.close

PostgresBackend.replace

PostgresBackend.delete

PostgresBackend.fetch

PostgresBackend.search

RAGBackend

RAGBackend.start

RAGBackend.close

RAGBackend.replace

RAGBackend.delete

RAGBackend.fetch

RAGBackend.search

RAGChunk

RAGDocument

RAGExtension

RAGExtension.start

RAGExtension.stop

RAGExtension.create_context

RAGExtension.context

RAGExtension.service

RAGExtension.memory

RAGExtension.postgres

RAGHit

RAGQuery

RAGService

RAGService.backend

RAGService.start

RAGService.close

RAGService.aenter

RAGService.aexit

RAGService.ingest

RAGService.delete

RAGService.fetch

RAGService.search

Scalar

SearchMode

extension

extension.start

extension.stop

extension.create_context

extension.context

extension.service

extension.memory

extension.postgres

rag

rag.start

rag.stop

rag.create_context

rag.context

rag.service

rag.memory

rag.postgres