Upload 41 files

api/adapter/__init__.py

@@ -0,0 +1 @@
+from api.adapter.template import get_prompt_adapter

api/adapter/model.py

@@ -0,0 +1,582 @@
+import os
+import sys
+from typing import List, Optional, Any, Dict, Tuple
+
+import torch
+from loguru import logger
+from peft import PeftModel
+from tqdm import tqdm
+from transformers import (
+    AutoModel,
+    AutoConfig,
+    AutoTokenizer,
+    AutoModelForCausalLM,
+    BitsAndBytesConfig,
+    PreTrainedTokenizer,
+    PreTrainedModel,
+)
+from transformers.utils.versions import require_version
+
+if sys.version_info >= (3, 9):
+    from functools import cache
+else:
+    from functools import lru_cache as cache
+
+
+class BaseModelAdapter:
+    """ The base and default model adapter. """
+
+    model_names = []
+
+    def match(self, model_name) -> bool:
+        """
+        Check if the given model name matches any of the predefined model names.
+
+        Args:
+            model_name (str): The model name to check.
+
+        Returns:
+            bool: True if the model name matches any of the predefined model names, False otherwise.
+        """
+
+        return any(m in model_name for m in self.model_names) if self.model_names else True
+
+    def load_model(
+        self,
+        model_name_or_path: Optional[str] = None,
+        adapter_model: Optional[str] = None,
+        **kwargs: Any,
+    ) -> Tuple[PreTrainedModel, PreTrainedTokenizer]:
+        """
+        Load a model and tokenizer based on the provided model name or path.
+
+        Args:
+            model_name_or_path (str, optional): The name or path of the model. Defaults to None.
+            adapter_model (str, optional): The adapter model to load the tokenizer from. Defaults to None.
+            **kwargs: Additional keyword arguments.
+
+        Returns:
+            Tuple[PreTrainedModel, PreTrainedTokenizer]: A tuple containing the loaded model and tokenizer.
+        """
+
+        model_name_or_path = model_name_or_path or self.default_model_name_or_path
+        tokenizer_kwargs = {"trust_remote_code": True, "use_fast": False}
+        tokenizer_kwargs.update(self.tokenizer_kwargs)
+        
+        # load a tokenizer from adapter model if it exists.
+        if adapter_model is not None:
+            try:
+                tokenizer = self.tokenizer_class.from_pretrained(
+                    adapter_model, **tokenizer_kwargs,
+                    )
+            except OSError:
+                tokenizer = self.tokenizer_class.from_pretrained(
+                    model_name_or_path, **tokenizer_kwargs,
+                    )
+        else:
+            tokenizer = self.tokenizer_class.from_pretrained(
+                model_name_or_path, **tokenizer_kwargs,
+                )
+
+        config_kwargs = self.model_kwargs
+        device = kwargs.get("device", "cuda")
+        num_gpus = kwargs.get("num_gpus", 1)
+        dtype = kwargs.get("dtype", "half")
+        if device == "cuda":
+            if "torch_dtype" not in config_kwargs:
+                if dtype == "half":
+                    config_kwargs["torch_dtype"] = torch.float16
+                elif dtype == "bfloat16":
+                    config_kwargs["torch_dtype"] = torch.bfloat16
+                elif dtype == "float32":
+                    config_kwargs["torch_dtype"] = torch.float32
+
+            if num_gpus != 1:
+                config_kwargs["device_map"] = "auto"
+                # model_kwargs["device_map"] = "sequential"  # This is important for not the same VRAM sizes
+
+        # Quantization configurations (using bitsandbytes library).
+        if kwargs.get("load_in_8bit", False):
+            require_version("bitsandbytes>=0.37.0", "To fix: pip install bitsandbytes>=0.37.0")
+
+            config_kwargs["load_in_8bit"] = True
+            config_kwargs["quantization_config"] = BitsAndBytesConfig(
+                load_in_8bit=True,
+                llm_int8_threshold=6.0,
+            )
+            config_kwargs["device_map"] = "auto" if device == "cuda" else None
+
+            logger.info("Quantizing model to 8 bit.")
+
+        elif kwargs.get("load_in_4bit", False):
+            require_version("bitsandbytes>=0.39.0", "To fix: pip install bitsandbytes>=0.39.0")
+            require_version("peft>=0.4.0.dev0", "To fix: pip install git+https://github.com/huggingface/peft.git")
+
+            config_kwargs["load_in_4bit"] = True
+            config_kwargs["quantization_config"] = BitsAndBytesConfig(
+                load_in_4bit=True,
+                bnb_4bit_compute_dtype=torch.float16,
+                bnb_4bit_use_double_quant=True,
+                bnb_4bit_quant_type="nf4",
+            )
+            config_kwargs["device_map"] = "auto" if device == "cuda" else None
+
+            logger.info("Quantizing model to 4 bit.")
+
+        if kwargs.get("device_map", None) == "auto":
+            config_kwargs["device_map"] = "auto"
+
+        config = AutoConfig.from_pretrained(model_name_or_path, trust_remote_code=True)
+
+        # Fix config (for Qwen)
+        if hasattr(config, "fp16") and hasattr(config, "bf16"):
+            setattr(config, "fp16", dtype == "half")
+            setattr(config, "bf16", dtype == "bfloat16")
+            config_kwargs.pop("torch_dtype", None)
+
+        if kwargs.get("using_ptuning_v2", False) and adapter_model:
+            config.pre_seq_len = kwargs.get("pre_seq_len", 128)
+
+        # Load and prepare pretrained models (without valuehead).
+        model = self.model_class.from_pretrained(
+            model_name_or_path,
+            config=config,
+            trust_remote_code=True,
+            **config_kwargs
+        )
+
+        if device == "cpu":
+            model = model.float()
+
+        # post process for special tokens
+        tokenizer = self.post_tokenizer(tokenizer)
+        is_chatglm = "chatglm" in str(type(model))
+
+        if adapter_model is not None:
+            model = self.load_adapter_model(model, tokenizer, adapter_model, is_chatglm, config_kwargs, **kwargs)
+
+        if is_chatglm or "baichuan" in str(type(model)) or "xverse" in str(type(model)):
+            quantize = kwargs.get("quantize", None)
+            if quantize and quantize != 16:
+                logger.info(f"Quantizing model to {quantize} bit.")
+                model = model.quantize(quantize)
+
+        if device == "cuda" and num_gpus == 1 and "device_map" not in config_kwargs:
+            model.to(device)
+
+        # inference mode
+        model.eval()
+
+        return model, tokenizer
+
+    def load_lora_model(
+        self, model: PreTrainedModel, adapter_model: str, model_kwargs: Dict,
+    ) -> PeftModel:
+        """
+        Load a LoRA model.
+
+        This function loads a LoRA model using the specified pretrained model and adapter model.
+
+        Args:
+            model (PreTrainedModel): The base pretrained model.
+            adapter_model (str): The name or path of the adapter model.
+            model_kwargs (dict): Additional keyword arguments for the model.
+
+        Returns:
+            PeftModel: The loaded LoRA model.
+        """
+        return PeftModel.from_pretrained(
+            model,
+            adapter_model,
+            torch_dtype=model_kwargs.get("torch_dtype", torch.float16),
+        )
+
+    def load_adapter_model(
+        self, 
+        model: PreTrainedModel, 
+        tokenizer: PreTrainedTokenizer, 
+        adapter_model: str, 
+        is_chatglm: bool,
+        model_kwargs: Dict,
+        **kwargs: Any,
+    ) -> PreTrainedModel:
+        using_ptuning_v2 = kwargs.get("using_ptuning_v2", False)
+        resize_embeddings = kwargs.get("resize_embeddings", False)
+        if adapter_model and resize_embeddings and not is_chatglm:
+            model_vocab_size = model.get_input_embeddings().weight.size(0)
+            tokenzier_vocab_size = len(tokenizer)
+            logger.info(f"Vocab of the base model: {model_vocab_size}")
+            logger.info(f"Vocab of the tokenizer: {tokenzier_vocab_size}")
+
+            if model_vocab_size != tokenzier_vocab_size:
+                assert tokenzier_vocab_size > model_vocab_size
+                logger.info("Resize model embeddings to fit tokenizer")
+                model.resize_token_embeddings(tokenzier_vocab_size)
+
+        if using_ptuning_v2:
+            prefix_state_dict = torch.load(os.path.join(adapter_model, "pytorch_model.bin"))
+            new_prefix_state_dict = {
+                k[len("transformer.prefix_encoder."):]: v
+                for k, v in prefix_state_dict.items()
+                if k.startswith("transformer.prefix_encoder.")
+            }
+            model.transformer.prefix_encoder.load_state_dict(new_prefix_state_dict)
+            model.transformer.prefix_encoder.float()
+        else:
+            model = self.load_lora_model(model, adapter_model, model_kwargs)
+
+        return model
+
+    def post_tokenizer(self, tokenizer) -> PreTrainedTokenizer:
+        return tokenizer
+
+    @property
+    def model_class(self):
+        return AutoModelForCausalLM
+
+    @property
+    def model_kwargs(self):
+        return {}
+
+    @property
+    def tokenizer_class(self):
+        return AutoTokenizer
+
+    @property
+    def tokenizer_kwargs(self):
+        return {}
+
+    @property
+    def default_model_name_or_path(self):
+        return "zpn/llama-7b"
+
+
+# A global registry for all model adapters
+model_adapters: List[BaseModelAdapter] = []
+
+
+def register_model_adapter(cls):
+    """ Register a model adapter. """
+    model_adapters.append(cls())
+
+
+@cache
+def get_model_adapter(model_name: str) -> BaseModelAdapter:
+    """
+    Get a model adapter for a given model name.
+
+    Args:
+        model_name (str): The name of the model.
+
+    Returns:
+        ModelAdapter: The model adapter that matches the given model name.
+    """
+    for adapter in model_adapters:
+        if adapter.match(model_name):
+            return adapter
+    raise ValueError(f"No valid model adapter for {model_name}")
+
+
+def load_model(
+    model_name: str,
+    model_name_or_path: Optional[str] = None,
+    adapter_model: Optional[str] = None,
+    quantize: Optional[int] = 16,
+    device: Optional[str] = "cuda",
+    load_in_8bit: Optional[bool] = False,
+    **kwargs: Any,
+) -> Tuple[PreTrainedModel, PreTrainedTokenizer]:
+    """
+    Load a pre-trained model and tokenizer.
+
+    Args:
+        model_name (str): The name of the model.
+        model_name_or_path (Optional[str], optional): The path or name of the pre-trained model. Defaults to None.
+        adapter_model (Optional[str], optional): The name of the adapter model. Defaults to None.
+        quantize (Optional[int], optional): The quantization level. Defaults to 16.
+        device (Optional[str], optional): The device to load the model on. Defaults to "cuda".
+        load_in_8bit (Optional[bool], optional): Whether to load the model in 8-bit mode. Defaults to False.
+        **kwargs (Any): Additional keyword arguments.
+
+    Returns:
+        Tuple[PreTrainedModel, PreTrainedTokenizer]: A tuple containing the loaded model and tokenizer.
+    """
+    model_name = model_name.lower()
+
+    if "tiger" in model_name:
+        def skip(*args, **kwargs):
+            pass
+
+        torch.nn.init.kaiming_uniform_ = skip
+        torch.nn.init.uniform_ = skip
+        torch.nn.init.normal_ = skip
+
+    # get model adapter
+    adapter = get_model_adapter(model_name)
+    model, tokenizer = adapter.load_model(
+        model_name_or_path,
+        adapter_model,
+        device=device,
+        quantize=quantize,
+        load_in_8bit=load_in_8bit,
+        **kwargs
+    )
+    return model, tokenizer
+
+
+class ChatglmModelAdapter(BaseModelAdapter):
+    """ https://github.com/THUDM/ChatGLM-6B """
+
+    model_names = ["chatglm"]
+
+    @property
+    def model_class(self):
+        return AutoModel
+
+    @property
+    def default_model_name_or_path(self):
+        return "THUDM/chatglm2-6b"
+
+
+class Chatglm3ModelAdapter(ChatglmModelAdapter):
+    """ https://github.com/THUDM/ChatGLM-6B """
+
+    model_names = ["chatglm3"]
+
+    @property
+    def tokenizer_kwargs(self):
+        return {"encode_special_tokens": True}
+
+    @property
+    def default_model_name_or_path(self):
+        return "THUDM/chatglm3-6b"
+
+
+class LlamaModelAdapter(BaseModelAdapter):
+    """ https://github.com/project-baize/baize-chatbot """
+
+    model_names = ["alpaca", "baize", "openbuddy-llama", "ziya-llama", "guanaco", "llama2"]
+
+    def post_tokenizer(self, tokenizer):
+        tokenizer.bos_token = "<s>"
+        tokenizer.eos_token = "</s>"
+        tokenizer.unk_token = "<unk>"
+        return tokenizer
+
+    @property
+    def model_kwargs(self):
+        return {"low_cpu_mem_usage": True}
+
+
+class MossModelAdapter(BaseModelAdapter):
+    """ https://github.com/OpenLMLab/MOSS """
+
+    model_names = ["moss"]
+
+    @property
+    def default_model_name_or_path(self):
+        return "fnlp/moss-moon-003-sft-int4"
+
+
+class PhoenixModelAdapter(BaseModelAdapter):
+    """ https://github.com/FreedomIntelligence/LLMZoo """
+
+    model_names = ["phoenix"]
+
+    @property
+    def model_kwargs(self):
+        return {"low_cpu_mem_usage": True}
+
+    @property
+    def tokenizer_kwargs(self):
+        return {"use_fast": True}
+
+    @property
+    def default_model_name_or_path(self):
+        return "FreedomIntelligence/phoenix-inst-chat-7b"
+
+
+class FireflyModelAdapter(BaseModelAdapter):
+    """ https://github.com/yangjianxin1/Firefly """
+
+    model_names = ["firefly"]
+
+    @property
+    def model_kwargs(self):
+        return {"torch_dtype": torch.float32}
+
+    @property
+    def tokenizer_kwargs(self):
+        return {"use_fast": True}
+
+    @property
+    def default_model_name_or_path(self):
+        return "YeungNLP/firefly-2b6"
+
+
+class YuLanChatModelAdapter(BaseModelAdapter):
+    """ https://github.com/RUC-GSAI/YuLan-Chat """
+
+    model_names = ["yulan"]
+
+    def post_tokenizer(self, tokenizer):
+        tokenizer.bos_token = "<s>"
+        tokenizer.eos_token = "</s>"
+        tokenizer.unk_token = "<unk>"
+        return tokenizer
+
+    @property
+    def model_kwargs(self):
+        return {"low_cpu_mem_usage": True}
+
+    def load_adapter_model(self, model, tokenizer, adapter_model, is_chatglm, model_kwargs, **kwargs):
+        adapter_model = AutoModelForCausalLM.from_pretrained(
+            adapter_model, torch_dtype=torch.float16, low_cpu_mem_usage=True
+        )
+        if model.model.embed_tokens.weight.size(0) + 1 == adapter_model.model.embed_tokens.weight.size(0):
+            model.resize_token_embeddings(len(tokenizer))
+            model.model.embed_tokens.weight.data[-1, :] = 0
+
+        logger.info("Applying the delta")
+        for name, param in tqdm(model.state_dict().items(), desc="Applying delta"):
+            assert name in model.state_dict()
+            param.data += model.state_dict()[name]
+
+        return model
+
+
+class TigerBotModelAdapter(BaseModelAdapter):
+    """ https://github.com/TigerResearch/TigerBot """
+
+    model_names = ["tiger"]
+
+    @property
+    def tokenizer_kwargs(self):
+        return {"use_fast": True}
+
+    @property
+    def default_model_name_or_path(self):
+        return "TigerResearch/tigerbot-7b-sft"
+
+
+class OpenBuddyFalconModelAdapter(BaseModelAdapter):
+    """ https://github.com/OpenBuddy/OpenBuddy """
+
+    model_names = ["openbuddy-falcon"]
+
+    @property
+    def default_model_name_or_path(self):
+        return "OpenBuddy/openbuddy-falcon-7b-v5-fp16"
+
+
+class AnimaModelAdapter(LlamaModelAdapter):
+
+    model_names = ["anima"]
+
+    def load_lora_model(self, model, adapter_model, model_kwargs):
+        return PeftModel.from_pretrained(model, adapter_model)
+
+
+class BaiChuanModelAdapter(BaseModelAdapter):
+    """ https://github.com/baichuan-inc/Baichuan-13B """
+
+    model_names = ["baichuan"]
+
+    def load_lora_model(self, model, adapter_model, model_kwargs):
+        return PeftModel.from_pretrained(model, adapter_model)
+
+    @property
+    def default_model_name_or_path(self):
+        return "baichuan-inc/Baichuan-13B-Chat"
+
+
+class InternLMModelAdapter(BaseModelAdapter):
+    """ https://github.com/InternLM/InternLM """
+
+    model_names = ["internlm"]
+
+    @property
+    def default_model_name_or_path(self):
+        return "internlm/internlm-chat-7b"
+
+
+class StarCodeModelAdapter(BaseModelAdapter):
+    """ https://github.com/bigcode-project/starcoder """
+
+    model_names = ["starcode", "starchat"]
+
+    @property
+    def tokenizer_kwargs(self):
+        return {}
+
+    @property
+    def default_model_name_or_path(self):
+        return "HuggingFaceH4/starchat-beta"
+
+
+class AquilaModelAdapter(BaseModelAdapter):
+    """ https://github.com/FlagAI-Open/FlagAI """
+
+    model_names = ["aquila"]
+
+    @property
+    def default_model_name_or_path(self):
+        return "BAAI/AquilaChat-7B"
+
+
+class QwenModelAdapter(BaseModelAdapter):
+    """ https://github.com/QwenLM/Qwen-7B """
+
+    model_names = ["qwen"]
+
+    @property
+    def default_model_name_or_path(self):
+        return "Qwen/Qwen-7B-Chat"
+
+
+class XverseModelAdapter(BaseModelAdapter):
+    """ https://github.com/xverse-ai/XVERSE-13B """
+
+    model_names = ["xverse"]
+
+    @property
+    def default_model_name_or_path(self):
+        return "xverse/XVERSE-13B-Chat"
+
+
+class CodeLlamaModelAdapter(LlamaModelAdapter):
+    """ https://github.com/project-baize/baize-chatbot """
+
+    model_names = ["code-llama"]
+
+    @property
+    def tokenizer_class(self):
+        require_version("transformers>=4.33.1", "To fix: pip install transformers>=4.33.1")
+        from transformers import CodeLlamaTokenizer
+
+        return CodeLlamaTokenizer
+
+    @property
+    def default_model_name_or_path(self):
+        return "codellama/CodeLlama-7b-Instruct-hf"
+
+
+register_model_adapter(ChatglmModelAdapter)
+register_model_adapter(Chatglm3ModelAdapter)
+register_model_adapter(LlamaModelAdapter)
+register_model_adapter(MossModelAdapter)
+register_model_adapter(PhoenixModelAdapter)
+register_model_adapter(FireflyModelAdapter)
+register_model_adapter(YuLanChatModelAdapter)
+register_model_adapter(TigerBotModelAdapter)
+register_model_adapter(OpenBuddyFalconModelAdapter)
+register_model_adapter(AnimaModelAdapter)
+register_model_adapter(BaiChuanModelAdapter)
+register_model_adapter(InternLMModelAdapter)
+register_model_adapter(AquilaModelAdapter)
+register_model_adapter(QwenModelAdapter)
+register_model_adapter(XverseModelAdapter)
+register_model_adapter(CodeLlamaModelAdapter)
+
+# After all adapters, try the default base adapter.
+register_model_adapter(BaseModelAdapter)

api/adapter/schema.py

@@ -0,0 +1,375 @@
+from typing import Any, Dict, List, Optional
+
+from openai.types.chat.completion_create_params import Function
+from pydantic import BaseModel
+
+from api.utils.compat import model_dump
+
+
+def convert_data_type(param_type: str) -> str:
+    """ convert data_type to typescript data type """
+    return "number" if param_type in {"integer", "float"} else param_type
+
+
+def get_param_type(param: Dict[str, Any]) -> str:
+    """ get param_type of parameter """
+    param_type = "any"
+    if "type" in param:
+        raw_param_type = param["type"]
+        param_type = (
+            " | ".join(raw_param_type)
+            if type(raw_param_type) is list
+            else raw_param_type
+        )
+    elif "oneOf" in param:
+        one_of_types = [
+            convert_data_type(item["type"])
+            for item in param["oneOf"]
+            if "type" in item
+        ]
+        one_of_types = list(set(one_of_types))
+        param_type = " | ".join(one_of_types)
+    return convert_data_type(param_type)
+
+
+def get_format_param(param: Dict[str, Any]) -> Optional[str]:
+    """ Get "format" from param. There are cases where format is not directly in param but in oneOf """
+    if "format" in param:
+        return param["format"]
+    if "oneOf" in param:
+        formats = [item["format"] for item in param["oneOf"] if "format" in item]
+        if formats:
+            return " or ".join(formats)
+    return None
+
+
+def get_param_info(param: Dict[str, Any]) -> Optional[str]:
+    """ get additional information about parameter such as: format, default value, min, max, ... """
+    param_type = param.get("type", "any")
+    info_list = []
+    if "description" in param:
+        desc = param["description"]
+        if not desc.endswith("."):
+            desc += "."
+        info_list.append(desc)
+
+    if "default" in param:
+        default_value = param["default"]
+        if param_type == "string":
+            default_value = f'"{default_value}"'  # if string --> add ""
+        info_list.append(f"Default={default_value}.")
+
+    format_param = get_format_param(param)
+    if format_param is not None:
+        info_list.append(f"Format={format_param}")
+
+    info_list.extend(
+        f"{field_name}={str(param[field])}"
+        for field, field_name in [
+            ("maximum", "Maximum"),
+            ("minimum", "Minimum"),
+            ("maxLength", "Maximum length"),
+            ("minLength", "Minimum length"),
+        ]
+        if field in param
+    )
+    if info_list:
+        result = "// " + " ".join(info_list)
+        return result.replace("\n", " ")
+    return None
+
+
+def append_new_param_info(info_list: List[str], param_declaration: str, comment_info: Optional[str], depth: int):
+    """ Append a new parameter with comment to the info_list """
+    offset = "".join(["    " for _ in range(depth)]) if depth >= 1 else ""
+    if comment_info is not None:
+        # if depth == 0:  # format: //comment\nparam: type
+        info_list.append(f"{offset}{comment_info}")
+    info_list.append(f"{offset}{param_declaration}")
+
+
+def get_enum_option_str(enum_options: List) -> str:
+    """get enum option separated by: "|"
+
+    Args:
+        enum_options (List): list of options
+
+    Returns:
+        _type_: concatenation of options separated by "|"
+    """
+    # if each option is string --> add quote
+    return " | ".join([f'"{v}"' if type(v) is str else str(v) for v in enum_options])
+
+
+def get_array_typescript(param_name: Optional[str], param_dic: dict, depth: int = 0) -> str:
+    """recursive implementation for generating type script of array
+
+    Args:
+        param_name (Optional[str]): name of param, optional
+        param_dic (dict): param_dic
+        depth (int, optional): nested level. Defaults to 0.
+
+    Returns:
+        _type_: typescript of array
+    """
+    offset = "".join(["    " for _ in range(depth)]) if depth >= 1 else ""
+    items_info = param_dic.get("items", {})
+
+    if len(items_info) == 0:
+        return f"{offset}{param_name}: []" if param_name is not None else "[]"
+    array_type = get_param_type(items_info)
+    if array_type == "object":
+        info_lines = []
+        child_lines = get_parameter_typescript(
+            items_info.get("properties", {}), items_info.get("required", []), depth + 1
+        )
+        # if comment_info is not None:
+        #    info_lines.append(f"{offset}{comment_info}")
+        if param_name is not None:
+            info_lines.append(f"{offset}{param_name}" + ": {")
+        else:
+            info_lines.append(f"{offset}" + "{")
+        info_lines.extend(child_lines)
+        info_lines.append(f"{offset}" + "}[]")
+        return "\n".join(info_lines)
+
+    elif array_type == "array":
+        item_info = get_array_typescript(None, items_info, depth + 1)
+        if param_name is None:
+            return f"{item_info}[]"
+        return f"{offset}{param_name}: {item_info.strip()}[]"
+
+    else:
+        if "enum" not in items_info:
+            return (
+                f"{array_type}[]"
+                if param_name is None
+                else f"{offset}{param_name}: {array_type}[],"
+            )
+        item_type = get_enum_option_str(items_info["enum"])
+        if param_name is None:
+            return f"({item_type})[]"
+        else:
+            return f"{offset}{param_name}: ({item_type})[]"
+
+
+def get_parameter_typescript(properties, required_params, depth=0) -> List[str]:
+    """Recursion, returning the information about parameters including data type, description and other information
+    These kinds of information will be put into the prompt
+
+    Args:
+        properties (_type_): properties in parameters
+        required_params (_type_): List of required parameters
+        depth (int, optional): the depth of params (nested level). Defaults to 0.
+
+    Returns:
+        _type_: list of lines containing information about all parameters
+    """
+    tp_lines = []
+    for param_name, param in properties.items():
+        # Sometimes properties have "required" field as a list of string.
+        # Even though it is supposed to be not under properties. So we skip it
+        if not isinstance(param, dict):
+            continue
+        # Param Description
+        comment_info = get_param_info(param)
+        # Param Name declaration
+        param_declaration = f"{param_name}"
+        if isinstance(required_params, list) and param_name not in required_params:
+            param_declaration += "?"
+        param_type = get_param_type(param)
+
+        offset = ""
+        if depth >= 1:
+            offset = "".join(["    " for _ in range(depth)])
+
+        if param_type == "object":  # param_type is object
+            child_lines = get_parameter_typescript(param.get("properties", {}), param.get("required", []), depth + 1)
+            if comment_info is not None:
+                tp_lines.append(f"{offset}{comment_info}")
+
+            param_declaration += ": {"
+            tp_lines.append(f"{offset}{param_declaration}")
+            tp_lines.extend(child_lines)
+            tp_lines.append(f"{offset}" + "},")
+
+        elif param_type == "array":  # param_type is an array
+            item_info = param.get("items", {})
+            if "type" not in item_info:  # don't know type of array
+                param_declaration += ": [],"
+                append_new_param_info(tp_lines, param_declaration, comment_info, depth)
+            else:
+                array_declaration = get_array_typescript(param_declaration, param, depth)
+                if not array_declaration.endswith(","):
+                    array_declaration += ","
+                if comment_info is not None:
+                    tp_lines.append(f"{offset}{comment_info}")
+                tp_lines.append(array_declaration)
+        else:
+            if "enum" in param:
+                param_type = " | ".join([f'"{v}"' for v in param["enum"]])
+            param_declaration += f": {param_type},"
+            append_new_param_info(tp_lines, param_declaration, comment_info, depth)
+
+    return tp_lines
+
+
+def generate_schema_from_functions(functions: List[Function], namespace="functions") -> str:
+    """
+    Convert functions schema to a schema that language models can understand.
+    """
+
+    schema = "// Supported function definitions that should be called when necessary.\n"
+    schema += f"namespace {namespace} {{\n\n"
+
+    for function in functions:
+        # Convert a Function object to dict, if necessary
+        if isinstance(function, BaseModel):
+            function = model_dump(function)
+        function_name = function.get("name", None)
+        if function_name is None:
+            continue
+
+        description = function.get("description", "")
+        schema += f"// {description}\n"
+        schema += f"type {function_name}"
+
+        parameters = function.get("parameters", None)
+        if parameters is not None and parameters.get("properties") is not None:
+            schema += " = (_: {\n"
+            required_params = parameters.get("required", [])
+            tp_lines = get_parameter_typescript(parameters.get("properties"), required_params, 0)
+            schema += "\n".join(tp_lines)
+            schema += "\n}) => any;\n\n"
+        else:
+            # Doesn't have any parameters
+            schema += " = () => any;\n\n"
+
+    schema += f"}} // namespace {namespace}"
+
+    return schema
+
+
+def generate_schema_from_openapi(specification: Dict[str, Any], description: str, namespace: str) -> str:
+    """
+    Convert OpenAPI specification object to a schema that language models can understand.
+
+    Input:
+    specification: can be obtained by json. loads of any OpanAPI json spec, or yaml.safe_load for yaml OpenAPI specs
+
+    Example output:
+
+    // General Description
+    namespace functions {
+
+    // Simple GET endpoint
+    type getEndpoint = (_: {
+    // This is a string parameter
+    param_string: string,
+    param_integer: number,
+    param_boolean?: boolean,
+    param_enum: "value1" | "value2" | "value3",
+    }) => any;
+
+    } // namespace functions
+    """
+
+    description_clean = description.replace("\n", "")
+
+    schema = f"// {description_clean}\n"
+    schema += f"namespace {namespace} {{\n\n"
+
+    for path_name, paths in specification.get("paths", {}).items():
+        for method_name, method_info in paths.items():
+            operationId = method_info.get("operationId", None)
+            if operationId is None:
+                continue
+            description = method_info.get("description", method_info.get("summary", ""))
+            schema += f"// {description}\n"
+            schema += f"type {operationId}"
+
+            if ("requestBody" in method_info) or (method_info.get("parameters") is not None):
+                schema += f"  = (_: {{\n"
+                # Body
+                if "requestBody" in method_info:
+                    try:
+                        body_schema = (
+                            method_info.get("requestBody", {})
+                            .get("content", {})
+                            .get("application/json", {})
+                            .get("schema", {})
+                        )
+                    except AttributeError:
+                        body_schema = {}
+                    for param_name, param in body_schema.get("properties", {}).items():
+                        # Param Description
+                        description = param.get("description")
+                        if description is not None:
+                            schema += f"// {description}\n"
+
+                        # Param Name
+                        schema += f"{param_name}"
+                        if (
+                            (not param.get("required", False))
+                            or (param.get("nullable", False))
+                            or (param_name in body_schema.get("required", []))
+                        ):
+                            schema += "?"
+
+                        # Param Type
+                        param_type = param.get("type", "any")
+                        if param_type == "integer":
+                            param_type = "number"
+                        if "enum" in param:
+                            param_type = " | ".join([f'"{v}"' for v in param["enum"]])
+                        schema += f": {param_type},\n"
+
+                # URL
+                for param in method_info.get("parameters", []):
+                    # Param Description
+                    if description := param.get("description"):
+                        schema += f"// {description}\n"
+
+                    # Param Name
+                    schema += f"{param['name']}"
+                    if (not param.get("required", False)) or (param.get("nullable", False)):
+                        schema += "?"
+                    if param.get("schema") is None:
+                        continue
+                    # Param Type
+                    param_type = param["schema"].get("type", "any")
+                    if param_type == "integer":
+                        param_type = "number"
+                    if "enum" in param["schema"]:
+                        param_type = " | ".join([f'"{v}"' for v in param["schema"]["enum"]])
+                    schema += f": {param_type},\n"
+
+                schema += f"}}) => any;\n\n"
+            else:
+                # Doesn't have any parameters
+                schema += f" = () => any;\n\n"
+
+    schema += f"}} // namespace {namespace}"
+
+    return schema
+
+
+if __name__ == "__main__":
+    functions = [
+        {
+            "name": "get_current_weather",
+            "description": "Get the current weather in a given location",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "location": {
+                        "type": "string",
+                        "description": "The city and state, e.g. San Francisco, CA",
+                    },
+                    "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]},
+                },
+                "required": ["location"],
+            },
+        }
+    ]
+    print(generate_schema_from_functions(functions))