mirror of
https://github.com/langgenius/dify.git
synced 2026-09-07 18:36:02 +08:00
Keep marketplace card-wrapper dialog and partner-only carousel autoplay. Take main's skills permissions, skill card actions, and agent orchestrate layout. Union oxlint suppressions from both sides.
1851 lines
78 KiB
Python
1851 lines
78 KiB
Python
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import json
|
|
import logging
|
|
from collections import defaultdict
|
|
from collections.abc import Callable, Sequence
|
|
from dataclasses import asdict, dataclass
|
|
from enum import StrEnum
|
|
from json import JSONDecodeError
|
|
from typing import TYPE_CHECKING, Any, Protocol, Self
|
|
|
|
from pydantic import TypeAdapter, ValidationError
|
|
from sqlalchemy import select
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
from configs import dify_config
|
|
from core.db.session_factory import session_factory
|
|
from core.entities.model_entities import DefaultModelEntity, DefaultModelProviderEntity
|
|
from core.entities.provider_configuration import ProviderConfiguration, ProviderConfigurations, ProviderModelBundle
|
|
from core.entities.provider_entities import (
|
|
CredentialConfiguration,
|
|
CustomConfiguration,
|
|
CustomModelConfiguration,
|
|
CustomProviderConfiguration,
|
|
ModelLoadBalancingConfiguration,
|
|
ModelSettings,
|
|
ProviderQuotaType,
|
|
QuotaConfiguration,
|
|
QuotaUnit,
|
|
SystemConfiguration,
|
|
UnaddedModelConfiguration,
|
|
)
|
|
from core.helper import encrypter
|
|
from core.helper.model_provider_cache import ProviderCredentialsCache, ProviderCredentialsCacheType
|
|
from core.helper.position_helper import is_filtered
|
|
from core.plugin.entities.plugin import PluginInstallationSource
|
|
from core.plugin.entities.plugin_daemon import PluginModelProviderDeclaration
|
|
from enums import DeploymentEdition
|
|
from extensions import ext_hosting_provider
|
|
from extensions.ext_database import db
|
|
from extensions.ext_redis import redis_client
|
|
from graphon.model_runtime.entities.model_entities import ModelType
|
|
from graphon.model_runtime.entities.provider_entities import (
|
|
ConfigurateMethod,
|
|
CredentialFormSchema,
|
|
FormType,
|
|
ProviderEntity,
|
|
)
|
|
from graphon.model_runtime.model_providers.model_provider_factory import ModelProviderFactory
|
|
from models.enums import CredentialSourceType
|
|
from models.provider import (
|
|
LoadBalancingModelConfig,
|
|
Provider,
|
|
ProviderCredential,
|
|
ProviderModel,
|
|
ProviderModelCredential,
|
|
ProviderModelSetting,
|
|
ProviderType,
|
|
TenantDefaultModel,
|
|
TenantPreferredModelProvider,
|
|
)
|
|
from models.provider_ids import ModelProviderID
|
|
from services.feature_service import FeatureService
|
|
|
|
if TYPE_CHECKING:
|
|
from graphon.model_runtime.protocols.runtime import ModelRuntime
|
|
from models.account import Account
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_credentials_adapter: TypeAdapter[dict[str, Any]] = TypeAdapter(dict[str, Any])
|
|
_PROVIDER_CONFIGURATION_CACHE_TTL_SECONDS = 300
|
|
_PROVIDER_CONFIGURATION_CACHE_VERSION_TTL_SECONDS = 360
|
|
_PROVIDER_CONFIGURATION_CACHE_VERSION_KEY = "provider_configurations:tenant:{tenant_id}:source:{source}:version"
|
|
_PROVIDER_CONFIGURATION_CACHE_SOURCE_KEY = "provider_configurations:v2:tenant:{tenant_id}:source:{source}:v:{version}"
|
|
|
|
|
|
class ProviderConfigurationCacheSource(StrEnum):
|
|
PROVIDER_MODELS = "provider_models"
|
|
PREFERRED_MODEL_PROVIDERS = "preferred_model_providers"
|
|
PROVIDER_MODEL_SETTINGS = "provider_model_settings"
|
|
PROVIDER_MODEL_CREDENTIALS = "provider_model_credentials"
|
|
PROVIDER_CREDENTIALS = "provider_credentials"
|
|
PROVIDER_LOAD_BALANCING_CONFIGS = "provider_load_balancing_configs"
|
|
|
|
|
|
_PROVIDER_CONFIGURATION_SOURCES = tuple(ProviderConfigurationCacheSource)
|
|
|
|
|
|
class _CacheEntry(Protocol):
|
|
@classmethod
|
|
def from_cache_row(cls, row: dict[str, Any]) -> Self: ...
|
|
|
|
def to_cache_row(self) -> dict[str, Any]: ...
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class _ProviderConfigurationCacheSourceSpec[T: _CacheEntry]:
|
|
name: ProviderConfigurationCacheSource
|
|
entry_cls: type[T]
|
|
load_records: Callable[[str], list[T]]
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class _ProviderModelCacheEntry:
|
|
id: str
|
|
provider_name: str
|
|
model_name: str
|
|
model_type: ModelType
|
|
credential_id: str | None
|
|
credential_name: str | None
|
|
encrypted_config: str | None
|
|
|
|
@classmethod
|
|
def from_record(cls, record: ProviderModel) -> _ProviderModelCacheEntry:
|
|
credential = record.__dict__.get("credential")
|
|
return cls(
|
|
id=record.id,
|
|
provider_name=record.provider_name,
|
|
model_name=record.model_name,
|
|
model_type=record.model_type,
|
|
credential_id=record.credential_id,
|
|
credential_name=credential.credential_name if credential else None,
|
|
encrypted_config=credential.encrypted_config if credential else None,
|
|
)
|
|
|
|
@classmethod
|
|
def from_cache_row(cls, row: dict[str, Any]) -> _ProviderModelCacheEntry:
|
|
return cls(
|
|
id=row["id"],
|
|
provider_name=row["provider_name"],
|
|
model_name=row["model_name"],
|
|
model_type=ModelType(row["model_type"]),
|
|
credential_id=row.get("credential_id"),
|
|
credential_name=row.get("credential_name"),
|
|
encrypted_config=row.get("encrypted_config"),
|
|
)
|
|
|
|
def to_cache_row(self) -> dict[str, Any]:
|
|
row = asdict(self)
|
|
row["model_type"] = self.model_type.value
|
|
return row
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class _TenantPreferredModelProviderCacheEntry:
|
|
provider_name: str
|
|
preferred_provider_type: ProviderType
|
|
|
|
@classmethod
|
|
def from_record(cls, record: TenantPreferredModelProvider) -> _TenantPreferredModelProviderCacheEntry:
|
|
return cls(
|
|
provider_name=record.provider_name,
|
|
preferred_provider_type=record.preferred_provider_type,
|
|
)
|
|
|
|
@classmethod
|
|
def from_cache_row(cls, row: dict[str, Any]) -> _TenantPreferredModelProviderCacheEntry:
|
|
return cls(
|
|
provider_name=row["provider_name"],
|
|
preferred_provider_type=ProviderType(row["preferred_provider_type"]),
|
|
)
|
|
|
|
def to_cache_row(self) -> dict[str, Any]:
|
|
return {
|
|
"provider_name": self.provider_name,
|
|
"preferred_provider_type": self.preferred_provider_type.value,
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class _ProviderModelSettingCacheEntry:
|
|
provider_name: str
|
|
model_name: str
|
|
model_type: ModelType
|
|
enabled: bool
|
|
load_balancing_enabled: bool
|
|
|
|
@classmethod
|
|
def from_record(cls, record: ProviderModelSetting) -> _ProviderModelSettingCacheEntry:
|
|
return cls(
|
|
provider_name=record.provider_name,
|
|
model_name=record.model_name,
|
|
model_type=record.model_type,
|
|
enabled=record.enabled,
|
|
load_balancing_enabled=record.load_balancing_enabled,
|
|
)
|
|
|
|
@classmethod
|
|
def from_cache_row(cls, row: dict[str, Any]) -> _ProviderModelSettingCacheEntry:
|
|
return cls(
|
|
provider_name=row["provider_name"],
|
|
model_name=row["model_name"],
|
|
model_type=ModelType(row["model_type"]),
|
|
enabled=row["enabled"],
|
|
load_balancing_enabled=row["load_balancing_enabled"],
|
|
)
|
|
|
|
def to_cache_row(self) -> dict[str, Any]:
|
|
return {
|
|
"provider_name": self.provider_name,
|
|
"model_name": self.model_name,
|
|
"model_type": self.model_type.value,
|
|
"enabled": self.enabled,
|
|
"load_balancing_enabled": self.load_balancing_enabled,
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class _ProviderModelCredentialCacheEntry:
|
|
id: str
|
|
provider_name: str
|
|
model_name: str
|
|
model_type: ModelType
|
|
credential_name: str
|
|
|
|
@classmethod
|
|
def from_record(cls, record: ProviderModelCredential) -> _ProviderModelCredentialCacheEntry:
|
|
return cls(
|
|
id=record.id,
|
|
provider_name=record.provider_name,
|
|
model_name=record.model_name,
|
|
model_type=record.model_type,
|
|
credential_name=record.credential_name,
|
|
)
|
|
|
|
@classmethod
|
|
def from_cache_row(cls, row: dict[str, Any]) -> _ProviderModelCredentialCacheEntry:
|
|
return cls(
|
|
id=row["id"],
|
|
provider_name=row["provider_name"],
|
|
model_name=row["model_name"],
|
|
model_type=ModelType(row["model_type"]),
|
|
credential_name=row["credential_name"],
|
|
)
|
|
|
|
def to_cache_row(self) -> dict[str, Any]:
|
|
return {
|
|
"id": self.id,
|
|
"provider_name": self.provider_name,
|
|
"model_name": self.model_name,
|
|
"model_type": self.model_type.value,
|
|
"credential_name": self.credential_name,
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class _ProviderCredentialCacheEntry:
|
|
id: str
|
|
provider_name: str
|
|
credential_name: str
|
|
|
|
@classmethod
|
|
def from_record(cls, record: ProviderCredential) -> _ProviderCredentialCacheEntry:
|
|
return cls(
|
|
id=record.id,
|
|
provider_name=record.provider_name,
|
|
credential_name=record.credential_name,
|
|
)
|
|
|
|
@classmethod
|
|
def from_cache_row(cls, row: dict[str, Any]) -> _ProviderCredentialCacheEntry:
|
|
return cls(
|
|
id=row["id"],
|
|
provider_name=row["provider_name"],
|
|
credential_name=row["credential_name"],
|
|
)
|
|
|
|
def to_cache_row(self) -> dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class _LoadBalancingModelConfigCacheEntry:
|
|
id: str
|
|
tenant_id: str
|
|
provider_name: str
|
|
model_name: str
|
|
model_type: ModelType
|
|
name: str
|
|
encrypted_config: str | None
|
|
credential_id: str | None
|
|
credential_source_type: CredentialSourceType | None
|
|
enabled: bool
|
|
|
|
@classmethod
|
|
def from_record(cls, record: LoadBalancingModelConfig) -> _LoadBalancingModelConfigCacheEntry:
|
|
return cls(
|
|
id=record.id,
|
|
tenant_id=record.tenant_id,
|
|
provider_name=record.provider_name,
|
|
model_name=record.model_name,
|
|
model_type=record.model_type,
|
|
name=record.name,
|
|
encrypted_config=record.encrypted_config,
|
|
credential_id=record.credential_id,
|
|
credential_source_type=record.credential_source_type,
|
|
enabled=record.enabled,
|
|
)
|
|
|
|
@classmethod
|
|
def from_cache_row(cls, row: dict[str, Any]) -> _LoadBalancingModelConfigCacheEntry:
|
|
return cls(
|
|
id=row["id"],
|
|
tenant_id=row["tenant_id"],
|
|
provider_name=row["provider_name"],
|
|
model_name=row["model_name"],
|
|
model_type=ModelType(row["model_type"]),
|
|
name=row["name"],
|
|
encrypted_config=row.get("encrypted_config"),
|
|
credential_id=row.get("credential_id"),
|
|
credential_source_type=CredentialSourceType(row["credential_source_type"])
|
|
if row.get("credential_source_type")
|
|
else None,
|
|
enabled=row["enabled"],
|
|
)
|
|
|
|
def to_cache_row(self) -> dict[str, Any]:
|
|
return {
|
|
"id": self.id,
|
|
"tenant_id": self.tenant_id,
|
|
"provider_name": self.provider_name,
|
|
"model_name": self.model_name,
|
|
"model_type": self.model_type.value,
|
|
"name": self.name,
|
|
"encrypted_config": self.encrypted_config,
|
|
"credential_id": self.credential_id,
|
|
"credential_source_type": self.credential_source_type.value if self.credential_source_type else None,
|
|
"enabled": self.enabled,
|
|
}
|
|
|
|
|
|
class _ProviderConfigurationSourceCache:
|
|
"""Redis-backed cache for tenant provider DB cache entries.
|
|
|
|
The assembled ``ProviderConfigurations`` object is intentionally not cached
|
|
here because it carries request-scoped runtime bindings. Cache only the DB
|
|
rows that are stable enough to reuse across processes, then let each
|
|
``ProviderManager`` assemble and bind fresh runtime-aware entities.
|
|
"""
|
|
|
|
@classmethod
|
|
def get_records[T: _CacheEntry](
|
|
cls,
|
|
*,
|
|
tenant_id: str,
|
|
source: ProviderConfigurationCacheSource,
|
|
entry_cls: type[T],
|
|
) -> tuple[list[T] | None, str | None]:
|
|
version: str | None = None
|
|
try:
|
|
version = cls._get_version(tenant_id=tenant_id, source=source)
|
|
cache_key = cls._source_key(tenant_id=tenant_id, source=source, version=version)
|
|
cached_records = redis_client.get(cache_key)
|
|
if cached_records is None:
|
|
return None, version
|
|
|
|
cached_text = cached_records.decode("utf-8") if isinstance(cached_records, bytes) else cached_records
|
|
rows = json.loads(cached_text)
|
|
if not isinstance(rows, list):
|
|
return None, version
|
|
|
|
return [entry_cls.from_cache_row(row) for row in rows if isinstance(row, dict)], version
|
|
except Exception:
|
|
logger.warning("Failed to read provider configuration source cache", exc_info=True)
|
|
return None, version
|
|
|
|
@classmethod
|
|
def set_records(
|
|
cls,
|
|
*,
|
|
tenant_id: str,
|
|
source: ProviderConfigurationCacheSource,
|
|
records: Sequence[_CacheEntry],
|
|
expected_version: str | None = None,
|
|
) -> None:
|
|
try:
|
|
version = cls._get_version(tenant_id=tenant_id, source=source)
|
|
if expected_version is not None and version != expected_version:
|
|
return
|
|
cache_key = cls._source_key(tenant_id=tenant_id, source=source, version=version)
|
|
rows = [record.to_cache_row() for record in records]
|
|
redis_client.setex(cache_key, _PROVIDER_CONFIGURATION_CACHE_TTL_SECONDS, json.dumps(rows))
|
|
except Exception:
|
|
logger.warning("Failed to write provider configuration source cache", exc_info=True)
|
|
|
|
@classmethod
|
|
def invalidate_tenant(
|
|
cls,
|
|
tenant_id: str,
|
|
sources: Sequence[ProviderConfigurationCacheSource] | None = None,
|
|
) -> None:
|
|
try:
|
|
if sources is None:
|
|
sources = _PROVIDER_CONFIGURATION_SOURCES
|
|
for source in sources:
|
|
version_key = _PROVIDER_CONFIGURATION_CACHE_VERSION_KEY.format(tenant_id=tenant_id, source=source.value)
|
|
redis_client.incr(version_key)
|
|
redis_client.expire(version_key, _PROVIDER_CONFIGURATION_CACHE_VERSION_TTL_SECONDS)
|
|
except Exception:
|
|
logger.warning("Failed to invalidate provider configuration source cache", exc_info=True)
|
|
|
|
@classmethod
|
|
def _get_version(cls, *, tenant_id: str, source: ProviderConfigurationCacheSource) -> str:
|
|
version_key = _PROVIDER_CONFIGURATION_CACHE_VERSION_KEY.format(tenant_id=tenant_id, source=source.value)
|
|
version = redis_client.get(version_key)
|
|
if version is None:
|
|
redis_client.set(version_key, "0", ex=_PROVIDER_CONFIGURATION_CACHE_VERSION_TTL_SECONDS)
|
|
return "0"
|
|
redis_client.expire(version_key, _PROVIDER_CONFIGURATION_CACHE_VERSION_TTL_SECONDS)
|
|
return version.decode("utf-8") if isinstance(version, bytes) else str(version)
|
|
|
|
@staticmethod
|
|
def _source_key(*, tenant_id: str, source: ProviderConfigurationCacheSource, version: str) -> str:
|
|
return _PROVIDER_CONFIGURATION_CACHE_SOURCE_KEY.format(
|
|
tenant_id=tenant_id,
|
|
source=source.value,
|
|
version=version,
|
|
)
|
|
|
|
|
|
def _get_cached_or_load_records[T: _CacheEntry](
|
|
*,
|
|
tenant_id: str,
|
|
cache_source: _ProviderConfigurationCacheSourceSpec[T],
|
|
) -> list[T]:
|
|
cached_records, cache_version = _ProviderConfigurationSourceCache.get_records(
|
|
tenant_id=tenant_id,
|
|
source=cache_source.name,
|
|
entry_cls=cache_source.entry_cls,
|
|
)
|
|
if cached_records is not None:
|
|
return cached_records
|
|
|
|
records = cache_source.load_records(tenant_id)
|
|
_ProviderConfigurationSourceCache.set_records(
|
|
tenant_id=tenant_id,
|
|
source=cache_source.name,
|
|
records=records,
|
|
expected_version=cache_version,
|
|
)
|
|
return records
|
|
|
|
|
|
def _attach_active_credentials(
|
|
*,
|
|
session: Any,
|
|
records: Sequence[Provider | ProviderModel],
|
|
credential_model_cls: type[ProviderCredential | ProviderModelCredential],
|
|
) -> None:
|
|
credential_ids = [record.credential_id for record in records if getattr(record, "credential_id", None)]
|
|
if not credential_ids:
|
|
return
|
|
|
|
credentials = session.scalars(select(credential_model_cls).where(credential_model_cls.id.in_(credential_ids))).all()
|
|
credential_by_id = {credential.id: credential for credential in credentials}
|
|
for record in records:
|
|
if getattr(record, "credential_id", None):
|
|
record.__dict__["credential"] = credential_by_id.get(record.credential_id)
|
|
|
|
|
|
def _load_provider_model_cache_entries(tenant_id: str) -> list[_ProviderModelCacheEntry]:
|
|
with session_factory.create_session() as session:
|
|
stmt = select(ProviderModel).where(ProviderModel.tenant_id == tenant_id, ProviderModel.is_valid == True)
|
|
provider_models = list(session.scalars(stmt))
|
|
_attach_active_credentials(
|
|
session=session,
|
|
records=provider_models,
|
|
credential_model_cls=ProviderModelCredential,
|
|
)
|
|
return [_ProviderModelCacheEntry.from_record(provider_model) for provider_model in provider_models]
|
|
|
|
|
|
def _load_preferred_model_provider_cache_entries(tenant_id: str) -> list[_TenantPreferredModelProviderCacheEntry]:
|
|
with session_factory.create_session() as session:
|
|
stmt = select(TenantPreferredModelProvider).where(TenantPreferredModelProvider.tenant_id == tenant_id)
|
|
return [
|
|
_TenantPreferredModelProviderCacheEntry.from_record(preferred_model_provider)
|
|
for preferred_model_provider in session.scalars(stmt)
|
|
]
|
|
|
|
|
|
def _load_provider_model_setting_cache_entries(tenant_id: str) -> list[_ProviderModelSettingCacheEntry]:
|
|
with session_factory.create_session() as session:
|
|
stmt = select(ProviderModelSetting).where(ProviderModelSetting.tenant_id == tenant_id)
|
|
return [
|
|
_ProviderModelSettingCacheEntry.from_record(provider_model_setting)
|
|
for provider_model_setting in session.scalars(stmt)
|
|
]
|
|
|
|
|
|
def _load_provider_model_credential_cache_entries(tenant_id: str) -> list[_ProviderModelCredentialCacheEntry]:
|
|
with session_factory.create_session() as session:
|
|
stmt = (
|
|
select(ProviderModelCredential)
|
|
.where(ProviderModelCredential.tenant_id == tenant_id)
|
|
.order_by(ProviderModelCredential.created_at.desc())
|
|
)
|
|
return [
|
|
_ProviderModelCredentialCacheEntry.from_record(provider_model_credential)
|
|
for provider_model_credential in session.scalars(stmt)
|
|
]
|
|
|
|
|
|
def _load_provider_credential_cache_entries(tenant_id: str) -> list[_ProviderCredentialCacheEntry]:
|
|
with session_factory.create_session() as session:
|
|
stmt = (
|
|
select(ProviderCredential)
|
|
.where(ProviderCredential.tenant_id == tenant_id)
|
|
.order_by(ProviderCredential.created_at.desc())
|
|
)
|
|
return [
|
|
_ProviderCredentialCacheEntry.from_record(provider_credential)
|
|
for provider_credential in session.scalars(stmt)
|
|
]
|
|
|
|
|
|
def _load_provider_load_balancing_config_cache_entries(tenant_id: str) -> list[_LoadBalancingModelConfigCacheEntry]:
|
|
with session_factory.create_session() as session:
|
|
stmt = select(LoadBalancingModelConfig).where(LoadBalancingModelConfig.tenant_id == tenant_id)
|
|
return [
|
|
_LoadBalancingModelConfigCacheEntry.from_record(load_balancing_model_config)
|
|
for load_balancing_model_config in session.scalars(stmt)
|
|
]
|
|
|
|
|
|
_PROVIDER_MODELS_CACHE_SOURCE = _ProviderConfigurationCacheSourceSpec(
|
|
name=ProviderConfigurationCacheSource.PROVIDER_MODELS,
|
|
entry_cls=_ProviderModelCacheEntry,
|
|
load_records=_load_provider_model_cache_entries,
|
|
)
|
|
_PREFERRED_MODEL_PROVIDERS_CACHE_SOURCE = _ProviderConfigurationCacheSourceSpec(
|
|
name=ProviderConfigurationCacheSource.PREFERRED_MODEL_PROVIDERS,
|
|
entry_cls=_TenantPreferredModelProviderCacheEntry,
|
|
load_records=_load_preferred_model_provider_cache_entries,
|
|
)
|
|
_PROVIDER_MODEL_SETTINGS_CACHE_SOURCE = _ProviderConfigurationCacheSourceSpec(
|
|
name=ProviderConfigurationCacheSource.PROVIDER_MODEL_SETTINGS,
|
|
entry_cls=_ProviderModelSettingCacheEntry,
|
|
load_records=_load_provider_model_setting_cache_entries,
|
|
)
|
|
_PROVIDER_MODEL_CREDENTIALS_CACHE_SOURCE = _ProviderConfigurationCacheSourceSpec(
|
|
name=ProviderConfigurationCacheSource.PROVIDER_MODEL_CREDENTIALS,
|
|
entry_cls=_ProviderModelCredentialCacheEntry,
|
|
load_records=_load_provider_model_credential_cache_entries,
|
|
)
|
|
_PROVIDER_CREDENTIALS_CACHE_SOURCE = _ProviderConfigurationCacheSourceSpec(
|
|
name=ProviderConfigurationCacheSource.PROVIDER_CREDENTIALS,
|
|
entry_cls=_ProviderCredentialCacheEntry,
|
|
load_records=_load_provider_credential_cache_entries,
|
|
)
|
|
_PROVIDER_LOAD_BALANCING_CONFIGS_CACHE_SOURCE = _ProviderConfigurationCacheSourceSpec(
|
|
name=ProviderConfigurationCacheSource.PROVIDER_LOAD_BALANCING_CONFIGS,
|
|
entry_cls=_LoadBalancingModelConfigCacheEntry,
|
|
load_records=_load_provider_load_balancing_config_cache_entries,
|
|
)
|
|
|
|
|
|
class ProviderManager:
|
|
"""
|
|
ProviderManager manages tenant-scoped model provider configuration.
|
|
|
|
The runtime adapter is injected by the composition layer so this class stays
|
|
focused on configuration assembly instead of constructing plugin runtimes.
|
|
Request-bound managers may carry caller identity in that runtime, and the
|
|
resulting ``ProviderConfiguration`` objects must reuse it for downstream
|
|
model-type and schema lookups.
|
|
|
|
Configuration assembly is cached per manager instance so call chains that
|
|
share one request-scoped manager can reuse the same provider graph instead
|
|
of rebuilding it for every lookup. Call ``clear_configurations_cache()``
|
|
when a long-lived manager needs to observe writes performed within the same
|
|
instance scope.
|
|
"""
|
|
|
|
# Keyed by tenant_id -- a single ProviderManager instance may be asked to decrypt
|
|
# credentials belonging to different tenants (e.g. load balancing configs each carry
|
|
# their own tenant_id), so this cache must not collapse to a single shared value.
|
|
_decoding_contexts: dict[str, Any]
|
|
_model_runtime: ModelRuntime
|
|
_configurations_cache: dict[str, ProviderConfigurations]
|
|
|
|
def __init__(self, model_runtime: ModelRuntime):
|
|
self._decoding_contexts = {}
|
|
self._model_runtime = model_runtime
|
|
self._configurations_cache = {}
|
|
|
|
def _get_decoding_context(self, tenant_id: str) -> Any:
|
|
"""Return this manager's cached decoding context for `tenant_id`, fetching it once if absent."""
|
|
if tenant_id not in self._decoding_contexts:
|
|
self._decoding_contexts[tenant_id] = encrypter.get_decrypt_decoding(tenant_id)
|
|
return self._decoding_contexts[tenant_id]
|
|
|
|
def clear_configurations_cache(self, tenant_id: str | None = None) -> None:
|
|
"""Drop assembled provider configurations cached on this manager instance."""
|
|
if tenant_id is None:
|
|
self._configurations_cache.clear()
|
|
return
|
|
|
|
self._configurations_cache.pop(tenant_id, None)
|
|
|
|
@staticmethod
|
|
def invalidate_configurations_cache(
|
|
tenant_id: str,
|
|
sources: Sequence[ProviderConfigurationCacheSource] | None = None,
|
|
) -> None:
|
|
"""Invalidate cross-process provider configuration source cache for a tenant."""
|
|
_ProviderConfigurationSourceCache.invalidate_tenant(tenant_id, sources=sources)
|
|
|
|
def get_configurations(self, tenant_id: str) -> ProviderConfigurations:
|
|
"""
|
|
Get model provider configurations.
|
|
|
|
Construct ProviderConfiguration objects for each provider
|
|
Including:
|
|
1. Basic information of the provider
|
|
2. Hosting configuration information, including:
|
|
(1. Whether to enable (support) hosting type, if enabled, the following information exists
|
|
(2. List of hosting type provider configurations
|
|
(including quota type, quota limit, current remaining quota, etc.)
|
|
(3. The current hosting type in use (whether there is a quota or not)
|
|
paid quotas > provider free quotas > hosting trial quotas
|
|
(4. Unified credentials for hosting providers
|
|
3. Custom configuration information, including:
|
|
(1. Whether to enable (support) custom type, if enabled, the following information exists
|
|
(2. Custom provider configuration (including credentials)
|
|
(3. List of custom provider model configurations (including credentials)
|
|
4. Hosting/custom preferred provider type.
|
|
Provide methods:
|
|
- Get the current configuration (including credentials)
|
|
- Get the availability and status of the hosting configuration: active available,
|
|
quota_exceeded insufficient quota, unsupported hosting
|
|
- Get the availability of custom configuration
|
|
Custom provider available conditions:
|
|
(1. custom provider credentials available
|
|
(2. at least one custom model credentials available
|
|
- Verify, update, and delete custom provider configuration
|
|
- Verify, update, and delete custom provider model configuration
|
|
- Get the list of available models (optional provider filtering, model type filtering)
|
|
Append custom provider models to the list
|
|
- Get provider instance
|
|
- Switch selection priority
|
|
|
|
:param tenant_id:
|
|
:return:
|
|
"""
|
|
cached_configurations = self._configurations_cache.get(tenant_id)
|
|
if cached_configurations is not None:
|
|
return cached_configurations
|
|
|
|
# Get all provider records of the workspace
|
|
provider_name_to_provider_records_dict = self._get_all_providers(tenant_id)
|
|
|
|
# Initialize trial provider records if not exist
|
|
provider_name_to_provider_records_dict = self._init_trial_provider_records(
|
|
tenant_id, provider_name_to_provider_records_dict
|
|
)
|
|
|
|
# append providers with langgenius/openai/openai
|
|
provider_name_list = list(provider_name_to_provider_records_dict.keys())
|
|
for provider_name in provider_name_list:
|
|
provider_id = ModelProviderID(provider_name)
|
|
if str(provider_id) not in provider_name_list:
|
|
provider_name_to_provider_records_dict[str(provider_id)] = provider_name_to_provider_records_dict[
|
|
provider_name
|
|
]
|
|
|
|
# Get all provider model records of the workspace
|
|
provider_name_to_provider_model_records_dict = self._get_all_provider_models(tenant_id)
|
|
for provider_name in list(provider_name_to_provider_model_records_dict.keys()):
|
|
provider_id = ModelProviderID(provider_name)
|
|
if str(provider_id) not in provider_name_to_provider_model_records_dict:
|
|
provider_name_to_provider_model_records_dict[str(provider_id)] = (
|
|
provider_name_to_provider_model_records_dict[provider_name]
|
|
)
|
|
|
|
# Get all provider entities
|
|
model_provider_factory = ModelProviderFactory(runtime=self._model_runtime)
|
|
provider_entities = model_provider_factory.get_providers()
|
|
|
|
# Get All preferred provider types of the workspace
|
|
provider_name_to_preferred_model_provider_records_dict = self._get_all_preferred_model_providers(tenant_id)
|
|
# Ensure that both the original provider name and its ModelProviderID string representation
|
|
# are present in the dictionary to handle cases where either form might be used
|
|
for provider_name in list(provider_name_to_preferred_model_provider_records_dict.keys()):
|
|
provider_id = ModelProviderID(provider_name)
|
|
if str(provider_id) not in provider_name_to_preferred_model_provider_records_dict:
|
|
# Add the ModelProviderID string representation if it's not already present
|
|
provider_name_to_preferred_model_provider_records_dict[str(provider_id)] = (
|
|
provider_name_to_preferred_model_provider_records_dict[provider_name]
|
|
)
|
|
|
|
# Get All provider model settings
|
|
provider_name_to_provider_model_settings_dict = self._get_all_provider_model_settings(tenant_id)
|
|
|
|
# Get All load balancing configs
|
|
provider_name_to_provider_load_balancing_model_configs_dict = self._get_all_provider_load_balancing_configs(
|
|
tenant_id
|
|
)
|
|
|
|
# Get All provider model credentials
|
|
provider_name_to_provider_model_credentials_dict = self._get_all_provider_model_credentials(tenant_id)
|
|
|
|
# Get All provider credentials
|
|
provider_name_to_provider_credentials_dict = self._get_all_provider_credentials(tenant_id)
|
|
|
|
provider_configurations = ProviderConfigurations(tenant_id=tenant_id)
|
|
|
|
# Construct ProviderConfiguration objects for each provider
|
|
for provider_entity in provider_entities:
|
|
# handle include, exclude
|
|
if is_filtered(
|
|
include_set=dify_config.POSITION_PROVIDER_INCLUDES_SET,
|
|
exclude_set=dify_config.POSITION_PROVIDER_EXCLUDES_SET,
|
|
data=provider_entity,
|
|
name_func=lambda x: x.provider,
|
|
):
|
|
continue
|
|
|
|
provider_name = provider_entity.provider
|
|
provider_records = provider_name_to_provider_records_dict.get(provider_entity.provider, [])
|
|
provider_model_records = provider_name_to_provider_model_records_dict.get(provider_entity.provider, [])
|
|
provider_id_entity = ModelProviderID(provider_name)
|
|
if provider_id_entity.is_langgenius():
|
|
provider_model_records.extend(
|
|
provider_name_to_provider_model_records_dict.get(provider_id_entity.provider_name, [])
|
|
)
|
|
provider_model_credentials = provider_name_to_provider_model_credentials_dict.get(
|
|
provider_entity.provider, []
|
|
)
|
|
provider_id_entity = ModelProviderID(provider_name)
|
|
if provider_id_entity.is_langgenius():
|
|
provider_model_credentials.extend(
|
|
provider_name_to_provider_model_credentials_dict.get(provider_id_entity.provider_name, [])
|
|
)
|
|
|
|
# Convert to custom configuration
|
|
custom_configuration = self._to_custom_configuration(
|
|
tenant_id,
|
|
provider_entity,
|
|
provider_records,
|
|
provider_model_records,
|
|
provider_model_credentials,
|
|
provider_name_to_provider_credentials_dict,
|
|
)
|
|
|
|
# Convert to system configuration
|
|
system_configuration = self._to_system_configuration(tenant_id, provider_entity, provider_records)
|
|
|
|
# Get preferred provider type
|
|
preferred_provider_type_record = provider_name_to_preferred_model_provider_records_dict.get(provider_name)
|
|
|
|
if preferred_provider_type_record:
|
|
preferred_provider_type = preferred_provider_type_record.preferred_provider_type
|
|
elif dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD and system_configuration.enabled:
|
|
preferred_provider_type = ProviderType.SYSTEM
|
|
elif custom_configuration.provider or custom_configuration.models:
|
|
preferred_provider_type = ProviderType.CUSTOM
|
|
elif system_configuration.enabled:
|
|
preferred_provider_type = ProviderType.SYSTEM
|
|
else:
|
|
preferred_provider_type = ProviderType.CUSTOM
|
|
|
|
using_provider_type = preferred_provider_type
|
|
has_valid_quota = any(quota_conf.is_valid for quota_conf in system_configuration.quota_configurations)
|
|
|
|
if preferred_provider_type == ProviderType.SYSTEM:
|
|
if not system_configuration.enabled or not system_configuration.quota_configurations:
|
|
using_provider_type = ProviderType.CUSTOM
|
|
elif not has_valid_quota and (custom_configuration.provider or custom_configuration.models):
|
|
# Only configured alternatives can serve as fallbacks; otherwise downstream checks must surface
|
|
# system quota exhaustion instead of reporting missing custom credentials.
|
|
using_provider_type = ProviderType.CUSTOM
|
|
|
|
else:
|
|
if not custom_configuration.provider and not custom_configuration.models:
|
|
if system_configuration.enabled and has_valid_quota:
|
|
using_provider_type = ProviderType.SYSTEM
|
|
|
|
# Get provider load balancing configs
|
|
provider_model_settings = provider_name_to_provider_model_settings_dict.get(provider_name)
|
|
|
|
# Get provider load balancing configs
|
|
provider_load_balancing_configs = provider_name_to_provider_load_balancing_model_configs_dict.get(
|
|
provider_name
|
|
)
|
|
|
|
provider_id_entity = ModelProviderID(provider_name)
|
|
|
|
if provider_id_entity.is_langgenius():
|
|
if provider_model_settings is not None:
|
|
provider_model_settings.extend(
|
|
provider_name_to_provider_model_settings_dict.get(provider_id_entity.provider_name, [])
|
|
)
|
|
if provider_load_balancing_configs is not None:
|
|
provider_load_balancing_configs.extend(
|
|
provider_name_to_provider_load_balancing_model_configs_dict.get(
|
|
provider_id_entity.provider_name, []
|
|
)
|
|
)
|
|
|
|
# Convert to model settings
|
|
model_settings = self._to_model_settings(
|
|
provider_entity=provider_entity,
|
|
provider_model_settings=provider_model_settings,
|
|
load_balancing_model_configs=provider_load_balancing_configs,
|
|
)
|
|
|
|
provider_configuration = ProviderConfiguration(
|
|
tenant_id=tenant_id,
|
|
provider=provider_entity,
|
|
preferred_provider_type=preferred_provider_type,
|
|
using_provider_type=using_provider_type,
|
|
system_configuration=system_configuration,
|
|
custom_configuration=custom_configuration,
|
|
model_settings=model_settings,
|
|
)
|
|
provider_configuration.bind_model_runtime(self._model_runtime)
|
|
|
|
provider_configurations[str(provider_id_entity)] = provider_configuration
|
|
|
|
self._configurations_cache[tenant_id] = provider_configurations
|
|
|
|
# Return the encapsulated object
|
|
return provider_configurations
|
|
|
|
def get_provider_model_bundle(self, tenant_id: str, provider: str, model_type: ModelType) -> ProviderModelBundle:
|
|
"""
|
|
Get provider model bundle.
|
|
:param tenant_id: workspace id
|
|
:param provider: provider name
|
|
:param model_type: model type
|
|
:return:
|
|
"""
|
|
provider_configurations = self.get_configurations(tenant_id)
|
|
|
|
# get provider instance
|
|
provider_configuration = provider_configurations.get(provider)
|
|
if not provider_configuration:
|
|
raise ValueError(f"Provider {provider} does not exist.")
|
|
|
|
model_type_instance = provider_configuration.get_model_type_instance(model_type)
|
|
|
|
return ProviderModelBundle(
|
|
configuration=provider_configuration,
|
|
model_type_instance=model_type_instance,
|
|
)
|
|
|
|
def get_default_model(self, tenant_id: str, model_type: ModelType) -> DefaultModelEntity | None:
|
|
"""
|
|
Get default model.
|
|
|
|
:param tenant_id: workspace id
|
|
:param model_type: model type
|
|
:return:
|
|
"""
|
|
stmt = select(TenantDefaultModel).where(
|
|
TenantDefaultModel.tenant_id == tenant_id,
|
|
TenantDefaultModel.model_type == model_type,
|
|
)
|
|
default_model = db.session.scalar(stmt)
|
|
|
|
# If it does not exist, get the first available provider model from get_configurations
|
|
# and update the TenantDefaultModel record
|
|
if not default_model:
|
|
# Get provider configurations
|
|
provider_configurations = self.get_configurations(tenant_id)
|
|
|
|
# get available models from provider_configurations
|
|
available_models = provider_configurations.get_models(model_type=model_type, only_active=True)
|
|
|
|
if available_models:
|
|
available_model = available_models[0]
|
|
|
|
default_model = TenantDefaultModel(
|
|
tenant_id=tenant_id,
|
|
model_type=model_type,
|
|
provider_name=available_model.provider.provider,
|
|
model_name=available_model.model,
|
|
)
|
|
db.session.add(default_model)
|
|
db.session.commit()
|
|
|
|
if not default_model:
|
|
return None
|
|
|
|
model_provider_factory = ModelProviderFactory(runtime=self._model_runtime)
|
|
provider_schema = model_provider_factory.get_provider_schema(provider=default_model.provider_name)
|
|
|
|
return DefaultModelEntity(
|
|
model=default_model.model_name,
|
|
model_type=model_type,
|
|
provider=DefaultModelProviderEntity(
|
|
provider=provider_schema.provider,
|
|
label=provider_schema.label,
|
|
icon_small=provider_schema.icon_small,
|
|
supported_model_types=provider_schema.supported_model_types,
|
|
),
|
|
)
|
|
|
|
def get_first_provider_first_model(self, tenant_id: str, model_type: ModelType) -> tuple[str | None, str | None]:
|
|
"""
|
|
Get names of first model and its provider
|
|
|
|
:param tenant_id: workspace id
|
|
:param model_type: model type
|
|
:return: provider name, model name
|
|
"""
|
|
provider_configurations = self.get_configurations(tenant_id)
|
|
|
|
# get available models from provider_configurations
|
|
all_models = provider_configurations.get_models(model_type=model_type, only_active=False)
|
|
|
|
if not all_models:
|
|
return None, None
|
|
|
|
return all_models[0].provider.provider, all_models[0].model
|
|
|
|
def update_default_model_record(
|
|
self, tenant_id: str, model_type: ModelType, provider: str, model: str
|
|
) -> TenantDefaultModel:
|
|
"""
|
|
Update default model record.
|
|
|
|
:param tenant_id: workspace id
|
|
:param model_type: model type
|
|
:param provider: provider name
|
|
:param model: model name
|
|
:return:
|
|
"""
|
|
provider_configurations = self.get_configurations(tenant_id)
|
|
if provider not in provider_configurations:
|
|
raise ValueError(f"Provider {provider} does not exist.")
|
|
|
|
# get available models from provider_configurations
|
|
available_models = provider_configurations.get_models(model_type=model_type, only_active=True)
|
|
|
|
# check if the model is exist in available models
|
|
model_names = [model.model for model in available_models]
|
|
if model not in model_names:
|
|
raise ValueError(f"Model {model} does not exist.")
|
|
stmt = select(TenantDefaultModel).where(
|
|
TenantDefaultModel.tenant_id == tenant_id,
|
|
TenantDefaultModel.model_type == model_type,
|
|
)
|
|
default_model = db.session.scalar(stmt)
|
|
|
|
# create or update TenantDefaultModel record
|
|
if default_model:
|
|
# update default model
|
|
default_model.provider_name = provider
|
|
default_model.model_name = model
|
|
db.session.commit()
|
|
else:
|
|
# create default model
|
|
default_model = TenantDefaultModel(
|
|
tenant_id=tenant_id,
|
|
model_type=model_type,
|
|
provider_name=provider,
|
|
model_name=model,
|
|
)
|
|
db.session.add(default_model)
|
|
db.session.commit()
|
|
|
|
return default_model
|
|
|
|
@staticmethod
|
|
def _get_all_providers(tenant_id: str) -> dict[str, list[Provider]]:
|
|
provider_name_to_provider_records_dict = defaultdict(list)
|
|
with session_factory.create_session() as session:
|
|
stmt = select(Provider).where(Provider.tenant_id == tenant_id, Provider.is_valid == True)
|
|
providers = list(session.scalars(stmt))
|
|
_attach_active_credentials(
|
|
session=session,
|
|
records=providers,
|
|
credential_model_cls=ProviderCredential,
|
|
)
|
|
for provider in providers:
|
|
# Use provider name with prefix after the data migration
|
|
provider_name_to_provider_records_dict[str(ModelProviderID(provider.provider_name))].append(provider)
|
|
return provider_name_to_provider_records_dict
|
|
|
|
@staticmethod
|
|
def _get_all_provider_models(tenant_id: str) -> dict[str, list[_ProviderModelCacheEntry]]:
|
|
"""
|
|
Get all provider model records of the workspace.
|
|
|
|
:param tenant_id: workspace id
|
|
:return:
|
|
"""
|
|
provider_models = _get_cached_or_load_records(
|
|
tenant_id=tenant_id,
|
|
cache_source=_PROVIDER_MODELS_CACHE_SOURCE,
|
|
)
|
|
|
|
provider_name_to_provider_model_records_dict = defaultdict(list)
|
|
for provider_model in provider_models:
|
|
provider_name_to_provider_model_records_dict[provider_model.provider_name].append(provider_model)
|
|
return provider_name_to_provider_model_records_dict
|
|
|
|
@staticmethod
|
|
def _get_all_preferred_model_providers(tenant_id: str) -> dict[str, _TenantPreferredModelProviderCacheEntry]:
|
|
"""
|
|
Get All preferred provider types of the workspace.
|
|
|
|
:param tenant_id: workspace id
|
|
:return:
|
|
"""
|
|
preferred_provider_types = _get_cached_or_load_records(
|
|
tenant_id=tenant_id,
|
|
cache_source=_PREFERRED_MODEL_PROVIDERS_CACHE_SOURCE,
|
|
)
|
|
|
|
return {
|
|
preferred_provider_type.provider_name: preferred_provider_type
|
|
for preferred_provider_type in preferred_provider_types
|
|
}
|
|
|
|
@staticmethod
|
|
def _get_all_provider_model_settings(tenant_id: str) -> dict[str, list[_ProviderModelSettingCacheEntry]]:
|
|
"""
|
|
Get All provider model settings of the workspace.
|
|
|
|
:param tenant_id: workspace id
|
|
:return:
|
|
"""
|
|
provider_model_settings = _get_cached_or_load_records(
|
|
tenant_id=tenant_id,
|
|
cache_source=_PROVIDER_MODEL_SETTINGS_CACHE_SOURCE,
|
|
)
|
|
|
|
provider_name_to_provider_model_settings_dict = defaultdict(list)
|
|
for provider_model_setting in provider_model_settings:
|
|
provider_name_to_provider_model_settings_dict[provider_model_setting.provider_name].append(
|
|
provider_model_setting
|
|
)
|
|
return provider_name_to_provider_model_settings_dict
|
|
|
|
@staticmethod
|
|
def _get_all_provider_model_credentials(tenant_id: str) -> dict[str, list[_ProviderModelCredentialCacheEntry]]:
|
|
"""
|
|
Get All provider model credentials of the workspace.
|
|
|
|
:param tenant_id: workspace id
|
|
:return:
|
|
"""
|
|
provider_model_credentials = _get_cached_or_load_records(
|
|
tenant_id=tenant_id,
|
|
cache_source=_PROVIDER_MODEL_CREDENTIALS_CACHE_SOURCE,
|
|
)
|
|
|
|
provider_name_to_provider_model_credentials_dict = defaultdict(list)
|
|
for provider_model_credential in provider_model_credentials:
|
|
provider_name_to_provider_model_credentials_dict[provider_model_credential.provider_name].append(
|
|
provider_model_credential
|
|
)
|
|
return provider_name_to_provider_model_credentials_dict
|
|
|
|
@staticmethod
|
|
def _get_all_provider_credentials(tenant_id: str) -> dict[str, list[_ProviderCredentialCacheEntry]]:
|
|
"""
|
|
Get All provider credentials of the workspace.
|
|
|
|
:param tenant_id: workspace id
|
|
:return:
|
|
"""
|
|
provider_credentials = _get_cached_or_load_records(
|
|
tenant_id=tenant_id,
|
|
cache_source=_PROVIDER_CREDENTIALS_CACHE_SOURCE,
|
|
)
|
|
|
|
provider_name_to_provider_credentials_dict = defaultdict(list)
|
|
for provider_credential in provider_credentials:
|
|
provider_name_to_provider_credentials_dict[provider_credential.provider_name].append(provider_credential)
|
|
return provider_name_to_provider_credentials_dict
|
|
|
|
@staticmethod
|
|
def _get_all_provider_load_balancing_configs(
|
|
tenant_id: str,
|
|
) -> dict[str, list[_LoadBalancingModelConfigCacheEntry]]:
|
|
"""
|
|
Get All provider load balancing configs of the workspace.
|
|
|
|
:param tenant_id: workspace id
|
|
:return:
|
|
"""
|
|
cache_key = f"tenant:{tenant_id}:model_load_balancing_enabled"
|
|
cache_result = redis_client.get(cache_key)
|
|
if cache_result is None:
|
|
model_load_balancing_enabled = FeatureService.get_features(
|
|
tenant_id, exclude_vector_space=True
|
|
).model_load_balancing_enabled
|
|
redis_client.setex(cache_key, 120, str(model_load_balancing_enabled))
|
|
else:
|
|
cache_result = cache_result.decode("utf-8")
|
|
model_load_balancing_enabled = cache_result == "True"
|
|
|
|
if not model_load_balancing_enabled:
|
|
return {}
|
|
|
|
provider_load_balancing_configs = _get_cached_or_load_records(
|
|
tenant_id=tenant_id,
|
|
cache_source=_PROVIDER_LOAD_BALANCING_CONFIGS_CACHE_SOURCE,
|
|
)
|
|
|
|
provider_name_to_provider_load_balancing_model_configs_dict = defaultdict(list)
|
|
for provider_load_balancing_config in provider_load_balancing_configs:
|
|
provider_name_to_provider_load_balancing_model_configs_dict[
|
|
provider_load_balancing_config.provider_name
|
|
].append(provider_load_balancing_config)
|
|
|
|
return provider_name_to_provider_load_balancing_model_configs_dict
|
|
|
|
@staticmethod
|
|
def _get_provider_names(provider_name: str) -> list[str]:
|
|
"""
|
|
provider_name: `openai` or `langgenius/openai/openai`
|
|
return: [`openai`, `langgenius/openai/openai`]
|
|
"""
|
|
provider_names = [provider_name]
|
|
model_provider_id = ModelProviderID(provider_name)
|
|
if model_provider_id.is_langgenius():
|
|
if "/" in provider_name:
|
|
provider_names.append(model_provider_id.provider_name)
|
|
else:
|
|
provider_names.append(str(model_provider_id))
|
|
return provider_names
|
|
|
|
@staticmethod
|
|
def get_provider_available_credentials(
|
|
tenant_id: str,
|
|
provider_name: str,
|
|
user: Account | None = None,
|
|
) -> list[CredentialConfiguration]:
|
|
"""
|
|
Get provider all credentials, filtered by visibility.
|
|
|
|
:param tenant_id: workspace id
|
|
:param provider_name: provider name
|
|
:param user: current user (id + admin flag drive the visibility filter)
|
|
:return:
|
|
"""
|
|
from models.credential_permission import CredentialType as CredPermType
|
|
from services.credential_permission_service import CredentialPermissionService
|
|
|
|
with session_factory.create_session() as session:
|
|
stmt = (
|
|
select(ProviderCredential)
|
|
.where(
|
|
ProviderCredential.tenant_id == tenant_id,
|
|
ProviderCredential.provider_name.in_(ProviderManager._get_provider_names(provider_name)),
|
|
)
|
|
.order_by(ProviderCredential.created_at.desc())
|
|
)
|
|
|
|
if user is not None:
|
|
stmt = CredentialPermissionService.apply_visibility_filter(
|
|
stmt,
|
|
model_id_column=ProviderCredential.id,
|
|
model_user_id_column=ProviderCredential.user_id,
|
|
model_visibility_column=ProviderCredential.visibility,
|
|
credential_type=CredPermType.PROVIDER_CREDENTIAL,
|
|
user=user,
|
|
)
|
|
|
|
available_credentials = session.scalars(stmt).all()
|
|
|
|
return [
|
|
CredentialConfiguration(credential_id=credential.id, credential_name=credential.credential_name)
|
|
for credential in available_credentials
|
|
]
|
|
|
|
@staticmethod
|
|
def get_provider_model_available_credentials(
|
|
tenant_id: str, provider_name: str, model_name: str, model_type: str
|
|
) -> list[CredentialConfiguration]:
|
|
"""
|
|
Get provider custom model all credentials.
|
|
|
|
:param tenant_id: workspace id
|
|
:param provider_name: provider name
|
|
:param model_name: model name
|
|
:param model_type: model type
|
|
:return:
|
|
"""
|
|
with session_factory.create_session() as session:
|
|
stmt = (
|
|
select(ProviderModelCredential)
|
|
.where(
|
|
ProviderModelCredential.tenant_id == tenant_id,
|
|
ProviderModelCredential.provider_name.in_(ProviderManager._get_provider_names(provider_name)),
|
|
ProviderModelCredential.model_name == model_name,
|
|
ProviderModelCredential.model_type == model_type,
|
|
)
|
|
.order_by(ProviderModelCredential.created_at.desc())
|
|
)
|
|
|
|
available_credentials = session.scalars(stmt).all()
|
|
|
|
return [
|
|
CredentialConfiguration(credential_id=credential.id, credential_name=credential.credential_name)
|
|
for credential in available_credentials
|
|
]
|
|
|
|
@staticmethod
|
|
def _init_trial_provider_records(
|
|
tenant_id: str, provider_name_to_provider_records_dict: dict[str, list[Provider]]
|
|
) -> dict[str, list[Provider]]:
|
|
"""
|
|
Initialize trial provider records if not exists.
|
|
|
|
:param tenant_id: workspace id
|
|
:param provider_name_to_provider_records_dict: provider name to provider records dict
|
|
:return:
|
|
"""
|
|
# Get hosting configuration
|
|
hosting_configuration = ext_hosting_provider.hosting_configuration
|
|
|
|
for provider_name, configuration in hosting_configuration.provider_map.items():
|
|
if not configuration.enabled:
|
|
continue
|
|
|
|
provider_records = provider_name_to_provider_records_dict.get(provider_name)
|
|
if not provider_records:
|
|
provider_records = []
|
|
|
|
provider_quota_to_provider_record_dict = {}
|
|
for provider_record in provider_records:
|
|
if provider_record.provider_type != ProviderType.SYSTEM:
|
|
continue
|
|
|
|
if provider_record.quota_type is not None:
|
|
provider_quota_to_provider_record_dict[provider_record.quota_type] = provider_record
|
|
|
|
for quota in configuration.quotas:
|
|
if quota.quota_type in (ProviderQuotaType.TRIAL, ProviderQuotaType.PAID):
|
|
# Init trial provider records if not exists
|
|
if quota.quota_type not in provider_quota_to_provider_record_dict:
|
|
try:
|
|
# FIXME ignore the type error, only TrialHostingQuota has limit need to change the logic
|
|
new_provider_record = Provider(
|
|
tenant_id=tenant_id,
|
|
# TODO: Use provider name with prefix after the data migration.
|
|
provider_name=ModelProviderID(provider_name).provider_name,
|
|
provider_type=ProviderType.SYSTEM,
|
|
quota_type=quota.quota_type, # type: ignore[arg-type]
|
|
quota_limit=0, # type: ignore
|
|
quota_used=0,
|
|
is_valid=True,
|
|
)
|
|
db.session.add(new_provider_record)
|
|
db.session.commit()
|
|
provider_name_to_provider_records_dict[provider_name].append(new_provider_record)
|
|
except IntegrityError:
|
|
db.session.rollback()
|
|
stmt = select(Provider).where(
|
|
Provider.tenant_id == tenant_id,
|
|
Provider.provider_name == ModelProviderID(provider_name).provider_name,
|
|
Provider.provider_type == ProviderType.SYSTEM.value,
|
|
Provider.quota_type == quota.quota_type,
|
|
)
|
|
existed_provider_record = db.session.scalar(stmt)
|
|
if not existed_provider_record:
|
|
continue
|
|
|
|
if not existed_provider_record.is_valid:
|
|
existed_provider_record.is_valid = True
|
|
db.session.commit()
|
|
|
|
provider_name_to_provider_records_dict[provider_name].append(existed_provider_record)
|
|
|
|
return provider_name_to_provider_records_dict
|
|
|
|
def _to_custom_configuration(
|
|
self,
|
|
tenant_id: str,
|
|
provider_entity: ProviderEntity,
|
|
provider_records: list[Provider],
|
|
provider_model_records: list[_ProviderModelCacheEntry],
|
|
provider_model_credentials: list[_ProviderModelCredentialCacheEntry],
|
|
provider_credentials_by_name: dict[str, list[_ProviderCredentialCacheEntry]],
|
|
) -> CustomConfiguration:
|
|
"""
|
|
Convert to custom configuration.
|
|
|
|
:param tenant_id: workspace id
|
|
:param provider_entity: provider entity
|
|
:param provider_records: provider records
|
|
:param provider_model_records: provider model records
|
|
:return:
|
|
"""
|
|
# Get custom provider configuration
|
|
custom_provider_configuration = self._get_custom_provider_configuration(
|
|
tenant_id,
|
|
provider_entity,
|
|
provider_records,
|
|
provider_credentials_by_name,
|
|
)
|
|
|
|
# Get custom models which have not been added to the model list yet
|
|
unadded_models = self._get_can_added_models(provider_model_records, provider_model_credentials)
|
|
|
|
# Get custom model configurations
|
|
custom_model_configurations = self._get_custom_model_configurations(
|
|
tenant_id, provider_entity, provider_model_records, unadded_models, provider_model_credentials
|
|
)
|
|
|
|
can_added_models = [
|
|
UnaddedModelConfiguration(model=model["model"], model_type=model["model_type"]) for model in unadded_models
|
|
]
|
|
|
|
return CustomConfiguration(
|
|
provider=custom_provider_configuration,
|
|
models=custom_model_configurations,
|
|
can_added_models=can_added_models,
|
|
)
|
|
|
|
def _get_custom_provider_configuration(
|
|
self,
|
|
tenant_id: str,
|
|
provider_entity: ProviderEntity,
|
|
provider_records: list[Provider],
|
|
provider_credentials_by_name: dict[str, list[_ProviderCredentialCacheEntry]],
|
|
) -> CustomProviderConfiguration | None:
|
|
"""Get custom provider configuration."""
|
|
# Find custom provider record (non-system)
|
|
custom_provider_record = next(
|
|
(record for record in provider_records if record.provider_type != ProviderType.SYSTEM), None
|
|
)
|
|
|
|
if not custom_provider_record:
|
|
return None
|
|
|
|
# Get provider credential secret variables
|
|
provider_credential_secret_variables = self._extract_secret_variables(
|
|
provider_entity.provider_credential_schema.credential_form_schemas
|
|
if provider_entity.provider_credential_schema
|
|
else []
|
|
)
|
|
|
|
# Get and decrypt provider credentials
|
|
provider_credentials = self._get_and_decrypt_credentials(
|
|
tenant_id=tenant_id,
|
|
record_id=custom_provider_record.id,
|
|
encrypted_config=custom_provider_record.encrypted_config,
|
|
secret_variables=provider_credential_secret_variables,
|
|
cache_type=ProviderCredentialsCacheType.PROVIDER,
|
|
is_provider=True,
|
|
)
|
|
|
|
return CustomProviderConfiguration(
|
|
credentials=provider_credentials,
|
|
current_credential_name=custom_provider_record.credential_name,
|
|
current_credential_id=custom_provider_record.credential_id,
|
|
available_credentials=self._get_provider_available_credentials_from_records(
|
|
custom_provider_record.provider_name,
|
|
provider_credentials_by_name,
|
|
),
|
|
)
|
|
|
|
@staticmethod
|
|
def _get_provider_available_credentials_from_records(
|
|
provider_name: str,
|
|
provider_credentials_by_name: dict[str, list[_ProviderCredentialCacheEntry]],
|
|
) -> list[CredentialConfiguration]:
|
|
available_credentials: list[CredentialConfiguration] = []
|
|
for candidate_provider_name in ProviderManager._get_provider_names(provider_name):
|
|
available_credentials.extend(
|
|
CredentialConfiguration(credential_id=credential.id, credential_name=credential.credential_name)
|
|
for credential in provider_credentials_by_name.get(candidate_provider_name, [])
|
|
)
|
|
return available_credentials
|
|
|
|
def _get_can_added_models(
|
|
self,
|
|
provider_model_records: list[_ProviderModelCacheEntry],
|
|
all_model_credentials: Sequence[_ProviderModelCredentialCacheEntry],
|
|
) -> list[dict]:
|
|
"""Get the custom models and credentials from enterprise version which haven't add to the model list"""
|
|
existing_model_set = {(record.model_name, record.model_type) for record in provider_model_records}
|
|
|
|
# Get not added custom models credentials
|
|
not_added_custom_models_credentials = [
|
|
credential
|
|
for credential in all_model_credentials
|
|
if (credential.model_name, credential.model_type) not in existing_model_set
|
|
]
|
|
|
|
# Group credentials by model
|
|
model_to_credentials = defaultdict(list)
|
|
for credential in not_added_custom_models_credentials:
|
|
model_to_credentials[(credential.model_name, credential.model_type)].append(credential)
|
|
|
|
return [
|
|
{
|
|
"model": model_key[0],
|
|
"model_type": ModelType(model_key[1]),
|
|
"available_model_credentials": [
|
|
CredentialConfiguration(credential_id=cred.id, credential_name=cred.credential_name)
|
|
for cred in creds
|
|
],
|
|
}
|
|
for model_key, creds in model_to_credentials.items()
|
|
]
|
|
|
|
def _get_custom_model_configurations(
|
|
self,
|
|
tenant_id: str,
|
|
provider_entity: ProviderEntity,
|
|
provider_model_records: list[_ProviderModelCacheEntry],
|
|
can_added_models: list[dict],
|
|
all_model_credentials: Sequence[_ProviderModelCredentialCacheEntry],
|
|
) -> list[CustomModelConfiguration]:
|
|
"""Get custom model configurations."""
|
|
# Get model credential secret variables
|
|
model_credential_secret_variables = self._extract_secret_variables(
|
|
provider_entity.model_credential_schema.credential_form_schemas
|
|
if provider_entity.model_credential_schema
|
|
else []
|
|
)
|
|
|
|
# Create credentials lookup for efficient access
|
|
credentials_map = defaultdict(list)
|
|
for credential in all_model_credentials:
|
|
credentials_map[(credential.model_name, credential.model_type)].append(credential)
|
|
|
|
custom_model_configurations = []
|
|
|
|
# Process existing model records
|
|
for provider_model_record in provider_model_records:
|
|
# Use pre-fetched credentials instead of individual database calls
|
|
available_model_credentials = [
|
|
CredentialConfiguration(credential_id=cred.id, credential_name=cred.credential_name)
|
|
for cred in credentials_map.get(
|
|
(provider_model_record.model_name, provider_model_record.model_type), []
|
|
)
|
|
]
|
|
|
|
# Get and decrypt model credentials
|
|
provider_model_credentials = self._get_and_decrypt_credentials(
|
|
tenant_id=tenant_id,
|
|
record_id=provider_model_record.id,
|
|
encrypted_config=provider_model_record.encrypted_config,
|
|
secret_variables=model_credential_secret_variables,
|
|
cache_type=ProviderCredentialsCacheType.MODEL,
|
|
is_provider=False,
|
|
)
|
|
|
|
custom_model_configurations.append(
|
|
CustomModelConfiguration(
|
|
model=provider_model_record.model_name,
|
|
model_type=provider_model_record.model_type,
|
|
credentials=provider_model_credentials,
|
|
current_credential_id=provider_model_record.credential_id,
|
|
current_credential_name=provider_model_record.credential_name,
|
|
available_model_credentials=available_model_credentials,
|
|
)
|
|
)
|
|
|
|
# Add models that can be added
|
|
for model in can_added_models:
|
|
custom_model_configurations.append(
|
|
CustomModelConfiguration(
|
|
model=model["model"],
|
|
model_type=model["model_type"],
|
|
credentials=None,
|
|
current_credential_id=None,
|
|
current_credential_name=None,
|
|
available_model_credentials=model["available_model_credentials"],
|
|
unadded_to_model_list=True,
|
|
)
|
|
)
|
|
|
|
return custom_model_configurations
|
|
|
|
def _get_and_decrypt_credentials(
|
|
self,
|
|
tenant_id: str,
|
|
record_id: str,
|
|
encrypted_config: str | None,
|
|
secret_variables: list[str],
|
|
cache_type: ProviderCredentialsCacheType,
|
|
is_provider: bool = False,
|
|
) -> dict[str, Any]:
|
|
"""Get and decrypt credentials with caching."""
|
|
credentials_cache = ProviderCredentialsCache(
|
|
tenant_id=tenant_id,
|
|
identity_id=record_id,
|
|
cache_type=cache_type,
|
|
)
|
|
|
|
# Try to get from cache first
|
|
cached_credentials = credentials_cache.get()
|
|
if cached_credentials:
|
|
return cached_credentials
|
|
|
|
# Parse encrypted config
|
|
if not encrypted_config:
|
|
return {}
|
|
|
|
if is_provider and not encrypted_config.startswith("{"):
|
|
return {"openai_api_key": encrypted_config}
|
|
|
|
try:
|
|
credentials = _credentials_adapter.validate_json(encrypted_config)
|
|
except (ValueError, JSONDecodeError):
|
|
return {}
|
|
|
|
# Decrypt secret variables
|
|
decoding_context = self._get_decoding_context(tenant_id)
|
|
|
|
for variable in secret_variables:
|
|
if variable in credentials:
|
|
with contextlib.suppress(ValueError):
|
|
credentials[variable] = encrypter.decrypt_token_with_decoding(
|
|
credentials.get(variable) or "",
|
|
decoding_context,
|
|
)
|
|
|
|
# Cache the decrypted credentials
|
|
credentials_cache.set(credentials=credentials)
|
|
return credentials
|
|
|
|
def _to_system_configuration(
|
|
self, tenant_id: str, provider_entity: ProviderEntity, provider_records: list[Provider]
|
|
) -> SystemConfiguration:
|
|
"""
|
|
Convert to system configuration.
|
|
|
|
:param tenant_id: workspace id
|
|
:param provider_entity: provider entity
|
|
:param provider_records: provider records
|
|
:return:
|
|
"""
|
|
# Get hosting configuration
|
|
hosting_configuration = ext_hosting_provider.hosting_configuration
|
|
|
|
provider_hosting_configuration = hosting_configuration.provider_map.get(provider_entity.provider)
|
|
if provider_hosting_configuration is None or not provider_hosting_configuration.enabled:
|
|
return SystemConfiguration(enabled=False)
|
|
|
|
try:
|
|
plugin_provider_entity = PluginModelProviderDeclaration.model_validate(provider_entity)
|
|
except ValidationError:
|
|
return SystemConfiguration(enabled=False)
|
|
|
|
if plugin_provider_entity.installation_source in (None, PluginInstallationSource.Package):
|
|
return SystemConfiguration(enabled=False)
|
|
|
|
from core.plugin.plugin_service import PluginService
|
|
|
|
if not PluginService.is_plugin_verified(tenant_id, plugin_provider_entity.plugin_unique_identifier):
|
|
return SystemConfiguration(enabled=False)
|
|
|
|
# Convert provider_records to dict
|
|
quota_type_to_provider_records_dict: dict[ProviderQuotaType, Provider] = {}
|
|
for provider_record in provider_records:
|
|
if provider_record.provider_type != ProviderType.SYSTEM:
|
|
continue
|
|
|
|
if provider_record.quota_type is not None:
|
|
quota_type_to_provider_records_dict[provider_record.quota_type] = provider_record # type: ignore[index]
|
|
quota_configurations = []
|
|
|
|
if dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD:
|
|
from services.credit_pool_service import CreditPoolService
|
|
|
|
with session_factory.create_session() as session:
|
|
trail_pool = CreditPoolService.get_pool(
|
|
tenant_id=tenant_id,
|
|
pool_type=ProviderQuotaType.TRIAL,
|
|
session=session,
|
|
)
|
|
paid_pool = CreditPoolService.get_pool(
|
|
tenant_id=tenant_id,
|
|
pool_type=ProviderQuotaType.PAID,
|
|
session=session,
|
|
)
|
|
else:
|
|
trail_pool = None
|
|
paid_pool = None
|
|
|
|
for provider_quota in provider_hosting_configuration.quotas:
|
|
if provider_quota.quota_type not in quota_type_to_provider_records_dict:
|
|
if provider_quota.quota_type == ProviderQuotaType.FREE:
|
|
quota_configuration = QuotaConfiguration(
|
|
quota_type=provider_quota.quota_type,
|
|
quota_unit=provider_hosting_configuration.quota_unit or QuotaUnit.TOKENS,
|
|
quota_used=0,
|
|
quota_limit=0,
|
|
is_valid=False,
|
|
restrict_models=provider_quota.restrict_models,
|
|
)
|
|
else:
|
|
continue
|
|
else:
|
|
provider_record = quota_type_to_provider_records_dict[provider_quota.quota_type]
|
|
|
|
if provider_record.quota_used is None:
|
|
raise ValueError("quota_used is None")
|
|
if provider_record.quota_limit is None:
|
|
raise ValueError("quota_limit is None")
|
|
match provider_quota.quota_type:
|
|
case ProviderQuotaType.TRIAL if trail_pool is not None:
|
|
quota_configuration = QuotaConfiguration(
|
|
quota_type=provider_quota.quota_type,
|
|
quota_unit=provider_hosting_configuration.quota_unit or QuotaUnit.TOKENS,
|
|
quota_used=trail_pool.quota_used,
|
|
quota_limit=trail_pool.quota_limit,
|
|
is_valid=trail_pool.quota_limit > trail_pool.quota_used or trail_pool.quota_limit == -1,
|
|
restrict_models=provider_quota.restrict_models,
|
|
)
|
|
|
|
case ProviderQuotaType.PAID if paid_pool is not None:
|
|
quota_configuration = QuotaConfiguration(
|
|
quota_type=provider_quota.quota_type,
|
|
quota_unit=provider_hosting_configuration.quota_unit or QuotaUnit.TOKENS,
|
|
quota_used=paid_pool.quota_used,
|
|
quota_limit=paid_pool.quota_limit,
|
|
is_valid=paid_pool.quota_limit > paid_pool.quota_used or paid_pool.quota_limit == -1,
|
|
restrict_models=provider_quota.restrict_models,
|
|
)
|
|
|
|
case _:
|
|
quota_configuration = QuotaConfiguration(
|
|
quota_type=provider_quota.quota_type,
|
|
quota_unit=provider_hosting_configuration.quota_unit or QuotaUnit.TOKENS,
|
|
quota_used=provider_record.quota_used,
|
|
quota_limit=provider_record.quota_limit,
|
|
is_valid=provider_record.quota_limit > provider_record.quota_used
|
|
or provider_record.quota_limit == -1,
|
|
restrict_models=provider_quota.restrict_models,
|
|
)
|
|
|
|
quota_configurations.append(quota_configuration)
|
|
|
|
if len(quota_configurations) == 0:
|
|
return SystemConfiguration(enabled=False)
|
|
|
|
current_quota_type = self._choice_current_using_quota_type(quota_configurations)
|
|
|
|
current_using_credentials = provider_hosting_configuration.credentials
|
|
if current_quota_type == ProviderQuotaType.FREE:
|
|
provider_record_quota_free = quota_type_to_provider_records_dict.get(current_quota_type)
|
|
|
|
if provider_record_quota_free:
|
|
provider_credentials_cache = ProviderCredentialsCache(
|
|
tenant_id=tenant_id,
|
|
identity_id=provider_record_quota_free.id,
|
|
cache_type=ProviderCredentialsCacheType.PROVIDER,
|
|
)
|
|
|
|
# Get cached provider credentials
|
|
# error occurs
|
|
cached_provider_credentials = provider_credentials_cache.get()
|
|
|
|
if not cached_provider_credentials:
|
|
provider_credentials: dict[str, Any] = {}
|
|
if provider_records and provider_records[0].encrypted_config:
|
|
provider_credentials = _credentials_adapter.validate_json(provider_records[0].encrypted_config)
|
|
|
|
# Get provider credential secret variables
|
|
provider_credential_secret_variables = self._extract_secret_variables(
|
|
provider_entity.provider_credential_schema.credential_form_schemas
|
|
if provider_entity.provider_credential_schema
|
|
else []
|
|
)
|
|
|
|
# Get decoding context for decrypting credentials
|
|
decoding_context = self._get_decoding_context(tenant_id)
|
|
|
|
for variable in provider_credential_secret_variables:
|
|
if variable in provider_credentials:
|
|
try:
|
|
provider_credentials[variable] = encrypter.decrypt_token_with_decoding(
|
|
provider_credentials.get(variable, ""),
|
|
decoding_context,
|
|
)
|
|
except ValueError:
|
|
pass
|
|
|
|
current_using_credentials = provider_credentials or {}
|
|
|
|
# cache provider credentials
|
|
provider_credentials_cache.set(credentials=current_using_credentials)
|
|
else:
|
|
current_using_credentials = cached_provider_credentials
|
|
else:
|
|
current_using_credentials = {}
|
|
quota_configurations = []
|
|
|
|
return SystemConfiguration(
|
|
enabled=True,
|
|
current_quota_type=current_quota_type,
|
|
quota_configurations=quota_configurations,
|
|
credentials=current_using_credentials,
|
|
)
|
|
|
|
@staticmethod
|
|
def _choice_current_using_quota_type(quota_configurations: list[QuotaConfiguration]) -> ProviderQuotaType:
|
|
"""
|
|
Choice current using quota type.
|
|
paid quotas > provider free quotas > hosting trial quotas
|
|
If there is still quota for the corresponding quota type according to the sorting,
|
|
|
|
:param quota_configurations:
|
|
:return:
|
|
"""
|
|
# convert to dict
|
|
quota_type_to_quota_configuration_dict = {
|
|
quota_configuration.quota_type: quota_configuration for quota_configuration in quota_configurations
|
|
}
|
|
|
|
last_quota_configuration = None
|
|
for quota_type in [ProviderQuotaType.PAID, ProviderQuotaType.FREE, ProviderQuotaType.TRIAL]:
|
|
if quota_type in quota_type_to_quota_configuration_dict:
|
|
last_quota_configuration = quota_type_to_quota_configuration_dict[quota_type]
|
|
if last_quota_configuration.is_valid:
|
|
return quota_type
|
|
|
|
if last_quota_configuration:
|
|
return last_quota_configuration.quota_type
|
|
|
|
raise ValueError("No quota type available")
|
|
|
|
@staticmethod
|
|
def _extract_secret_variables(credential_form_schemas: list[CredentialFormSchema]) -> list[str]:
|
|
"""
|
|
Extract secret input form variables.
|
|
|
|
:param credential_form_schemas:
|
|
:return:
|
|
"""
|
|
secret_input_form_variables = []
|
|
for credential_form_schema in credential_form_schemas:
|
|
if credential_form_schema.type == FormType.SECRET_INPUT:
|
|
secret_input_form_variables.append(credential_form_schema.variable)
|
|
|
|
return secret_input_form_variables
|
|
|
|
def _to_model_settings(
|
|
self,
|
|
provider_entity: ProviderEntity,
|
|
provider_model_settings: list[_ProviderModelSettingCacheEntry] | None = None,
|
|
load_balancing_model_configs: list[_LoadBalancingModelConfigCacheEntry] | None = None,
|
|
) -> list[ModelSettings]:
|
|
"""
|
|
Convert to model settings.
|
|
:param provider_entity: provider entity
|
|
:param provider_model_settings: provider model settings include enabled, load balancing enabled
|
|
:param load_balancing_model_configs: load balancing model configs
|
|
:return:
|
|
"""
|
|
# Get provider model credential secret variables
|
|
if ConfigurateMethod.PREDEFINED_MODEL in provider_entity.configurate_methods:
|
|
model_credential_secret_variables = self._extract_secret_variables(
|
|
provider_entity.provider_credential_schema.credential_form_schemas
|
|
if provider_entity.provider_credential_schema
|
|
else []
|
|
)
|
|
else:
|
|
model_credential_secret_variables = self._extract_secret_variables(
|
|
provider_entity.model_credential_schema.credential_form_schemas
|
|
if provider_entity.model_credential_schema
|
|
else []
|
|
)
|
|
|
|
model_settings: list[ModelSettings] = []
|
|
if not provider_model_settings:
|
|
return model_settings
|
|
|
|
for provider_model_setting in provider_model_settings:
|
|
load_balancing_configs = []
|
|
if provider_model_setting.load_balancing_enabled and load_balancing_model_configs:
|
|
for load_balancing_model_config in load_balancing_model_configs:
|
|
if (
|
|
load_balancing_model_config.model_name == provider_model_setting.model_name
|
|
and load_balancing_model_config.model_type == provider_model_setting.model_type
|
|
):
|
|
if not load_balancing_model_config.enabled:
|
|
continue
|
|
|
|
if not load_balancing_model_config.encrypted_config:
|
|
if load_balancing_model_config.name == "__inherit__":
|
|
load_balancing_configs.append(
|
|
ModelLoadBalancingConfiguration(
|
|
id=load_balancing_model_config.id,
|
|
name=load_balancing_model_config.name,
|
|
credentials={},
|
|
)
|
|
)
|
|
continue
|
|
|
|
provider_model_credentials_cache = ProviderCredentialsCache(
|
|
tenant_id=load_balancing_model_config.tenant_id,
|
|
identity_id=load_balancing_model_config.id,
|
|
cache_type=ProviderCredentialsCacheType.LOAD_BALANCING_MODEL,
|
|
)
|
|
|
|
# Get cached provider model credentials
|
|
cached_provider_model_credentials = provider_model_credentials_cache.get()
|
|
|
|
if not cached_provider_model_credentials:
|
|
try:
|
|
provider_model_credentials = _credentials_adapter.validate_json(
|
|
load_balancing_model_config.encrypted_config
|
|
)
|
|
except (ValueError, JSONDecodeError):
|
|
continue
|
|
|
|
# Get decoding context for decrypting credentials
|
|
decoding_context = self._get_decoding_context(load_balancing_model_config.tenant_id)
|
|
|
|
for variable in model_credential_secret_variables:
|
|
if variable in provider_model_credentials:
|
|
try:
|
|
provider_model_credentials[variable] = encrypter.decrypt_token_with_decoding(
|
|
provider_model_credentials.get(variable) or "",
|
|
decoding_context,
|
|
)
|
|
except ValueError:
|
|
pass
|
|
|
|
# cache provider model credentials
|
|
provider_model_credentials_cache.set(credentials=provider_model_credentials)
|
|
else:
|
|
provider_model_credentials = cached_provider_model_credentials
|
|
|
|
load_balancing_configs.append(
|
|
ModelLoadBalancingConfiguration(
|
|
id=load_balancing_model_config.id,
|
|
name=load_balancing_model_config.name,
|
|
credentials=provider_model_credentials,
|
|
credential_source_type=load_balancing_model_config.credential_source_type,
|
|
credential_id=load_balancing_model_config.credential_id,
|
|
)
|
|
)
|
|
|
|
model_settings.append(
|
|
ModelSettings(
|
|
model=provider_model_setting.model_name,
|
|
model_type=provider_model_setting.model_type,
|
|
enabled=provider_model_setting.enabled,
|
|
load_balancing_enabled=provider_model_setting.load_balancing_enabled,
|
|
load_balancing_configs=load_balancing_configs if len(load_balancing_configs) > 1 else [],
|
|
)
|
|
)
|
|
|
|
return model_settings
|