// Example 11 · Advanced

A full RAG system.

Retrieval-augmented generation from Cosmonapse primitives only: four Neurons and three Engrams, a node each, run as retrieve, rerank, generate on one trace. Hybrid semantic and lexical recall fused by reciprocal rank, and an answer cache two Neurons share.
cosmonapse-examples/11-rag
Install & run

Clone the examples repo, install, and run from inside the example's folder.

terminal
$ git clone https://github.com/Cosmonapse/cosmonapse-examples
$ pip install 'cosmonapse[receptor]' httpx python-dotenv
# a Hugging Face token with read scope, in cosmonapse-examples/.env
$ echo "HF_TOKEN=hf_xxxxxxxxxxxxxxxx" >> cosmonapse-examples/.env

$ cd cosmonapse-examples/11-rag
$ python brain.py "What is a Dendrite and what role does it play?"
$ python brain.py stats
$ python brain.py            # REPL + the HTTP API on :8000 - ask twice to hit the cache

$ curl -s localhost:8000/ask -H 'content-type: application/json' -d '{"question": "What is a Dendrite?"}'
$ curl -s localhost:8000/ingest -H 'content-type: application/json' -d '{"doc_id": "notes", "text": "..."}'
$ curl -s localhost:8000/stats
  indexed cosmonapse-core          2 chunks
  indexed memory-and-pathways      2 chunks
  indexed routing-and-bidding      2 chunks
A Dendrite is the Synapse-side participant that hosts Axons ... [cosmonapse-core#0]
sources: cosmonapse-core#0 (1.577), memory-and-pathways#1 (1.4569), ...

The answer text varies; the pipeline shape does not.

Layout

One component per file, one Dendrite per node.

The skeleton cosmo init scaffolds. Modules under neurons/, engram/, effector/ and receptors/ declare behaviour; brain.py owns deployment - which node hosts what.

11-rag
11-rag/
  config.py              models, TOP_K / FETCH_K / MIN_SCORE, cache_key()
  embeddings.py          HF embedding client + chunker
  engram/vector_engram.py   VectorEngram - cosine, over the Engram ABC
  engram/keyword_engram.py  KeywordEngram - BM25
  engram/indexes.py      VECTORS, KEYWORDS, CACHE
  neurons/               ingester, retriever, reranker, generator
  helpers.py             ask_pipeline(), ingest_samples(), stats()
  receptors/terminal.py  ask, stats
  receptors/api.py       POST /ingest, POST /ask, GET /stats
  brain.py               a node per Engram, per Neuron, per interface
01 · The indexes

Two Engrams, one protocol.

VectorEngram and KeywordEngram implement the Engram ABC, so RECALL and IMPRINT carry vector search and BM25 without new Signal types. Each answers only queries in its own language (can_serve), so they never answer each other's. The cache is a stock InMemoryEngram.

engram/indexes.py
from cosmonapse import InMemoryEngram

from engram.keyword_engram import KeywordEngram
from engram.vector_engram import VectorEngram

VECTORS = VectorEngram(engram_id="rag-vectors", engram_kind="semantic")
KEYWORDS = KeywordEngram(engram_id="rag-keywords", engram_kind="lexical")
CACHE = InMemoryEngram(engram_id="rag-cache", engram_kind="context")
engram/vector_engram.py
class VectorEngram(Engram):
    """In-memory vector store behind the Engram protocol.

    Entry shape (imprint):
        {"text": str, "doc_id": str, "chunk_index": int,
         "embedding": list[float], ...any metadata}

    Query shape (recall):
        {"embedding": list[float], "top_k": int = 4}
        filters: {"doc_id": str} narrows to one document.
    """

    def __init__(
        self,
        *,
        engram_id: str = "rag-vectors",
        engram_kind: str = "semantic",
        version: str | None = "0.1.0",
    ) -> None:
        self.engram_id = engram_id
        self.engram_kind = engram_kind
        self.capabilities = ["vector", "cosine", "merge_key"]
        self.version = version
        self._entries: dict[str, dict[str, Any]] = {}
        self._merge_keys: dict[str, str] = {}  # merge_key -> entry id
        self._lock = asyncio.Lock()

    # -- lifecycle -----------------------------------------------------
    async def connect(self) -> None: ...

    async def close(self) -> None:
        self._entries.clear()
        self._merge_keys.clear()

    # -- capability negotiation -----------------------------------------
    async def can_serve(self, query: dict[str, Any]) -> bool:
        return "embedding" in query

    # -- read ------------------------------------------------------------
    async def recall(
        self,
        query: dict[str, Any],
        *,
        filters: dict[str, Any] | None = None,
        context_ref: str | None = None,
        deadline_ms: int | None = None,
        min_confidence: float | None = None,
    ) -> list[Hit]:
        t0 = time.monotonic()
        qvec = query.get("embedding")
        if not qvec:
            return []
        top_k = int(query.get("top_k", 4))

        async with self._lock:
            items = list(self._entries.items())

        scored: list[Hit] = []
        for eid, entry in items:
            if filters and any(entry.get(k) != v for k, v in filters.items()):
                continue
            score = cosine(qvec, entry.get("embedding", []))
            if min_confidence is not None and score < min_confidence:
                continue
            # Strip the embedding from the wire reply - it is bulky and
            # the caller only needs text + metadata + score.
            slim = {k: v for k, v in entry.items() if k != "embedding"}
            scored.append(Hit(id=eid, entry=slim, score=score))
            if deadline_ms and (time.monotonic() - t0) * 1000 > deadline_ms:
                break

        scored.sort(key=lambda h: h.score, reverse=True)
        return scored[:top_k]

    # ... imprint(): add / merge / upsert / delete, with a per-trace saga journal
02 · Ingest

One chunk, two indexes.

The ingester chunks and embeds, then imprints every chunk into both indexes under the same merge_key, so re-ingesting a document upserts instead of duplicating.

neurons/ingester.py
from cosmonapse import Axon, EngramBinding

from config import hf_token
from embeddings import chunk_text, embed


async def ingest_neuron(input, context, *, imprint):
    """{"doc_id", "text"} -> {"doc_id", "chunks"}. Dual-writes both indexes."""
    doc_id = input["doc_id"]
    chunks = chunk_text(input["text"])
    vectors = await embed(chunks, api_key=hf_token())

    for i, (chunk, vec) in enumerate(zip(chunks, vectors)):
        last = i == len(chunks) - 1
        entry = {"doc_id": doc_id, "chunk_index": i, "text": chunk}
        await imprint(
            "vectors", op="upsert",
            entry={**entry, "embedding": vec},
            merge_key=f"{doc_id}:{i}", await_ack=last, deadline_ms=2000,
        )
        await imprint(
            "keywords", op="upsert",
            entry=entry,
            merge_key=f"{doc_id}:{i}", await_ack=last, deadline_ms=2000,
        )
    return {"doc_id": doc_id, "chunks": len(chunks)}


AXON = Axon(
    neuron_id="ingester", neuron_fn=ingest_neuron,
    capabilities=["rag-ingest"],
    engrams=[EngramBinding(name="vectors", directed_id="rag-vectors"),
             EngramBinding(name="keywords", directed_id="rag-keywords")],
)
03 · Retrieve

Recall both, fuse the ranks.

The retriever checks the answer cache, then recalls from both indexes in their native query shapes and merges the ranked lists with reciprocal-rank fusion.

neurons/retriever.py
from cosmonapse import Axon, EngramBinding

from config import FETCH_K, MIN_SCORE, RRF_K, TOP_K, cache_key, hf_token
from embeddings import embed


async def retrieve_neuron(input, context, *, recall):
    """{"question", "top_k"?} -> {"cache_hit"} | {"candidates": [...]}.

    Hybrid recall: semantic (vectors) + lexical (keywords), fused with
    reciprocal-rank fusion. Checks the answer cache first.
    """
    question = input["question"]
    top_k = int(input.get("top_k", TOP_K))

    # 0. Answer-cache short-circuit (exact normalised question).
    cached = await recall(
        "cache", query={"merge_key": cache_key(question)}, deadline_ms=500,
    )
    if cached.hits:
        return {"cache_hit": True, "answer": cached.hits[0].entry["content"]}

    # 1. Both backends in their native query language.
    qvec = (await embed([question], api_key=hf_token()))[0]
    sem = await recall(
        "vectors", query={"embedding": qvec, "top_k": FETCH_K},
        min_confidence=MIN_SCORE, deadline_ms=2000,
    )
    lex = await recall(
        "keywords", query={"text": question, "top_k": FETCH_K},
        deadline_ms=2000,
    )

    # 2. Reciprocal-rank fusion across the two ranked lists.
    fused: dict[str, dict] = {}
    for hits in (sem.hits, lex.hits):
        for rank, h in enumerate(hits):
            key = f"{h.entry['doc_id']}#{h.entry['chunk_index']}"
            slot = fused.setdefault(key, {
                "id": key, "doc_id": h.entry["doc_id"],
                "chunk_index": h.entry["chunk_index"],
                "text": h.entry["text"], "rrf": 0.0,
            })
            slot["rrf"] += 1.0 / (RRF_K + rank + 1)

    candidates = sorted(fused.values(), key=lambda c: c["rrf"], reverse=True)
    return {"cache_hit": False, "candidates": candidates[: top_k * 2]}


AXON = Axon(
    neuron_id="retriever", neuron_fn=retrieve_neuron,
    capabilities=["rag-retrieve"],
    engrams=[EngramBinding(name="vectors", directed_id="rag-vectors"),
             EngramBinding(name="keywords", directed_id="rag-keywords"),
             EngramBinding(name="cache", directed_id="rag-cache")],
)
04 · Rerank and generate

A cheap rescore, a grounded answer, cached on the way out.

The reranker is lexical overlap; replace it with a cross-encoder Neuron and nothing else changes. The generator answers from the chunks only and upserts the answer into the same cache the retriever reads.

neurons/reranker.py
from cosmonapse import Axon

from config import RRF_K, TOP_K
from engram.keyword_engram import tokenize


async def rerank_neuron(input, context):
    """{"question", "candidates", "top_k"?} -> {"chunks": [...]}."""
    qterms = set(tokenize(input["question"]))
    top_k = int(input.get("top_k", TOP_K))

    rescored = []
    for c in input["candidates"]:
        terms = set(tokenize(c["text"]))
        overlap = len(qterms & terms) / (len(qterms) or 1)
        rescored.append({**c, "score": round(0.7 * c["rrf"] * RRF_K + 0.3 * overlap, 4)})
    rescored.sort(key=lambda c: c["score"], reverse=True)
    return {"chunks": rescored[:top_k]}


AXON = Axon(
    neuron_id="reranker", neuron_fn=rerank_neuron,
    capabilities=["rag-rerank"],
)
neurons/generator.py
from cosmonapse import Axon, EngramBinding, Neuron

from config import GEN_MODEL, SYSTEM_PROMPT, cache_key, hf_token


def make_generate_neuron():
    """{"question", "chunks"} -> {"answer", "sources"}. Caches the answer."""
    llm = Neuron(
        source="huggingface",
        endpoint="https://router.huggingface.co",
        model=GEN_MODEL,
        api_key=hf_token(),
        use_chat_api=True,
        max_new_tokens=512,
        temperature=0.2,
    )

    async def generate_neuron(input, context, *, imprint):
        question = input["question"]
        chunks = input["chunks"]
        if not chunks:
            return {"answer": "No relevant context found in the index.",
                    "sources": []}

        blocks = [f"[{c['id']}]\n{c['text']}" for c in chunks]
        messages = [
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": (
                "Context:\n\n" + "\n\n---\n\n".join(blocks)
                + f"\n\nQuestion: {question}"
            )},
        ]
        out = await llm({"messages": messages}, [])
        answer = out["response"].strip()

        await imprint(
            "cache", op="upsert",
            entry={"content": answer, "tags": ["answer"]},
            merge_key=cache_key(question),
            await_ack=False,
        )
        sources = [{"id": c["id"], "doc_id": c["doc_id"],
                    "chunk_index": c["chunk_index"], "score": c["score"]}
                   for c in chunks]
        return {"answer": answer, "sources": sources}

    return generate_neuron


def make_axon() -> Axon:
    return Axon(
        neuron_id="generator", neuron_fn=make_generate_neuron(),
        capabilities=["rag-generate"],
        engrams=[EngramBinding(name="cache", directed_id="rag-cache")],
    )
05 · The pipeline

Three TASKs, one trace.

ask_pipeline chains the stages with an explicit trace_id and parent_id, so Prism shows one lineage per question. Both interfaces call it from the node they are mounted on.

helpers.py
async def ask_pipeline(orchestrator, question: str, *, top_k: int = TOP_K,
                       timeout_s: float = 60.0) -> dict:
    """retrieve -> rerank -> generate, chained on ONE trace_id, so Prism
    shows the whole workflow as a single lineage."""
    tid = new_trace_id()

    r = await orchestrator.dispatch_and_wait(
        neuron="retriever", input={"question": question, "top_k": top_k},
        trace_id=tid, timeout_s=timeout_s,
    )
    out = r.payload["output"]
    if out.get("cache_hit"):
        return {"answer": out["answer"], "sources": [], "cached": True}

    r = await orchestrator.dispatch_and_wait(
        neuron="reranker",
        input={"question": question, "candidates": out["candidates"],
               "top_k": top_k},
        trace_id=tid, parent_id=r.id, timeout_s=timeout_s,
    )
    chunks = r.payload["output"]["chunks"]

    r = await orchestrator.dispatch_and_wait(
        neuron="generator",
        input={"question": question, "chunks": chunks},
        trace_id=tid, parent_id=r.id, timeout_s=timeout_s,
    )
    return {**r.payload["output"], "cached": False}
06 · Two interfaces

A terminal and an API onto the same nodes.

POST /ingest is the ApiReceptor's own endpoint: one TASK, with send, wait and stream modes. /ask and /stats are extra routes, because the pipeline is three TASKs rather than one. See Receptors for the interface layer on its own.

receptors/terminal.py
from cosmonapse import CliReceptor

from engram.indexes import VECTORS
from helpers import ask_pipeline, ingest_samples, stats as index_stats

RECEPTOR = CliReceptor(
    prog="rag",
    description="Hybrid-retrieval RAG over sample_docs/.",
    timeout_s=120.0,
)


@RECEPTOR.command(local=True, default=True, help="ask a question")
async def ask(question: str):
    if VECTORS.size == 0:
        for doc in await ingest_samples(RECEPTOR.dendrite):
            print(f"  indexed {doc['doc_id']:<24} {doc['chunks']} chunks")
    out = await ask_pipeline(RECEPTOR.dendrite, question)
    lines = [("(cached) " if out["cached"] else "") + out["answer"]]
    if out["sources"]:
        lines.append("sources: " + ", ".join(f"{s['id']} ({s['score']})"
                                             for s in out["sources"]))
    return "\n".join(lines)


@RECEPTOR.command(local=True, help="index sizes")
def stats():
    return index_stats()
receptors/api.py
from cosmonapse import ApiReceptor

from engram.indexes import VECTORS
from helpers import ask_pipeline, ingest_samples, stats as index_stats

RECEPTOR = ApiReceptor(neuron="ingester", path="/ingest", timeout_s=120.0)


@RECEPTOR.route("/ask", methods=["POST"])
async def ask(body: dict):
    if VECTORS.size == 0:
        await ingest_samples(RECEPTOR.dendrite)
    return await ask_pipeline(RECEPTOR.dendrite, body["question"],
                              top_k=int(body.get("top_k", 4)))


@RECEPTOR.route("/stats")
async def stats():
    return index_stats()
07 · The brain

Nine nodes, three builders.

brain.py
import asyncio
import contextlib
import signal

from cosmonapse import Dendrite, MemorySynapse, connect_synapse, run_brain

from config import NAMESPACE, SYNAPSE_URL
from engram.indexes import CACHE, KEYWORDS, VECTORS
from neurons import generator, ingester, reranker, retriever
from receptors import api, terminal


def build_engram(synapse, engram) -> Dendrite:
    node = Dendrite(synapse=synapse, namespace=NAMESPACE,
                    dendrite_id=f"{engram.engram_id}-node", role="worker")
    node.attach_engram(engram)
    return node


def build_neuron(synapse, axon) -> Dendrite:
    node = Dendrite(synapse=synapse, namespace=NAMESPACE,
                    dendrite_id=f"{axon.neuron_id}-node", role="worker")
    node.attach_axon(axon)
    return node


def build_interface(synapse, name: str, receptor) -> Dendrite:
    """Interfaces run the pipeline, so they sit on orchestrator nodes."""
    node = Dendrite(synapse=synapse, namespace=NAMESPACE,
                    dendrite_id=f"{name}-node", heartbeat_s=0)
    node.attach_receptor(receptor)
    return node

main() passes every Engram, every Neuron and both interfaces to run_brain. Set RAG_GEN_MODEL, RAG_TOP_K, RAG_FETCH_K or RAG_MIN_SCORE to tune.

Over a real synapse

Same code, different transport.

With SYNAPSE_URL unset the brain runs on an in-process bus. Point it at a synapse and nothing in the modules changes. Prism attaches to any shared synapse and animates every Signal as it crosses.

terminal
# terminal 1 - a dev synapse (TCP, single host)
$ cosmo synapse start memory --namespace=rag

# terminal 2 - the same brain, over that synapse
$ SYNAPSE_URL=cosmo://127.0.0.1:7070 python brain.py "What is a Dendrite?"

# terminal 3 - Prism, the live view (http://127.0.0.1:7071)
$ cosmo prism --url=cosmo://127.0.0.1:7070 -n rag

# production transports - only the URL changes
$ SYNAPSE_URL=nats://127.0.0.1:4222 python brain.py "What is a Dendrite?"
$ SYNAPSE_URL=kafka://127.0.0.1:9092 python brain.py "What is a Dendrite?"
http://127.0.0.1:7071 · -n rag
Prism renders every Signal on the bus as it fires - REGISTER, TASK, AGENT_OUTPUT, FINAL.
Related
→

RAG + MCP

The same retrieval stack grounding a coding agent.

→

Retry & rollback

Undo a half-finished ingest with the saga journal.

→

RAG CLI

One Neuron, one Engram, one Effector - a RAG that fills its own memory.