Openai decorators

CODEOWNERS

@@ -11,4 +11,3 @@
 #
 
 * @vrdmr @gavin-aguiar @YunchuWang @pdthummar @hallvictoria
-

azure/functions/decorators/constants.py

@@ -28,9 +28,16 @@
 DAPR_SECRET = "daprSecret"
 DAPR_PUBLISH = "daprPublish"
 DAPR_INVOKE = "daprInvoke"
-DAPR_PUBLISH = "daprPublish"
 DAPR_BINDING = "daprBinding"
 ORCHESTRATION_TRIGGER = "orchestrationTrigger"
 ACTIVITY_TRIGGER = "activityTrigger"
 ENTITY_TRIGGER = "entityTrigger"
 DURABLE_CLIENT = "durableClient"
+ASSISTANT_SKILLS_TRIGGER = "assistantSkillsTrigger"
+TEXT_COMPLETION = "TextCompletion"
+ASSISTANT_QUERY = "assistantQuery"
+EMBEDDINGS = "embeddings"
+EMBEDDINGS_STORE = "embeddingsStore"
+ASSISTANT_CREATE = "assistantCreate"
+ASSISTANT_POST = "assistantPost"
+SEMANTIC_SEARCH = "semanticSearch"

azure/functions/decorators/function_app.py

@@ -35,6 +35,9 @@
     parse_iterable_param_to_enums, StringifyEnumJsonEncoder
 from azure.functions.http import HttpRequest
 from .generic import GenericInputBinding, GenericTrigger, GenericOutputBinding
+from .openai import AssistantSkillTrigger, OpenAIModels, TextCompletionInput, AssistantCreateOutput, \
+    AssistantQueryInput, AssistantPostInput, InputType, EmbeddingsInput, semantic_search_system_prompt, \
+    SemanticSearchInput, EmbeddingsStoreOutput
 from .retry_policy import RetryPolicy
 from .function_name import FunctionName
 from .warmup import WarmUpTrigger
@@ -1435,6 +1438,32 @@ def decorator():
 
         return wrap
 
+    def assistant_skill_trigger(self,
+                                arg_name: str,
+                                task_description: str,
+                                data_type: Optional[
+                                    Union[DataType, str]] = None,
+                                **kwargs: Any) -> Callable[..., Any]:
+        """
+        TODO: PYDocs
+        """
+
+        @self._configure_function_builder
+        def wrap(fb):
+            def decorator():
+                fb.add_trigger(
+                    trigger=AssistantSkillTrigger(
+                        name=arg_name,
+                        task_description=task_description,
+                        data_type=parse_singular_param_to_enum(data_type,
+                                                               DataType),
+                        **kwargs))
+                return fb
+
+            return decorator()
+
+        return wrap
+
 
 class BindingApi(DecoratorApi, ABC):
     """Interface to extend for using existing binding decorator functions."""
@@ -2740,6 +2769,232 @@ def decorator():
 
         return wrap
 
+    def text_completion_input(self,
+                              arg_name: str,
+                              prompt: str,
+                              model: Optional[str] = OpenAIModels.DefaultChatModel,
+                              temperature: Optional[str] = "0.5",
+                              top_p: Optional[str] = None,
+                              max_tokens: Optional[str] = "100",
+                              data_type: Optional[
+                                  Union[DataType, str]] = None,
+                              **kwargs) \
+            -> Callable[..., Any]:
+        """
+        TODO: pydocs
+        """
+        @self._configure_function_builder
+        def wrap(fb):
+            def decorator():
+                fb.add_binding(
+                    binding=TextCompletionInput(
+                        name=arg_name,
+                        prompt=prompt,
+                        model=model,
+                        temperature=temperature,
+                        top_p=top_p,
+                        max_tokens=max_tokens,
+                        data_type=parse_singular_param_to_enum(data_type,
+                                                               DataType),
+                        **kwargs))
+                return fb
+
+            return decorator()
+
+        return wrap
+
+    def assistant_create_output(self, arg_name: str,
+                                data_type: Optional[
+                                    Union[DataType, str]] = None,
+                                **kwargs) \
+            -> Callable[..., Any]:
+        """
+        TODO: pydocs
+        """
+        @self._configure_function_builder
+        def wrap(fb):
+            def decorator():
+                fb.add_binding(
+                    binding=AssistantCreateOutput(
+                        name=arg_name,
+                        data_type=parse_singular_param_to_enum(data_type,
+                                                               DataType),
+                        **kwargs))
+                return fb
+
+            return decorator()
+
+        return wrap
+
+    def assistant_query_input(self,
+                              arg_name: str,
+                              id: str,
+                              timestamp_utc: str,
+                              data_type: Optional[
+                                  Union[DataType, str]] = None,
+                              **kwargs) \
+            -> Callable[..., Any]:
+        """
+        TODO: pydocs
+        """
+        @self._configure_function_builder
+        def wrap(fb):
+            def decorator():
+                fb.add_binding(
+                    binding=AssistantQueryInput(
+                        name=arg_name,
+                        id=id,
+                        timestamp_utc=timestamp_utc,
+                        data_type=parse_singular_param_to_enum(data_type,
+                                                               DataType),
+                        **kwargs))
+                return fb
+
+            return decorator()
+
+        return wrap
+
+    def assistant_post_input(self, arg_name: str,
+                             id: str,
+                             user_message: str,
+                             data_type: Optional[
+                                 Union[DataType, str]] = None,
+                             **kwargs) \
+            -> Callable[..., Any]:
+        """
+        TODO: pydocs
+        """
+
+        @self._configure_function_builder
+        def wrap(fb):
+            def decorator():
+                fb.add_binding(
+                    binding=AssistantPostInput(
+                        name=arg_name,
+                        id=id,
+                        user_message=user_message,
+                        data_type=parse_singular_param_to_enum(data_type,
+                                                               DataType),
+                        **kwargs))
+                return fb
+
+            return decorator()
+
+        return wrap
+
+    def embeddings_input(self,
+                         arg_name: str,
+                         input: str,
+                         input_type: InputType,
+                         model: Optional[str] = None,
+                         max_chunk_length: Optional[int] = 8 * 1024,
+                         max_overlap: Optional[int] = 128,
+                         data_type: Optional[
+                             Union[DataType, str]] = None,
+                         **kwargs) \
+            -> Callable[..., Any]:
+        """
+        TODO: pydocs
+        """
+
+        @self._configure_function_builder
+        def wrap(fb):
+            def decorator():
+                fb.add_binding(
+                    binding=EmbeddingsInput(
+                        name=arg_name,
+                        input=input,
+                        input_type=input_type,
+                        model=model,
+                        max_chunk_length=max_chunk_length,
+                        max_overlap=max_overlap,
+                        data_type=parse_singular_param_to_enum(data_type,
+                                                               DataType),
+                        **kwargs))
+                return fb
+
+            return decorator()
+
+        return wrap
+
+    def semantic_search_input(self,
+                              arg_name: str,
+                              connection_name: str,
+                              collection: str,
+                              query: Optional[str] = None,
+                              embeddings_model: Optional[str] = OpenAIModels.DefaultEmbeddingsModel,
+                              chat_model: Optional[str] = OpenAIModels.DefaultChatModel,
+                              system_prompt: Optional[str] = semantic_search_system_prompt,
+                              max_knowledge_count: Optional[int] = 1,
+                              data_type: Optional[
+                                  Union[DataType, str]] = None,
+                              **kwargs) \
+            -> Callable[..., Any]:
+        """
+        TODO: pydocs
+        """
+
+        @self._configure_function_builder
+        def wrap(fb):
+            def decorator():
+                fb.add_binding(
+                    binding=SemanticSearchInput(
+                        name=arg_name,
+                        connnection_name=connection_name,
+                        collection=collection,
+                        query=query,
+                        embeddings_model=embeddings_model,
+                        chat_model=chat_model,
+                        system_prompt=system_prompt,
+                        max_knowledge_count=max_knowledge_count,
+                        data_type=parse_singular_param_to_enum(data_type,
+                                                               DataType),
+                        **kwargs))
+                return fb
+
+            return decorator()
+
+        return wrap
+
+    def embeddings_store_output(self,
+                                arg_name: str,
+                                input: str,
+                                input_type: InputType,
+                                connection_name: str,
+                                collection: str,
+                                model: Optional[str] = OpenAIModels.DefaultEmbeddingsModel,
+                                max_chunk_length: Optional[int] = 8 * 1024,
+                                max_overlap: Optional[int] = 128,
+                                data_type: Optional[
+                                    Union[DataType, str]] = None,
+                                **kwargs) \
+            -> Callable[..., Any]:
+        """
+        TODO: pydocs
+        """
+
+        @self._configure_function_builder
+        def wrap(fb):
+            def decorator():
+                fb.add_binding(
+                    binding=EmbeddingsStoreOutput(
+                        name=arg_name,
+                        input=input,
+                        input_type=input_type,
+                        connnection_name=connection_name,
+                        collection=collection,
+                        model=model,
+                        max_chunk_length=max_chunk_length,
+                        max_overlap=max_overlap,
+                        data_type=parse_singular_param_to_enum(data_type,
+                                                               DataType),
+                        **kwargs))
+                return fb
+
+            return decorator()
+
+        return wrap
+
 
 class SettingsApi(DecoratorApi, ABC):
     """Interface to extend for using existing settings decorator in