Spaces:
Running
Running
| 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)) | |