Implement message setup and feedback handling in AgentExecutor (#6465)

* Implement message setup and feedback handling in AgentExecutor

- Added  method to streamline message preparation for agent execution, allowing for integration with human input providers.
- Introduced  and  methods to manage the state during feedback processing.
- Enhanced  and  methods to re-run the executor flow using existing feedback messages.
- Updated tests to verify the new message setup and feedback handling functionality, ensuring compatibility with human input providers.

* dont commit runner

* Remove xfail marker from test_crew_train_success as training feedback migration to AgentExecutor is complete.

* fix runtype errors

* fix test

* revert

* mypy fix

* handled reset iterations

lib/crewai/src/crewai/experimental/agent_executor.py

@@ -106,6 +106,7 @@ from crewai.utilities.planning_types import (
     TodoItem,
     TodoList,
 )
+from crewai.utilities.prompts import StandardPromptResult, SystemPromptResult
 from crewai.utilities.step_execution_context import StepExecutionContext, StepResult
 from crewai.utilities.string_utils import sanitize_tool_name
 from crewai.utilities.tool_utils import execute_tool_and_check_finality
@@ -118,7 +119,6 @@ if TYPE_CHECKING:
     from crewai.agents.tools_handler import ToolsHandler
     from crewai.llms.base_llm import BaseLLM
     from crewai.tools.tool_types import ToolResult
-    from crewai.utilities.prompts import StandardPromptResult, SystemPromptResult
 
 _RouteT = TypeVar("_RouteT", bound=str)
 
@@ -218,6 +218,7 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
     _instance_id: str = PrivateAttr(default_factory=lambda: str(uuid4())[:8])
     _step_executor: Any = PrivateAttr(default=None)
     _planner_observer: PlannerObserver | None = PrivateAttr(default=None)
+    _is_feedback_iteration: bool = PrivateAttr(default=False)
 
     @model_validator(mode="after")
     def _setup_executor(self) -> Self:
@@ -296,6 +297,33 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
         """Set state messages."""
         self._state.messages = value
 
+    def _setup_messages(self, inputs: dict[str, Any]) -> None:
+        """Set up messages for the agent execution."""
+        provider = get_provider()
+        if provider.setup_messages(cast("ExecutorContext", self)):
+            return
+
+        from crewai.llms.cache import mark_cache_breakpoint
+
+        if isinstance(self.prompt, SystemPromptResult):
+            system_prompt = self._format_prompt(self.prompt["system"], inputs)
+            user_prompt = self._format_prompt(self.prompt["user"], inputs)
+            self.state.messages.append(
+                mark_cache_breakpoint(
+                    format_message_for_llm(system_prompt, role="system")
+                )
+            )
+            self.state.messages.append(
+                mark_cache_breakpoint(format_message_for_llm(user_prompt))
+            )
+        elif isinstance(self.prompt, StandardPromptResult):
+            user_prompt = self._format_prompt(self.prompt["prompt"], inputs)
+            self.state.messages.append(
+                mark_cache_breakpoint(format_message_for_llm(user_prompt))
+            )
+
+        provider.post_setup_messages(cast("ExecutorContext", self))
+
     @property
     def ask_for_human_input(self) -> bool:
         """Compatibility property - returns state ask_for_human_input."""
@@ -314,6 +342,8 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
         enabled on the agent, it generates a plan before execution begins.
         The plan is stored in state and todos are created from the steps.
         """
+        if self._is_feedback_iteration:
+            return
         if not getattr(self.agent, "planning_enabled", False):
             return
 
@@ -2761,27 +2791,7 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
                     "AgentExecutor.llm or .prompt is unset; the executor was "
                     "not fully restored or initialized before execution."
                 )
-            if "system" in self.prompt:
-                from crewai.llms.cache import mark_cache_breakpoint
-
-                prompt = cast("SystemPromptResult", self.prompt)
-                system_prompt = self._format_prompt(prompt["system"], inputs)
-                user_prompt = self._format_prompt(prompt["user"], inputs)
-                self.state.messages.append(
-                    mark_cache_breakpoint(
-                        format_message_for_llm(system_prompt, role="system")
-                    )
-                )
-                self.state.messages.append(
-                    mark_cache_breakpoint(format_message_for_llm(user_prompt))
-                )
-            else:
-                from crewai.llms.cache import mark_cache_breakpoint
-
-                user_prompt = self._format_prompt(self.prompt["prompt"], inputs)
-                self.state.messages.append(
-                    mark_cache_breakpoint(format_message_for_llm(user_prompt))
-                )
+            self._setup_messages(inputs)
 
             self._inject_files_from_inputs(inputs)
 
@@ -2867,27 +2877,7 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
                     "AgentExecutor.llm or .prompt is unset; the executor was "
                     "not fully restored or initialized before execution."
                 )
-            if "system" in self.prompt:
-                from crewai.llms.cache import mark_cache_breakpoint
-
-                prompt = cast("SystemPromptResult", self.prompt)
-                system_prompt = self._format_prompt(prompt["system"], inputs)
-                user_prompt = self._format_prompt(prompt["user"], inputs)
-                self.state.messages.append(
-                    mark_cache_breakpoint(
-                        format_message_for_llm(system_prompt, role="system")
-                    )
-                )
-                self.state.messages.append(
-                    mark_cache_breakpoint(format_message_for_llm(user_prompt))
-                )
-            else:
-                from crewai.llms.cache import mark_cache_breakpoint
-
-                user_prompt = self._format_prompt(self.prompt["prompt"], inputs)
-                self.state.messages.append(
-                    mark_cache_breakpoint(format_message_for_llm(user_prompt))
-                )
+            self._setup_messages(inputs)
 
             await self._ainject_files_from_inputs(inputs)
 
@@ -3169,8 +3159,13 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
         Returns:
             Final answer after feedback.
         """
+        self.messages = self.state.messages
         provider = get_provider()
-        return provider.handle_feedback(formatted_answer, cast("ExecutorContext", self))
+        final_answer = provider.handle_feedback(
+            formatted_answer, cast("ExecutorContext", self)
+        )
+        self._complete_feedback(final_answer)
+        return final_answer
 
     async def _ahandle_human_feedback(
         self, formatted_answer: AgentFinish
@@ -3183,10 +3178,63 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
         Returns:
             Final answer after feedback.
         """
+        self.messages = self.state.messages
         provider = get_provider()
-        return await provider.handle_feedback_async(
+        final_answer = await provider.handle_feedback_async(
             formatted_answer, cast("AsyncExecutorContext", self)
         )
+        self._complete_feedback(final_answer)
+        return final_answer
+
+    def _complete_feedback(self, final_answer: AgentFinish) -> None:
+        """Mark the final reviewed answer as the completed executor state."""
+        self.state.current_answer = final_answer
+        self.state.is_finished = True
+        self.state.ask_for_human_input = False
+        self._finalize_called = True
+
+    def _prepare_feedback_iteration(self) -> None:
+        """Reset flow completion state before rerunning with feedback."""
+        self._finalize_called = False
+        self._is_feedback_iteration = True
+        self.state.current_answer = None
+        self.state.is_finished = False
+        self.state.iterations = 0
+        self.state.use_native_tools = False
+        self.state.pending_tool_calls = []
+        self.state.plan = None
+        self.state.plan_ready = False
+        self.state.todos = TodoList()
+        self.state.replan_count = 0
+        self.state.last_replan_reason = None
+        self.state.observations = {}
+        self.state.execution_log = []
+
+    def _invoke_loop(self) -> AgentFinish:
+        """Re-run the executor flow using the existing feedback messages."""
+        self._prepare_feedback_iteration()
+        try:
+            self.kickoff()
+        finally:
+            self._is_feedback_iteration = False
+
+        if not isinstance(self.state.current_answer, AgentFinish):
+            raise RuntimeError("Agent execution ended without reaching a final answer.")
+
+        return self.state.current_answer
+
+    async def _ainvoke_loop(self) -> AgentFinish:
+        """Re-run the executor flow asynchronously using feedback messages."""
+        self._prepare_feedback_iteration()
+        try:
+            await self.kickoff_async()
+        finally:
+            self._is_feedback_iteration = False
+
+        if not isinstance(self.state.current_answer, AgentFinish):
+            raise RuntimeError("Agent execution ended without reaching a final answer.")
+
+        return self.state.current_answer
 
     def _is_training_mode(self) -> bool:
         """Check if training mode is active.
@@ -3196,6 +3244,12 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
         """
         return bool(self.crew and self.crew._train)
 
+    def _format_feedback_message(self, feedback: str) -> LLMMessage:
+        """Format human feedback as an LLM message."""
+        return format_message_for_llm(
+            I18N_DEFAULT.slice("feedback_instructions").format(feedback=feedback)
+        )
+
 
 # Backward compatibility alias (deprecated)
 CrewAgentExecutorFlow = AgentExecutor

lib/crewai/tests/agents/test_agent.py

@@ -3,13 +3,14 @@
 import os
 import threading
 from unittest import mock
-from unittest.mock import MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 import warnings
 
 from crewai.agents.crew_agent_executor import AgentFinish, CrewAgentExecutor
 from crewai.constants import DEFAULT_LLM_MODEL
 from crewai.events.event_bus import crewai_event_bus
 from crewai.events.types.tool_usage_events import ToolUsageFinishedEvent
+from crewai.experimental.agent_executor import AgentExecutor
 from crewai.knowledge.knowledge import Knowledge
 from crewai.knowledge.knowledge_config import KnowledgeConfig
 from crewai.knowledge.source.base_knowledge_source import BaseKnowledgeSource
@@ -802,6 +803,97 @@ def test_agent_human_input():
         assert output.strip().lower() == "hello"
 
 
+def test_agent_default_executor_human_input():
+    from crewai.core.providers.human_input import SyncHumanInputProvider
+
+    agent = Agent(
+        role="test role",
+        goal="test goal",
+        backstory="test backstory",
+    )
+    task = Task(
+        agent=agent,
+        description="Say the word: Hi",
+        expected_output="The word: Hi",
+        human_input=True,
+    )
+    answers = iter(
+        [
+            AgentFinish(output="Hi", thought="", text="Hi"),
+            AgentFinish(output="Hello", thought="", text="Hello"),
+        ]
+    )
+    feedback_responses = iter(["Don't say hi, say Hello instead!", ""])
+
+    def kickoff_side_effect(executor, *_args, **_kwargs):
+        executor.state.current_answer = next(answers)
+        executor.state.is_finished = True
+
+    with (
+        patch.object(
+            SyncHumanInputProvider,
+            "_prompt_input",
+            side_effect=lambda *_args, **_kwargs: next(feedback_responses),
+        ) as mock_prompt_input,
+        patch.object(
+            AgentExecutor, "kickoff", autospec=True, side_effect=kickoff_side_effect
+        ) as mock_kickoff,
+    ):
+        output = agent.execute_task(task)
+
+    assert output == "Hello"
+    assert mock_prompt_input.call_count == 2
+    assert mock_kickoff.call_count == 2
+
+
+@pytest.mark.asyncio
+async def test_agent_default_executor_async_human_input():
+    from crewai.core.providers.human_input import SyncHumanInputProvider
+
+    agent = Agent(
+        role="test role",
+        goal="test goal",
+        backstory="test backstory",
+    )
+    task = Task(
+        agent=agent,
+        description="Say the word: Hi",
+        expected_output="The word: Hi",
+        human_input=True,
+    )
+    answers = iter(
+        [
+            AgentFinish(output="Hi", thought="", text="Hi"),
+            AgentFinish(output="Hello", thought="", text="Hello"),
+        ]
+    )
+    feedback_responses = iter(["Don't say hi, say Hello instead!", ""])
+
+    async def kickoff_side_effect(executor, *_args, **_kwargs):
+        executor.state.current_answer = next(answers)
+        executor.state.is_finished = True
+
+    with (
+        patch.object(
+            SyncHumanInputProvider,
+            "_prompt_input_async",
+            new_callable=AsyncMock,
+            side_effect=lambda *_args, **_kwargs: next(feedback_responses),
+        ) as mock_prompt_input,
+        patch.object(
+            AgentExecutor,
+            "kickoff_async",
+            autospec=True,
+            side_effect=kickoff_side_effect,
+        ) as mock_kickoff,
+    ):
+        output = await agent.aexecute_task(task)
+
+    assert output == "Hello"
+    assert mock_prompt_input.await_count == 2
+    assert mock_kickoff.await_count == 2
+
+
 def test_interpolate_inputs():
     agent = Agent(
         role="{topic} specialist",

lib/crewai/tests/agents/test_agent_executor.py

@@ -18,6 +18,7 @@ import pytest
 from pydantic import BaseModel
 
 from crewai.agents.tools_handler import ToolsHandler as _ToolsHandler
+from crewai.core.providers.human_input import SyncHumanInputProvider
 from crewai.agents.step_executor import StepExecutor
 
 
@@ -27,6 +28,13 @@ def _build_executor(**kwargs: Any) -> AgentExecutor:
     Uses model_construct to skip Pydantic validators so plain Mock()
     objects are accepted for typed fields like llm, agent, crew, task.
     """
+    prompt = kwargs.get("prompt")
+    if isinstance(prompt, dict):
+        if "system" in prompt:
+            kwargs["prompt"] = SystemPromptResult(**prompt)
+        else:
+            kwargs["prompt"] = StandardPromptResult(**prompt)
+
     executor = AgentExecutor.model_construct(**kwargs)
     executor._state = AgentExecutorState()
     executor._methods = {}
@@ -50,6 +58,7 @@ def _build_executor(**kwargs: Any) -> AgentExecutor:
     executor._last_context_error = None
     executor._step_executor = None
     executor._planner_observer = None
+    executor._is_feedback_iteration = False
     return executor
 from crewai.agents.planner_observer import PlannerObserver
 from crewai.experimental.agent_executor import (
@@ -68,7 +77,8 @@ from crewai.events.types.tool_usage_events import (
 )
 from crewai.tools.tool_types import ToolResult
 from crewai.utilities.step_execution_context import StepExecutionContext
-from crewai.utilities.planning_types import TodoItem
+from crewai.utilities.planning_types import TodoItem, TodoList
+from crewai.utilities.prompts import StandardPromptResult, SystemPromptResult
 from crewai.utilities.file_store import clear_files, clear_task_files, store_files
 from crewai_files import TextFile
 
@@ -119,6 +129,189 @@ class TestAgentExecutor:
     class StructuredResult(BaseModel):
         value: str
 
+    def test_setup_messages_calls_human_input_provider_hooks(self):
+        """Message setup should preserve the HumanInputProvider hook contract."""
+        executor = _build_executor(
+            prompt=StandardPromptResult(prompt="Original task: {input}"),
+        )
+        provider = Mock()
+        provider.setup_messages.return_value = False
+
+        def post_setup(context: AgentExecutor) -> None:
+            context.messages.append(
+                {"role": "system", "content": "provider post setup"}
+            )
+
+        provider.post_setup_messages.side_effect = post_setup
+
+        with patch(
+            "crewai.experimental.agent_executor.get_provider", return_value=provider
+        ):
+            executor._setup_messages(
+                {"input": "draft this", "tool_names": "", "tools": ""}
+            )
+
+        provider.setup_messages.assert_called_once_with(executor)
+        provider.post_setup_messages.assert_called_once_with(executor)
+        assert executor.state.messages[0]["role"] == "user"
+        assert executor.state.messages[0]["content"] == "Original task: draft this"
+        assert executor.state.messages[1] == {
+            "role": "system",
+            "content": "provider post setup",
+        }
+
+    def test_setup_messages_can_be_owned_by_human_input_provider(self):
+        """Providers can skip standard prompt setup by returning True."""
+        executor = _build_executor(
+            prompt=StandardPromptResult(prompt="Original task: {input}"),
+        )
+        provider = Mock()
+
+        def setup(context: AgentExecutor) -> bool:
+            context.messages.append({"role": "user", "content": "provider message"})
+            return True
+
+        provider.setup_messages.side_effect = setup
+
+        with patch(
+            "crewai.experimental.agent_executor.get_provider", return_value=provider
+        ):
+            executor._setup_messages(
+                {"input": "draft this", "tool_names": "", "tools": ""}
+            )
+
+        provider.setup_messages.assert_called_once_with(executor)
+        provider.post_setup_messages.assert_not_called()
+        assert executor.state.messages == [
+            {"role": "user", "content": "provider message"}
+        ]
+
+    def test_human_feedback_reruns_flow_with_state_messages(self):
+        """Human feedback should use AgentExecutor state messages."""
+        executor = _build_executor(agent=SimpleNamespace(verbose=False), crew=None)
+        executor.state.messages = [{"role": "user", "content": "original task"}]
+        executor.state.current_answer = AgentFinish(
+            thought="", output="draft", text="draft"
+        )
+        executor.state.is_finished = True
+        executor._finalize_called = True
+        executor.ask_for_human_input = True
+        executor.state.iterations = executor.max_iter
+        executor.state.plan = "completed plan"
+        executor.state.plan_ready = True
+        executor.state.todos = TodoList(
+            items=[TodoItem(step_number=1, description="Done", status="completed")]
+        )
+
+        improved_answer = AgentFinish(thought="", output="improved", text="improved")
+        feedback_responses = iter(["make it friendlier", ""])
+
+        def finish_feedback_iteration(*_args: Any, **_kwargs: Any) -> None:
+            assert executor._is_feedback_iteration is True
+            assert executor.state.iterations == 0
+            assert executor.state.plan is None
+            assert executor.state.todos.items == []
+            executor.state.current_answer = improved_answer
+            executor.state.is_finished = True
+
+        with (
+            patch.object(
+                SyncHumanInputProvider,
+                "_prompt_input",
+                side_effect=lambda *_args, **_kwargs: next(feedback_responses),
+            ) as mock_prompt_input,
+            patch.object(
+                AgentExecutor, "kickoff", side_effect=finish_feedback_iteration
+            ) as mock_kickoff,
+        ):
+            result = executor._handle_human_feedback(
+                AgentFinish(thought="", output="draft", text="draft")
+            )
+
+        assert result is improved_answer
+        assert mock_prompt_input.call_count == 2
+        mock_kickoff.assert_called_once()
+        assert executor.messages is executor.state.messages
+        assert "make it friendlier" in executor.state.messages[-1]["content"]
+        assert executor.ask_for_human_input is False
+        assert executor.state.current_answer is improved_answer
+        assert executor.state.is_finished is True
+        assert executor._finalize_called is True
+        assert executor._is_feedback_iteration is False
+
+    @pytest.mark.asyncio
+    async def test_async_human_feedback_reruns_flow_with_state_messages(self):
+        """Async human feedback should use AgentExecutor state messages."""
+        executor = _build_executor(agent=SimpleNamespace(verbose=False), crew=None)
+        executor.state.messages = [{"role": "user", "content": "original task"}]
+        executor.state.current_answer = AgentFinish(
+            thought="", output="draft", text="draft"
+        )
+        executor.state.is_finished = True
+        executor._finalize_called = True
+        executor.ask_for_human_input = True
+        executor.state.iterations = executor.max_iter
+        executor.state.plan = "completed plan"
+        executor.state.plan_ready = True
+        executor.state.todos = TodoList(
+            items=[TodoItem(step_number=1, description="Done", status="completed")]
+        )
+
+        improved_answer = AgentFinish(thought="", output="improved", text="improved")
+        feedback_responses = iter(["make it friendlier", ""])
+
+        async def finish_feedback_iteration(*_args: Any, **_kwargs: Any) -> None:
+            assert executor._is_feedback_iteration is True
+            assert executor.state.iterations == 0
+            assert executor.state.plan is None
+            assert executor.state.todos.items == []
+            executor.state.current_answer = improved_answer
+            executor.state.is_finished = True
+
+        with (
+            patch.object(
+                SyncHumanInputProvider,
+                "_prompt_input_async",
+                new_callable=AsyncMock,
+                side_effect=lambda *_args, **_kwargs: next(feedback_responses),
+            ) as mock_prompt_input,
+            patch.object(
+                AgentExecutor,
+                "kickoff_async",
+                new_callable=AsyncMock,
+                side_effect=finish_feedback_iteration,
+            ) as mock_kickoff,
+        ):
+            result = await executor._ahandle_human_feedback(
+                AgentFinish(thought="", output="draft", text="draft")
+            )
+
+        assert result is improved_answer
+        assert mock_prompt_input.await_count == 2
+        mock_kickoff.assert_awaited_once()
+        assert executor.messages is executor.state.messages
+        assert "make it friendlier" in executor.state.messages[-1]["content"]
+        assert executor.ask_for_human_input is False
+        assert executor.state.current_answer is improved_answer
+        assert executor.state.is_finished is True
+        assert executor._finalize_called is True
+        assert executor._is_feedback_iteration is False
+
+    def test_feedback_iteration_skips_plan_generation(self):
+        """Feedback reruns should reason over feedback without regenerating a plan."""
+        executor = _build_executor(
+            agent=SimpleNamespace(planning_enabled=True, verbose=False),
+            task=SimpleNamespace(),
+        )
+        executor._is_feedback_iteration = True
+
+        with patch("crewai.utilities.reasoning_handler.AgentReasoning") as reasoning:
+            executor.generate_plan()
+
+        reasoning.assert_not_called()
+        assert executor.state.plan is None
+        assert executor.state.todos.items == []
+
     def test_inject_files_from_crew_task_store(self):
         """Crew-level input_files should attach to the LLM user message."""
         crew_id = uuid4()

lib/crewai/tests/test_crew.py

@@ -2908,12 +2908,6 @@ def test_manager_agent_with_tools_raises_exception(researcher, writer):
         crew.kickoff()
 
 
-@pytest.mark.xfail(
-    strict=True,
-    reason="crew.train() relies on CrewAgentExecutor._format_feedback_message; "
-    "AgentExecutor (the new default) does not implement training feedback yet. "
-    "Remove this xfail once training is migrated to AgentExecutor.",
-)
 @pytest.mark.vcr()
 def test_crew_train_success(researcher, writer, monkeypatch):
     task = Task(

lib/crewai/tests/test_flow_human_input_integration.py

@@ -138,4 +138,27 @@ class TestFlowHumanInputIntegration:
                 for call in call_args
                 if call[0]
             )
-            assert training_panel_found
+            assert training_panel_found
+
+    @patch("builtins.input", return_value="please make it warmer")
+    def test_non_empty_input_prints_processing_feedback(self, mock_input):
+        """Non-empty input should be displayed as feedback to process."""
+        provider = SyncHumanInputProvider()
+        crew = MagicMock()
+        crew._train = False
+
+        formatter = event_listener.formatter
+
+        with (
+            patch.object(formatter, "pause_live_updates"),
+            patch.object(formatter, "resume_live_updates"),
+            patch.object(formatter.console, "print") as mock_console_print,
+        ):
+            result = provider._prompt_input(crew)
+
+        assert result == "please make it warmer"
+        mock_input.assert_called_once()
+        printed_text = "\n".join(
+            str(call.args[0]) for call in mock_console_print.call_args_list
+        )
+        assert "Processing your feedback" in printed_text