feat: Prompt Profiles for subordinate agent, prompt folder metadata, prompt plugins

prompts/default/agent.system.tool.call_sub.md

@@ -6,21 +6,51 @@ message field: always describe role, task details goal overview for new subordin
 delegate specific subtasks not entire task
 reset arg usage:
   "true": spawn new subordinate
-  "false": continue existing subordinate
+  "false": ask respond to subordinate
+prompt_profile defines subordinate specialization
 if superior, orchestrate
-respond to existing subordinates using call_subordinate tool with reset false
+respond to existing subordinates using call_subordinate tool with reset: "false"
 
-example usage
+#### if you are subordinate:
+- superior is {{agent_name}} minus 1
+- execute the task you were assigned
+- delegate further if asked
+
+#### Arguments:
+- message (string): The detailed task for the subordinate to accomplish
+- reset (boolean): Whether to discard current subordinate dialog and spawn a fresh subordinate. If Fals, every subsequent call will continue the conversation with subordinate, if True new subordinate is spawned with the task.
+- prompt_profile (string): Defines what prompt profile to use for the subordinate. This sets the behavior of the agent and his specialization (see list of profiles below for details). Choose a profile best suited for the task
+
+##### Prompt Profiles (prompt_profile options)
+{{prompt_profiles}}
+
+#### example usage
 ~~~json
 {
     "thoughts": [
         "The result seems to be ok but...",
-        "I will ask a coder subordinate to fix...",
+        "I will ask a subordinate to fix...",
+    ],
+    "tool_name": "call_subordinate",
+    "tool_args": {
+        "message": "...",
+        "reset": "true",
+        "prompt_profile": "default",
+    }
+}
+~~~
+
+~~~json
+{
+    "thoughts": [
+        "This task is challenging and requires a data analyst",
+        "The research_agent profile supports data analysis",
     ],
     "tool_name": "call_subordinate",
     "tool_args": {
         "message": "...",
-        "reset": "true"
+        "reset": "true",
+        "prompt_profile": "research_agent",
     }
 }
-~~~
+~~~

prompts/default/agent.system.tool.call_sub.py

@@ -0,0 +1,20 @@
+import json
+from typing import Any
+from python.helpers.files import VariablesPlugin
+from python.helpers import files
+from python.helpers.print_style import PrintStyle
+
+
+class CallSubordinate(VariablesPlugin):
+    def get_variables(self) -> dict[str, Any]:
+        meta = files.read_file(files.get_abs_path("prompts", "meta.json"))
+        profiles = ""
+        try:
+            for profile in json.loads(meta):
+                profiles += f"- {profile['name']}: {profile['description']}\n"
+        except Exception as e:
+            PrintStyle().error(f"Error loading prompt profiles: {e}")
+            profiles = "- default: Default Agent-Zero AI Assistant"
+        return {
+            "prompt_profiles": profiles
+        }

prompts/meta.json

@@ -0,0 +1,14 @@
+[
+    {
+        "name": "default",
+        "description": "Default Agent-Zero AI Assistant"
+    },
+    {
+        "name": "reflection",
+        "description": "Agent-Zero AI Assistant with self-reflection and reasoning capabilities"
+    },
+    {
+        "name": "research_agent",
+        "description": "Agent-Zero AI Assistant with academic and corporate research, writing, data analysis and reporting capabilities"
+    }
+]

python/helpers/files.py

@@ -1,12 +1,74 @@
+from abc import ABC, abstractmethod
 from fnmatch import fnmatch
 import json
-import os, re
-import base64
-
+import os
+import sys
 import re
+import base64
 import shutil
 import tempfile
+from typing import Any
 import zipfile
+import importlib
+import importlib.util
+import inspect
+from python.helpers.print_style import PrintStyle
+
+
+class VariablesPlugin(ABC):
+    @abstractmethod
+    def get_variables(self) -> dict[str, Any]:  # type: ignore
+        pass
+
+
+def load_plugin_variables(file: str, backup_dirs: list[str] | None = None) -> dict[str, Any]:
+    if not file.endswith(".md"):
+        return {}
+
+    if backup_dirs is None:
+        backup_dirs = []
+
+    try:
+        PrintStyle().debug(f"Loading prompt variables for {file}")
+        plugin_file = find_file_in_dirs(
+            get_abs_path(dirname(file), basename(file, ".md") + ".py"),
+            backup_dirs
+        )
+    except FileNotFoundError:
+        PrintStyle().debug(f"No plugin file found: {get_abs_path(dirname(file), basename(file, '.md') + '.py')}")
+        plugin_file = None
+
+    if plugin_file and exists(plugin_file):
+        # load python code and extract variables variables from it
+        PrintStyle().debug(f"Importing module from file: {plugin_file}")
+        module = None
+        module_name = dirname(plugin_file).replace("/", ".") + "." + basename(plugin_file, '.py')
+        try:
+            spec = importlib.util.spec_from_file_location(module_name, plugin_file)
+            if not spec:
+                PrintStyle().debug(f"Module not found: {plugin_file} {module_name}")
+                return {}
+            module = importlib.util.module_from_spec(spec)
+            sys.modules[spec.name] = module
+            spec.loader.exec_module(module)  # type: ignore
+        except ImportError:
+            PrintStyle().debug(f"Module not found: {plugin_file} {module_name}")
+            return {}
+
+        if module is None:
+            PrintStyle().debug(f"Module not found: {plugin_file} {module_name}")
+            return {}
+
+        # Get all classes in the module
+        class_list = inspect.getmembers(module, inspect.isclass)
+        PrintStyle().debug(f"Found {len(class_list)} classes in module: {basename(plugin_file, '.py')}")
+        # Filter for classes that are subclasses of VariablesPlugin
+        # iterate backwards to skip imported superclasses
+        for cls in reversed(class_list):
+            if cls[1] is not VariablesPlugin and issubclass(cls[1], VariablesPlugin):
+                PrintStyle().debug(f"Loading prompt variables from {plugin_file}")
+                return cls[1]().get_variables()  # type: ignore
+    return {}
 
 from python.helpers.strings import sanitize_string
 
@@ -15,13 +77,15 @@ def parse_file(_relative_path, _backup_dirs=None, _encoding="utf-8", **kwargs):
     content = read_file(_relative_path, _backup_dirs, _encoding)
     is_json = is_full_json_template(content)
     content = remove_code_fences(content)
+    variables = load_plugin_variables(_relative_path, _backup_dirs) or {}  # type: ignore
+    variables.update(kwargs)
     if is_json:
-        content = replace_placeholders_json(content, **kwargs)
+        content = replace_placeholders_json(content, **variables)
         obj = json.loads(content)
-        # obj = replace_placeholders_dict(obj, **kwargs)
+        # obj = replace_placeholders_dict(obj, **variables)
         return obj
     else:
-        content = replace_placeholders_text(content, **kwargs)
+        content = replace_placeholders_text(content, **variables)
         return content
 
 
@@ -37,11 +101,15 @@ def read_file(_relative_path, _backup_dirs=None, _encoding="utf-8", **kwargs):
         # content = remove_code_fences(f.read())
         content = f.read()
 
+    variables = load_plugin_variables(_relative_path, _backup_dirs) or {}  # type: ignore
+    variables.update(kwargs)
+
     # Replace placeholders with values from kwargs
-    content = replace_placeholders_text(content, **kwargs)
+    content = replace_placeholders_text(content, **variables)
 
     # Process include statements
     content = process_includes(
+        # here we use kwargs, the plugin variables are not inherited
         content, os.path.dirname(_relative_path), _backup_dirs, **kwargs
     )
 
@@ -162,9 +230,6 @@ def find_file_in_dirs(file_path, backup_dirs):
     )
 
 
-import re
-
-
 def remove_code_fences(text):
     # Pattern to match code fences with optional language specifier
     pattern = r"(```|~~~)(.*?\n)(.*?)(\1)"
@@ -179,9 +244,6 @@ def remove_code_fences(text):
     return result
 
 
-import re
-
-
 def is_full_json_template(text):
     # Pattern to match the entire text enclosed in ```json or ~~~json fences
     pattern = r"^\s*(```|~~~)\s*json\s*\n(.*?)\n\1\s*$"
@@ -275,6 +337,16 @@ def get_base_dir():
     return base_dir
 
 
+def basename(path: str, suffix: str | None = None):
+    if suffix:
+        return os.path.basename(path).removesuffix(suffix)
+    return os.path.basename(path)
+
+
+def dirname(path: str):
+    return os.path.dirname(path)
+
+
 def is_in_base_dir(path: str):
     # check if the given path is within the base directory
     base_dir = get_base_dir()
@@ -327,6 +399,4 @@ def move_file(relative_path: str, new_path: str):
 
 def safe_file_name(filename: str) -> str:
     # Replace any character that's not alphanumeric, dash, underscore, or dot with underscore
-    import re
-
-    return re.sub(r"[^a-zA-Z0-9-._]", "_", filename)
+    return re.sub(r'[^a-zA-Z0-9-._]', '_', filename)

python/tools/call_subordinate.py

@@ -16,10 +16,20 @@ class Delegation(Tool):
             sub.set_data(Agent.DATA_NAME_SUPERIOR, self.agent)
             self.agent.set_data(Agent.DATA_NAME_SUBORDINATE, sub)
 
+        prompt_profile = kwargs.get("prompt_profile", "default")
+        agent_prompt_profile = self.agent.config.prompts_subdir
+        if agent_prompt_profile != prompt_profile:
+            self.agent.config.prompts_subdir = prompt_profile
+
         # add user message to subordinate agent
         subordinate: Agent = self.agent.get_data(Agent.DATA_NAME_SUBORDINATE)
         subordinate.hist_add_user_message(UserMessage(message=message, attachments=[]))
+
         # run subordinate monologue
         result = await subordinate.monologue()
+
+        # reset prompt profile
+        self.agent.config.prompts_subdir = agent_prompt_profile
+
         # result
         return Response(message=result, break_loop=False)