From e55e710df0e7ae80f4cf62c6ef52648f66db4143 Mon Sep 17 00:00:00 2001 From: Lorenze Jay <63378463+lorenzejay@users.noreply.github.com> Date: Mon, 6 Jul 2026 15:28:37 -0700 Subject: [PATCH] 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 --- .../src/crewai/experimental/agent_executor.py | 144 +++++++++---- lib/crewai/tests/agents/test_agent.py | 94 ++++++++- .../tests/agents/test_agent_executor.py | 195 +++++++++++++++++- lib/crewai/tests/test_crew.py | 6 - .../test_flow_human_input_integration.py | 25 ++- 5 files changed, 410 insertions(+), 54 deletions(-) diff --git a/lib/crewai/src/crewai/experimental/agent_executor.py b/lib/crewai/src/crewai/experimental/agent_executor.py index 0ecd8e63a..3c76bd8d8 100644 --- a/lib/crewai/src/crewai/experimental/agent_executor.py +++ b/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 diff --git a/lib/crewai/tests/agents/test_agent.py b/lib/crewai/tests/agents/test_agent.py index daffb7a20..2b126016c 100644 --- a/lib/crewai/tests/agents/test_agent.py +++ b/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", diff --git a/lib/crewai/tests/agents/test_agent_executor.py b/lib/crewai/tests/agents/test_agent_executor.py index e4de4a484..80bd5fe85 100644 --- a/lib/crewai/tests/agents/test_agent_executor.py +++ b/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() diff --git a/lib/crewai/tests/test_crew.py b/lib/crewai/tests/test_crew.py index 82f3207dd..557f826d5 100644 --- a/lib/crewai/tests/test_crew.py +++ b/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( diff --git a/lib/crewai/tests/test_flow_human_input_integration.py b/lib/crewai/tests/test_flow_human_input_integration.py index 461111e09..93a74fcac 100644 --- a/lib/crewai/tests/test_flow_human_input_integration.py +++ b/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 \ No newline at end of file + 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