Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-llm/src/lexigram/ai/llm/registry/core.py: 57%
131 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""Provider registry for LLM clients.
3This module provides a registry for discovering and managing LLM providers,
4making it easy to add custom providers and query available options.
6Example:
7 >>> from lexigram.ai.llm.registry.core import ProviderRegistry
8 >>> from lexigram.di.container import Container
9 >>>
10 >>> # Resolve via the DI container (recommended)
11 >>> container = Container()
12 >>> registry = await container.resolve(ProviderRegistry)
13 >>>
14 >>> # Get built-in provider
15 >>> info = registry.get_provider("openai")
16 >>> print(f"Models: {info.default_models}")
17 >>>
18 >>> # Register custom provider
19 >>> from my_package import CustomLLMClient
20 >>> registry.register(
21 ... name="custom",
22 ... client_class=CustomLLMClient,
23 ... default_models=["custom-gpt-1"],
24 ... supports_streaming=True
25 ... )
27"""
29from __future__ import annotations
31from dataclasses import dataclass, field
32from typing import TYPE_CHECKING, cast
34if TYPE_CHECKING:
35 from lexigram.contracts.ai import LLMClientProtocol
36 from lexigram.contracts.ai.providers import ModelInfo
37 from lexigram.contracts.ai.types import ModelCapability
39from lexigram.logging import (
40 get_logger,
41)
42from lexigram.primitives.registry import Registry
44logger = get_logger(__name__)
47__all__ = [
48 "ProviderInfo",
49 "ProviderRegistry",
50]
53@dataclass
54class ProviderInfo:
55 """Information about an LLM provider.
57 Attributes:
58 name: Provider identifier (e.g., "openai", "anthropic").
59 client_class: LLMClientProtocol implementation class.
60 default_models: List of default/recommended models.
61 supports_streaming: Whether streaming is supported.
62 supports_tools: Whether function/tool calling is supported.
63 supports_vision: Whether vision/image inputs are supported.
64 base_url: Default base URL for API (optional).
65 docs_url: Documentation URL (optional).
66 pricing_url: Pricing page URL (optional).
67 description: Human-readable description.
69 """
71 name: str
72 # Concrete provider implementations are stored as classes for metadata and
73 # instantiation. Structural protocol conformance is enforced on instances.
74 client_class: type[object]
75 default_models: list[str] = field(default_factory=list)
76 supports_streaming: bool = True
77 supports_tools: bool = False
78 supports_vision: bool = False
79 base_url: str | None = None
80 docs_url: str | None = None
81 pricing_url: str | None = None
82 description: str = ""
85class ProviderRegistry(Registry[str, ProviderInfo]):
86 """Registry for LLM providers.
88 Singleton registry that maintains information about all available
89 LLM providers, both built-in and custom.
90 """
92 def __init__(self) -> None:
93 """Initialize provider registry."""
94 if getattr(self, "_initialized", False):
95 return
96 super().__init__(name="providers")
97 self._provider_clients: dict[str, LLMClientProtocol] = {}
98 self._provider_models: dict[str, list[ModelInfo]] = {}
99 self._initialize_builtin_providers()
100 self._initialized = True
102 @staticmethod
103 def _validate_provider_name(name: str) -> None:
104 """Validate a provider registry key."""
105 if not name or " " in name or name != name.lower():
106 msg = "Provider name must be lowercase with no spaces (e.g., 'my-provider')"
107 raise ValueError(msg)
109 def _initialize_builtin_providers(self) -> None:
110 """Register built-in providers using per-provider lazy imports.
112 Each client module is imported only when its block executes inside
113 this method (which is called once at registry construction time).
114 This keeps all SDK imports deferred away from the module top-level
115 while still loading all providers eagerly enough to populate the
116 registry catalogue — a good trade-off since the registry is a
117 singleton that is only constructed when first requested.
118 """
119 # OpenAI
120 from lexigram.ai.llm.clients.openai import OpenAIClient
122 self._items["openai"] = ProviderInfo(
123 name="openai",
124 client_class=OpenAIClient,
125 default_models=[
126 "gpt-4-turbo",
127 "gpt-4o",
128 "gpt-4o-mini",
129 "gpt-3.5-turbo",
130 ],
131 supports_streaming=True,
132 supports_tools=True,
133 supports_vision=True,
134 base_url="https://api.openai.com/v1",
135 docs_url="https://platform.openai.com/docs",
136 pricing_url="https://openai.com/pricing",
137 description="OpenAI GPT models - industry standard for general purpose AI",
138 )
140 # Anthropic (Claude)
141 from lexigram.ai.llm.clients.anthropic import AnthropicClient
143 self._items["anthropic"] = ProviderInfo(
144 name="anthropic",
145 client_class=AnthropicClient,
146 default_models=[
147 "claude-3-5-sonnet-20241022",
148 "claude-3-opus-20240229",
149 "claude-3-sonnet-20240229",
150 "claude-3-haiku-20240307",
151 ],
152 supports_streaming=True,
153 supports_tools=True,
154 supports_vision=True,
155 base_url="https://api.anthropic.com",
156 docs_url="https://docs.anthropic.com",
157 pricing_url="https://www.anthropic.com/pricing",
158 description="Anthropic Claude - strong reasoning and long context (200k tokens)",
159 )
161 # Groq
162 from lexigram.ai.llm.clients.groq import GroqClient
164 self._items["groq"] = ProviderInfo(
165 name="groq",
166 client_class=GroqClient,
167 default_models=[
168 "llama-3.1-70b-versatile",
169 "llama-3.1-8b-instant",
170 "mixtral-8x7b-32768",
171 ],
172 supports_streaming=True,
173 supports_tools=True,
174 supports_vision=False,
175 base_url="https://api.groq.com/openai/v1",
176 docs_url="https://console.groq.com/docs",
177 pricing_url="https://groq.com/pricing",
178 description="Groq - ultra-fast inference with LPU hardware (100+ tokens/sec, currently free)",
179 )
181 # Mistral
182 from lexigram.ai.llm.clients.mistral import MistralClient
184 self._items["mistral"] = ProviderInfo(
185 name="mistral",
186 client_class=MistralClient,
187 default_models=[
188 "mistral-large-latest",
189 "mistral-medium-latest",
190 "mistral-small-latest",
191 "open-mixtral-8x7b",
192 ],
193 supports_streaming=True,
194 supports_tools=True,
195 supports_vision=False,
196 base_url="https://api.mistral.ai/v1",
197 docs_url="https://docs.mistral.ai",
198 pricing_url="https://mistral.ai/technology/#pricing",
199 description="Mistral AI - GDPR-compliant EU provider with strong multilingual support",
200 )
202 # Cohere
203 from lexigram.ai.llm.clients.cohere import CohereClient
205 self._items["cohere"] = ProviderInfo(
206 name="cohere",
207 client_class=CohereClient,
208 default_models=[
209 "command-r-plus",
210 "command-r",
211 "command",
212 "embed-english-v3.0",
213 ],
214 supports_streaming=True,
215 supports_tools=True,
216 supports_vision=False,
217 base_url="https://api.cohere.ai/v1",
218 docs_url="https://docs.cohere.com",
219 pricing_url="https://cohere.com/pricing",
220 description="Cohere - best-in-class embeddings, reranking, and RAG-optimized models",
221 )
223 # Ollama
224 from lexigram.ai.llm.clients.ollama import OllamaClient
226 self._items["ollama"] = ProviderInfo(
227 name="ollama",
228 client_class=OllamaClient,
229 default_models=[
230 "llama3:8b",
231 "llama3:70b",
232 "mistral:7b",
233 "codellama:13b",
234 ],
235 supports_streaming=True,
236 supports_tools=False,
237 supports_vision=False,
238 base_url="http://localhost:11434",
239 docs_url="https://ollama.ai/docs",
240 pricing_url=None,
241 description="Ollama - run LLMs locally with zero API costs and full privacy",
242 )
244 # OpenRouter
245 from lexigram.ai.llm.clients.openrouter import OpenRouterClient
247 self.register(
248 name="openrouter",
249 client_class=OpenRouterClient,
250 default_models=["gpt-4o", "gpt-4o-mini"],
251 supports_streaming=True,
252 supports_tools=True,
253 supports_vision=False,
254 base_url="https://api.openrouter.ai/v1",
255 docs_url="https://docs.openrouter.ai",
256 pricing_url=None,
257 description="OpenRouter - open and compatible routing layer for many models",
258 )
260 # Gemini
261 from lexigram.ai.llm.clients.gemini import GeminiClient
263 self._items["gemini"] = ProviderInfo(
264 name="gemini",
265 client_class=GeminiClient,
266 default_models=[
267 "gemini-2.5-flash",
268 "gemini-2.0-flash",
269 "gemini-1.5-flash",
270 "gemini-1.5-pro",
271 ],
272 supports_streaming=False,
273 supports_tools=False,
274 supports_vision=True,
275 base_url="https://generativelanguage.googleapis.com",
276 docs_url="https://ai.google.dev/docs",
277 pricing_url="https://ai.google.dev/pricing",
278 description="Google Gemini — multimodal models with vision and long context; free tier available",
279 )
281 # Cloudflare Workers AI
282 from lexigram.ai.llm.clients.cloudflare_workers import CloudflareWorkersClient
284 self._items["cloudflare"] = ProviderInfo(
285 name="cloudflare",
286 client_class=CloudflareWorkersClient,
287 default_models=[
288 "@cf/meta/llama-3.2-11b-vision-instruct",
289 "@cf/meta/llama-3.1-8b-instruct",
290 "@cf/mistral/mistral-7b-instruct-v0.2",
291 ],
292 supports_streaming=False,
293 supports_tools=False,
294 supports_vision=True,
295 base_url="https://api.cloudflare.com",
296 docs_url="https://developers.cloudflare.com/workers-ai",
297 pricing_url="https://developers.cloudflare.com/workers-ai/platform/pricing/",
298 description="Cloudflare Workers AI — serverless inference with vision models; free tier available",
299 )
301 # Azure OpenAI
302 from lexigram.ai.llm.clients.azure_openai import AzureOpenAIClient
304 self.register(
305 name="azure-openai",
306 client_class=AzureOpenAIClient,
307 default_models=["gpt-4o", "gpt-4o-mini", "gpt-4-turbo"],
308 supports_streaming=True,
309 supports_tools=True,
310 supports_vision=True,
311 base_url="https://{resource}.openai.azure.com/openai",
312 docs_url="https://learn.microsoft.com/azure/ai-services/openai/",
313 pricing_url="https://azure.microsoft.com/pricing/details/cognitive-services/openai-service/",
314 description="Azure OpenAI — enterprise OpenAI deployment with Azure identity and networking controls",
315 )
317 # AWS Bedrock
318 from lexigram.ai.llm.clients.aws_bedrock import BedrockClient
320 self.register(
321 name="aws-bedrock",
322 client_class=BedrockClient,
323 default_models=[
324 "anthropic.claude-3-5-sonnet-20240620-v1:0",
325 "meta.llama3-70b-instruct-v1:0",
326 "amazon.titan-text-express-v1",
327 ],
328 supports_streaming=True,
329 supports_tools=True,
330 supports_vision=True,
331 base_url="https://bedrock-runtime.{region}.amazonaws.com",
332 docs_url="https://docs.aws.amazon.com/bedrock/latest/userguide/what-is-bedrock.html",
333 pricing_url="https://aws.amazon.com/bedrock/pricing/",
334 description="AWS Bedrock — managed foundation model platform with native AWS IAM and enterprise controls",
335 )
337 # Google Vertex AI
338 from lexigram.ai.llm.clients.vertex_ai import VertexAIClient
340 self.register(
341 name="google-vertex",
342 client_class=VertexAIClient,
343 default_models=["gemini-2.0-flash", "gemini-1.5-pro", "gemini-1.5-flash"],
344 supports_streaming=True,
345 supports_tools=True,
346 supports_vision=True,
347 base_url="https://{region}-aiplatform.googleapis.com",
348 docs_url="https://cloud.google.com/vertex-ai/docs/generative-ai/",
349 pricing_url="https://cloud.google.com/vertex-ai/pricing",
350 description="Google Vertex AI — enterprise Gemini deployment with GCP IAM and regional controls",
351 )
353 # OpenAI-compatible providers
354 from lexigram.ai.llm.clients.openai_compatible import (
355 DeepSeekClient,
356 FireworksClient,
357 TogetherClient,
358 )
360 self.register(
361 name="deepseek",
362 client_class=DeepSeekClient,
363 default_models=["deepseek-chat", "deepseek-coder"],
364 supports_streaming=True,
365 supports_tools=True,
366 supports_vision=False,
367 base_url="https://api.deepseek.com/v1",
368 docs_url="https://api-docs.deepseek.com",
369 pricing_url="https://www.deepseek.com/",
370 description="DeepSeek — cost-efficient OpenAI-compatible inference provider",
371 )
373 self.register(
374 name="together",
375 client_class=TogetherClient,
376 default_models=[
377 "meta-llama/Llama-3-8b-chat-hf",
378 "mistralai/Mixtral-8x7B-Instruct-v0.1",
379 ],
380 supports_streaming=True,
381 supports_tools=True,
382 supports_vision=False,
383 base_url="https://api.together.xyz/v1",
384 docs_url="https://docs.together.ai",
385 pricing_url="https://www.together.ai/pricing",
386 description="Together AI — OpenAI-compatible serving for open-source and fine-tuned models",
387 )
389 self.register(
390 name="fireworks",
391 client_class=FireworksClient,
392 default_models=[
393 "accounts/fireworks/models/llama-v3-70b-instruct",
394 "accounts/fireworks/models/mixtral-8x7b-instruct",
395 ],
396 supports_streaming=True,
397 supports_tools=True,
398 supports_vision=False,
399 base_url="https://api.fireworks.ai/inference/v1",
400 docs_url="https://docs.fireworks.ai",
401 pricing_url="https://fireworks.ai/pricing",
402 description="Fireworks AI — optimized OpenAI-compatible inference platform",
403 )
405 logger.info("Initialized %d built-in providers", len(self))
407 def register( # type: ignore[override]
408 self,
409 name: str,
410 client_class: type[object],
411 default_models: list[str] | None = None,
412 supports_streaming: bool = True,
413 supports_tools: bool = False,
414 supports_vision: bool = False,
415 base_url: str | None = None,
416 docs_url: str | None = None,
417 pricing_url: str | None = None,
418 description: str = "",
419 ) -> ProviderInfo:
420 """Register a new LLM provider."""
421 self._validate_provider_name(name)
423 info = ProviderInfo(
424 name=name,
425 client_class=client_class,
426 default_models=default_models or [],
427 supports_streaming=supports_streaming,
428 supports_tools=supports_tools,
429 supports_vision=supports_vision,
430 base_url=base_url,
431 docs_url=docs_url,
432 pricing_url=pricing_url,
433 description=description,
434 )
435 return cast("ProviderInfo", super().register(name, info))
437 def get_provider(self, name: str) -> ProviderInfo:
438 """Get provider information."""
439 info = cast("ProviderInfo | None", super().get(name))
440 if info is None:
441 available = ", ".join(self.list_providers())
442 msg = f"Provider '{name}' not found. Available providers: {available}"
443 raise KeyError(msg)
445 return info
447 def list_providers(self) -> list[str]:
448 """List all registered provider names."""
449 return sorted(super().keys())
451 def search_providers(
452 self,
453 supports_streaming: bool | None = None,
454 supports_tools: bool | None = None,
455 supports_vision: bool | None = None,
456 ) -> list[ProviderInfo]:
457 """Search providers by capabilities."""
458 results = []
460 for info in self.values():
461 if (
462 supports_streaming is not None
463 and info.supports_streaming != supports_streaming
464 ):
465 continue
466 if supports_tools is not None and info.supports_tools != supports_tools:
467 continue
468 if supports_vision is not None and info.supports_vision != supports_vision:
469 continue
471 results.append(info)
473 return results
475 def unregister(self, name: str) -> None:
476 """Unregister a provider."""
477 self._provider_clients.pop(name, None)
478 self._provider_models.pop(name, None)
479 if super().unregister(name) is None:
480 msg = f"Provider '{name}' not found"
481 raise KeyError(msg)
483 # --- Protocol Implementation ---
485 async def register_provider(
486 self, name: str, client: LLMClientProtocol, models: list[ModelInfo]
487 ) -> None:
488 """Register a provider following the ProviderRegistryProtocol."""
489 from lexigram.contracts.ai.types import ModelCapability
491 self._validate_provider_name(name)
493 existing = super().get(name)
494 supports_streaming = (
495 any(ModelCapability.STREAMING in model.capabilities for model in models)
496 if models
497 else (existing.supports_streaming if existing is not None else True)
498 )
499 supports_tools = (
500 any(
501 ModelCapability.FUNCTION_CALLING in model.capabilities
502 for model in models
503 )
504 if models
505 else (existing.supports_tools if existing is not None else False)
506 )
507 supports_vision = (
508 any(ModelCapability.VISION in model.capabilities for model in models)
509 if models
510 else (existing.supports_vision if existing is not None else False)
511 )
513 info = ProviderInfo(
514 name=name,
515 client_class=type(client),
516 default_models=(
517 [model.model_id for model in models]
518 if models
519 else (list(existing.default_models) if existing is not None else [])
520 ),
521 supports_streaming=supports_streaming,
522 supports_tools=supports_tools,
523 supports_vision=supports_vision,
524 base_url=existing.base_url if existing is not None else None,
525 docs_url=existing.docs_url if existing is not None else None,
526 pricing_url=existing.pricing_url if existing is not None else None,
527 description=existing.description if existing is not None else "",
528 )
529 super().register(name, info, allow_overwrite=True)
530 self._provider_clients[name] = client
531 if models:
532 self._provider_models[name] = list(models)
534 async def get_client(self, provider: str) -> LLMClientProtocol | None:
535 """Get an initialized client for a provider."""
536 return self._provider_clients.get(provider)
538 def list_models(
539 self, capabilities: set[ModelCapability] | None = None
540 ) -> list[ModelInfo]:
541 """List all models matching capabilities."""
542 from lexigram.contracts.ai.providers import ModelInfo
543 from lexigram.contracts.ai.types import ModelCapability
545 models: list[ModelInfo] = []
546 for provider_info in self.values():
547 registered_models = self._provider_models.get(provider_info.name)
548 if registered_models is not None:
549 provider_models = registered_models
550 else:
551 provider_capabilities = {ModelCapability.CHAT}
552 if provider_info.supports_streaming:
553 provider_capabilities.add(ModelCapability.STREAMING)
554 if provider_info.supports_tools:
555 provider_capabilities.add(ModelCapability.FUNCTION_CALLING)
556 if provider_info.supports_vision:
557 provider_capabilities.add(ModelCapability.VISION)
558 provider_models = [
559 ModelInfo(
560 model_id=model_id,
561 provider=provider_info.name,
562 display_name=model_id,
563 capabilities=frozenset(provider_capabilities),
564 context_window=8192,
565 max_output_tokens=4096,
566 input_cost_per_million=0,
567 output_cost_per_million=0,
568 )
569 for model_id in provider_info.default_models
570 ]
572 for model in provider_models:
573 if capabilities is not None and not capabilities.issubset(
574 model.capabilities
575 ):
576 continue
577 models.append(model)
578 return models
580 def get_model_info(self, model_id: str) -> ModelInfo | None:
581 """Get information about a specific model."""
582 models = self.list_models()
583 return next((m for m in models if m.model_id == model_id), None)