Provider Development Guide
This document describes how to build, register, configure, and test providers for Face Intel. A provider is the unit of pluggable capability in the platform β anything from a face detector to a reverse image search engine to a forensics analyzer.
Read this end-to-end before adding your first provider. The contract is small but strict β providers that follow it get caching, retries, circuit breaking, metrics, and structured logging for free.
Table of Contents
- Architecture at a Glance
- The Provider Protocol
- The Capability Enum
- ProviderResult Contract
- BaseProvider Helpers
- Step-by-Step: Adding a New Provider
- Manifest Entry Format
- Settings Flag Convention
- Health Check Pattern
- Error Handling Pattern (
_safe_execute) - PipelineOutput β What You Receive
- Testing Pattern for Providers
- Reference Providers
- Common Pitfalls
1. Architecture at a Glance
HTTP request β Service β Pipeline (validate/preprocess/hash/feature-extract)
β
PipelineOutput
β
Orchestrator.run(...)
β
βββββββββββββββ¬ββββββββββββββ¬ββββββββββββββ
βΌ βΌ βΌ βΌ
Provider A Provider B Provider C Provider D
β β β β
βββββββββββββββ΄ββββββββββββββ΄ββββββββββββββ
β
dict[str, ProviderResult]
β
ReportMerger.merge(...)
β
UnifiedFaceReport
Key invariants:
- Providers receive a
PipelineOutput, never raw user input. - Providers are isolated β a failing optional dependency in one provider must never affect another.
- Every result (success or failure) is preserved as
Evidencein the final report β nothing is silently discarded. - Providers do not import from siblings. They import only from
models/,utils/,config/,pipeline/, andproviders/base.py.
See docs/ARCHITECTURE.md for the full layer
dependency graph and execution flow.
2. The Provider Protocol
Defined in providers/base.py:
@runtime_checkable
class Provider(Protocol):
name: str
capability: ProviderCapability
def is_available(self) -> bool: ...
def execute(self, input_data: Any) -> ProviderResult: ...
This is a structural protocol, not an ABC. Any object that exposes
those four members satisfies the contract β no inheritance required.
In practice, almost all providers subclass BaseProvider to inherit
timing and error handling (see Β§5).
| Member | Purpose |
|---|---|
name |
Stable string identifier used in logs, metrics, cache keys, and the manifest. Lowercase, snake_case. |
capability |
One of the 7 ProviderCapability values. Determines which job kinds invoke this provider. |
is_available() |
Returns False if optional deps or API keys are missing. The orchestrator still calls execute(), but BaseProvider._safe_execute short-circuits and returns a NotConfigured result. |
execute(input_data) |
Receives a PipelineOutput. Returns a ProviderResult. Never raises β see Β§10. |
3. The Capability Enum
Defined in models/providers.py:
class ProviderCapability(str, enum.Enum):
DETECTION = "detection"
RECOGNITION = "recognition"
SCRAPING = "scraping"
REVERSE_SEARCH = "reverse_search"
IMAGE_ANALYSIS = "image_analysis" # quality, properties, visual features
METADATA = "metadata" # EXIF, XMP, IPTC
FORENSICS = "forensics" # integrity, duplicates, manipulation
The orchestrator routes a request to providers by querying the registry
with list_by_capability(cap). The services layer requests the
following capabilities per job kind:
JobKind |
Capabilities invoked |
|---|---|
DETECTION |
DETECTION |
RECOGNITION |
RECOGNITION |
SEARCH |
SCRAPING, REVERSE_SEARCH |
IMAGE_ANALYSIS |
IMAGE_ANALYSIS |
METADATA |
METADATA |
FORENSICS |
FORENSICS |
FULL_PIPELINE |
All of the above (concurrently) |
Adding a new capability
If you need an 8th capability (e.g. AGE_ESTIMATION):
- Add the enum value to
models/providers.py::ProviderCapability. - Add a corresponding list field on
UnifiedFaceReportinmodels/reports.py(e.g.age_estimations: List[...]). - Add a
Normalized*DTO innormalization/schema.py. - Add a
_collect_*collector method innormalization/merger.py. - Add a
score_*method inconfidence/engine.py. - Map any new
JobKindto the new capability inservices/analysis_service.py::_KIND_TO_CAPABILITY(or create a new service).
Capability additions are rare β confirm with the architecture owner first. The seven current capabilities cover all planned Phase 3 work.
4. ProviderResult Contract
Defined in providers/base.py:
@dataclass
class ProviderResult:
provider: str
capability: ProviderCapability
success: bool
elapsed_ms: float
raw: Any = None # verbatim provider response (evidence)
normalized: dict = field(default_factory=dict) # cleaned fields
error: Optional[str] = None
error_type: Optional[str] = None # e.g. "RuntimeError", "TimeoutError"
metadata: dict = field(default_factory=dict)
retry_count: int = 0
| Field | Required? | Purpose |
|---|---|---|
provider |
yes | Same string as name. |
capability |
yes | Same enum value as capability. |
success |
yes | True if the provider produced a usable result. |
elapsed_ms |
yes | Wall-clock time in milliseconds. |
raw |
recommended | The verbatim provider response (API payload, model output dict, etc.). Stored as Evidence β never modified. |
normalized |
recommended | Cleaned fields in the schema the merger expects (see Β§11). |
error |
on failure | Human-readable error message. |
error_type |
on failure | Python exception class name. |
metadata |
optional | Free-form dict (e.g. {"cache_hit": True}). |
retry_count |
set by orchestrator | How many retries occurred before this result. |
Evidence-first principle: raw is preserved verbatim in every job's
UnifiedFaceReport.evidence list. Even if normalized is wrong or the
schema changes, the original data survives for forensic review.
5. BaseProvider Helpers
BaseProvider is the recommended base class. It provides:
- Constructor that accepts an optional
Settingsinstance (DI-friendly). execute()that delegates to_safe_execute()._safe_execute()wraps_run()with timing + try/except (see Β§10).- Default
is_available()returnsTrue(override when needed).
Subclasses implement only:
def _run(self, pipeline_output: PipelineOutput) -> tuple[Any, dict]:
"""Return (raw, normalized). Raise on error β _safe_execute catches."""
The orchestrator ALWAYS passes a
pipeline.feature_extraction.PipelineOutput.
Providers extract what they need from it (image, face crops, gallery,
scrape URL, original bytes for EXIF/forensics).
6. Step-by-Step: Adding a New Provider
This is the complete process. After these three changes, the orchestrator, services, API, and UI pick up the new provider automatically β zero changes elsewhere.
Step 1 β Implement the provider
Create providers/<category>/<your_provider>.py. The category folder
must already exist (detection/, recognition/, scraper/,
reverse/, image_analysis/, metadata/, forensics/).
"""
MyProvider β one-line description.
Optional: implementation notes, references to papers or APIs,
verified facts, failure modes, rate limits, etc.
"""
from __future__ import annotations
import cv2
import numpy as np
from config.settings import Settings, settings as _default_settings
from pipeline.feature_extraction import PipelineOutput
from providers.base import BaseProvider, ProviderCapability, ProviderResult
from utils.image import BBox
class MyProvider(BaseProvider):
"""What this provider does, in one sentence."""
name = "my_provider"
capability = ProviderCapability.DETECTION # or RECOGNITION / etc.
def __init__(self, settings: Settings | None = None) -> None:
super().__init__(settings=settings or _default_settings)
# Initialize your model / API client / detector here.
# If a required dependency is missing, store the error and
# return False from is_available() β do NOT raise.
self._init_error: str | None = None
try:
# ... your initialization ...
self._model = self._load_model()
except Exception as e:
self._init_error = str(e)
# ------------------------------------------------------------------ #
# Health check
# ------------------------------------------------------------------ #
def is_available(self) -> bool:
"""Return False if optional deps / API keys / models are missing."""
return self._model is not None and self._init_error is None
# ------------------------------------------------------------------ #
# Core logic β return (raw, normalized). Raise on error.
# ------------------------------------------------------------------ #
def _run(self, pipeline_output: PipelineOutput) -> tuple[dict, dict]:
img: np.ndarray = pipeline_output.image
# ... do your work ...
boxes = [BBox(0, 0, 100, 100).to_dict()]
raw = {
"model": "my_model_v1",
"detections": [{"box": [0, 0, 100, 100], "confidence": 0.95}],
"num_faces": 1,
}
normalized = {
"boxes": boxes,
"num_faces": len(boxes),
"confidences": [0.95],
"landmarks": None,
}
return raw, normalized
Step 2 β Add one manifest entry
Edit providers/registry.py and append to
PROVIDER_MANIFEST:
PROVIDER_MANIFEST: list[ManifestEntry] = [
# ... existing entries ...
ManifestEntry(
name="my_provider",
module_path="providers.detection.my_provider",
class_name="MyProvider",
capability=ProviderCapability.DETECTION,
enable_flag="enable_my_provider",
description="My provider β one-line description",
optional_dependency=True, # True if it imports an optional package
),
]
Step 3 β Add the enable flag
Edit config/settings.py and add the flag in
the "Provider enable flags" section:
enable_my_provider: bool = False
# ... any tuning knobs your provider needs, e.g.:
my_provider_threshold: float = 0.7
Also add the corresponding line to .env.example:
FI_ENABLE_MY_PROVIDER=false
Step 4 β Done
You're finished. On next startup:
ProviderRegistry.discover()imports your module and instantiatesMyProvider(settings=...).- If
enable_my_provider=False, the provider isDISABLEDand never instantiated. - If
enable_my_provider=Truebutis_available()returnsFalse, the provider shows up asNOT_CONFIGUREDin/providers. - If both pass, the orchestrator invokes it for matching capability queries.
/providerslists it,/statstracks its metrics, the circuit breaker protects against its failures, and the cache dedupes identical invocations.
No other file needs to change. This is the core extensibility guarantee of the platform.
7. Manifest Entry Format
ManifestEntry is a dataclass declared in providers/registry.py:
@dataclass
class ManifestEntry:
name: str # stable id, matches Provider.name
module_path: str # dotted import path to the module
class_name: str # class to instantiate
capability: ProviderCapability # routes to orchestrator
enable_flag: str # Settings attribute name
description: str = "" # shown in /providers
optional_dependency: bool = False # affects logging on ImportError
Conventions
namemust matchProvider.nameexactly (case-sensitive).module_pathfollows theproviders.<category>.<name>pattern. Use underscores in the file name; the manifest entry name itself should also use underscores.enable_flagis the Settings attribute name (e.g."enable_my_provider"), NOT the env var (FI_ENABLE_MY_PROVIDER). Pydantic handles the prefix translation.- Set
optional_dependency=Trueif your module imports an optional package (e.g.insightface,dlib,tensorflow). The registry then logs the missing dependency atwarninglevel and marks the provider asNOT_CONFIGUREDrather than crashing startup.
Existing manifest (24 entries)
See the full list in
providers/registry.py. The 24 providers
span all 7 capabilities:
| Capability | Providers |
|---|---|
| DETECTION | haar, dnn, mtcnn, retinaface |
| RECOGNITION | face_recognition, deepface, insightface |
| SCRAPING | beautifulsoup, selenium, bing, duckduckgo |
| REVERSE_SEARCH | google_lens, serpapi, yandex, tineye |
| IMAGE_ANALYSIS | image_quality, image_properties, visual_features |
| METADATA | exif, xmp |
| FORENSICS | image_integrity, duplicate_detector, manipulation_analyzer |
8. Settings Flag Convention
Naming
- Enable flag:
enable_<provider_name>: bool = <default> - Tuning knob:
<provider_name>_<knob>: <type> = <default>
Examples from config/settings.py:
enable_haar: bool = True
haar_scale_factor: float = 1.1
haar_min_neighbors: int = 5
enable_dnn: bool = True
dnn_confidence_threshold: float = 0.7
enable_serpapi: bool = False
serpapi_key: str = "" # API keys live alongside enable flag
Defaults
- Default
Truefor providers that ship with the platform and have no external dependencies (haar,dnn,image_quality,image_properties,exif,image_integrity,duplicate_detector,beautifulsoup,duckduckgo). - Default
Falsefor providers that require optional Python packages, paid API keys, or a Chrome/Chromium binary (retinaface,deepface,insightface,bing,serpapi,yandex,tineye,visual_features,xmp,manipulation_analyzer). mtcnn,face_recognition,selenium,google_lensdefault toTruebecause their deps are inrequirements.txt, but they may fail tois_available()at runtime (e.g. Chrome missing).
Env-var mapping
Pydantic-Settings auto-translates enable_my_provider to the env var
FI_ENABLE_MY_PROVIDER (case-insensitive). See
docs/CONFIGURATION.md for the full table.
9. Health Check Pattern
is_available() is called by:
- The registry's
info()method (used by/providersand/health/providers). BaseProvider._safe_execute()at the start of every invocation β returns aNotConfiguredresult if it returnsFalse.- The
FeatureExtractor(only the chosen default detector is used for pre-extraction of face crops).
Three valid patterns
Pattern A β always available (pure-OpenCV providers)
def is_available(self) -> bool:
return True
Used by: haar, image_quality, image_properties,
image_integrity, duplicate_detector.
Pattern B β model loaded successfully
def __init__(self, settings):
super().__init__(settings=settings or _default_settings)
self._net = None
self._init_error: str | None = None
try:
self._net = cv2.dnn.readNetFromCaffe(...)
except Exception as e:
self._init_error = str(e)
def is_available(self) -> bool:
return self._net is not None and self._init_error is None
Used by: dnn (also auto-downloads model files in __init__).
Pattern C β optional dependency / API key
def __init__(self, settings):
super().__init__(settings=settings or _default_settings)
self._api_key = self._settings.serpapi_key
def is_available(self) -> bool:
return bool(self._api_key)
Used by: serpapi, bing, tineye, exif (imports PIL
lazily).
What is_available() must NOT do
- β Make network calls.
- β Run the model on an image.
- β Raise exceptions β wrap in try/except and return
False. - β Block for more than a few milliseconds.
The orchestrator calls this on every request; expensive checks defeat the circuit breaker.
10. Error Handling Pattern (_safe_execute)
BaseProvider.execute() calls _safe_execute(), which:
- Records
time.perf_counter()ast0. - Calls
is_available(). IfFalse, returns aProviderResultwithsuccess=False,error_type="NotConfigured",error="Provider not available (disabled or missing dependencies)". - Calls
self._run(input_data)inside atryblock. - On success, returns a
ProviderResultwithsuccess=True, the raw + normalized dicts, and the measuredelapsed_ms. - On any
Exception, returns aProviderResultwithsuccess=False,error=str(e),error_type=type(e).__name__.
def _safe_execute(self, input_data: Any) -> ProviderResult:
t0 = time.perf_counter()
try:
if not self.is_available():
return ProviderResult(
provider=self.name,
capability=self.capability,
success=False,
elapsed_ms=0.0,
error="Provider not available (disabled or missing dependencies)",
error_type="NotConfigured",
)
raw, normalized = self._run(input_data)
elapsed = (time.perf_counter() - t0) * 1000.0
return ProviderResult(
provider=self.name,
capability=self.capability,
success=True,
elapsed_ms=round(elapsed, 3),
raw=raw,
normalized=normalized,
)
except Exception as e:
elapsed = (time.perf_counter() - t0) * 1000.0
return ProviderResult(
provider=self.name,
capability=self.capability,
success=False,
elapsed_ms=round(elapsed, 3),
error=str(e),
error_type=type(e).__name__,
)
Implications for _run
- You may raise any exception. It will be caught and converted to
a
ProviderResultwithsuccess=False. - You should NOT catch broad
Exceptionyourself unless you want to convert specific errors into richernormalizedoutput. Let_safe_executedo the generic catching. - You should NOT mutate global state that survives across
invocations. The orchestrator may invoke you concurrently from
multiple threads (it uses
asyncio.to_thread). - You should NOT call other providers. Cross-provider composition happens in the orchestrator and merger layers.
Retryable errors
The orchestrator's RetryPolicy (see
orchestrator/retry.py) only retries
TimeoutError, ConnectionError, and OSError. If your provider
raises RuntimeError for a transient network issue, it will not be
retried β re-raise as ConnectionError instead:
import requests
def _run(self, pipeline_output):
try:
resp = self._session.post(...)
resp.raise_for_status()
except requests.ConnectionError as e:
raise # retriable
except requests.HTTPError as e:
if e.response.status_code == 429:
raise ConnectionError("rate limited") from e # retriable
raise # non-retriable (4xx other than 429)
11. PipelineOutput β What You Receive
Defined in pipeline/feature_extraction.py:
@dataclass
class PipelineOutput:
image: np.ndarray # BGR, preprocessed (max dim 1024)
image_hash: str # SHA-256 cache key
width: int
height: int
source: str # "url" | "base64" | "bytes"
original_bytes: Optional[bytes] = None # for EXIF / forensics
original_format: Optional[str] = None # ".jpg", ".png", etc.
face_crops: List[FaceCrop] = field(default_factory=list)
primary_detector: str = "" # name of detector used for crops
gallery: Optional[dict] = None # set by RecognitionService
scrape_url: Optional[str] = None # set by SearchService
What each capability should read
| Capability | Read from PipelineOutput |
|---|---|
| DETECTION | image |
| RECOGNITION | image, face_crops, gallery (dict: person_name -> List[np.ndarray]) |
| SCRAPING | scrape_url (or image for the source page) |
| REVERSE_SEARCH | image (uploaded as JPEG to the reverse-search service) |
| IMAGE_ANALYSIS | image |
| METADATA | original_bytes, original_format |
| FORENSICS | original_bytes, image |
Normalized output schemas
The merger expects specific keys in normalized per capability. If
you omit keys, the merger fills defaults. If you use different keys,
your data won't reach the final UnifiedFaceReport.
| Capability | Required normalized keys |
|---|---|
| DETECTION | boxes: list[{x,y,w,h}], num_faces: int, confidences: list[float], landmarks: list|dict|None |
| RECOGNITION | matches: list[{query_face_index, best_match, distance, distances}] |
| SCRAPING | images: list[{url, alt, source_page, width, height}] |
| REVERSE_SEARCH | results: list[{image_url, source_page, title, snippet, thumbnail}] |
| IMAGE_ANALYSIS | quality_score, brightness, contrast, sharpness, noise_level, width, height, channels, color_profile, dominant_colors, aspects |
| METADATA | format, exif, xmp, iptc, gps, camera_make, camera_model, software, capture_time |
| FORENSICS | integrity_score, is_duplicate, duplicate_of, similarity_score, manipulation_indicators, ela_score, noise_inconsistency, details |
Use BBox(x, y, w, h).to_dict() from utils/image.py
to produce correctly-shaped boxes entries for detection providers.
12. Testing Pattern for Providers
Provider tests live in tests/providers/test_<provider_name>.py and
follow this shape (see tests/providers/test_haar.py for the canonical
example):
"""Provider tests for MyProvider."""
from __future__ import annotations
import cv2
import numpy as np
import pytest
from config.settings import Settings
from pipeline.feature_extraction import PipelineOutput
from providers.detection.my_provider import MyProvider
from providers.base import ProviderResult
@pytest.fixture
def my_provider():
return MyProvider(settings=Settings(environment="test", db_path=":memory:"))
@pytest.fixture
def pipeline_output(sample_image_bytes):
img = cv2.imdecode(np.frombuffer(sample_image_bytes, np.uint8), cv2.IMREAD_COLOR)
return PipelineOutput(
image=img,
image_hash="test-hash",
width=img.shape[1],
height=img.shape[0],
source="bytes",
original_bytes=sample_image_bytes,
original_format=".jpg",
)
class TestMyProvider:
def test_name(self, my_provider):
assert my_provider.name == "my_provider"
def test_capability(self, my_provider):
from models.providers import ProviderCapability
assert my_provider.capability == ProviderCapability.DETECTION
def test_is_available(self, my_provider):
assert my_provider.is_available() is True
def test_execute_returns_provider_result(self, my_provider, pipeline_output):
result = my_provider.execute(pipeline_output)
assert isinstance(result, ProviderResult)
assert result.provider == "my_provider"
assert result.success is True
assert result.elapsed_ms > 0
def test_normalized_has_expected_keys(self, my_provider, pipeline_output):
result = my_provider.execute(pipeline_output)
assert "boxes" in result.normalized
assert "num_faces" in result.normalized
def test_black_image_finds_no_faces(self, my_provider):
img = np.zeros((200, 200, 3), dtype=np.uint8)
po = PipelineOutput(image=img, image_hash="x",
width=200, height=200, source="bytes")
result = my_provider.execute(po)
assert result.success is True
assert result.normalized["num_faces"] == 0
def test_execute_does_not_raise_on_bad_input(self, my_provider):
"""1x1 image β must not crash."""
img = np.zeros((1, 1, 3), dtype=np.uint8)
po = PipelineOutput(image=img, image_hash="x",
width=1, height=1, source="bytes")
result = my_provider.execute(po)
assert result.success is True
Checklist for a new provider test file
- Fixture constructs the provider with
Settings(environment="test", db_path=":memory:"). - Fixture builds a
PipelineOutput(use thesample_image_bytesconftest fixture). - Tests
name,capability,is_available(). - Tests that
execute()returns aProviderResultwith the rightproviderandsuccess=True. - Tests the
normalizeddict has the required keys for the capability. - Tests a "no faces / empty input" case returns
success=Truewith empty results (not a crash). - Tests a tiny/edge-case image (1Γ1, fully black, fully white) does not raise.
Running provider tests
python -m pytest tests/providers/ -v
python -m pytest tests/providers/test_my_provider.py -v
See docs/TESTING.md for the full test guide including
fixtures, coverage goals, and CI integration.
13. Reference Providers
Each existing provider demonstrates a different pattern. Read the source for the one closest to what you're building:
| Provider | File | Pattern demonstrated |
|---|---|---|
haar |
providers/detection/haar.py |
Minimal β always available, reads pipeline_output.image, returns boxes/confidences. |
dnn |
providers/detection/dnn.py |
Model auto-download, CUDA fallback, is_available() checks loaded model, threshold from settings. |
image_quality |
providers/image_analysis/image_quality.py |
Pure-OpenCV analysis, composite heuristic score, static methods for sub-computations. |
image_properties |
providers/image_analysis/image_properties.py |
K-means dominant colors with downsampling for speed. |
exif |
providers/metadata/exif.py |
Reads original_bytes, lazy PIL import, GPS coordinate conversion, graceful empty fallback. |
image_integrity |
providers/forensics/image_integrity.py |
Multi-check forensics (Pillow decode, JPEG APP marker scan, SHA-256, size sanity), issue list β score. |
duplicate_detector |
providers/forensics/duplicate_detector.py |
pHash + dHash, in-memory hash registry, Hamming-distance matching. |
serpapi |
providers/reverse/serpapi.py |
Paid API provider: API-key check in is_available(), two-step upload+search flow, uses utils.http.shared_session(). |
14. Common Pitfalls
β Importing from siblings
# providers/detection/my_provider.py
from providers.detection.haar import HaarDetector # WRONG
Providers must not depend on each other. If you need shared logic,
put it in utils/ or pipeline/.
β Reading Settings() directly instead of using injected settings
class MyProvider(BaseProvider):
def _run(self, ...):
threshold = Settings().my_threshold # WRONG β bypasses DI
threshold = self._settings.my_threshold # RIGHT
β Raising from is_available()
def is_available(self) -> bool:
return self._net.is_ready() # WRONG β could raise
Wrap it:
def is_available(self) -> bool:
try:
return self._net.is_ready()
except Exception:
return False
β Returning ad-hoc normalized schemas
normalized = {"faces_found": 1} # WRONG β merger expects "num_faces"
Match the schema in Β§11. Use BBox(...).to_dict() for boxes.
β Blocking the event loop
Provider execute() runs in asyncio.to_thread, so synchronous I/O
is fine β but if your provider makes a long HTTP call, set a sensible
timeout (requests defaults to none). The orchestrator enforces a
hard cap of orchestrator_timeout_seconds (default 90s) per
provider.
β Forgetting to add the enable flag
If you add a manifest entry with enable_flag="enable_my_provider"
but forget to add the field to Settings, the registry logs:
Provider my_provider disabled by config (enable_my_provider=False)
β¦and silently skips it. Pydantic won't raise because extra="ignore"
is set on the Settings model. Always add the field.
β Heavy work in __init__
The registry instantiates every enabled provider at startup. If your
constructor downloads a 100 MB model, the app takes 100 MB longer to
boot. Prefer lazy initialization: download in the first _run() call,
or use a flag like dnn does (_ensure_models_downloaded()).
See Also
docs/ARCHITECTURE.mdβ layer dependency graph, execution flow, lifecycle diagrams.docs/CONFIGURATION.mdβ everyFI_*env var including all provider enable flags.docs/API_REFERENCE.mdβ the/providersendpoint that exposes your provider once registered.docs/TESTING.mdβ full test layout and fixtures.docs/TROUBLESHOOTING.mdβ provider not_configured, circuit breaker, model download failures.