functionary/schema.py at main · jjhw/functionary · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
from typing import Any, Dict, Optional


def generate_schema_from_functions(functions, namespace="functions"):
    """
    Convert functions array to a schema that language models can understand.
    """

    schema = f"// Supported function definitions that should be called when necessary.\n"
    schema += f"namespace {namespace} {{\n\n"

    for function in functions:
        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:
            schema += f" = (_: {{\n"
            required_params = parameters.get("required", [])
            for param_name, param in parameters.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 param_name not in required_params:
                    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"

            schema += f"}}) => any;\n\n"
        else:
            # Doesnt have any parameters
            schema += f" = () => 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:
                    body_schema = method_info.get("requestBody", {}).get("content", {}).get("application/json", {}).get("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:
                # Doesnt have any parameters
                schema += f" = () => any;\n\n"


    schema += f"}} // namespace {namespace}"

    return schema