Update Assistant Api

phi/assistant/assistant.py

@@ -1,10 +1,12 @@
-from typing import List, Any, Optional, Dict
+from typing import List, Any, Optional, Dict, Union, Callable
 
-from pydantic import BaseModel, ConfigDict, field_validator
+from pydantic import BaseModel, ConfigDict, field_validator, model_validator
 
 from phi.assistant.file import File
 from phi.assistant.tool import Tool
+from phi.assistant.tool.registry import ToolRegistry
 from phi.assistant.row import AssistantRow
+from phi.assistant.function import Function, FunctionCall
 from phi.assistant.storage import AssistantStorage
 from phi.assistant.exceptions import AssistantIdNotSet
 from phi.knowledge.base import KnowledgeBase
@@ -40,7 +42,9 @@ class Assistant(BaseModel):
     # -*- Assistant Tools
     # A list of tool enabled on the assistant. There can be a maximum of 128 tools per assistant.
     # Tools can be of types code_interpreter, retrieval, or function.
-    tools: Optional[List[Tool | Dict]] = None
+    tools: Optional[List[Union[Tool, Dict, Callable, ToolRegistry]]] = None
+    # Functions the Assistant may call.
+    _function_map: Optional[Dict[str, Function]] = None
 
     # -*- Assistant Files
     # A list of file IDs attached to this assistant.
@@ -90,6 +94,26 @@ def set_log_level(cls, v: bool) -> bool:
     def client(self) -> OpenAI:
         return self.openai or OpenAI()
 
+    def add_function(self, f: Function) -> None:
+        if self._function_map is None:
+            self._function_map = {}
+        self._function_map[f.name] = f
+        logger.debug(f"Added function {f.name} to Assistant")
+
+    @model_validator(mode="after")
+    def add_functions_to_assistant(self) -> "Assistant":
+        if self.tools is not None:
+            for tool in self.tools:
+                if callable(tool):
+                    f = Function.from_callable(tool)
+                    self.add_function(f)
+                elif isinstance(tool, ToolRegistry):
+                    if self._function_map is None:
+                        self._function_map = {}
+                    self._function_map.update(tool.functions)
+                    logger.debug(f"Tools from {tool.name} added to Assistant.")
+        return self
+
     def load_from_storage(self):
         pass
 
@@ -98,6 +122,7 @@ def load_from_openai(self, openai_assistant: OpenAIAssistant):
         self.object = openai_assistant.object
         self.created_at = openai_assistant.created_at
         self.file_ids = openai_assistant.file_ids
+        self.openai_assistant = openai_assistant
 
     def create(self) -> "Assistant":
         request_body: Dict[str, Any] = {}
@@ -112,8 +137,14 @@ def create(self) -> "Assistant":
             for _tool in self.tools:
                 if isinstance(_tool, Tool):
                     _tools.append(_tool.to_dict())
-                else:
+                elif isinstance(_tool, dict):
                     _tools.append(_tool)
+                elif callable(_tool):
+                    func = Function.from_callable(_tool)
+                    _tools.append({"type": "function", "function": func.to_dict()})
+                elif isinstance(_tool, ToolRegistry):
+                    for _f in _tool.functions.values():
+                        _tools.append({"type": "function", "function": _f.to_dict()})
             request_body["tools"] = _tools
         if self.file_ids is not None or self.files is not None:
             _file_ids = self.file_ids or []
@@ -139,10 +170,7 @@ def get_id(self) -> Optional[str]:
             _id = self.id
         return _id
 
-    def get(self, use_cache: bool = True) -> "Assistant":
-        if self.openai_assistant is not None and use_cache:
-            return self
-
+    def get_from_openai(self) -> OpenAIAssistant:
         _assistant_id = self.get_id()
         if _assistant_id is None:
             raise AssistantIdNotSet("Assistant.id not set")
@@ -151,6 +179,13 @@ def get(self, use_cache: bool = True) -> "Assistant":
             assistant_id=_assistant_id,
         )
         self.load_from_openai(self.openai_assistant)
+        return self.openai_assistant
+
+    def get(self, use_cache: bool = True) -> "Assistant":
+        if self.openai_assistant is not None and use_cache:
+            return self
+
+        self.get_from_openai()
         return self
 
     def get_or_create(self, use_cache: bool = True) -> "Assistant":
@@ -161,7 +196,7 @@ def get_or_create(self, use_cache: bool = True) -> "Assistant":
 
     def update(self) -> "Assistant":
         try:
-            assistant_to_update = self.get()
+            assistant_to_update = self.get_from_openai()
             if assistant_to_update is not None:
                 request_body: Dict[str, Any] = {}
                 if self.name is not None:
@@ -175,8 +210,14 @@ def update(self) -> "Assistant":
                     for _tool in self.tools:
                         if isinstance(_tool, Tool):
                             _tools.append(_tool.to_dict())
-                        else:
+                        elif isinstance(_tool, dict):
                             _tools.append(_tool)
+                        elif callable(_tool):
+                            func = Function.from_callable(_tool)
+                            _tools.append({"type": "function", "function": func.to_dict()})
+                        elif isinstance(_tool, ToolRegistry):
+                            for _f in _tool.functions.values():
+                                _tools.append({"type": "function", "function": _f.to_dict()})
                     request_body["tools"] = _tools
                 if self.file_ids is not None or self.files is not None:
                     _file_ids = self.file_ids or []
@@ -201,7 +242,7 @@ def update(self) -> "Assistant":
 
     def delete(self) -> OpenAIAssistantDeleted:
         try:
-            assistant_to_delete = self.get()
+            assistant_to_delete = self.get_from_openai()
             if assistant_to_delete is not None:
                 deletion_status = self.client.beta.assistants.delete(
                     assistant_id=assistant_to_delete.id,
@@ -239,3 +280,58 @@ def __str__(self) -> str:
         import json
 
         return json.dumps(self.to_dict(), indent=4)
+
+    def get_function_call(self, name: str, arguments: Optional[str] = None) -> Optional[FunctionCall]:
+        import json
+
+        logger.debug(f"Getting function {name}. Args: {arguments}")
+        if self._function_map is None:
+            return None
+
+        function_to_call: Optional[Function] = None
+        if name in self._function_map:
+            function_to_call = self._function_map[name]
+        if function_to_call is None:
+            logger.error(f"Function {name} not found")
+            return None
+
+        function_call = FunctionCall(function=function_to_call)
+        if arguments is not None and arguments != "":
+            try:
+                if "None" in arguments:
+                    arguments = arguments.replace("None", "null")
+                if "True" in arguments:
+                    arguments = arguments.replace("True", "true")
+                if "False" in arguments:
+                    arguments = arguments.replace("False", "false")
+                _arguments = json.loads(arguments)
+            except Exception as e:
+                logger.error(f"Unable to decode function arguments {arguments}: {e}")
+                return None
+
+            if not isinstance(_arguments, dict):
+                logger.error(f"Function arguments {arguments} is not a valid JSON object")
+                return None
+
+            try:
+                clean_arguments: Dict[str, Any] = {}
+                for k, v in _arguments.items():
+                    if isinstance(v, str):
+                        _v = v.strip().lower()
+                        if _v in ("none", "null"):
+                            clean_arguments[k] = None
+                        elif _v == "true":
+                            clean_arguments[k] = True
+                        elif _v == "false":
+                            clean_arguments[k] = False
+                        else:
+                            clean_arguments[k] = v.strip()
+                    else:
+                        clean_arguments[k] = v
+
+                function_call.arguments = clean_arguments
+            except Exception as e:
+                logger.error(f"Unable to parse function arguments {arguments}: {e}")
+                return None
+
+        return function_call

phi/assistant/file/__init__.py

@@ -0,0 +1 @@
+from phi.assistant.file.file import File

phi/assistant/function.py

@@ -0,0 +1,97 @@
+from typing import Any, Dict, Optional, Callable, get_type_hints
+from pydantic import BaseModel, validate_call
+
+from phi.utils.log import logger
+
+
+class Tool(BaseModel):
+    """Model for Assistant Tools"""
+
+    # The type of tool
+    type: str
+    function: Optional[Dict[str, Any]] = None
+
+    def to_dict(self) -> Dict[str, Any]:
+        return self.model_dump(exclude_none=True)
+
+
+class Function(BaseModel):
+    """Model for Assistant functions"""
+
+    # The name of the function to be called.
+    # Must be a-z, A-Z, 0-9, or contain underscores and dashes, with a maximum length of 64.
+    name: str
+    # A description of what the function does, used by the model to choose when and how to call the function.
+    description: Optional[str] = None
+    # The parameters the functions accepts, described as a JSON Schema object.
+    # To describe a function that accepts no parameters, provide the value {"type": "object", "properties": {}}.
+    parameters: Dict[str, Any] = {"type": "object", "properties": {}}
+    entrypoint: Optional[Callable] = None
+
+    def to_dict(self) -> Dict[str, Any]:
+        return self.model_dump(exclude_none=True, exclude={"entrypoint"})
+
+    @classmethod
+    def from_callable(cls, c: Callable) -> "Function":
+        from inspect import getdoc
+        from phi.utils.json_schema import get_json_schema
+
+        parameters = {"type": "object", "properties": {}}
+        try:
+            type_hints = get_type_hints(c)
+            parameters = get_json_schema(type_hints)
+            # logger.debug(f"Type hints for {c.__name__}: {type_hints}")
+        except Exception as e:
+            logger.warning(f"Could not parse args for {c.__name__}: {e}")
+
+        return cls(
+            name=c.__name__,
+            description=getdoc(c),
+            parameters=parameters,
+            entrypoint=validate_call(c),
+        )
+
+
+class FunctionCall(BaseModel):
+    """Model for Assistant function calls"""
+
+    # The function to be called.
+    function: Function
+    # The arguments to call the function with.
+    arguments: Optional[Dict[str, Any]] = None
+    # The result of the function call.
+    result: Optional[Any] = None
+
+    def get_call_str(self) -> str:
+        """Returns a string representation of the function call."""
+        if self.arguments is None:
+            return f"{self.function.name}()"
+        return f"{self.function.name}({', '.join([f'{k}={v}' for k, v in self.arguments.items()])})"
+
+    def execute(self) -> bool:
+        """Runs the function call.
+
+        @return: True if the function call was successful, False otherwise.
+        """
+        if self.function.entrypoint is None:
+            return False
+
+        logger.debug(f"Running: {self.get_call_str()}")
+
+        # Call the function with no arguments if none are provided.
+        if self.arguments is None:
+            try:
+                self.result = self.function.entrypoint()
+                return True
+            except Exception as e:
+                logger.warning(f"Could not run function {self.get_call_str()}")
+                logger.error(e)
+                return False
+
+        try:
+            self.result = self.function.entrypoint(**self.arguments)
+            return True
+        except Exception as e:
+            logger.warning(f"Could not run function {self.get_call_str()}")
+            logger.error(e)
+            return False