|
|
""" |
|
|
Embedding Provider Factory - Dynamic Provider Creation |
|
|
""" |
|
|
|
|
|
from typing import Dict, Any, Optional |
|
|
from warbler_cda.embeddings.base_provider import EmbeddingProvider |
|
|
from warbler_cda.embeddings.local_provider import LocalEmbeddingProvider |
|
|
from warbler_cda.embeddings.openai_provider import OpenAIEmbeddingProvider |
|
|
from warbler_cda.embeddings.sentence_transformer_provider import SentenceTransformerEmbeddingProvider |
|
|
|
|
|
|
|
|
class EmbeddingProviderFactory: |
|
|
"""Factory for creating embedding providers.""" |
|
|
|
|
|
PROVIDERS = { |
|
|
"local": LocalEmbeddingProvider, |
|
|
"openai": OpenAIEmbeddingProvider, |
|
|
"sentence_transformer": SentenceTransformerEmbeddingProvider, |
|
|
} |
|
|
|
|
|
@classmethod |
|
|
def create_provider(cls, provider_type: str, config: Optional[Dict[str, Any]] = None) -> EmbeddingProvider: |
|
|
"""Create an embedding provider of the specified type.""" |
|
|
if provider_type not in cls.PROVIDERS: |
|
|
available = list(cls.PROVIDERS.keys()) |
|
|
raise ValueError(f"Unknown provider type '{provider_type}'. Available: {available}") |
|
|
|
|
|
provider_class = cls.PROVIDERS[provider_type] |
|
|
return provider_class(config) |
|
|
|
|
|
@classmethod |
|
|
def get_default_provider(cls, config: Optional[Dict[str, Any]] = None) -> EmbeddingProvider: |
|
|
"""Get the default embedding provider (SentenceTransformer with fallback to local).""" |
|
|
try: |
|
|
return cls.create_provider("sentence_transformer", config) |
|
|
except ImportError: |
|
|
print("Warning: SentenceTransformer not available, falling back to LocalEmbeddingProvider") |
|
|
return cls.create_provider("local", config) |
|
|
|
|
|
@classmethod |
|
|
def list_available_providers(cls) -> list[str]: |
|
|
"""List all available provider types.""" |
|
|
return list(cls.PROVIDERS.keys()) |
|
|
|
|
|
@classmethod |
|
|
def create_from_config(cls, full_config: Dict[str, Any]) -> EmbeddingProvider: |
|
|
"""Create provider from configuration dict.""" |
|
|
provider_type = full_config.get("provider", "local") |
|
|
provider_config = full_config.get("config", {}) |
|
|
|
|
|
return cls.create_provider(provider_type, provider_config) |
