refactor: enhance final answer synthesis logic in AgentExecutor

- Updated the finalization process to conditionally skip synthesis when the last todo result is sufficient as a complete answer.
- Introduced a new method to determine if the last todo result can be used directly, improving efficiency.
- Added tests to verify the new behavior, ensuring synthesis is skipped when appropriate and maintained when a response model is set.

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

@@ -2146,20 +2146,21 @@ class AgentExecutor(Flow[AgentReActState], CrewAgentExecutorMixin):
                 color="magenta",
             )
 
-        # Plan-and-Execute path: synthesize from completed todos
-        # Check for todos with results (even if not all marked "completed" —
-        # the goal_achieved path may skip marking some as completed)
-        todos_with_results = [t for t in self.state.todos.items if t.result]
-        if todos_with_results and self.state.current_answer is None:
-            self._synthesize_final_answer_from_todos()
-
-        # Legacy path: synthesize if todos are all formally complete
-        if (
-            self.state.todos.items
-            and self.state.todos.is_complete
-            and self.state.current_answer is None
-        ):
-            self._synthesize_final_answer_from_todos()
+        if self.state.current_answer is None:
+            # Plan-and-Execute path: todos may have results even if not all are
+            # marked "completed" (e.g., goal_achieved early).
+            todos_with_results = [t for t in self.state.todos.items if t.result]
+            if todos_with_results:
+                if self._can_use_last_todo_result_as_final_answer(todos_with_results):
+                    last_todo = max(todos_with_results, key=lambda todo: todo.step_number)
+                    final_text = str(last_todo.result or "")
+                    self.state.current_answer = AgentFinish(
+                        thought="Final answer returned directly from last completed todo",
+                        output=final_text,
+                        text=final_text,
+                    )
+                else:
+                    self._synthesize_final_answer_from_todos()
 
         if self.state.current_answer is None:
             skip_text = Text()
@@ -2185,6 +2186,32 @@ class AgentExecutor(Flow[AgentReActState], CrewAgentExecutorMixin):
 
         return "completed"
 
+    def _can_use_last_todo_result_as_final_answer(
+        self, todos_with_results: list[TodoItem]
+    ) -> bool:
+        """Determine whether synthesis can be skipped for planning results."""
+        # Keep synthesis when structured output is requested.
+        if self.response_model is not None:
+            return False
+        if not todos_with_results:
+            return False
+
+        last_todo = max(todos_with_results, key=lambda todo: todo.step_number)
+        if last_todo.tool_to_use:
+            return False
+
+        last_result = str(last_todo.result or "").strip()
+        if not last_result:
+            return False
+
+        lowered_result = last_result.lower()
+        if lowered_result.startswith("error:") or "tool execution error" in lowered_result:
+            return False
+
+        word_count = len(last_result.split())
+        has_sentence_punctuation = any(ch in last_result for ch in ".!?")
+        return (len(last_result) >= 200 or word_count >= 30) and has_sentence_punctuation
+
     def _synthesize_final_answer_from_todos(self) -> None:
         """Synthesize a coherent final answer from all todo results.
 

lib/crewai/tests/agents/test_agent_executor.py

@@ -202,6 +202,88 @@ class TestAgentExecutor:
         assert result == "skipped"
         assert executor.state.is_finished is False
 
+    def test_finalize_skips_synthesis_for_strong_last_todo_result(
+        self, mock_dependencies
+    ):
+        """Finalize should skip synthesis when last todo is already a complete answer."""
+        with patch.object(AgentExecutor, "_show_logs") as mock_show_logs:
+            executor = AgentExecutor(**mock_dependencies)
+            executor.state.todos.items = [
+                TodoItem(
+                    step_number=1,
+                    description="Gather source details",
+                    tool_to_use="search_tool",
+                    status="completed",
+                    result="Source A and Source B identified.",
+                ),
+                TodoItem(
+                    step_number=2,
+                    description="Write final response",
+                    tool_to_use=None,
+                    status="completed",
+                    result=(
+                        "The final recommendation is to adopt a phased rollout plan with "
+                        "weekly checkpoints, explicit ownership, and a rollback path for "
+                        "each milestone. This approach keeps risk controlled while still "
+                        "moving quickly, and it aligns delivery metrics with stakeholder "
+                        "communication and operational readiness."
+                    ),
+                ),
+            ]
+
+            with patch.object(
+                executor, "_synthesize_final_answer_from_todos"
+            ) as mock_synthesize:
+                result = executor.finalize()
+
+            assert result == "completed"
+            assert isinstance(executor.state.current_answer, AgentFinish)
+            assert (
+                executor.state.current_answer.output
+                == executor.state.todos.items[1].result
+            )
+            assert executor.state.is_finished is True
+            mock_synthesize.assert_not_called()
+            mock_show_logs.assert_called_once()
+
+    def test_finalize_keeps_synthesis_when_response_model_is_set(
+        self, mock_dependencies
+    ):
+        """Finalize should still synthesize when response_model is configured."""
+        with patch.object(AgentExecutor, "_show_logs"):
+            executor = AgentExecutor(**mock_dependencies)
+            executor.response_model = Mock()
+            executor.state.todos.items = [
+                TodoItem(
+                    step_number=1,
+                    description="Write final response",
+                    tool_to_use=None,
+                    status="completed",
+                    result=(
+                        "This is already detailed prose with multiple sentences. "
+                        "It should still run synthesis because structured output "
+                        "was requested via response_model."
+                    ),
+                )
+            ]
+
+            def _set_current_answer() -> None:
+                executor.state.current_answer = AgentFinish(
+                    thought="Synthesized",
+                    output="structured-like-answer",
+                    text="structured-like-answer",
+                )
+
+            with patch.object(
+                executor,
+                "_synthesize_final_answer_from_todos",
+                side_effect=_set_current_answer,
+            ) as mock_synthesize:
+                result = executor.finalize()
+
+            assert result == "completed"
+            mock_synthesize.assert_called_once()
+
     def test_format_prompt(self, mock_dependencies):
         """Test prompt formatting."""
         executor = AgentExecutor(**mock_dependencies)