RAG extension

The official RAG Harnest Extension gives Agent Tools and Tasks one typed API for PostgreSQL retrieval-augmented generation. It also owns the RAGBackend contract used by additional datastore extensions, so agents do not change their document, chunk, query, or hit types when storage changes. The extension chunks and embeds documents, enforces namespaces and result bounds, and delegates filtering, ranking, ordering, and limiting to the backend. Your agent decides which retrieval operations become Tools and which ingestion operations remain trusted Task code.

Choose RAG or long-term memory

Use this extension for application knowledge: documents are chunked, optionally embedded, filtered, and ranked for grounded retrieval. Use Harnest’s built-in context.memory for deliberate, user-scoped facts and preferences that Agent Tools save and retrieve across sessions. Core memory uses literal text search, while RAG supports keyword, semantic, and hybrid ranking. Neither system adds data to a model prompt automatically. The similarly named rag.memory(...) factory is only the process-local test backend for this RAG API; it is unrelated to context.memory. One agent may use both: core memory for explicit user facts and RAG for a searchable document corpus.

Install the extension

harnest extensions install rag --project my-agent
harnest env sync my-agent

Version 0.1.3 requires Harnest >=1.0.0,<2. It depends on asyncpg and imports it only when a PostgreSQL lifecycle resource starts. It does not install Elasticsearch, Neo4j, or a PostgreSQL server extension.

Supply an embedder

Implement the small provider-neutral Embedder contract in application code:

lib/embeddings.py

from collections.abc import Sequence


class Embeddings:
    """Adapt the application's selected embedding client to RAG."""

    def __init__(self, client) -> None:
        self._client = client

    async def embed(
        self, texts: Sequence[str]
    ) -> Sequence[Sequence[float]]:
        return await self._client.embed(texts)

The extension requires exactly one finite vector per input. Keyword-only retrieval does not require an embedder.

Configure PostgreSQL

Create one application lifecycle resource and publish it to trusted agent code:

lifecycle/knowledge.py

import os

from harnest import context, lifecycle
from harnest.extensions.rag import RAGService, rag
from lib.embeddings import Embeddings
from lib.provider_clients import embedding_client


knowledge = rag.postgres(
    os.environ["DATABASE_URL"],
    table="product_knowledge",
    embedder=Embeddings(embedding_client),
    namespace=lambda: context.current().user_id,
)


@lifecycle.resource
@context.provider("knowledge")
async def knowledge_resource():
    async with knowledge:
        yield knowledge

The backend creates an extension-owned table with a tenant-qualified primary key, a document index, and a generated full-text index. Keyword ranking uses PostgreSQL text search. Semantic and hybrid ranking compute cosine similarity over portable float arrays inside PostgreSQL, so the base extension works on stock PostgreSQL without pgvector. Pass an existing asyncpg pool with pool=... when the application already owns connection lifecycle. Set setup_schema=False only when operators provision the same table shape before startup.

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

Retrieve from an Agent Tool

tools/search_knowledge.py

from harnest import context
from harnest.agent import tool
from harnest.extensions.rag import RAGService, SearchMode


@tool
async def search_knowledge(query: str, limit: int = 6) -> list[dict]:
    """Find grounded passages in the caller's knowledge namespace."""

    knowledge = context.resource("knowledge", RAGService)
    hits = await knowledge.search(query, mode=SearchMode.HYBRID, limit=limit)
    return [
        {
            "chunk_id": hit.chunk.id,
            "document_id": hit.chunk.document_id,
            "title": hit.chunk.title,
            "uri": hit.chunk.uri,
            "text": hit.chunk.text,
            "score": hit.score,
        }
        for hit in hits
    ]

Keep stable source identity and URI fields in the Tool result so the model can cite retrieved evidence. Use portable metadata filters instead of accepting SQL from model input.

Choose a search mode

Mode	Embedder required	Behavior
`SearchMode.KEYWORD`	No	Full-text candidates ordered by PostgreSQL `ts_rank_cd`
`SearchMode.SEMANTIC`	Yes	Compatible stored vectors ordered by cosine similarity
`SearchMode.HYBRID`	Yes	Lexical or vector candidates scored with equal lexical and non-negative cosine weights

search accepts a result limit from 1 to 100, portable metadata filters, and an optional finite min_score. A scalar filter requires equality. A tuple such as {"kind": ("manual", "release-note")} performs membership matching. Different fields combine with AND; tuple values within one field combine with OR. Nested metadata and caller-authored SQL or datastore query syntax are rejected.

Fetch exact chunks

Use await knowledge.fetch(("manual-2026:0", "manual-2026:3")) when application code already knows chunk IDs. PostgreSQL returns existing chunks in caller order, omits missing IDs, and applies the active namespace without relevance ranking.

Replace documents from trusted code

from harnest.extensions.rag import RAGDocument, RAGService


async def index_manuals(knowledge: RAGService) -> int:
    return await knowledge.ingest(
        (
            RAGDocument(
                id="manual-2026",
                title="Product manual",
                uri="https://docs.example.com/manual",
                text="...",
                metadata={"kind": "manual", "version": "2026"},
            ),
        ),
        trigger="user",
    )

Ingestion atomically replaces every supplied document in its namespace, including removal of obsolete chunks when content becomes shorter. Harnest emits payload-free extension mutation audit events after committed writes and on failures. Do not expose ingest or delete directly to the model without explicit permission and approval policy. Delete complete documents with await knowledge.delete(("manual-2025",), trigger="user"). The active namespace applies to the complete mutation.

Customize chunking

The default FixedSizeChunker(size=2000, overlap=200) creates overlapping passages at nearby whitespace and stable IDs from document identity and source order. Supply a synchronous Chunker implementation when Markdown, source code, or another structured format needs application-specific boundaries. A custom chunker must return a non-empty sequence of non-empty strings, and shared ingestion bounds still apply. Use rag.memory(max_chunks=...) for deterministic unit tests and small local demonstrations. It implements the same replacement, filtering, scoring, and result validation contracts, but it is process-local, non-durable, and not intended for production knowledge.

Add another datastore extension

A provider extension depends on rag instead of redefining its public API:

extensions/company-search/extension.yaml

requires:
  extensions: [rag]

It then implements the shared contract:

from collections.abc import Sequence
from harnest.extensions.rag import RAGChunk, RAGHit, RAGQuery


class CompanySearchBackend:
    async def start(self) -> None: ...
    async def close(self) -> None: ...
    async def replace(
        self,
        namespace: str,
        document_ids: Sequence[str],
        chunks: Sequence[RAGChunk],
    ) -> None: ...
    async def delete(self, namespace: str, document_ids: Sequence[str]) -> None: ...
    async def fetch(
        self, namespace: str, chunk_ids: Sequence[str]
    ) -> Sequence[RAGChunk]: ...
    async def search(self, query: RAGQuery) -> Sequence[RAGHit]: ...

The provider backend must apply predicates, projection, ranking, ordering, and query.limit in its datastore. RAGService validates returned types, namespace isolation, score threshold, ordering, and result count before exposing hits to agent code. Harnest rejects missing or cyclic extension dependencies, starts dependencies first, and stops them last. A published provider also declares harnest-extension-rag as a Python distribution dependency.

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.

Overview

Get started

Guides

RAG extension

Choose RAG or long-term memory

Install the extension

Supply an embedder

Configure PostgreSQL

Retrieve from an Agent Tool

Choose a search mode

Fetch exact chunks

Replace documents from trusted code

Customize chunking

Add another datastore extension

Public API and bounds

​Choose RAG or long-term memory

​Install the extension

​Supply an embedder

​Configure PostgreSQL

​Retrieve from an Agent Tool

​Choose a search mode

​Fetch exact chunks

​Replace documents from trusted code

​Customize chunking

​Add another datastore extension

​Public API and bounds

Choose RAG or long-term memory

Install the extension

Supply an embedder

Configure PostgreSQL

Retrieve from an Agent Tool

Choose a search mode

Fetch exact chunks

Replace documents from trusted code

Customize chunking

Add another datastore extension

Public API and bounds