Retrieval-augmented generation (RAG)

A language model knows what was in its training data. It does not know your company handbook, last week’s tickets, or anything about a village that exists only in your database. Ask it anyway and you get a confident guess.

Retrieval-augmented generation fixes that without retraining anything:

  1. Retrieve the handful of passages most relevant to the question.

  2. Generate an answer from those passages, handed to the model in its prompt.

The retrieval half runs on embeddings — vectors positioned so that text with similar meaning lands close together (see LLM clients). Store those vectors, and finding relevant passages becomes a nearest-neighbour search instead of a keyword match.

Kaval.AI provides a BaseRagService that handles the whole loop:

  • Indexing — embed text and persist it.

  • Querying — embed a question and return the closest items, ranked by cosine similarity.

  • Organising — group items into collections, tag them with a source_id and arbitrary JSON metadata, and filter retrieval on either.

Two backends implement it: PostgresRagService (pgvector, for production) and SqliteRagService (a single portable file). This notebook uses Postgres, then shows the SQLite one at the end.

Setup

KAVALAI_DB_URI points at a PostgreSQL instance with the pgvector extension; KAVALAI_DB_SCHEMA says which schema to use. The service creates and migrates its own tables, so nothing needs provisioning by hand. A database role that may not issue DDL is served by provision=False, with the tables created beforehand by ensure_registry() and create_collection() — see Deployment.

import os
import sys

import dotenv
from loguru import logger

dotenv.load_dotenv("../.env")

logger.remove()
_ = logger.add(sys.stderr, level="WARNING")

Choosing an embedding model

An embedding model is named provider/model, like a chat model. Five providers ship with Kaval.AI — openai, gemini, ollama, fastembed and browser — and the list is open: register_embedding_provider adds your own. To see what the current process can resolve, ask the registry:

from kavalai import registered_embedding_providers

print(registered_embedding_providers())
# ['browser', 'fastembed', 'gemini', 'ollama', 'openai']

That gives you the providers. The model names themselves come from each provider’s own catalogue — Kaval.AI keeps no list of its own, so a name it cannot resolve fails with a message naming the providers that exist, while an unknown model fails at the provider. Model providers has the catalogue links, the one-liner that asks each provider what it currently offers, and what to weigh when choosing.

This notebook uses fastembed, which runs locally with no API key, so it is reproducible. (The model downloads from the Hugging Face Hub on first use.)

Important

Vectors from two different embedding models are not comparable: a search that mixes them returns unrelated passages rather than failing. The service therefore records the model with each collection and always embeds with the recorded one — the service’s own model is only the model new collections are created with. The collection’s model shows this in practice.

Each collection lives in its own table, provisioned on first index — the embedding dimension is taken from the first batch. hit.model and hit.embedding_size on any result say which model built the index, and list_collections() reports it per collection.

from kavalai.rag import PostgresRagService

EMBEDDING_MODEL = "fastembed/BAAI/bge-small-en-v1.5"
COLLECTION = "green_village"

rag = PostgresRagService.from_uri(
    os.environ["KAVALAI_DB_URI"],
    model=EMBEDDING_MODEL,
    schema=os.environ.get("KAVALAI_DB_SCHEMA"),
)

# Start from an empty collection, so re-running the notebook indexes nothing twice.
await rag.drop_collection(COLLECTION)

Indexing a corpus

The corpus is the collected knowledge of Green Village — 104 residents, one pub, one dalmatian.

index_batch embeds every text and stores it in one round trip. Each item carries:

  • texts — the content to embed and store;

  • metadata_list — a JSON dict per item, for filtering and bookkeeping;

  • source_ids — an external identifier per item (a document id, say). Chunks of one document share a source_id;

  • collection_name — the logical group, which maps to a table.

For a single item there is index(text, ...).

FACTS = [
    "President of Green Village is Thomas Cook (born 12.04.1994).",
    "Green Village has 104 residents.",
    "Green Village was founded on 03.09.1887 by shepherd Elias Thornbury.",
    "The tallest building in Green Village is the Old Grain Tower "
    "at 23 metres.",
    "Green Village's official flower is the marsh marigold.",
    "The village bakery, run by Greta Lindqvist, sells exactly 340 "
    "loaves every week.",
    "Green Village has one school with 14 pupils and 2 teachers.",
    "The annual Turnip Festival takes place every year on the third "
    "Saturday of October.",
    "Green Village's fire brigade consists of 7 volunteers and one "
    "dalmatian named Pepper.",
    "The village pond, Lake Miller, is 1.2 metres deep at its deepest point.",
    "Green Village's oldest resident is Agnes Whitlow (born 02.06.1929).",
    "The village has 3 streets: Main Road, Willow Lane, and Cobbler's Path.",
    "The local church bell weighs 412 kilograms and was cast in 1901.",
    "Green Village produces 8 tons of honey per year from its 26 beehives.",
    "The village library owns 1,847 books and is open on Tuesdays and Fridays.",
    "The speed limit everywhere in Green Village is 30 km/h.",
    "Green Village's only pub, The Rusty Anchor, has been operating "
    "since 1923.",
]

TOPICS = [
    "people", "people", "history", "buildings", "nature", "business", "school",
    "events", "safety", "nature", "people", "streets", "buildings", "business",
    "culture", "traffic", "business",
]

rows = await rag.index_batch(
    texts=FACTS,
    metadata_list=[{"topic": topic} for topic in TOPICS],
    source_ids=[f"fact-{i:02d}" for i in range(len(FACTS))],
    collection_name=COLLECTION,
)

print(f"indexed {len(rows)} facts into collection {COLLECTION!r}")

indexed 17 facts into collection 'green_village'

Querying

query embeds the question and returns the top_k closest items, ordered from most to least similar.

Retrieval is semantic, not lexical. The question below never says “oldest”, “born” or “Agnes”, and a keyword search for “lived the longest” would match nothing at all — yet the right fact comes back first.

hits = await rag.query(
    "Who has lived in the village the longest?",
    top_k=4,
    collection_name=COLLECTION,
)

for hit in hits:
    print(f"{hit.similarity:.3f}  [{hit.source_id}]  {hit.content}")

0.676  [fact-10]  Green Village's oldest resident is Agnes Whitlow (born 02.06.1929).
0.628  [fact-01]  Green Village has 104 residents.
0.576  [fact-11]  The village has 3 streets: Main Road, Willow Lane, and Cobbler's Path.
0.558  [fact-04]  Green Village's official flower is the marsh marigold.

Every hit is a RagServiceResult, carrying the stored content plus everything you need to filter, cite or debug it:

top = hits[0]
print("source_id   :", top.source_id)
print("similarity  :", round(top.similarity, 4))
print("collection  :", top.collection_name)
print("metadata    :", top.rag_metadata)
print("model       :", top.model)
print("dimensions  :", top.embedding_size)

source_id   : fact-10
similarity  : 0.6762
collection  : green_village
metadata    : {'topic': 'people'}
model       : fastembed/BAAI/bge-small-en-v1.5
dimensions  : 384

The generation half

Retrieval on its own is useful — but the point is to answer questions. First, watch what happens without retrieval. Green Village is fictional, so the model has nothing to draw on:

import textwrap

from kavalai import make_client

client = make_client("openai/gpt-5.6-luna")

QUESTION = ("How old was Green Village's oldest resident at the "
            "2025 Turnip Festival?")

print(textwrap.fill(
    await client.prompt(QUESTION + " Answer in one sentence."), 80
))

Green Village’s oldest resident was 102 years old at the 2025 Turnip Festival.

Now retrieve first, and hand the model only what it needs. This is the entire RAG pattern — a search, a prompt template, and a call:

import textwrap

PROMPT = """Answer the question using the facts below.

Every village-specific detail must come from these facts — do not invent any.
You may use general knowledge such as the calendar to reason about them. If the
facts are not enough to answer, say exactly what is missing.

Facts:
{context}

Question: {question}
"""


async def answer_with_rag(question: str, top_k: int) -> str:
    retrieved = await rag.query(
        question, top_k=top_k, collection_name=COLLECTION
    )
    print(f"retrieved (top_k={top_k}):")
    for hit in retrieved:
        print(f"  {hit.similarity:.3f}  {hit.content}")
    context = "\n".join(f"- {hit.content}" for hit in retrieved)
    answer = await client.prompt(
        PROMPT.format(context=context, question=question)
    )
    return "\n".join(
        textwrap.fill(line, 80) for line in answer.splitlines()
    )


print("\nanswer:\n" + await answer_with_rag(QUESTION, top_k=5))

retrieved (top_k=5):
  0.787  Green Village's oldest resident is Agnes Whitlow (born 02.06.1929).
  0.700  Green Village has 104 residents.
  0.681  President of Green Village is Thomas Cook (born 12.04.1994).
  0.668  Green Village was founded on 03.09.1887 by shepherd Elias Thornbury.
  0.615  Green Village's official flower is the marsh marigold.

answer:
The date of the 2025 Turnip Festival is not provided, so her exact age cannot be
determined. Agnes Whitlow, born 2 June 1929, would have been **95 before 2 June
2025** and **96 on or after 2 June 2025**.

Not the intended answer, but the behaviour the prompt asked for. Answering this needs two facts: Agnes Whitlow’s birth date and the date of the Turnip Festival. The top 5 hits carried only the first, so the model hedged and named exactly what was missing instead of inventing a date.

This is the single most important property of RAG to understand: the answer can only be as good as the retrieval. Widen the search and the second fact comes along:

print("\nanswer:\n" + await answer_with_rag(QUESTION, top_k=8))

retrieved (top_k=8):
  0.787  Green Village's oldest resident is Agnes Whitlow (born 02.06.1929).
  0.700  Green Village has 104 residents.
  0.681  President of Green Village is Thomas Cook (born 12.04.1994).
  0.668  Green Village was founded on 03.09.1887 by shepherd Elias Thornbury.
  0.615  Green Village's official flower is the marsh marigold.
  0.614  The annual Turnip Festival takes place every year on the third Saturday of October.
  0.588  The tallest building in Green Village is the Old Grain Tower at 23 metres.
  0.574  Green Village has one school with 14 pupils and 2 teachers.

answer:
The 2025 Turnip Festival was on **18 October 2025**, the third Saturday of
October. Agnes Whitlow, born **2 June 1929**, was therefore **96 years old**.

Now the model has both facts, and does the part it is good at — the third Saturday of October 2025, minus a birthday in June 1929. Every fact came from your database; the model supplied the language and the arithmetic.

That division of labour is the whole idea. It also explains the usual failure mode: when a RAG system answers badly, the retrieval is usually at fault, not the model. Tune top_k, the chunking, and the embedding model before you blame the prompt.

Note also the “say so” instruction. Grounding a model reduces invention; it does not abolish it.

Collections, metadata and filters

A collection is, quite literally, a table. Each collection gets its own table, registered in a rag_collections table together with its embedding model and dimension. Under PostgreSQL the table has a typed vector column and its own HNSW index; the SQLite backend keeps the registry and every collection table in one file. The layout is described in The database and data model. Ask the service what it is holding:

for collection in await rag.list_collections():
    print(collection)

print("\nentries in this collection:", await rag.count_entries(COLLECTION))

{'name': 'green_village', 'model': 'fastembed/BAAI/bge-small-en-v1.5', 'embedding_size': 384, 'schema_version': 1, 'count': 17}
{'name': 'songs', 'model': 'fastembed/BAAI/bge-small-en-v1.5', 'embedding_size': 384, 'schema_version': 1, 'count': 23684}

entries in this collection: 17

source_ids narrows a search to specific documents — useful once candidates have been pre-filtered some other way (by metadata, permissions, recency). None, the default, does not filter; a list restricts the search to the sources it names:

narrowed = await rag.query(
    "how tall is it?",
    top_k=3,
    collection_name=COLLECTION,
    source_ids=["fact-03", "fact-12"],  # grain tower, church bell
)
for hit in narrowed:
    print(f"{hit.similarity:.3f}  [{hit.source_id}]  {hit.content}")

0.485  [fact-03]  The tallest building in Green Village is the Old Grain Tower at 23 metres.
0.450  [fact-12]  The local church bell weighs 412 kilograms and was cast in 1901.

An empty list matches nothing. A pre-filter that finds no candidates — a tenant without documents, a permission check that passed nothing — therefore returns no hits instead of widening into a search of the whole collection, and the question is not even embedded. query_batch returns one empty list per question in the same case.

empty = await rag.query(
    "how tall is it?", collection_name=COLLECTION, source_ids=[]
)
everything = await rag.query(
    "how tall is it?", top_k=2, collection_name=COLLECTION, source_ids=None
)
print("source_ids=[]  :", empty)
print("source_ids=None:", [hit.source_id for hit in everything])

source_ids=[]  : []
source_ids=None: ['fact-09', 'fact-03']

Chunked documents and keep_best

Long documents get split into chunks that share one source_id. A plain query can then return several chunks of the same document, crowding everything else out of the top-k. keep_best=True collapses them to the best chunk per source_id.

CHUNKS = [
    "The Rusty Anchor, chapter 1: the pub opened in 1923 in a "
    "former grain store.",
    "The Rusty Anchor, chapter 2: its oak bar was carved by Elias "
    "Thornbury's grandson.",
    "The Rusty Anchor, chapter 3: the pub serves 40 pints on a typical Friday.",
]

await rag.index_batch(
    texts=CHUNKS,
    metadata_list=[{"chunk": i} for i in range(len(CHUNKS))],
    source_ids=["pub-history"] * len(CHUNKS),
    collection_name=COLLECTION,
)

question = "Tell me about the village pub."
crowded = await rag.query(question, top_k=4, collection_name=COLLECTION)
collapsed = await rag.query(
    question, top_k=4, collection_name=COLLECTION, keep_best=True
)

print("without keep_best:", [h.source_id for h in crowded])
print("with keep_best   :", [h.source_id for h in collapsed])

without keep_best: ['pub-history', 'fact-16', 'pub-history', 'fact-01']
with keep_best   : ['pub-history', 'fact-16', 'fact-01', 'fact-11']

Batched queries

For several questions at once, query_batch embeds and searches them in a single database round trip (a CROSS JOIN LATERAL over the query vectors), returning one result list per query, in input order.

questions = ["What do the bees produce?", "When is the library open?"]

for question, results in zip(
    questions,
    await rag.query_batch(
        questions, top_k=2, collection_name=COLLECTION
    ),
):
    print(f"{question!r}")
    for hit in results:
        print(f"   {hit.similarity:.3f}  {hit.content}")

'What do the bees produce?'
   0.617  Green Village produces 8 tons of honey per year from its 26 beehives.
   0.479  Green Village's official flower is the marsh marigold.
'When is the library open?'
   0.624  The village library owns 1,847 books and is open on Tuesdays and Fridays.
   0.479  The Rusty Anchor, chapter 1: the pub opened in 1923 in a former grain store.

Similarity matrix

compute_similarity_matrix scores every text against every listed source_id and returns a 2-D matrix — useful for ranking, clustering or deduplication, where all pairwise scores are needed at once rather than a top-k.

texts = ["honey production", "opening hours", "village leadership"]
targets = ["fact-13", "fact-14", "fact-00"]  # bees, library, president

matrix = await rag.compute_similarity_matrix(
    texts=texts, source_ids=targets, collection_name=COLLECTION
)

print(f"{'':>20}" + "".join(f"{t:>10}" for t in targets))
for text, row in zip(texts, matrix):
    print(f"{text:>20}" + "".join(f"{value:10.3f}" for value in row))

                       fact-13   fact-14   fact-00
    honey production     0.657     0.337     0.415
       opening hours     0.420     0.568     0.468
  village leadership     0.548     0.558     0.663

The diagonal dominates — each phrase scores highest against the fact it is about.

Similarity thresholds

top_k bounds how many hits come back, not how close they are: a question the corpus cannot answer still returns top_k hits, the nearest of which may be far from it. min_similarity drops hits whose similarity is lower. similarity is higher-is-better cosine on both backends, so one number means the same on each; a useful threshold still depends on the embedding model and on whether a normaliser is used, and is best read off queries against the corpus itself. The filter applies after top_k, so fewer than top_k hits may come back.

for question in ["How deep is the pond?", "What is the capital of Peru?"]:
    nearest = await rag.query(question, top_k=3, collection_name=COLLECTION)
    kept = await rag.query(
        question, top_k=3, collection_name=COLLECTION, min_similarity=0.6
    )
    print(question)
    print("  top 3   :", [round(hit.similarity, 3) for hit in nearest])
    print("  >= 0.6  :", [hit.source_id for hit in kept])

How deep is the pond?
  top 3   : [0.735, 0.421, 0.405]
  >= 0.6  : ['fact-09']

What is the capital of Peru?
  top 3   : [0.445, 0.424, 0.414]
  >= 0.6  : []

Filtering by metadata

source_ids narrows a search to particular documents. match narrows it by what was recorded in the metadata: it keeps the entries whose metadata has every key of match equal — top-level keys with string, number or boolean values, the condition delete_by_metadata deletes by. A category, a language or a tenant recorded at indexing time is then a filter at query time.

The filter is part of the nearest-neighbour search, not applied to its result. The top_k nearest entries that match come back, however far from the question the matching entries lie, which a filter over the top_k result could not promise. On PostgreSQL it is metadata @> match, served by the GIN index every collection table carries, with pgvector’s iterative index scan enabled as it is for source_ids; on SQLite it joins the WHERE clause of the scan. An empty mapping or a nested value is refused with ValueError.

question = "What is there to see in the village?"
for match in [None, {"topic": "buildings"}, {"topic": "nature"}]:
    hits = await rag.query(
        question, top_k=2, collection_name=COLLECTION, match=match
    )
    print(f"match={match}")
    for hit in hits:
        print(f"  {hit.similarity:.3f} [{hit.rag_metadata['topic']}] {hit.content}")

match=None
  0.630 [people] Green Village has 104 residents.
  0.602 [streets] The village has 3 streets: Main Road, Willow Lane, and Cobbler's Path.
match={'topic': 'buildings'}
  0.531 [buildings] The tallest building in Green Village is the Old Grain Tower at 23 metres.
  0.388 [buildings] The local church bell weighs 412 kilograms and was cast in 1901.

match={'topic': 'nature'}
  0.591 [nature] Green Village's official flower is the marsh marigold.
  0.557 [nature] The village pond, Lake Miller, is 1.2 metres deep at its deepest point.

The collection’s model

list_collections() above reports the model each collection was built with. That record is authoritative: query, query_batch and index_batch on an existing collection embed with the recorded model, whichever model the service was given, and a warning is logged when the two differ. The service keeps one embedding client per model, so one service serves collections of different models.

The service’s model is therefore only the model new collections are created with. A service built without one — as the backoffice builds it, for indexes it did not create — still queries and indexes existing collections, and refuses only to create a new one:

reader = PostgresRagService.from_uri(
    os.environ["KAVALAI_DB_URI"], schema=os.environ.get("KAVALAI_DB_SCHEMA")
)
hit = (
    await reader.query(
        "Who runs the bakery?", top_k=1, collection_name=COLLECTION
    )
)[0]
print(f"{hit.model}: {hit.content}")

try:
    await reader.index("A new fact.", collection_name="scratch")
except ValueError as error:
    print(error)

fastembed/BAAI/bge-small-en-v1.5: The village bakery, run by Greta Lindqvist, sells exactly 340 loaves every week.
This PostgresRagService was created without an embedding model; creating a collection requires one.

Updating and deleting

Besides delete (one item by id) and delete_by_source_id, two methods remove items by selection. delete_many removes several items by id in one statement. delete_by_metadata removes the items whose metadata has every key of match equal — top-level keys with scalar values (strings, numbers, booleans), which is what both backends serve the same way: metadata @> match on PostgreSQL, answered by the GIN index every collection table carries, and json_extract on SQLite.

notes = await rag.index_batch(
    texts=[
        "The pond freezes over most Januaries.",
        "The pond is home to 30 ducks.",
        "Skating on the pond is forbidden.",
    ],
    metadata_list=[{"topic": "pond", "draft": True}] * 3,
    source_ids=["pond-notes"] * 3,
    collection_name=COLLECTION,
)
print("with the notes          :", await rag.count_entries(COLLECTION))

await rag.delete_many([row["id"] for row in notes[:2]], COLLECTION)
print("after delete_many       :", await rag.count_entries(COLLECTION))

await rag.delete_by_metadata(COLLECTION, {"draft": True})
print("after delete_by_metadata:", await rag.count_entries(COLLECTION))

with the notes          : 23
after delete_many       : 21
after delete_by_metadata: 20

replace re-indexes a document in one step. It selects the old items with match=, source_id= or both, embeds the new texts first — outside any transaction, as the embedding call may go over the network — and then deletes the old items and inserts the new ones in a single transaction. A failure therefore leaves the old items in place rather than a document half indexed. An empty texts deletes the selection. Here the three chapters of the pub’s history, indexed under keep_best above, become one passage:

await rag.replace(
    COLLECTION,
    texts=[
        "The Rusty Anchor, the village pub, opened in 1923 in a former grain "
        "store and serves 40 pints on a typical Friday."
    ],
    metadata_list=[{"chunk": 0}],
    source_ids=["pub-history"],
    source_id="pub-history",
)

for hit in await rag.query(
    "Tell me about the village pub.", top_k=2, collection_name=COLLECTION
):
    print(f"{hit.similarity:.3f}  [{hit.source_id}]  {hit.content}")
print("entries:", await rag.count_entries(COLLECTION))

0.650  [pub-history]  The Rusty Anchor, the village pub, opened in 1923 in a former grain store and serves 40 pints on a typical Friday.
0.624  [fact-16]  Green Village's only pub, The Rusty Anchor, has been operating since 1923.
entries: 18

Recording embedding usage

A RAG service records nothing about its embedding calls by itself. Any object with a receive_model_stats(stats) method — a ModelStatsReceiver — passed as stats_receiver= to index, index_batch, query or query_batch receives the ModelCallStat of that call’s embedding; passed to the constructor, it is the default for every call that does not name its own. Inside a workflow, a rag_query node passes the run’s own receiver, so the query embedding is attributed to the run like any other model call. An indexing script that wants a row in the agent database passes a StatsBridge over a task logger.

class EmbeddingLedger:
    def __init__(self):
        self.calls = []

    def receive_model_stats(self, stats):
        self.calls.append(stats)


ledger = EmbeddingLedger()
await rag.query_batch(
    ["Who is the president?", "How many pupils are there?"],
    collection_name=COLLECTION,
    stats_receiver=ledger,
)
await rag.query("Who is the president?", collection_name=COLLECTION)

for call in ledger.calls:
    print(call.call_type, call.model, "batch_size:", call.batch_size)

embedding fastembed/BAAI/bge-small-en-v1.5 batch_size: 2

Only the call that named the ledger reached it; the second query had no receiver, and nothing was recorded for it.

A portable index: SQLite

SqliteRagService implements the same interface over a single SQLite file (using the sqlite-vector extension), with the same storage model as the Postgres service — a rag_collections registry and one table per collection — so the backoffice RAG explorer reads either. No server, and the file can be built offline and copied wherever it is needed.

The API is identical; only the constructor changes.

from kavalai.rag import SqliteRagService

portable = SqliteRagService(":memory:", model=EMBEDDING_MODEL)

await portable.index_batch(
    texts=FACTS[:6],
    metadata_list=[{"topic": topic} for topic in TOPICS[:6]],
    source_ids=[f"fact-{i:02d}" for i in range(6)],
)

for hit in await portable.query("who bakes the bread?", top_k=2):
    print(f"{hit.similarity:.3f}  {hit.content}")

0.530  The village bakery, run by Greta Lindqvist, sells exactly 340 loaves every week.
0.506  President of Green Village is Thomas Cook (born 12.04.1994).

Pass a path instead of ":memory:" and you get a file you can copy, version or serve.

Retrieval inside a workflow

Everything so far has been Python calling a RagService directly. A workflow can do the retrieval itself, with a rag_query node — so the search becomes part of the document, shows up in the diagram, and is recorded as its own task alongside the model call.

The node is read-only: it calls query and nothing else. Indexing stays in Python, where you can see what you are writing and to which collection.

query is a template, exactly like an llm node’s prompt, so it can pull the question straight out of the run context. store="content" keeps just the hit texts — which is what the next prompt wants; the default, store="results", keeps the full RagServiceResult list so scores and metadata stay available for routing. top_k, source_ids, match, keep_best and min_similarity mean what they mean on query; the string values of match are templates, so a filter an earlier node extracted can drive the search.

The service is passed to the engine, not named in the document: a workflow is served over HTTP and edited in the backoffice, so it carries a name, never a connection string. With one index you do not even need the name — a single service is registered as "default", which is what a node resolves to when neither it nor the workflow says otherwise.

from kavalai import WorkflowBuilder

village_index = SqliteRagService(":memory:", model=EMBEDDING_MODEL)
await village_index.index_batch(
    texts=FACTS,
    metadata_list=[{"topic": topic} for topic in TOPICS],
    source_ids=[f"fact-{i:02d}" for i in range(len(FACTS))],
)

ANSWER_PROMPT = """Answer the question using the facts below.

Every village-specific detail must come from these facts — do not invent any.
You may use general knowledge such as the calendar to reason about them.

Facts:
{{ context.facts }}

Question: {{ context.input.question }}
"""

village_desk = (
    WorkflowBuilder("green-village-desk", llm_model="openai/gpt-5.6-luna")
    .data_type("input", {"question": str})
    .data_type("output", {"answer": str})
    .start(next="retrieve")
    .rag_query(
        "retrieve",
        query="{{ context.input.question }}",
        output="facts",
        next="answer",
        top_k=8,
        min_similarity=0.5,
        store="content",
    )
    .llm("answer", prompt=ANSWER_PROMPT, output="output", next="done")
    .end(name="done", output="output")
    .build_engine(rag_services=village_index)
)

state = await village_desk.run({"question": QUESTION})

print("trace:  ", " -> ".join(state.trace))
print(textwrap.fill("answer:  " + state.output_data["answer"], 80))

trace:   start -> retrieve -> answer -> done
answer:  Agnes Whitlow was 96 years old at the 2025 Turnip Festival, held on 18
October 2025.

Two nodes, and the whole RAG loop is a document. The retrieval step is visible in the rendered graph, and the run recorded a task for it — how long the search took and what it returned — next to the model call it fed.

To point several nodes at different indexes, pass a dict (rag_services={"handbook": ..., "tickets": ...}) and name one per node with service=, or set a workflow-wide default with WorkflowBuilder(..., rag_service="handbook"). Under python -m kavalai.server, where nobody constructs the engine, register the service instead — see Cookbook.

Whatever store says, the node records its hits — id, source_id, similarity and metadata, without the text, which the context already holds. They are written to the node’s task row and carried in the output_data of its node_completed event, so the passages behind an answer can be cited and audited afterwards. run_stream yields that event:

async for event in village_desk.run_stream({"question": QUESTION}):
    if event.type == "node_completed" and event.name == "retrieve":
        recorded = event.output_data["hits"]
        print(len(recorded), "hits recorded; the first three:")
        for hit in recorded[:3]:
            print(f"  {hit['similarity']:.3f}  {hit['source_id']}  "
                  f"{hit['metadata']}")

8 hits recorded; the first three:
  0.787  fact-10  {'topic': 'people'}
  0.700  fact-01  {'topic': 'people'}
  0.681  fact-00  {'topic': 'people'}

Cleanup

drop_collection removes the collection’s table and its registry entry. delete_by_source_id, delete_many and delete_by_metadata remove individual documents.

await rag.drop_collection(COLLECTION)
print("collections now:", [c["name"] for c in await rag.list_collections()])

collections now: ['songs']

Where to next