The registry pattern

A registry is just a dict from a string name to a function. That is the whole pattern. The power comes from what you can do with it: iterate it, look up by name, validate task lists against its keys, expose its keys as the API contract.

agent/tasks.py
python
TASK_REGISTRY = {
    "sentiment": classify_sentiment,
    "entities": extract_entities,
    "keywords": extract_keywords,
    "summary": summarize,
    "emotion": detect_emotion,
}

Six lines. This is the entire orchestration contract. Adding a sarcasm detector means writing one more async function and adding "sarcasm": detect_sarcasm to this dict.

agent/graph.py
python
from agent.tasks import TASK_REGISTRY
from models import VALID_TASKS


def _validate_tasks(tasks: list[str]) -> list[str]:
    seen = []
    for t in tasks:
        t = t.lower().strip()
        if t in VALID_TASKS and t not in seen:
            seen.append(t)
    if not seen:
        seen = ["sentiment", "summary"]
    return seen


async def _run_plain(text: str, tasks: list[str]) -> dict:
    out: dict = {"errors": {}}
    for name in tasks:
        fn = TASK_REGISTRY[name]
        try:
            out[name] = await fn(text)
        except Exception as e:
            out["errors"][name] = str(e)
    return out

The dispatch loop never mentions sentiment, entities, or any specific task by name. It iterates whatever the caller requested, looks up the function in the registry, and runs it. Adding tasks never requires touching this code.

You could, but every new task would mean editing the dispatch logic, the validation logic, and the API documentation by hand. The registry makes the task list the single source of truth. VALID_TASKS can be computed as set(TASK_REGISTRY.keys()). The FastAPI docs can list supported tasks automatically. Tests can iterate the registry. One dict replaces a dozen scattered updates.

Flashcards: Flashcards

Loading practice…