Add Coze function tool factory

pqai_agent/toolkit/coze_function_tools.py

@@ -0,0 +1,61 @@
+import types
+from typing import List, Dict
+import textwrap
+from pqai_agent.toolkit.function_tool import FunctionTool
+from pqai_agent.chat_service import Coze, TokenAuth, Message, CozeChat
+
+
+class FunctionToolFactory:
+    @staticmethod
+    def create_tool(bot_id: str, coze_client: CozeChat, func_name: str, func_desc: str) -> FunctionTool:
+        """
+        Create a FunctionTool for a specific Coze Bot.
+
+        Args:
+            bot_id (str): The ID of the Coze Bot.
+            coze_client (CozeChat): The Coze client instance to interact with the bot.
+            func_name (str): The name of the function to be used in the FunctionTool.
+            func_desc (str): A description of the function to be used in the FunctionTool.
+        Returns:
+            FunctionTool: A FunctionTool wrapping the Coze Bot interaction.
+        """
+
+        func_doc = f"""
+            {func_desc}
+
+            Args:
+                messages (List[Dict]): A list of messages to send to the bot.
+                custom_variables (Dict): Custom variables for the bot.
+            Returns:
+                str: The final response from the bot.
+            """
+        func_doc = textwrap.dedent(func_doc).strip()
+
+        def coze_func(messages: List[Dict], custom_variables: Dict = None) -> str:
+            # ATTENTION:
+            # custom_variables (Dict): Custom variables for the bot. THIS IS A TRICK.
+            # THIS PARAMETER SHOULD NOT BE VISIBLE TO THE AGENT AND FILLED BY THE SYSTEM.
+
+            # FIXME: Coze bot can return multimodal content.
+
+            response = coze_client.create(
+                bot_id=bot_id,
+                user_id='agent_tool_call',
+                messages=[Message.build_user_question_text(msg["text"]) for msg in messages],
+                custom_variables=custom_variables
+            )
+            if not response:
+                return 'Error in calling the function.'
+            return response
+
+        dynamic_function = types.FunctionType(
+            coze_func.__code__,
+            globals(),
+            name=func_name,
+            argdefs=coze_func.__defaults__,
+            closure=coze_func.__closure__
+        )
+        dynamic_function.__doc__ = func_doc
+
+        # Wrap the function in a FunctionTool
+        return FunctionTool(dynamic_function)