Merge pull request #119 from rgbkrk/automatic-model-creation

automatically detect pydantic models and instantiate them

Author: rgbkrk

CHANGELOG.md

@@ -7,6 +7,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+- ⏃ Automatically instantiate pydantic models when they are passed as parameters to a function call
+
 ## [1.1.1]
 
 - Support setting a custom `base_url`

chatlab/registry.py

@@ -42,7 +42,21 @@ class WhatTime(BaseModel):
 import asyncio
 import inspect
 import json
-from typing import Any, Callable, Dict, Iterable, List, Optional, Type, TypedDict, Union, get_args, get_origin, overload
+from typing import (
+    Any,
+    Callable,
+    Dict,
+    Iterable,
+    List,
+    Optional,
+    Type,
+    TypedDict,
+    Union,
+    cast,
+    get_args,
+    get_origin,
+    overload,
+)
 
 from openai.types import FunctionDefinition
 from openai.types.chat.completion_create_params import Function, FunctionCall
@@ -419,9 +433,6 @@ async def call(self, name: str, arguments: Optional[str] = None) -> Any:
         if name is None:
             raise UnknownFunctionError("Function name must be provided")
 
-        function = self.get(name)
-        parameters: dict = {}
-
         # Handle the code interpreter hallucination
         if name == "python" and self.python_hallucination_function is not None:
             function = self.python_hallucination_function
@@ -433,24 +444,38 @@ async def call(self, name: str, arguments: Optional[str] = None) -> Any:
             if asyncio.iscoroutinefunction(function):
                 return await function(arguments)
             return function(arguments)
-        elif function is None:
+
+        possible_function = self.get(name)
+
+        if possible_function is None:
             raise UnknownFunctionError(f"Function {name} is not registered")
-        elif arguments is None or arguments == "":
-            parameters = {}
-        else:
+
+        function = possible_function
+
+        parameters: dict = {}
+
+        if arguments is not None:
             try:
                 parameters = json.loads(arguments)
-                # TODO: Validate parameters against schema
             except json.JSONDecodeError:
                 raise FunctionArgumentError(f"Invalid Function call on {name}. Arguments must be a valid JSON object")
 
-        if function is None:
-            raise UnknownFunctionError(f"Function {name} is not registered")
+        prepared_arguments = {}
+
+        for param_name, param in inspect.signature(function).parameters.items():
+            param_type = param.annotation
+            arg_value = parameters.get(param_name)
+
+            # Check if parameter type is a subclass of BaseModel and deserialize JSON into Pydantic model
+            if inspect.isclass(param_type) and issubclass(param_type, BaseModel):
+                prepared_arguments[param_name] = param_type.model_validate(arg_value)
+            else:
+                prepared_arguments[param_name] = cast(Any, arg_value)
 
         if asyncio.iscoroutinefunction(function):
-            result = await function(**parameters)
+            result = await function(**prepared_arguments)
         else:
-            result = function(**parameters)
+            result = function(**prepared_arguments)
         return result
 
     def __contains__(self, name) -> bool: