# ========= Copyright 2023-2026 @ CAMEL-AI.org. All Rights Reserved. ========= # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # ========= Copyright 2023-2026 @ CAMEL-AI.org. All Rights Reserved. ========= import json import os from typing import Any, ClassVar, Dict, Optional, Type, Union from camel.models.aihubmix_model import AihubMixModel from camel.models.aiml_model import AIMLModel from camel.models.amd_model import AMDModel from camel.models.anthropic_model import AnthropicModel from camel.models.atlascloud_model import AtlasCloudModel from camel.models.avian_model import AvianModel from camel.models.aws_bedrock_model import AWSBedrockModel from camel.models.azure_openai_model import AzureOpenAIModel from camel.models.base_model import BaseModelBackend from camel.models.cerebras_model import CerebrasModel from camel.models.cohere_model import CohereModel from camel.models.cometapi_model import CometAPIModel from camel.models.crynux_model import CrynuxModel from camel.models.deepseek_model import DeepSeekModel from camel.models.function_gemma_model import FunctionGemmaModel from camel.models.gemini_model import GeminiModel from camel.models.groq_model import GroqModel from camel.models.internlm_model import InternLMModel from camel.models.litellm_model import LiteLLMModel from camel.models.lmstudio_model import LMStudioModel from camel.models.minimax_model import MinimaxModel from camel.models.mistral_model import MistralModel from camel.models.modelscope_model import ModelScopeModel from camel.models.moonshot_model import MoonshotModel from camel.models.nebius_model import NebiusModel from camel.models.netmind_model import NetmindModel from camel.models.novita_model import NovitaModel from camel.models.nvidia_model import NvidiaModel from camel.models.ollama_model import OllamaModel from camel.models.openai_compatible_model import OpenAICompatibleModel from camel.models.openai_model import OpenAIModel from camel.models.openrouter_model import OpenRouterModel from camel.models.ppio_model import PPIOModel from camel.models.qianfan_model import QianfanModel from camel.models.qwen_model import QwenModel from camel.models.reka_model import RekaModel from camel.models.samba_model import SambaModel from camel.models.sglang_model import SGLangModel from camel.models.siliconflow_model import SiliconFlowModel from camel.models.stub_model import StubModel from camel.models.togetherai_model import TogetherAIModel from camel.models.vllm_model import VLLMModel from camel.models.volcano_model import VolcanoModel from camel.models.watsonx_model import WatsonXModel from camel.models.yi_model import YiModel from camel.models.zhipuai_model import ZhipuAIModel from camel.types import ModelPlatformType, ModelType, UnifiedModelType from camel.utils import BaseTokenCounter, configure_langfuse class ModelFactory: r"""Factory of backend models. Raises: ValueError: in case the provided model type is unknown. """ _MODEL_PLATFORM_TO_CLASS_MAP: ClassVar[ Dict[ModelPlatformType, Type[BaseModelBackend]] ] = { ModelPlatformType.OLLAMA: OllamaModel, ModelPlatformType.VLLM: VLLMModel, ModelPlatformType.SGLANG: SGLangModel, ModelPlatformType.OPENAI_COMPATIBLE_MODEL: OpenAICompatibleModel, ModelPlatformType.SAMBA: SambaModel, ModelPlatformType.TOGETHER: TogetherAIModel, ModelPlatformType.LITELLM: LiteLLMModel, ModelPlatformType.AWS_BEDROCK: AWSBedrockModel, ModelPlatformType.NVIDIA: NvidiaModel, ModelPlatformType.SILICONFLOW: SiliconFlowModel, ModelPlatformType.AMD: AMDModel, ModelPlatformType.AIML: AIMLModel, ModelPlatformType.VOLCANO: VolcanoModel, ModelPlatformType.NETMIND: NetmindModel, ModelPlatformType.OPENAI: OpenAIModel, ModelPlatformType.AZURE: AzureOpenAIModel, ModelPlatformType.ANTHROPIC: AnthropicModel, ModelPlatformType.GROQ: GroqModel, ModelPlatformType.CEREBRAS: CerebrasModel, ModelPlatformType.COMETAPI: CometAPIModel, ModelPlatformType.NEBIUS: NebiusModel, ModelPlatformType.LMSTUDIO: LMStudioModel, ModelPlatformType.MINIMAX: MinimaxModel, ModelPlatformType.OPENROUTER: OpenRouterModel, ModelPlatformType.ATLASCLOUD: AtlasCloudModel, ModelPlatformType.ZHIPU: ZhipuAIModel, ModelPlatformType.GEMINI: GeminiModel, ModelPlatformType.MISTRAL: MistralModel, ModelPlatformType.REKA: RekaModel, ModelPlatformType.COHERE: CohereModel, ModelPlatformType.YI: YiModel, ModelPlatformType.QWEN: QwenModel, ModelPlatformType.DEEPSEEK: DeepSeekModel, ModelPlatformType.PPIO: PPIOModel, ModelPlatformType.INTERNLM: InternLMModel, ModelPlatformType.MOONSHOT: MoonshotModel, ModelPlatformType.MODELSCOPE: ModelScopeModel, ModelPlatformType.NOVITA: NovitaModel, ModelPlatformType.WATSONX: WatsonXModel, ModelPlatformType.QIANFAN: QianfanModel, ModelPlatformType.CRYNUX: CrynuxModel, ModelPlatformType.AIHUBMIX: AihubMixModel, ModelPlatformType.AVIAN: AvianModel, ModelPlatformType.FUNCTION_GEMMA: FunctionGemmaModel, } @staticmethod def create( model_platform: Union[ModelPlatformType, str], model_type: Union[ModelType, str, UnifiedModelType], model_config_dict: Optional[Dict] = None, token_counter: Optional[BaseTokenCounter] = None, api_key: Optional[str] = None, url: Optional[str] = None, timeout: Optional[float] = None, max_retries: int = 3, client: Optional[Any] = None, async_client: Optional[Any] = None, **kwargs, ) -> BaseModelBackend: r"""Creates an instance of `BaseModelBackend` of the specified type. Args: model_platform (Union[ModelPlatformType, str]): Platform from which the model originates. Can be a string or ModelPlatformType enum. model_type (Union[ModelType, str, UnifiedModelType]): Model for which a backend is created. Can be a string, ModelType enum, or UnifiedModelType. model_config_dict (Optional[Dict]): A dictionary that will be fed into the backend constructor. (default: :obj:`None`) token_counter (Optional[BaseTokenCounter], optional): Token counter to use for the model. If not provided, :obj:`OpenAITokenCounter(ModelType.GPT_4O_MINI)` will be used if the model platform didn't provide official token counter. (default: :obj:`None`) api_key (Optional[str], optional): The API key for authenticating with the model service. (default: :obj:`None`) url (Optional[str], optional): The url to the model service. (default: :obj:`None`) timeout (Optional[float], optional): The timeout value in seconds for API calls. (default: :obj:`None`) max_retries (int, optional): Maximum number of retries for API calls. (default: :obj:`3`) client (Optional[Any], optional): A custom synchronous client instance. Supported by models that use OpenAI-compatible APIs . The client should implement the appropriate client interface for the platform. (default: :obj:`None`) async_client (Optional[Any], optional): A custom asynchronous client instance. Supported by models that use OpenAI-compatible APIs. The client should implement the appropriate async client interface for the platform. (default: :obj:`None`) **kwargs: Additional model-specific parameters that will be passed to the model constructor. For example, Azure OpenAI models may require `api_version`, `azure_deployment_name`, `azure_ad_token_provider`, and `azure_ad_token`. Returns: BaseModelBackend: The initialized backend. Raises: ValueError: If there is no backend for the model. """ # Auto-configure Langfuse only if explicitly enabled env_enabled_str = os.environ.get("LANGFUSE_ENABLED") if env_enabled_str and env_enabled_str.lower() == "true": configure_langfuse() # Convert string to ModelPlatformType enum if needed if isinstance(model_platform, str): try: model_platform = ModelPlatformType(model_platform) except ValueError: raise ValueError(f"Unknown model platform: {model_platform}") # Convert string to ModelType enum or UnifiedModelType if needed if isinstance(model_type, str): try: model_type = ModelType(model_type) except ValueError: # If not in ModelType, create a UnifiedModelType model_type = UnifiedModelType(model_type) model_class: Optional[Type[BaseModelBackend]] = None model_type = UnifiedModelType(model_type) model_class = ModelFactory._MODEL_PLATFORM_TO_CLASS_MAP.get( model_platform ) if model_type == ModelType.STUB: model_class = StubModel if model_class is None: raise ValueError(f"Unknown model platform `{model_platform}`") # Pass client and async_client via kwargs if provided if client is not None: kwargs['client'] = client if async_client is not None: kwargs['async_client'] = async_client return model_class( model_type=model_type, model_config_dict=model_config_dict, api_key=api_key, url=url, token_counter=token_counter, timeout=timeout, max_retries=max_retries, **kwargs, ) @classmethod def __parse_model_platform( cls, model_platform_str: str ) -> ModelPlatformType: r"""Parses a string and returns the corresponding ModelPlatformType enum. Args: model_platform_str (str): The platform name as a string. Can be in the form "ModelPlatformType." or simply "". Returns: ModelPlatformType: The matching enum value. Raises: ValueError: If the platform name is not a valid member of ModelPlatformType. """ try: if model_platform_str.startswith("ModelPlatformType."): platform_name = model_platform_str.split('.')[-1] else: platform_name = model_platform_str.upper() if platform_name not in ModelPlatformType.__members__: raise ValueError( f"Invalid model platform: {platform_name}. " f"Valid options: " f"{', '.join(ModelPlatformType.__members__.keys())}" ) return ModelPlatformType[platform_name] except KeyError: raise KeyError(f"Invalid model platform: {model_platform_str}") @classmethod def __load_yaml(cls, filepath: str) -> Dict: import yaml r"""Loads and parses a YAML file into a dictionary. Args: filepath (str): Path to the YAML configuration file. Returns: Dict: The parsed YAML content as a dictionary. """ with open(filepath, 'r') as file: config = yaml.safe_load(file) return config @classmethod def __load_json(cls, filepath: str) -> Dict: r"""Loads and parses a JSON file into a dictionary. Args: filepath (str): Path to the JSON configuration file. Returns: Dict: The parsed JSON content as a dictionary. """ with open(filepath, 'r') as file: config = json.load(file) return config @classmethod def create_from_yaml(cls, filepath: str) -> BaseModelBackend: r"""Creates and returns a model base backend instance from a YAML configuration file. Args: filepath (str): Path to the YAML file containing model configuration. Returns: BaseModelBackend: An instance of the model backend based on the configuration. """ config = cls.__load_yaml(filepath) config["model_platform"] = cls.__parse_model_platform( config["model_platform"] ) model = ModelFactory.create(**config) return model @classmethod def create_from_json(cls, filepath: str) -> BaseModelBackend: r"""Creates and returns a base model backend instance from a JSON configuration file. Args: filepath (str): Path to the JSON file containing model configuration. Returns: BaseModelBackend: An instance of the model backend based on the configuration. """ config = cls.__load_json(filepath) config["model_platform"] = cls.__parse_model_platform( config["model_platform"] ) model = ModelFactory.create(**config) return model