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.

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?")

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().

See Also¶

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