RagFlow_Plugs/conf/config.py

"""配置管理模块，从.env文件加载环境变量"""

import os
from dotenv import load_dotenv

# 加载.env文件中的环境变量
load_dotenv()

class ModelConfig:
    """模型配置类"""
    
    @staticmethod
    def get_model_provider() -> str:
        """获取模型提供商"""
        return os.getenv("MODEL_PROVIDER", "openai")
    
    @staticmethod
    def get_model_name() -> str:
        """获取模型名称"""
        return os.getenv("MODEL_NAME", "Qwen/Qwen3-VL-8B-Instruct")

    @staticmethod
    def get_embedding_model_name() -> str:
        """获取模型名称"""
        return os.getenv("EMBEDDING_MODEL_NAME", "Qwen/Qwen3-Embedding-0.6B")
    
    @staticmethod
    def get_base_url() -> str:
        """获取模型API基础URL"""
        return os.getenv("BASE_URL", "https://api.openai.com/v1")
    
    @staticmethod
    def get_api_key() -> str:
        """获取模型API密钥"""
        return os.getenv("API_KEY", "")
    
    @staticmethod
    def get_model_config() -> dict:
        """获取完整的模型配置"""
        return {
            "model_provider": ModelConfig.get_model_provider(),
            "model": ModelConfig.get_model_name(),
            "base_url": ModelConfig.get_base_url(),
            "api_key": ModelConfig.get_api_key()
        }

    @staticmethod
    def get_embedding_model_config() -> dict:
        """获取完整的模型配置"""
        return {
            "model_provider": ModelConfig.get_model_provider(),
            "model": ModelConfig.get_embedding_model_name(),
            "base_url": ModelConfig.get_base_url(),
            "api_key": ModelConfig.get_api_key()
        }

    @staticmethod
    def get_multimodal_embedding_model_name() -> str:
        """获取多模态嵌入模型名称"""
        return os.getenv("MULTIMODAL_EMBEDDING_MODEL_NAME", "qwen2.5-vl-embedding")
    
    @staticmethod
    def get_dashscope_api_key() -> str:
        """获取DASHSCOPE API密钥"""
        return os.getenv("DASHSCOPE", "")

    # RAGFLOW配置
    @staticmethod
    def get_ragflow_api_url() -> str:
        """获取RAGFLOW API基础URL"""
        return os.getenv("RAGFLOW_API_URL", "http://192.168.16.134/")
    
    @staticmethod
    def get_ragflow_api_key() -> str:
        """获取RAGFLOW API密钥"""
        return os.getenv("RAGFLOW_API_KEY", "")
    
    @staticmethod
    def get_dataset_id() -> str:
        """获取数据集ID"""
        return os.getenv("DATASET_ID", "")

class AppConfig:
    """应用配置类"""
    
    @staticmethod
    def get_log_level() -> str:
        """获取日志级别"""
        return os.getenv("LOG_LEVEL", "INFO")

class MinioConfig:
    """MinIO配置类"""
    
    @staticmethod
    def get_minio_endpoint() -> str:
        """获取MinIO端点"""
        return os.getenv("MINIO_ENDPOINT", "http://localhost:9000")
    
    @staticmethod
    def get_minio_access_key() -> str:
        """获取MinIO访问密钥"""
        return os.getenv("MINIO_ACCESS_KEY", "minioadmin")
    
    @staticmethod
    def get_minio_secret_key() -> str:
        """获取MinIO密钥"""
        return os.getenv("MINIO_SECRET_KEY", "minioadmin")
    
    @staticmethod
    def get_minio_bucket_name() -> str:
        """获取MinIO存储桶名称"""
        return os.getenv("MINIO_BUCKET_NAME", "ragflow")

    @staticmethod
    def get_minio_secure() -> bool:
        """获取MinIO是否使用HTTPS"""
        return os.getenv("MINIO_SECURE", "False")

class VectorDBConfig:
    """向量数据库配置类"""
    
    @staticmethod
    def get_vector_db_type() -> str:
        """获取向量数据库类型"""
        return os.getenv("VECTOR_DB_TYPE", "es")
    
    @staticmethod
    def get_infinity_host() -> str:
        """获取Infinity向量数据库主机"""
        return os.getenv("INFINITY_HOST", "localhost")
    
    @staticmethod
    def get_infinity_port() -> int:
        """获取Infinity向量数据库端口"""
        return int(os.getenv("INFINITY_PORT", "23810"))
    
    @staticmethod
    def get_infinity_user() -> str:
        """获取Infinity向量数据库用户名"""
        return os.getenv("INFINITY_USER", "admin")
    
    @staticmethod
    def get_infinity_password() -> str:
        """获取Infinity向量数据库密码"""
        return os.getenv("INFINITY_PASSWORD", "admin")

    @staticmethod
    def get_infinity_database() -> str:
        """获取Infinity向量数据库名称"""
        return os.getenv("INFINITY_DATABASE", "test")

    @staticmethod
    def get_infinity_table_name() -> str:
        """获取Infinity向量数据库表名"""
        return os.getenv("INFINITY_TABLE_NAME", "test")

# 导出配置实例
model_config = ModelConfig.get_model_config()
app_config = {
    "log_level": AppConfig.get_log_level()
}

vector_db_config = {
    "type": VectorDBConfig.get_vector_db_type(),
    "infinity": {
        "host": VectorDBConfig.get_infinity_host(),
        "port": VectorDBConfig.get_infinity_port(),
        "database": VectorDBConfig.get_infinity_database(),
        "user": VectorDBConfig.get_infinity_user(),
        "password": VectorDBConfig.get_infinity_password()
    }
}