feat: functions

utils/main.py

@@ -4,6 +4,9 @@ import time
 from typing import List
 from schemas import OpenAIChatMessage
 
+import inspect
+from typing import get_type_hints, Literal
+
 
 def stream_message_template(model: str, message: str):
     return {
@@ -42,3 +45,95 @@ def get_last_assistant_message(messages: List[dict]) -> str:
                         return item["text"]
             return message["content"]
     return None
+
+
+def add_or_update_system_message(content: str, messages: List[dict]):
+    """
+    Adds a new system message at the beginning of the messages list
+    or updates the existing system message at the beginning.
+
+    :param msg: The message to be added or appended.
+    :param messages: The list of message dictionaries.
+    :return: The updated list of message dictionaries.
+    """
+
+    if messages and messages[0].get("role") == "system":
+        messages[0]["content"] += f"{content}\n{messages[0]['content']}"
+    else:
+        # Insert at the beginning
+        messages.insert(0, {"role": "system", "content": content})
+
+    return messages
+
+
+def doc_to_dict(docstring):
+    lines = docstring.split("\n")
+    description = lines[1].strip()
+    param_dict = {}
+
+    for line in lines:
+        if ":param" in line:
+            line = line.replace(":param", "").strip()
+            param, desc = line.split(":", 1)
+            param_dict[param.strip()] = desc.strip()
+    ret_dict = {"description": description, "params": param_dict}
+    return ret_dict
+
+
+def get_function_specs(functions) -> List[dict]:
+
+    function_list = [
+        {"name": func, "function": getattr(functions, func)}
+        for func in dir(functions)
+        if callable(getattr(functions, func)) and not func.startswith("__")
+    ]
+
+    specs = []
+
+    for function_item in function_list:
+        function_name = function_item["name"]
+        function = function_item["function"]
+
+        function_doc = doc_to_dict(function.__doc__ or function_name)
+        specs.append(
+            {
+                "name": function_name,
+                # TODO: multi-line desc?
+                "description": function_doc.get("description", function_name),
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        param_name: {
+                            "type": param_annotation.__name__.lower(),
+                            **(
+                                {
+                                    "enum": (
+                                        param_annotation.__args__
+                                        if hasattr(param_annotation, "__args__")
+                                        else None
+                                    )
+                                }
+                                if hasattr(param_annotation, "__args__")
+                                else {}
+                            ),
+                            "description": function_doc.get("params", {}).get(
+                                param_name, param_name
+                            ),
+                        }
+                        for param_name, param_annotation in get_type_hints(
+                            function
+                        ).items()
+                        if param_name != "return"
+                    },
+                    "required": [
+                        name
+                        for name, param in inspect.signature(
+                            function
+                        ).parameters.items()
+                        if param.default is param.empty
+                    ],
+                },
+            }
+        )
+
+    return specs