feat: enhance tool execution guidance with user interruption and repetition notices

Author: Soulter

astrbot/core/agent/runners/tool_loop_agent_runner.py

@@ -95,16 +95,62 @@ class _ToolExecutionInterrupted(Exception):
 
 ToolExecutorResultT = T.TypeVar("ToolExecutorResultT")
 
-USER_INTERRUPTION_MESSAGE = (
-    "[SYSTEM: User actively interrupted the response generation. "
-    "Partial output before interruption is preserved.]"
-)
-
 
 class ToolLoopAgentRunner(BaseAgentRunner[TContext]):
     EMPTY_OUTPUT_RETRY_ATTEMPTS = 3
     EMPTY_OUTPUT_RETRY_WAIT_MIN_S = 1
     EMPTY_OUTPUT_RETRY_WAIT_MAX_S = 4
+    USER_INTERRUPTION_MESSAGE = (
+        "[SYSTEM: User actively interrupted the response generation. "
+        "Partial output before interruption is preserved.]"
+    )
+    FOLLOW_UP_NOTICE_TEMPLATE = (
+        "\n\n[SYSTEM NOTICE] User sent follow-up messages while tool execution "
+        "was in progress. Prioritize these follow-up instructions in your next "
+        "actions. In your very next action, briefly acknowledge to the user "
+        "that their follow-up message(s) were received before continuing.\n"
+        "{follow_up_lines}"
+    )
+    MAX_STEPS_REACHED_PROMPT = (
+        "Maximum tool call limit reached. "
+        "Stop calling tools, and based on the information you have gathered, "
+        "summarize your task and findings, and reply to the user directly."
+    )
+    SKILLS_LIKE_REQUERY_INSTRUCTION_TEMPLATE = (
+        "You have decided to call tool(s): {tool_names}. Now call the tool(s) "
+        "with required arguments using the tool schema, and follow the existing "
+        "tool-use rules."
+    )
+    SKILLS_LIKE_REQUERY_REPAIR_INSTRUCTION = (
+        "This is the second-stage tool execution step. "
+        "You must do exactly one of the following: "
+        "1. Call one of the selected tools using the provided tool schema. "
+        "2. If calling a tool is no longer possible or appropriate, reply to the user "
+        "with a brief explanation of why. "
+        "Do not return an empty response. "
+        "Do not ignore the selected tools without explanation."
+    )
+    REPEATED_TOOL_NOTICE_L1_THRESHOLD = 2
+    REPEATED_TOOL_NOTICE_L2_THRESHOLD = 3
+    REPEATED_TOOL_NOTICE_L3_THRESHOLD = 5
+    REPEATED_TOOL_NOTICE_L1_TEMPLATE = (
+        "\n\n[SYSTEM NOTICE] By the way, you have executed the same tool "
+        "`{tool_name}` {streak} times consecutively. Double-check whether another "
+        "tool, different arguments, or a summary would move the task forward better."
+    )
+    REPEATED_TOOL_NOTICE_L2_TEMPLATE = (
+        "\n\n[SYSTEM NOTICE] Important: you have executed the same tool "
+        "`{tool_name}` {streak} times consecutively. Unless this repetition is "
+        "clearly necessary, stop repeating the same action and either switch "
+        "tools, refine parameters, or summarize what is still missing."
+    )
+    REPEATED_TOOL_NOTICE_L3_TEMPLATE = (
+        "\n\n[SYSTEM NOTICE] Important: you have executed the same tool "
+        "`{tool_name}` {streak} times consecutively. Repetition is now very "
+        "high. Continue only if each call is clearly producing new information. "
+        "Otherwise, change strategy, adjust arguments, or explain the limitation "
+        "to the user."
+    )
 
     def _get_persona_custom_error_message(self) -> str | None:
         """Read persona-level custom error message from event extras when available."""
@@ -415,12 +461,8 @@ def _consume_follow_up_notice(self) -> str:
         follow_up_lines = "\n".join(
             f"{idx}. {ticket.text}" for idx, ticket in enumerate(follow_ups, start=1)
         )
-        return (
-            "\n\n[SYSTEM NOTICE] User sent follow-up messages while tool execution "
-            "was in progress. Prioritize these follow-up instructions in your next "
-            "actions. In your very next action, briefly acknowledge to the user "
-            "that their follow-up message(s) were received before continuing.\n"
-            f"{follow_up_lines}"
+        return self.FOLLOW_UP_NOTICE_TEMPLATE.format(
+            follow_up_lines=follow_up_lines,
         )
 
     def _merge_follow_up_notice(self, content: str) -> str:
@@ -438,30 +480,24 @@ def _track_tool_call_streak(self, tool_name: str) -> int:
         return self._same_tool_streak
 
     def _build_same_tool_guidance(self, tool_name: str, streak: int) -> str:
-        if streak < 3:
+        if streak < self.REPEATED_TOOL_NOTICE_L1_THRESHOLD:
             return ""
 
-        if streak >= 5:
-            return (
-                "\n\n[SYSTEM NOTICE] Important: you have executed the same tool "
-                f"`{tool_name}` {streak} times consecutively. Repetition is now very "
-                "high. Continue only if each call is clearly producing new information. "
-                "Otherwise, change strategy, adjust arguments, or explain the limitation "
-                "to the user."
+        if streak >= self.REPEATED_TOOL_NOTICE_L3_THRESHOLD:
+            return self.REPEATED_TOOL_NOTICE_L3_TEMPLATE.format(
+                tool_name=tool_name,
+                streak=streak,
             )
 
-        if streak >= 3:
-            return (
-                "\n\n[SYSTEM NOTICE] Important: you have executed the same tool "
-                f"`{tool_name}` {streak} times consecutively. Unless this repetition is "
-                "clearly necessary, stop repeating the same action and either switch "
-                "tools, refine parameters, or summarize what is still missing."
+        if streak >= self.REPEATED_TOOL_NOTICE_L2_THRESHOLD:
+            return self.REPEATED_TOOL_NOTICE_L2_TEMPLATE.format(
+                tool_name=tool_name,
+                streak=streak,
             )
 
-        return (
-            "\n\n[SYSTEM NOTICE] By the way, you have executed the same tool "
-            f"`{tool_name}` {streak} times consecutively. Double-check whether another "
-            "tool, different arguments, or a summary would move the task forward better."
+        return self.REPEATED_TOOL_NOTICE_L1_TEMPLATE.format(
+            tool_name=tool_name,
+            streak=streak,
         )
 
     @override
@@ -520,7 +556,7 @@ async def step(self):
                 if self._is_stop_requested():
                     llm_resp_result = LLMResponse(
                         role="assistant",
-                        completion_text=USER_INTERRUPTION_MESSAGE,
+                        completion_text=self.USER_INTERRUPTION_MESSAGE,
                         reasoning_content=llm_response.reasoning_content,
                         reasoning_signature=llm_response.reasoning_signature,
                     )
@@ -718,7 +754,7 @@ async def step_until_done(
             self.run_context.messages.append(
                 Message(
                     role="user",
-                    content="工具调用次数已达到上限，请停止使用工具，并根据已经收集到的信息，对你的任务和发现进行总结，然后直接回复用户。",
+                    content=self.MAX_STEPS_REACHED_PROMPT,
                 )
             )
             # 再执行最后一步
@@ -990,11 +1026,8 @@ def _build_tool_requery_context(
                 contexts.append(msg.model_dump())  # type: ignore[call-arg]
             elif isinstance(msg, dict):
                 contexts.append(copy.deepcopy(msg))
-        instruction = (
-            "You have decided to call tool(s): "
-            + ", ".join(tool_names)
-            + ". Now call the tool(s) with required arguments using the tool schema, "
-            "and follow the existing tool-use rules."
+        instruction = self.SKILLS_LIKE_REQUERY_INSTRUCTION_TEMPLATE.format(
+            tool_names=", ".join(tool_names)
         )
         if extra_instruction:
             instruction = f"{instruction}\n{extra_instruction}"
@@ -1065,14 +1098,7 @@ async def _resolve_tool_exec(
                     )
                     repair_contexts = self._build_tool_requery_context(
                         tool_names,
-                        extra_instruction=(
-                            "This is the second-stage tool execution step. "
-                            "You must do exactly one of the following: "
-                            "1. Call one of the selected tools using the provided tool schema. "
-                            "2. If calling a tool is no longer possible or appropriate, reply to the user with a brief explanation of why. "
-                            "Do not return an empty response. "
-                            "Do not ignore the selected tools without explanation."
-                        ),
+                        extra_instruction=self.SKILLS_LIKE_REQUERY_REPAIR_INSTRUCTION,
                     )
                     repair_resp = await self.provider.text_chat(
                         contexts=repair_contexts,
@@ -1114,7 +1140,7 @@ async def _finalize_aborted_step(
         if llm_resp.role != "assistant":
             llm_resp = LLMResponse(
                 role="assistant",
-                completion_text=USER_INTERRUPTION_MESSAGE,
+                completion_text=self.USER_INTERRUPTION_MESSAGE,
             )
         self.final_llm_resp = llm_resp
         self._aborted = True

tests/test_tool_loop_agent_runner.py

@@ -11,8 +11,8 @@
 sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
 
 from astrbot.core.agent.agent import Agent
-from astrbot.core.agent.hooks import BaseAgentRunHooks
 from astrbot.core.agent.handoff import HandoffTool
+from astrbot.core.agent.hooks import BaseAgentRunHooks
 from astrbot.core.agent.run_context import ContextWrapper
 from astrbot.core.agent.runners.tool_loop_agent_runner import ToolLoopAgentRunner
 from astrbot.core.agent.tool import FunctionTool, ToolSet
@@ -599,14 +599,25 @@ async def test_same_tool_consecutive_results_include_escalating_guidance(
     assert len(tool_messages) == 5
 
     tool_contents = [str(message.content) for message in tool_messages]
-    assert "same tool" not in tool_contents[0]
-    assert "By the way" in tool_contents[1]
-    assert "2 times consecutively" in tool_contents[1]
-    assert "Important" in tool_contents[2]
-    assert "3 times consecutively" in tool_contents[2]
-    assert "Important" in tool_contents[4]
-    assert "5 times consecutively" in tool_contents[4]
-    assert "very high" in tool_contents[4]
+    runner_cls = type(runner)
+    level_1_notice = runner_cls.SAME_TOOL_NOTICE_LEVEL_1_TEMPLATE.format(
+        tool_name="test_tool",
+        streak=runner_cls.SAME_TOOL_NOTICE_LEVEL_1_THRESHOLD,
+    )
+    level_2_notice = runner_cls.SAME_TOOL_NOTICE_LEVEL_2_TEMPLATE.format(
+        tool_name="test_tool",
+        streak=runner_cls.SAME_TOOL_NOTICE_LEVEL_2_THRESHOLD,
+    )
+    level_3_notice = runner_cls.SAME_TOOL_NOTICE_LEVEL_3_TEMPLATE.format(
+        tool_name="test_tool",
+        streak=runner_cls.SAME_TOOL_NOTICE_LEVEL_3_THRESHOLD,
+    )
+
+    assert level_1_notice not in tool_contents[0]
+    assert level_2_notice not in tool_contents[0]
+    assert level_1_notice in tool_contents[1]
+    assert level_2_notice in tool_contents[2]
+    assert level_3_notice in tool_contents[4]
 
 
 @pytest.mark.asyncio
@@ -652,11 +663,21 @@ async def test_same_tool_streak_resets_after_switching_tools(
     assert len(tool_messages) == 4
 
     tool_contents = [str(message.content) for message in tool_messages]
-    assert "same tool" not in tool_contents[0]
-    assert "same tool" not in tool_contents[1]
-    assert "same tool" not in tool_contents[2]
-    assert "By the way" in tool_contents[3]
-    assert "`test_tool` 2 times consecutively" in tool_contents[3]
+    runner_cls = type(runner)
+    level_1_notice = runner_cls.SAME_TOOL_NOTICE_LEVEL_1_TEMPLATE.format(
+        tool_name="test_tool",
+        streak=runner_cls.SAME_TOOL_NOTICE_LEVEL_1_THRESHOLD,
+    )
+    level_2_notice = runner_cls.SAME_TOOL_NOTICE_LEVEL_2_TEMPLATE.format(
+        tool_name="test_tool",
+        streak=runner_cls.SAME_TOOL_NOTICE_LEVEL_2_THRESHOLD,
+    )
+
+    assert level_1_notice not in tool_contents[0]
+    assert level_1_notice not in tool_contents[1]
+    assert level_1_notice not in tool_contents[2]
+    assert level_2_notice not in tool_contents[2]
+    assert level_1_notice in tool_contents[3]
 
 
 @pytest.mark.asyncio
@@ -1084,7 +1105,9 @@ async def test_follow_up_accepted_when_active_and_not_stopping(
 
     ticket = runner.follow_up(message_text="valid follow-up message")
 
-    assert ticket is not None, "Follow-up should be accepted when runner is active and not stopping"
+    assert ticket is not None, (
+        "Follow-up should be accepted when runner is active and not stopping"
+    )
     assert ticket.text == "valid follow-up message"
     assert ticket.consumed is False
     assert ticket in runner._pending_follow_ups