feat: add file_save tool

Author: zhayujie

agentmesh/protocal/agent.py

@@ -9,6 +9,7 @@
 from agentmesh.protocal.context import TeamContext, AgentOutput
 from agentmesh.protocal.result import AgentAction, AgentActionType, ToolResult, AgentResult
 from agentmesh.tools.base_tool import BaseTool
+from agentmesh.tools.base_tool import ToolStage
 
 
 class Agent:
@@ -48,10 +49,11 @@ def add_tool(self, tool: BaseTool):
         self.tools.append(tool)
 
     def _build_tools_prompt(self) -> str:
-        """Build the tool list description"""
+        """Build the tool list description, only including pre-process tools"""
         return "\n".join([
             f"{tool.name}: {tool.description} (parameters: {tool.params})"
             for tool in self.tools
+            if tool.stage == ToolStage.PRE_PROCESS  # Only include pre-process tools
         ])
 
     def _build_react_prompt(self) -> str:
@@ -98,10 +100,19 @@ def _build_react_prompt(self) -> str:
         return tools_prompt + ext_data_prompt + current_task_prompt
 
     def _find_tool(self, tool_name: str):
+        """Find and return a tool with the specified name"""
         for tool in self.tools:
             if tool.name == tool_name:
-                tool.model = self.model
-                return tool
+                # Only pre-process stage tools can be actively called
+                if tool.stage == ToolStage.PRE_PROCESS:
+                    tool.model = self.model
+                    tool.context = self  # Set tool context
+                    return tool
+                else:
+                    # If it's a post-process tool, return None to prevent direct calling
+                    logger.warning(f"Tool {tool_name} is a post-process tool and cannot be called directly.")
+                    return None
+        return None
 
     # output function based on mode
     def output(self, message="", end="\n"):
@@ -233,6 +244,16 @@ def step(self):
             if "final_answer" in parsed and parsed["final_answer"] and parsed["final_answer"].lower() not in ["null",
                                                                                                               "none"]:
                 final_answer = parsed["final_answer"]
+                self.final_answer = final_answer
+
+                # Store the final answer in team context
+                self.team_context.agent_outputs.append(
+                    AgentOutput(agent_name=self.name, output=final_answer)
+                )
+
+                # Execute all post-process tools
+                self._execute_post_process_tools()
+
                 break
 
             # Handle tool invocation
@@ -273,19 +294,31 @@ def step(self):
 
             current_step += 1
 
-        # Save final result
-        result = final_answer if final_answer else raw_response
-        self.final_answer = result
-        self.team_context.agent_outputs.append(
-            AgentOutput(agent_name=self.name, output=result)
-        )
-
         # Return a StepResult object
         return AgentResult.success(
             final_answer=self.final_answer,
             step_count=current_step + 1  # +1 because we count steps starting from 1
         )
 
+    def _execute_post_process_tools(self):
+        """Execute all post-process stage tools"""
+        # Get all post-process stage tools
+        post_process_tools = [tool for tool in self.tools if tool.stage == ToolStage.POST_PROCESS]
+
+        # Execute each tool
+        for tool in post_process_tools:
+            # Set tool context
+            tool.context = self
+
+            # Execute tool (with empty parameters, tool will extract needed info from context)
+            result = tool.execute({})
+
+            # Log result
+            if result.status == "success":
+                logger.info(f"Post-process tool {tool.name} executed successfully: {result.result.get('message', '')}")
+            else:
+                logger.warning(f"Post-process tool {tool.name} failed: {result.result}")
+
     def should_invoke_next_agent(self) -> int:
         """
         Determine if the next agent should be invoked based on the reply.

agentmesh/protocal/context.py

@@ -1,7 +1,7 @@
 class TeamContext:
     def __init__(self, name: str, description: str, rule: str, agents: list):
         """
-        Initialize the GroupContext with a name, description, rules, a list of agents, and a user question.
+        Initialize the TeamContext with a name, description, rules, a list of agents, and a user question.
         :param name: The name of the group context.
         :param description: A description of the group context.
         :param rule: The rules governing the group context.
@@ -14,6 +14,7 @@ def __init__(self, name: str, description: str, rule: str, agents: list):
         self.user_task = ""  # For backward compatibility
         self.task = None  # Will be a Task instance
         self.model = None  # Will be an instance of LLMModel
+        self.task_short_name = None  # Store the task directory name
         # List of agents that have been executed
         self.agent_outputs: list = []
 

agentmesh/protocal/team.py

@@ -1,5 +1,6 @@
 from typing import Union, Literal
 import json
+import re
 
 from agentmesh.common import LoadingIndicator
 from agentmesh.common.utils import string_util
@@ -29,6 +30,7 @@ def __init__(self, name: str, description: str, rule: str = "", model: LLMModel
         self.context = TeamContext(name, description, rule, agents=self.agents)
         self.model: LLMModel = model  # Instance of LLMModel
         self.max_steps = max_steps  # Maximum total steps across all agents
+        self.task_short_name = ""
 
     def add(self, agent: Agent):
         """
@@ -92,7 +94,7 @@ def output(message, end="\n"):
 
             prompt = GROUP_DECISION_PROMPT.format(group_name=self.name, group_description=self.description,
                                                   group_rules=self.rule, agents_str=agents_str,
-                                                  user_question=task.get_text())
+                                                  user_task=task.get_text())
 
             # Start loading animation (only in print mode)
             loading = None
@@ -132,6 +134,8 @@ def output(message, end="\n"):
                 decision_res = string_util.json_loads(reply_text)
                 selected_agent_id = decision_res.get("id")  # Extract the id from the response
                 subtask = decision_res.get("subtask")
+                task_short_name = decision_res.get("task_short_name")
+                self.context.task_short_name = task_short_name
 
                 # Find the selected agent based on the id
                 selected_agent: Agent = self.agents[selected_agent_id]
@@ -263,19 +267,24 @@ def output(message, end="\n"):
 
 
 GROUP_DECISION_PROMPT = """## Role
-As an expert in team task allocation, your role is to select the most suitable team member to initially address the task at hand, and give the subtask that need to be answered by this member. 
-After the task is completed, the results will pass to next member.
+You are the coordinator for a team of AI agents. Your job is to analyze the user's task and decide which agent in the team should handle it first, and give the subtask that need to be answered by this member.
 
-## Team
-Team Name: {group_name}
-Team Description: {group_description}
-Team Rules: {group_rules}
+## Team Information
+Team name: {group_name}
+Team description: {group_description}
+Team rules: {group_rules}
 
-## List of team members:
+## Available Agents
 {agents_str}
 
-## User Question:
-{user_question}
+## User Task
+{user_task}
+
+## Output Format
+Return your response in JSON format with the following fields:
+- id: The ID of the selected agent
+- subtask: the subtask that need to be answered by this member (use the same language as the user's task and preserve all key information from the original task)
+- task_short_name: A descriptive name for the user's original task (lowercase with underscores, max 5 English words)
 
 Please return the result in the following JSON structure which can be parsed directly by json.loads(), no extra content:
-{{"id": <member_id>, "subtask": ""}}"""
+{{"id": <member_id>, "subtask": "", "task_short_name": ""}}"""

agentmesh/tools/__init__.py

@@ -6,7 +6,8 @@
 from agentmesh.tools.google_search.google_search import GoogleSearch
 from agentmesh.tools.calculator.calculator import Calculator
 from agentmesh.tools.current_time.current_time import CurrentTime
-from agentmesh.tools.file_output.file_output import FileOutput
+from agentmesh.tools.file_save.file_save import FileSave
+
 
 # Delayed import for BrowserTool
 def _import_browser_tool():
@@ -22,8 +23,10 @@ def __init__(self, *args, **kwargs):
                     "Please install it with 'pip install browser-use>=0.1.40' or "
                     "'pip install agentmesh-sdk[full]'."
                 )
+
         return BrowserToolPlaceholder
 
+
 # Dynamically set BrowserTool
 BrowserTool = _import_browser_tool()
 

agentmesh/tools/base_tool.py

@@ -1,9 +1,16 @@
+from enum import Enum
 from typing import Any
 from pydantic import BaseModel, Field
 from agentmesh.models.llm.base_model import LLMModel
 from agentmesh.common import logger
 
 
+class ToolStage(Enum):
+    """Enum representing tool decision stages"""
+    PRE_PROCESS = "pre_process"  # Tools that need to be actively selected by the agent
+    POST_PROCESS = "post_process"  # Tools that automatically execute after final_answer
+
+
 class ToolResult(BaseModel):
     status: str = Field(default=None)
     result: Any = Field(default=None)
@@ -19,6 +26,11 @@ def fail(result, ext_data: Any = None):
 
 
 class BaseTool:
+    """Base class for all tools."""
+
+    # Default decision stage is pre-process
+    stage = ToolStage.PRE_PROCESS
+
     # Class attributes must be inherited
     name: str = "base_tool"
     description: str = "Base tool"
@@ -63,3 +75,13 @@ def _parse_schema(cls) -> dict:
                 prop.get("default", ...)
             )
         return fields
+
+    def should_auto_execute(self, context) -> bool:
+        """
+        Determine if this tool should be automatically executed based on context.
+
+        :param context: The agent context
+        :return: True if the tool should be executed, False otherwise
+        """
+        # Only tools in post-process stage will be automatically executed
+        return self.stage == ToolStage.POST_PROCESS

agentmesh/tools/browser/browser_action.py

@@ -50,3 +50,10 @@ class SwitchTab(BrowserAction):
     """Switch to a tab"""
     code = "switch_tab"
     description = "Switched to tab"
+
+
+class SendKeys(BrowserAction):
+    """Switch to a tab"""
+    code = "send_keys"
+    description = "Send strings of special keyboard keys like Escape, Backspace, Insert, PageDown, Delete, Enter, " \
+                  "ArrowRight, ArrowUp, etc"

agentmesh/tools/browser/browser_tool.py

@@ -29,7 +29,8 @@ def _import_browser_use():
 
 
 def _get_action_prompt():
-    action_classes = [Navigate, ClickElement, ExtractContent, InputText, OpenTab, SwitchTab, ScrollDown, ScrollUp]
+    action_classes = [Navigate, ClickElement, ExtractContent, InputText, OpenTab, SwitchTab, ScrollDown, ScrollUp,
+                      SendKeys]
     action_prompt = ""
     for action_class in action_classes:
         action_prompt += f"{action_class.code}: {action_class.description}\n"
@@ -76,6 +77,10 @@ class BrowserTool(BaseTool):
             "scroll_amount": {
                 "type": "integer",
                 "description": f"The number of pixels to scroll (required for '{ScrollDown.code}', '{ScrollUp.code}' action)."
+            },
+            "keys": {
+                "type": "string",
+                "description": f"Keys to send (required for '{SendKeys.code}' action)"
             }
         },
         "required": ["operation"]
@@ -246,6 +251,14 @@ async def _execute_async(self, action: str, params: Dict[str, Any]) -> ToolResul
             msg = f"{action} by {scroll_amount} pixels"
             return ToolResult.success(result=msg, ext_data=await self._get_page_info(context))
 
+        elif action == SendKeys.code:
+            keys = params.get("keys")
+            page = await context.get_current_page()
+            await page.keyboard.press(keys)
+            msg = f"Sent keys: {keys}"
+            print(msg)
+            return ToolResult(output=f"Sent keys: {keys}")
+
         else:
             msg = "Failed to operate the browser"
             return ToolResult.fail(result=msg)