122 lines
4.7 KiB
Python
122 lines
4.7 KiB
Python
"""Provider registry for the general AIOS Search Gateway.
|
|
|
|
The registry makes provider availability explicit and auditable. Provider
|
|
implementations are pluggable: production can register real public search,
|
|
official-site, vertical-source, browser-fetch, authorized-file or archive
|
|
providers without changing the Search Gateway core.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import asdict, dataclass, field
|
|
from typing import Any
|
|
|
|
from .search_gateway import SearchEvidence, SearchProviderAttempt, SearchRequest
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SearchProviderConfig:
|
|
name: str
|
|
provider_type: str
|
|
enabled: bool = True
|
|
requires_secret: bool = False
|
|
supports_public_web: bool = True
|
|
supports_authorized_sources: bool = False
|
|
health: str = "unknown"
|
|
metadata: dict[str, Any] = field(default_factory=dict)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
|
|
class NullSearchProvider:
|
|
"""Safe placeholder provider used until runtime wires a real adapter."""
|
|
|
|
def __init__(self, config: SearchProviderConfig):
|
|
self.config = config
|
|
self.name = config.name
|
|
|
|
def search(self, request: SearchRequest, queries: list[str]):
|
|
return [], SearchProviderAttempt(
|
|
provider=self.name,
|
|
status="not_configured" if self.config.enabled else "disabled",
|
|
result_count=0,
|
|
reason="provider adapter not wired in runtime",
|
|
)
|
|
|
|
|
|
class StaticArchiveProvider:
|
|
"""Small deterministic provider for tests and internal archived evidence."""
|
|
|
|
name = "internal_archive"
|
|
|
|
def __init__(self, records: list[dict[str, Any]] | None = None):
|
|
self.records = records or []
|
|
|
|
def search(self, request: SearchRequest, queries: list[str]):
|
|
evidence: list[SearchEvidence] = []
|
|
for record in self.records:
|
|
haystack = " ".join([record.get("title", ""), record.get("snippet", ""), record.get("tags", "")]).lower()
|
|
if any(q.lower() in haystack for q in queries):
|
|
evidence.append(
|
|
SearchEvidence(
|
|
title=record.get("title", "archive result"),
|
|
url=record.get("url"),
|
|
source=self.name,
|
|
snippet=record.get("snippet", ""),
|
|
evidence_type="internal_archive",
|
|
confidence=record.get("confidence", "medium"),
|
|
)
|
|
)
|
|
return evidence, SearchProviderAttempt(provider=self.name, status="ok", result_count=len(evidence))
|
|
|
|
|
|
class SearchProviderRegistry:
|
|
DEFAULT_PROVIDER_CONFIGS = [
|
|
SearchProviderConfig("cache", "cache", supports_public_web=False),
|
|
SearchProviderConfig("official_site", "official_site"),
|
|
SearchProviderConfig("general_web", "general_web"),
|
|
SearchProviderConfig("vertical_source", "vertical_source"),
|
|
SearchProviderConfig("browser_fetch", "browser_fetch"),
|
|
SearchProviderConfig("user_authorized_file", "authorized_file", supports_public_web=False, supports_authorized_sources=True),
|
|
SearchProviderConfig("internal_archive", "archive", supports_public_web=False),
|
|
]
|
|
|
|
def __init__(self):
|
|
self.configs: dict[str, SearchProviderConfig] = {}
|
|
self.providers: dict[str, Any] = {}
|
|
for config in self.DEFAULT_PROVIDER_CONFIGS:
|
|
self.register_config(config)
|
|
|
|
def register_config(self, config: SearchProviderConfig) -> None:
|
|
self.configs[config.name] = config
|
|
self.providers.setdefault(config.name, NullSearchProvider(config))
|
|
|
|
def register_provider(self, name: str, provider: Any, config: SearchProviderConfig | None = None) -> None:
|
|
if config:
|
|
self.configs[name] = config
|
|
elif name not in self.configs:
|
|
self.configs[name] = SearchProviderConfig(name=name, provider_type="custom")
|
|
self.providers[name] = provider
|
|
|
|
def ordered(self, order: list[str] | None = None) -> list[Any]:
|
|
order = order or list(self.configs)
|
|
result = [self.providers[name] for name in order if name in self.providers and self.configs[name].enabled]
|
|
result.extend([provider for name, provider in self.providers.items() if name not in order and self.configs[name].enabled])
|
|
return result
|
|
|
|
def status(self) -> dict[str, Any]:
|
|
return {
|
|
"providers": [self.configs[name].to_dict() for name in sorted(self.configs)],
|
|
"enabled": len([c for c in self.configs.values() if c.enabled]),
|
|
"configured": len([name for name, provider in self.providers.items() if not isinstance(provider, NullSearchProvider)]),
|
|
}
|
|
|
|
|
|
__all__ = [
|
|
"SearchProviderConfig",
|
|
"NullSearchProvider",
|
|
"StaticArchiveProvider",
|
|
"SearchProviderRegistry",
|
|
]
|