Support function edit sdk

Signed-off-by: junjie.jiang <junjie.jiang@zilliz.com>

Author: junjiejiangjjj

examples/function_edit.py

@@ -0,0 +1,62 @@
+from pymilvus import (
+    MilvusClient,
+    Function, DataType, FunctionType,
+)
+
+collection_name = "text_embedding"
+
+milvus_client = MilvusClient("http://localhost:19530")
+
+has_collection = milvus_client.has_collection(collection_name, timeout=5)
+if has_collection:
+    milvus_client.drop_collection(collection_name)
+
+schema = milvus_client.create_schema()
+schema.add_field("id", DataType.INT64, is_primary=True, auto_id=False)
+schema.add_field("document", DataType.VARCHAR, max_length=9000)
+schema.add_field("dense", DataType.FLOAT_VECTOR, dim=1536)
+
+text_embedding_function = Function(
+    name="openai",
+    function_type=FunctionType.TEXTEMBEDDING,
+    input_field_names=["document"],
+    output_field_names="dense",
+    params={
+        "provider": "openai",
+        "model_name": "text-embedding-3-small",
+    }
+)
+
+schema.add_function(text_embedding_function)
+
+index_params = milvus_client.prepare_index_params()
+index_params.add_index(
+    field_name="dense",
+    index_name="dense_index",
+    index_type="AUTOINDEX",
+    metric_type="IP",
+)
+
+ret = milvus_client.create_collection(collection_name, schema=schema, index_params=index_params, consistency_level="Strong")
+
+ret = milvus_client.describe_collection(collection_name)
+print(ret["functions"][0])
+
+text_embedding_function.params["user"] = "user123"
+
+milvus_client.alter_collection_function(collection_name, "openai", text_embedding_function)
+
+ret = milvus_client.describe_collection(collection_name)
+print(ret["functions"][0])
+
+milvus_client.drop_collection_function(collection_name, "openai")
+
+ret = milvus_client.describe_collection(collection_name)
+print(ret["functions"])
+
+text_embedding_function.params["user"] = "user1234"
+
+milvus_client.add_collection_function(collection_name, text_embedding_function)
+
+ret = milvus_client.describe_collection(collection_name)
+print(ret["functions"][0])

pymilvus/client/async_grpc_handler.py

@@ -1306,6 +1306,60 @@ async def add_collection_field(
         )
         check_status(status)
 
+    @retry_on_rpc_failure()
+    async def drop_collection_function(
+        self,
+        collection_name: str,
+        function_name: str,
+        timeout: Optional[float] = None,
+        **kwargs,
+    ):
+        await self.ensure_channel_ready()
+        check_pass_param(collection_name=collection_name, timeout=timeout)
+        request = Prepare.drop_collection_function_request(collection_name, function_name)
+
+        status = await self._async_stub.DropCollectionFunction(
+            request, timeout=timeout, metadata=_api_level_md(**kwargs)
+        )
+        check_status(status)
+
+    @retry_on_rpc_failure()
+    async def add_collection_function(
+        self,
+        collection_name: str,
+        function: Function,
+        timeout: Optional[float] = None,
+        **kwargs,
+    ):
+        await self.ensure_channel_ready()
+        check_pass_param(collection_name=collection_name, timeout=timeout)
+        request = Prepare.add_collection_function_request(collection_name, function)
+
+        status = await self._async_stub.AddCollectionFunction(
+            request, timeout=timeout, metadata=_api_level_md(**kwargs)
+        )
+        check_status(status)
+
+    @retry_on_rpc_failure()
+    async def alter_collection_function(
+        self,
+        collection_name: str,
+        function_name: str,
+        function: Function,
+        timeout: Optional[float] = None,
+        **kwargs,
+    ):
+        await self.ensure_channel_ready()
+        check_pass_param(collection_name=collection_name, timeout=timeout)
+        request = Prepare.alter_collection_function_request(
+            collection_name, function_name, function
+        )
+
+        status = await self._async_stub.AlterCollectionFunction(
+            request, timeout=timeout, metadata=_api_level_md(**kwargs)
+        )
+        check_status(status)
+
     @retry_on_rpc_failure()
     async def list_indexes(self, collection_name: str, timeout: Optional[float] = None, **kwargs):
         await self.ensure_channel_ready()

pymilvus/client/grpc_handler.py

@@ -408,6 +408,57 @@ def add_collection_field(
         )
         check_status(status)
 
+    @retry_on_rpc_failure()
+    def drop_collection_function(
+        self,
+        collection_name: str,
+        function_name: str,
+        timeout: Optional[float] = None,
+        **kwargs,
+    ):
+        check_pass_param(collection_name=collection_name, timeout=timeout)
+        request = Prepare.drop_collection_function_request(collection_name, function_name)
+
+        status = self._stub.DropCollectionFunction(
+            request, timeout=timeout, metadata=_api_level_md(**kwargs)
+        )
+        check_status(status)
+
+    @retry_on_rpc_failure()
+    def add_collection_function(
+        self,
+        collection_name: str,
+        function: Function,
+        timeout: Optional[float] = None,
+        **kwargs,
+    ):
+        check_pass_param(collection_name=collection_name, timeout=timeout)
+        request = Prepare.add_collection_function_request(collection_name, function)
+
+        status = self._stub.AddCollectionFunction(
+            request, timeout=timeout, metadata=_api_level_md(**kwargs)
+        )
+        check_status(status)
+
+    @retry_on_rpc_failure()
+    def alter_collection_function(
+        self,
+        collection_name: str,
+        function_name: str,
+        function: Function,
+        timeout: Optional[float] = None,
+        **kwargs,
+    ):
+        check_pass_param(collection_name=collection_name, timeout=timeout)
+        request = Prepare.alter_collection_function_request(
+            collection_name, function_name, function
+        )
+
+        status = self._stub.AlterCollectionFunction(
+            request, timeout=timeout, metadata=_api_level_md(**kwargs)
+        )
+        check_status(status)
+
     @retry_on_rpc_failure()
     def alter_collection_properties(
         self, collection_name: str, properties: List, timeout: Optional[float] = None, **kwargs

pymilvus/client/prepare.py

@@ -242,16 +242,7 @@ def get_schema_from_collection_schema(
             schema.struct_array_fields.append(struct_schema)
 
         for f in fields.functions:
-            function_schema = schema_types.FunctionSchema(
-                name=f.name,
-                description=f.description,
-                type=f.type,
-                input_field_names=f.input_field_names,
-                output_field_names=f.output_field_names,
-            )
-            for k, v in f.params.items():
-                kv_pair = common_types.KeyValuePair(key=str(k), value=str(v))
-                function_schema.params.append(kv_pair)
+            function_schema = cls.convert_function_to_function_schema(f)
             schema.functions.append(function_schema)
 
         return schema
@@ -364,6 +355,34 @@ def get_schema(
     def drop_collection_request(cls, collection_name: str) -> milvus_types.DropCollectionRequest:
         return milvus_types.DropCollectionRequest(collection_name=collection_name)
 
+    @classmethod
+    def drop_collection_function_request(
+        cls, collection_name: str, function_name: str
+    ) -> milvus_types.DropCollectionFunctionRequest:
+        return milvus_types.DropCollectionFunctionRequest(
+            collection_name=collection_name, function_name=function_name
+        )
+
+    @classmethod
+    def add_collection_function_request(
+        cls, collection_name: str, f: Function
+    ) -> milvus_types.AddCollectionFunctionRequest:
+        function_schema = cls.convert_function_to_function_schema(f)
+        return milvus_types.AddCollectionFunctionRequest(
+            collection_name=collection_name, functionSchema=function_schema
+        )
+
+    @classmethod
+    def alter_collection_function_request(
+        cls, collection_name: str, function_name: str, f: Function
+    ) -> milvus_types.AlterCollectionFunctionRequest:
+        function_schema = cls.convert_function_to_function_schema(f)
+        return milvus_types.AlterCollectionFunctionRequest(
+            collection_name=collection_name,
+            function_name=function_name,
+            functionSchema=function_schema,
+        )
+
     @classmethod
     def add_collection_field_request(
         cls,
@@ -2450,3 +2469,17 @@ def update_replicate_configuration_request(
         return milvus_types.UpdateReplicateConfigurationRequest(
             replicate_configuration=replicate_configuration
         )
+
+    @staticmethod
+    def convert_function_to_function_schema(f: Function) -> schema_types.FunctionSchema:
+        function_schema = schema_types.FunctionSchema(
+            name=f.name,
+            description=f.description,
+            type=f.type,
+            input_field_names=f.input_field_names,
+            output_field_names=f.output_field_names,
+        )
+        for k, v in f.params.items():
+            kv_pair = common_types.KeyValuePair(key=str(k), value=str(v))
+            function_schema.params.append(kv_pair)
+        return function_schema

pymilvus/milvus_client/async_milvus_client.py

@@ -702,6 +702,105 @@ async def add_collection_field(
             **kwargs,
         )
 
+
+    async def add_collection_function(
+        self, collection_name: str, function: Function, timeout: Optional[float] = None, **kwargs
+    ):
+        """Add a new function to the collection.
+
+        Args:
+            collection_name(``string``): The name of collection.
+            function(``Function``):  The function schema.
+            timeout (``float``, optional): A duration of time in seconds to allow for the RPC.
+                If timeout is set to None, the client keeps waiting until the server
+                responds or an error occurs.
+            **kwargs (``dict``): Optional field params
+
+        Raises:
+            MilvusException: If anything goes wrong
+        """
+        conn = self._get_connection()
+        await conn.add_collection_function(
+            collection_name,
+            function,
+            timeout=timeout,
+            **kwargs,
+        )
+
+    async def alter_collection_function(
+        self,
+        collection_name: str,
+        function_name: str,
+        function: Function,
+        timeout: Optional[float] = None,
+        **kwargs,
+    ):
+        """Alter a function in the collection.
+
+        Args:
+            collection_name(``string``): The name of collection.
+            function_name(``string``): The function name that needs to be modified
+            function(``Function``):  The function schema.
+            timeout (``float``, optional): A duration of time in seconds to allow for the RPC.
+                If timeout is set to None, the client keeps waiting until the server
+                responds or an error occurs.
+            **kwargs (``dict``): Optional field params
+
+        Raises:
+            MilvusException: If anything goes wrong
+        """
+        conn = self._get_connection()
+        await conn.alter_collection_function(
+            collection_name,
+            function_name,
+            function,
+            timeout=timeout,
+            **kwargs,
+        )
+
+    async def drop_collection_function(
+        self, collection_name: str, function_name: str, timeout: Optional[float] = None, **kwargs
+    ):
+        """Drop a function from the collection.
+
+        Args:
+            collection_name(``string``): The name of collection.
+            function_name(``string``): The function name that needs to be dropped
+            timeout (``float``, optional): A duration of time in seconds to allow for the RPC.
+                If timeout is set to None, the client keeps waiting until the server
+                responds or an error occurs.
+            **kwargs (``dict``): Optional field params
+
+        Raises:
+            MilvusException: If anything goes wrong
+        """
+        conn = self._get_connection()
+        await conn.drop_collection_function(
+            collection_name,
+            function_name,
+            timeout=timeout,
+            **kwargs,
+        )
+
+    @classmethod
+    def create_schema(cls, **kwargs):
+        kwargs["check_fields"] = False  # do not check fields for now
+        return CollectionSchema([], **kwargs)
+
+    @classmethod
+    def create_field_schema(
+        cls, name: str, data_type: DataType, desc: str = "", **kwargs
+    ) -> FieldSchema:
+        return FieldSchema(name, data_type, desc, **kwargs)
+
+    @classmethod
+    def prepare_index_params(cls, field_name: str = "", **kwargs) -> IndexParams:
+        index_params = IndexParams()
+        if field_name:
+            validate_param("field_name", field_name, str)
+            index_params.add_index(field_name, **kwargs)
+        return index_params
+
     async def close(self):
         await connections.async_remove_connection(self._using)