diff --git a/api/terraform/python/openai_api/lambda_openai_function/custom_config.py b/api/terraform/python/openai_api/lambda_openai_function/custom_config.py index eb94ab18..b25a9a54 100644 --- a/api/terraform/python/openai_api/lambda_openai_function/custom_config.py +++ b/api/terraform/python/openai_api/lambda_openai_function/custom_config.py @@ -1,4 +1,5 @@ # -*- coding: utf-8 -*- +# pylint: disable=E1101 """ This module contains the CustomConfig class, which is used to parse YAML config objects for function_refers_to.get_additional_info(). @@ -6,194 +7,163 @@ import json import logging import os +from typing import Optional import yaml from openai_api.common.conf import settings from openai_api.common.const import PYTHON_ROOT +from pydantic import BaseModel, Field, ValidationError, field_validator, root_validator log = logging.getLogger(__name__) CONFIG_PATH = PYTHON_ROOT + "/openai_api/lambda_openai_function/config/" -class CustomConfigBase: +def do_error(class_name: str, err: str) -> None: + """Print the error message and raise a ValueError""" + err = f"{class_name} - {err}" + print(err) + log.error(err) + raise ValueError(err) + + +def validate_required_keys(class_name: str, required_keys: list, config_json: dict) -> None: + """Validate the required keys""" + for key in required_keys: + if key not in config_json: + do_error(class_name, err=f"Invalid search_terms: {config_json}. Missing key: {key}.") + + +class CustomConfigBase(BaseModel): """Base class for CustomConfig and CustomConfigs""" - def __init__(self) -> None: - pass + class Config: + """Pydantic config""" + + extra = "forbid" # this will forbid any extra attributes during model initialization def __repr__(self): return f"{self.__class__.__name__}({self.__dict__})" def __str__(self): - return f"{self.__dict__}" + return f"{self.to_json()}" - def do_error(self, err: str) -> None: - """Print the error message and raise a ValueError""" - err = f"{self.__class__.__name__} - {err}" - print(err) - log.error(err) - raise ValueError(err) + def to_json(self) -> json: + """Return the config as a JSON object""" + raise NotImplementedError class SystemPrompt(CustomConfigBase): """System prompt of a CustomConfig object""" - config: str = None + system_prompt: str = Field(..., description="System prompt") - def __init__(self, system_prompt=None): - super().__init__() - system_prompt = system_prompt or "" - self.config = system_prompt - self.validate() + @field_validator("system_prompt") + @classmethod + def validate_system_prompt(cls, system_prompt) -> str: + """Validate the system_prompt field""" + if not isinstance(system_prompt, str): + do_error(class_name=cls.__name__, err=f"Expected a string but received {type(system_prompt)}") + return system_prompt - @property - def is_valid(self) -> bool: - """Return True if the config object is valid""" - try: - self.validate() - return True - except ValueError: - return False + def to_json(self) -> json: + """Return the config as a JSON object""" + return self.system_prompt - def validate(self) -> None: - """Validate the config object""" - if not isinstance(self.system_prompt, str): - self.do_error(f"Expected a string but received {type(self.system_prompt)}") - @property - def system_prompt(self) -> str: - """Return the system prompt""" - return self.config +class SearchTerms(CustomConfigBase): + """Search terms of a CustomConfig object""" - def __str__(self): - return f"{self.config}" + config_json: dict = Field(..., description="Config object") + @field_validator("config_json") + @classmethod + def validate_config_json(cls, config_json) -> dict: + """Validate the config_json object""" + if not isinstance(config_json, dict): + do_error(class_name=cls.__name__, err=f"Expected a dict but received {type(config_json)}") -class SearchTerms(CustomConfigBase): - """Search terms of a CustomConfig object""" + required_keys = ["strings", "pairs"] + validate_required_keys(class_name=cls.__name__, required_keys=required_keys, config_json=config_json) - def __init__(self, config_json: dict = None): - super().__init__() - self.config_json = config_json - self.validate() + strings = config_json["strings"] + pairs = config_json["pairs"] + if not all(isinstance(item, str) for item in strings): + do_error( + class_name=cls.__name__, err=f"Invalid config object: {strings}. 'strings' should be a list of strings." + ) + + if not all( + isinstance(pair, list) and len(pair) == 2 and all(isinstance(item, str) for item in pair) for pair in pairs + ): + do_error( + class_name=cls.__name__, + err=f"Invalid config object: {config_json}. 'pairs' should be a list of pairs of strings.", + ) + return config_json @property def strings(self) -> list: """Return a list of search terms""" - return self.config_json["strings"] + return self.config_json.get("strings") @property def pairs(self) -> list: """Return a list of search terms""" - return self.config_json["pairs"] - - @property - def is_valid(self) -> bool: - """Return True if the config object is valid""" - try: - self.validate() - return True - except ValueError: - return False - - def validate(self) -> None: - """Validate the config object""" - if not isinstance(self.config_json, dict): - self.do_error(f"Expected a dict but received {type(self.config_json)}") - - required_keys = ["strings", "pairs"] - for key in required_keys: - if key not in self.config_json: - self.do_error(f"Invalid search_terms: {self.config_json}. Missing key: {key}.") - - if not all(isinstance(item, str) for item in self.strings): - self.do_error(f"Invalid config object: {self.config_json}. 'strings' should be a list of strings.") - - if not all( - isinstance(pair, list) and len(pair) == 2 and all(isinstance(item, str) for item in pair) - for pair in self.pairs - ): - self.do_error(f"Invalid config object: {self.config_json}. 'pairs' should be a list of pairs of strings.") + return self.config_json.get("pairs") def to_json(self) -> json: """Return the config as a JSON object""" return self.config_json - def __str__(self): - """Return the config object as a string""" - return f"{self.to_json()}" - class AdditionalInformation(CustomConfigBase): """Additional information of a CustomConfig object""" - config_json: dict = None + config_json: dict = Field(..., description="Config object") - def __init__(self, config_json: dict = None): - super().__init__() - self.config_json = config_json - self.validate() + @field_validator("config_json") + @classmethod + def validate_config_json(cls, config_json) -> dict: + """Validate the config object""" + if not isinstance(config_json, dict): + do_error(class_name=cls.__name__, err=f"Expected a dict but received {type(config_json)}") + return config_json @property def keys(self) -> list: """Return a list of keys for additional information""" return list(self.config_json.keys()) - @property - def is_valid(self) -> bool: - """Return True if the config object is valid""" - try: - self.validate() - return True - except ValueError: - return False - - def validate(self) -> None: - """Validate the config object""" - if not isinstance(self.config_json, dict): - self.do_error(f"Expected a dict but received {type(self.config_json)}") - def to_json(self) -> json: """Return the config as a JSON object""" return self.config_json - def __str__(self): - """Return the config object as a string""" - return f"{self.to_json()}" - class Prompting(CustomConfigBase): """Prompting child class of a CustomConfig object""" - config_json: dict = None - search_terms: SearchTerms = None - system_prompt: SystemPrompt = None - - def __init__(self, config_json: dict = None): - super().__init__() - self.config_json = config_json - self.validate() - self.search_terms = SearchTerms(config_json=self.config_json["search_terms"]) - self.system_prompt = SystemPrompt(system_prompt=self.config_json["system_prompt"]) - - @property - def is_valid(self) -> bool: - """Return True if the config object is valid""" - try: - self.validate() - except ValueError: - return False - return True - - def validate(self): + config_json: dict = Field(..., description="Config object") + search_terms: SearchTerms = Field(None, description="Search terms of the config object") + system_prompt: SystemPrompt = Field(None, description="System prompt of the config object") + + @root_validator(pre=True) + def set_fields(cls, values): + """proxy for __init__() - Set the fields""" + config_json = values.get("config_json") + if not isinstance(config_json, dict): + raise ValueError(f"Expected config_json to be a dict but received {type(config_json)}") + if config_json: + values["search_terms"] = SearchTerms(config_json=config_json["search_terms"]) + values["system_prompt"] = SystemPrompt(system_prompt=config_json["system_prompt"]) + return values + + @field_validator("config_json") + @classmethod + def validate_config_json(cls, config_json) -> dict: """Validate the config object""" - if not isinstance(self.config_json, dict): - self.do_error(f"Expected a dict but received {type(self.config_json)}") required_keys = ["search_terms", "system_prompt"] - for key in required_keys: - if key not in self.config_json: - self.do_error(f"Invalid config object: {self.config_json}. Missing key: {key}.") + validate_required_keys(class_name=cls.__name__, required_keys=required_keys, config_json=config_json) def to_json(self) -> json: """Return the config as a JSON object""" @@ -206,39 +176,40 @@ def to_json(self) -> json: class FunctionCalling(CustomConfigBase): """FunctionCalling child class of a CustomConfig""" - config_json: dict = None - function_description: str = None - additional_information: AdditionalInformation = None - - def __init__(self, config_json: dict = None): - super().__init__() - - self.config_json = config_json - self.validate() - self.function_description = self.config_json["function_description"] - self.additional_information = AdditionalInformation(config_json=self.config_json["additional_information"]) - - @property - def is_valid(self) -> bool: - """Return True if the config object is valid""" - try: - self.validate() - except ValueError: - return False - return True - - def validate(self): + config_json: dict = Field(..., description="Config object") + function_description: str = Field(None, description="Description of the function") + additional_information: AdditionalInformation = Field(None, description="Additional information of the function") + + @root_validator(pre=True) + def set_fields(cls, values): + """proxy for __init__() - Set the fields""" + config_json = values.get("config_json") + if not isinstance(config_json, dict): + raise ValueError(f"Expected config_json to be a dict but received {type(config_json)}") + if config_json: + values["function_description"] = config_json["function_description"] + values["additional_information"] = AdditionalInformation(config_json=config_json["additional_information"]) + return values + + @field_validator("config_json") + @classmethod + def validate_config_json(cls, config_json) -> dict: """Validate the config object""" - if not isinstance(self.config_json, dict): - self.do_error(f"Expected a dict but received {type(self.config_json)}") + if not isinstance(config_json, dict): + do_error(class_name=cls.__name__, err=f"Expected a dict but received {type(config_json)}") required_keys = ["function_description", "additional_information"] - for key in required_keys: - if key not in self.config_json: - self.do_error(f"Invalid config object: {self.config_json}. Missing key: {key}.") - if not isinstance(self.config_json["function_description"], str): - self.do_error(f"Invalid config object: {self.config_json}. 'function_description' should be a string.") - if not isinstance(self.config_json["additional_information"], dict): - self.do_error(f"Invalid config object: {self.config_json}. 'additional_information' should be a dict.") + validate_required_keys(class_name=cls.__name__, required_keys=required_keys, config_json=config_json) + if not isinstance(config_json["function_description"], str): + do_error( + class_name=cls.__name__, + err=f"Invalid config object: {config_json}. 'function_description' should be a string.", + ) + if not isinstance(config_json["additional_information"], dict): + do_error( + class_name=cls.__name__, + err=f"Invalid config object: {config_json}. 'additional_information' should be a dict.", + ) + return config_json def to_json(self) -> json: """Return the config as a JSON object""" @@ -251,111 +222,96 @@ def to_json(self) -> json: class MetaData(CustomConfigBase): """Metadata of a CustomConfig object""" - config_json: dict = None + config_json: dict = Field(..., description="Config object") - def __init__(self, config_json: dict = None): - super().__init__() - self.config_json = config_json - self.validate() + @root_validator(pre=True) + def set_fields(cls, values): + """proxy for __init__() - Set the fields""" + config_json = values.get("config_json") + if not isinstance(config_json, dict): + raise ValueError(f"Expected config_json to be a dict but received {type(config_json)}") + return values - @property - def is_valid(self) -> bool: - """Return True if the config object is valid""" - try: - self.validate() - return True - except ValueError: - return False + @field_validator("config_json") + @classmethod + def validate_config_json(cls, config_json) -> dict: + """Validate the config object""" + if not isinstance(config_json, dict): + do_error(class_name=cls.__name__, err=f"Expected a dict but received {type(config_json)}") + + required_keys = ["config_path", "name", "description", "version", "author"] + validate_required_keys(class_name=cls.__name__, required_keys=required_keys, config_json=config_json) + return config_json @property def name(self) -> str: """Return the name of the config object""" - return self.config_json["name"] + return self.config_json.get("name") if self.config_json else None @property def config_path(self) -> str: """Return the path of the config object""" - return self.config_json["config_path"] + return self.config_json.get("config_path") if self.config_json else None @property def description(self) -> str: """Return the description of the config object""" - return self.config_json["description"] + return self.config_json.get("description") if self.config_json else None @property def version(self) -> str: """Return the version of the config object""" - return self.config_json["version"] + return self.config_json.get("version") if self.config_json else None @property def author(self) -> str: """Return the author of the config object""" - return self.config_json["author"] - - def validate(self) -> None: - """Validate the config object""" - if not isinstance(self.config_json, dict): - self.do_error(f"Expected a dict but received {type(self.config_json)}") - - required_keys = ["config_path", "name", "description", "version", "author"] - for key in required_keys: - if key not in self.config_json: - self.do_error(f"Invalid config object: {self.config_path}. Missing key: {key}.") + return self.config_json.get("author") if self.config_json else None def to_json(self) -> json: """Return the config as a JSON object""" return self.config_json - def __str__(self): - """Return the config object as a string""" - return f"{self.to_json()}" - class CustomConfig(CustomConfigBase): - """Parse the YAML config object for a given Lambda function""" - - config_json: dict = None - meta_data: MetaData = None - prompting: Prompting = None - function_calling: FunctionCalling = None - - def __init__(self, config_json: dict = None, index: int = 0): - super().__init__() - self.index = index - self.config_json = config_json - self.validate() - self.meta_data = MetaData(config_json=self.config_json["meta_data"]) - self.prompting = Prompting(config_json=self.config_json["prompting"]) - self.function_calling = FunctionCalling(config_json=self.config_json["function_calling"]) + """A json object that contains the config for a function_refers_to.get_additional_info() function""" + + index: int = Field(0, description="Index of the config object") + config_json: dict = Field(..., description="Config object") + meta_data: Optional[MetaData] = Field(None, description="Metadata of the config object") + prompting: Optional[Prompting] = Field(None, description="Prompting of the config object") + function_calling: Optional[FunctionCalling] = Field(None, description="FunctionCalling of the config object") @property def name(self) -> str: """Return a name in the format: "WillyWonka""" return self.meta_data.name - @property - def is_valid(self) -> bool: - """Return True if the config object is valid""" - try: - self.validate() - return True - except ValueError: - pass - return False - - def validate(self) -> None: + @root_validator(pre=True) + def set_fields(cls, values): + """proxy for __init__() - Set the fields""" + config_json = values.get("config_json") + if not isinstance(config_json, dict): + raise ValueError(f"Expected config_json to be a dict but received {type(config_json)}") + if config_json: + values["meta_data"] = MetaData(config_json=config_json.get("meta_data")) + values["prompting"] = Prompting(config_json=config_json.get("prompting")) + values["function_calling"] = FunctionCalling(config_json=config_json.get("function_calling")) + return values + + @field_validator("config_json") + @classmethod + def validate_config_json(cls, config_json) -> None: """Validate the config object""" - if not isinstance(self.config_json, dict): - self.do_error(f"Expected a dict but received {type(self.config_json)}") - required_keys = ["meta_data", "prompting", "function_calling"] for key in required_keys: - if key not in self.config_json: - self.do_error(f"Invalid config object: {self.config_json}. Missing key: {key}.") - if not isinstance(self.config_json[key], dict): - self.do_error( - f"Expected a dict for {key} but received {type(self.config_json[key])}: {self.config_json[key]}" + if key not in config_json: + cls.do_error(f"Invalid config object: {config_json}. Missing key: {key}.") + if not isinstance(config_json[key], dict): + do_error( + class_name=cls.__name__, + err=f"Expected a dict for {key} but received {type(config_json[key])}: {config_json[key]}", ) def to_json(self) -> json: @@ -367,9 +323,6 @@ def to_json(self) -> json: "function_calling": self.function_calling.to_json(), } - def __str__(self): - return f"{self.to_json()}" - class CustomConfigs: """List of CustomConfig objects""" @@ -409,19 +362,19 @@ def __init__(self, config_path: str = None, aws_s3_bucket_name: str = None): if config_json: custom_config = CustomConfig(config_json=config_json, index=i) self._custom_configs.append(custom_config) - print( - f"Loaded custom configuration from AWS S3 bucket: {custom_config.name} {custom_config.meta_data.version} created by {custom_config.meta_data.author}" - ) + # print( + # f"Loaded custom configuration from AWS S3 bucket: {custom_config.name} {custom_config.meta_data.version} created by {custom_config.meta_data.author}" + # ) @property def valid_configs(self) -> list[CustomConfig]: """Return a list of valid configs""" - return [config for config in self._custom_configs if config.is_valid] + return self._custom_configs @property def invalid_configs(self) -> list[CustomConfig]: """Return a list of invalid configs""" - return [config for config in self._custom_configs if not config.is_valid] + return [] @property def aws_bucket_path(self) -> str: @@ -480,8 +433,4 @@ def custom_configs(self) -> CustomConfigs: return self._custom_configs -_custom_configs = SingletonCustomConfigs().custom_configs -config = _custom_configs.valid_configs -if len(_custom_configs.invalid_configs) > 0: - invalid_configurations = [config.name for config in _custom_configs.invalid_configs] - log.error("Invalid custom config objects: %s", invalid_configurations) +custom_configs = SingletonCustomConfigs().custom_configs.valid_configs diff --git a/api/terraform/python/openai_api/lambda_openai_function/function_refers_to.py b/api/terraform/python/openai_api/lambda_openai_function/function_refers_to.py index 344d0af4..44f39cbf 100644 --- a/api/terraform/python/openai_api/lambda_openai_function/function_refers_to.py +++ b/api/terraform/python/openai_api/lambda_openai_function/function_refers_to.py @@ -6,12 +6,11 @@ import json from openai_api.common.const import PYTHON_ROOT -from openai_api.lambda_openai_function.custom_config import CustomConfig -from openai_api.lambda_openai_function.custom_config import config as refers_to_config +from openai_api.lambda_openai_function.custom_config import CustomConfig, custom_configs from openai_api.lambda_openai_function.natural_language_processing import does_refer_to -def search_terms_are_in_messages(messages: list, search_terms: list = None, search_pairs: list = None) -> bool: +def search_terms_are_in_messages(messages: list, search_terms: list, search_pairs: list) -> bool: """ Return True the user has mentioned Lawrence McDaniel or FullStackWithLawrence at any point in the history of the conversation. @@ -54,7 +53,7 @@ def customized_prompt(config: CustomConfig, messages: list) -> list: def get_additional_info(inquiry_type: str) -> str: """Return select info from custom config object""" - for config in refers_to_config: + for config in custom_configs: try: additional_information = config.function_calling.additional_information.to_json() retval = additional_information[inquiry_type] diff --git a/api/terraform/python/openai_api/lambda_openai_function/lambda_handler.py b/api/terraform/python/openai_api/lambda_openai_function/lambda_handler.py index d57eec3d..393e4f98 100644 --- a/api/terraform/python/openai_api/lambda_openai_function/lambda_handler.py +++ b/api/terraform/python/openai_api/lambda_openai_function/lambda_handler.py @@ -38,7 +38,7 @@ validate_completion_request, validate_item, ) -from openai_api.lambda_openai_function.custom_config import config as refers_to_config +from openai_api.lambda_openai_function.custom_config import custom_configs from openai_api.lambda_openai_function.function_refers_to import ( customized_prompt, get_additional_info, @@ -77,7 +77,7 @@ def handler(event, context): request_meta_data = request_meta_data_factory(model, object_type, temperature, max_tokens, input_text) # does the prompt have anything to do with any of the search terms defined in a custom configuration? - for config in refers_to_config: + for config in custom_configs: if search_terms_are_in_messages( messages=messages, search_terms=config.prompting.search_terms.strings, diff --git a/api/terraform/python/openai_api/lambda_openai_function/tests/test_custom_config.py b/api/terraform/python/openai_api/lambda_openai_function/tests/test_custom_config.py index 87dc74f6..8522142b 100644 --- a/api/terraform/python/openai_api/lambda_openai_function/tests/test_custom_config.py +++ b/api/terraform/python/openai_api/lambda_openai_function/tests/test_custom_config.py @@ -1,10 +1,9 @@ # -*- coding: utf-8 -*- # pylint: disable=wrong-import-position -# pylint: disable=R0801 +# pylint: disable=R0801,E1101 """Test lambda_openai_v2 function.""" # python stuff -import json import os import sys import unittest @@ -83,7 +82,12 @@ def test_search_terms_invalid(self): def test_additional_information(self): """Test additional_information.""" config_json = self.everlasting_gobbstopper["function_calling"]["additional_information"] + print("test_additional_information type config_json: ", type(config_json)) additional_information = AdditionalInformation(config_json=config_json) + print( + "test_additional_information type additional_information.config_json: ", + type(additional_information.config_json), + ) self.assertTrue(isinstance(additional_information, AdditionalInformation)) self.assertTrue(isinstance(additional_information.config_json, dict)) diff --git a/api/terraform/python/openai_api/lambda_openai_function/tests/test_lambda_openai_custom_config.py b/api/terraform/python/openai_api/lambda_openai_function/tests/test_lambda_openai_custom_config.py index f946d460..5133c859 100644 --- a/api/terraform/python/openai_api/lambda_openai_function/tests/test_lambda_openai_custom_config.py +++ b/api/terraform/python/openai_api/lambda_openai_function/tests/test_lambda_openai_custom_config.py @@ -1,6 +1,6 @@ # -*- coding: utf-8 -*- # pylint: disable=wrong-import-position -# pylint: disable=R0801 +# pylint: disable=R0801,E1101 """Test lambda_openai_v2 function.""" # python stuff diff --git a/api/terraform/python/openai_api/lambda_openai_function/tests/test_lambda_openai_function.py b/api/terraform/python/openai_api/lambda_openai_function/tests/test_lambda_openai_function.py index a66d8c9a..7935f300 100644 --- a/api/terraform/python/openai_api/lambda_openai_function/tests/test_lambda_openai_function.py +++ b/api/terraform/python/openai_api/lambda_openai_function/tests/test_lambda_openai_function.py @@ -1,6 +1,6 @@ # -*- coding: utf-8 -*- # pylint: disable=wrong-import-position -# pylint: disable=R0801 +# pylint: disable=R0801,E1101 # pylint: disable=broad-exception-caught """Test lambda_openai_v2 function."""