Protocol-Based Architecture

anchor uses Python Protocols (PEP 544) to define all extension points. This page explains what protocols are, why anchor chose them over class inheritance, and how to implement your own.

What Are Protocols?

A Protocol is a way to declare an interface in Python using structural subtyping -- also known as "static duck typing." A class satisfies a protocol if it has the right methods with the right signatures. No base class or registration is needed.

from typing import Protocol, runtime_checkable

@runtime_checkable
class Retriever(Protocol):
    def retrieve(self, query, top_k=10):
        ...

Any object with a retrieve(query, top_k) method satisfies the Retriever protocol -- even if it has never seen the protocol definition.

[!NOTE] PEP 544 Protocols were introduced in Python 3.8 via PEP 544. They are part of the typing module and are fully supported by mypy, pyright, and other type checkers.

Why Protocols Over Inheritance?

Traditional frameworks use abstract base classes (ABCs) to define interfaces. This creates problems:

With protocols, you can wrap any existing object -- a Pinecone client, a custom database class, a test stub -- without modifying its inheritance chain. If it has the right methods, it works.

Protocol Families

anchor defines protocols across seven families. Every protocol is @runtime_checkable, so you can use isinstance() checks at runtime.

Retrieval Protocols

For fetching and ranking context items.

Memory Protocols

For conversation history, persistent facts, and memory lifecycle.

Storage Protocols

For persisting context items, vectors, documents, and memory entries.

Observability Protocols

For tracing, metrics, and monitoring.

Query Protocols

For query transformation, classification, and routing.

Ingestion Protocols

For document parsing and chunking.

Evaluation Protocols

For assessing retrieval and generation quality.

Infrastructure Protocols

For caching and tokenization.

Implementing a Protocol

To implement a protocol, just write a class with the matching methods. Here is a concrete example implementing the Retriever protocol:

from anchor import ContextItem, SourceType
from anchor.models.query import QueryBundle

class KeywordRetriever:
    """A simple keyword-based retriever -- satisfies the Retriever protocol."""

    def __init__(self, documents: list[str]):
        self._docs = documents

    def retrieve(self, query: QueryBundle, top_k: int = 10) -> list[ContextItem]:
        query_words = set(query.query_str.lower().split())
        scored = []
        for doc in self._docs:
            overlap = len(query_words & set(doc.lower().split()))
            if overlap > 0:
                scored.append((doc, overlap))
        scored.sort(key=lambda x: -x[1])
        return [
            ContextItem(
                content=doc,
                source=SourceType.RETRIEVAL,
                score=min(1.0, count / len(query_words)) if query_words else 0.0,
            )
            for doc, count in scored[:top_k]
        ]

This class works with retriever_step() and ContextPipeline without importing any base class:

from anchor import ContextPipeline
from anchor.pipeline.step import retriever_step

retriever = KeywordRetriever(["Python is great", "RAG combines retrieval with generation"])
pipeline = ContextPipeline(max_tokens=8192).add_step(retriever_step("keyword", retriever))
result = pipeline.build("What is RAG?")

[!TIP] Runtime checking All protocols are @runtime_checkable, so you can verify at runtime:

from anchor.protocols import Retriever
assert isinstance(retriever, Retriever)  # True

Async Protocol Pairs

Many protocols come in sync/async pairs. The async variant uses a different method name (prefixed with a) to avoid ambiguity:

Use the sync variant with pipeline.build() and the async variant with pipeline.abuild().

[!CAUTION] Sync steps in async pipelines abuild() can run both sync and async steps -- sync functions are called directly. But build() cannot run async steps and will raise TypeError if it encounters one.

See Also

• Architecture -- how the pipeline uses protocols
• Token Budgets -- per-source allocation
• Retrieval Guide -- practical retriever examples