[feature] Accumulator response builder
diff --git a/docs/guides/primer/usage_pattern.md b/docs/guides/primer/usage_pattern.md
index bdd30446aebd2..109d29add66c0 100644
--- a/docs/guides/primer/usage_pattern.md
+++ b/docs/guides/primer/usage_pattern.md
@@ -330,13 +330,17 @@ query_engine = RetrieverQueryEngine.from_args(retriever, response_mode=<response
 
 Right now, we support the following options:
 - `default`: "create and refine" an answer by sequentially going through each retrieved `Node`; 
-    This make a separate LLM call per Node. Good for more detailed answers.
+    This makes a separate LLM call per Node. Good for more detailed answers.
 - `compact`: "compact" the prompt during each LLM call by stuffing as 
     many `Node` text chunks that can fit within the maximum prompt size. If there are 
     too many chunks to stuff in one prompt, "create and refine" an answer by going through
     multiple prompts.
 - `tree_summarize`: Given a set of `Node` objects and the query, recursively construct a tree 
     and return the root node as the response. Good for summarization purposes.
+- `accumulate`: Given a set of `Node` objects and the query, apply the query to each `Node` text
+    chunk while accumulating the responses into an array. Returns a concatenated string of all
+    responses. Good for when you need to run the same query separately against each text
+    chunk.
 
 ```python
 index = GPTListIndex.from_documents(documents)
diff --git a/llama_index/indices/response/response_builder.py b/llama_index/indices/response/response_builder.py
index 124c6df56d9da..e18a423a6eebb 100644
--- a/llama_index/indices/response/response_builder.py
+++ b/llama_index/indices/response/response_builder.py
@@ -9,7 +9,17 @@
 """
 from abc import ABC, abstractmethod
 import logging
-from typing import Any, Dict, Generator, List, Optional, Sequence, Tuple, cast
+from typing import (
+    Any,
+    Dict,
+    Generator,
+    List,
+    Optional,
+    Sequence,
+    Tuple,
+    cast,
+)
+import asyncio
 
 from llama_index.data_structs.data_structs import IndexGraph
 from llama_index.data_structs.node import Node
@@ -33,6 +43,7 @@
 from llama_index.token_counter.token_counter import llm_token_counter
 from llama_index.types import RESPONSE_TEXT_TYPE
 from llama_index.utils import temp_set_attrs
+from llama_index.async_utils import run_async_tasks
 
 logger = logging.getLogger(__name__)
 
@@ -576,6 +587,107 @@ def get_response(
             return stream_response
 
 
+class Accumulate(BaseResponseBuilder):
+    def __init__(
+        self,
+        service_context: ServiceContext,
+        text_qa_template: QuestionAnswerPrompt,
+        streaming: bool = False,
+        use_async: bool = False,
+    ) -> None:
+        super().__init__(service_context=service_context, streaming=streaming)
+        self.text_qa_template = text_qa_template
+        self._use_async = use_async
+
+    def flatten_list(self, md_array: List[List[Any]]) -> List[Any]:
+        return list(item for sublist in md_array for item in sublist)
+
+    def format_response(self, outputs: List[Any], separator: str) -> str:
+        responses: List[str] = []
+        for response, formatted_prompt in outputs:
+            self._log_prompt_and_response(
+                formatted_prompt, response, log_prefix="Initial"
+            )
+            responses.append(response or "Empty Response")
+
+        return separator.join(
+            [f"Response {index + 1}: {item}" for index, item in enumerate(responses)]
+        )
+
+    @llm_token_counter("aget_response")
+    async def aget_response(
+        self,
+        query_str: str,
+        text_chunks: Sequence[str],
+        separator: str = "\n---------------------\n",
+        **kwargs: Any,
+    ) -> RESPONSE_TEXT_TYPE:
+        """Apply the same prompt to text chunks and return async responses"""
+
+        if self._streaming:
+            raise ValueError("Unable to stream in Accumulate response mode")
+
+        tasks = [
+            self._give_responses(query_str, text_chunk, use_async=True)
+            for text_chunk in text_chunks
+        ]
+
+        flattened_tasks = self.flatten_list(tasks)
+        outputs = await asyncio.gather(*flattened_tasks)
+
+        return self.format_response(outputs, separator)
+
+    @llm_token_counter("get_response")
+    def get_response(
+        self,
+        query_str: str,
+        text_chunks: Sequence[str],
+        separator: str = "\n---------------------\n",
+    ) -> RESPONSE_TEXT_TYPE:
+        """Apply the same prompt to text chunks and return responses"""
+
+        if self._streaming:
+            raise ValueError("Unable to stream in Accumulate response mode")
+
+        tasks = [
+            self._give_responses(query_str, text_chunk, use_async=self._use_async)
+            for text_chunk in text_chunks
+        ]
+
+        outputs = self.flatten_list(tasks)
+
+        if self._use_async:
+            outputs = run_async_tasks(outputs)
+
+        return self.format_response(outputs, separator)
+
+    def _give_responses(
+        self, query_str: str, text_chunk: str, use_async: bool = False
+    ) -> List[Any]:
+        """Give responses given a query and a corresponding text chunk."""
+        text_qa_template = self.text_qa_template.partial_format(query_str=query_str)
+        qa_text_splitter = (
+            self._service_context.prompt_helper.get_text_splitter_given_prompt(
+                text_qa_template, 1
+            )
+        )
+        text_chunks = qa_text_splitter.split_text(text_chunk)
+
+        predictor = (
+            self._service_context.llm_predictor.apredict
+            if use_async
+            else self._service_context.llm_predictor.predict
+        )
+
+        return [
+            predictor(
+                text_qa_template,
+                context_str=cur_text_chunk,
+            )
+            for cur_text_chunk in text_chunks
+        ]
+
+
 def get_response_builder(
     service_context: ServiceContext,
     text_qa_template: Optional[QuestionAnswerPrompt] = None,
@@ -622,5 +734,12 @@ def get_response_builder(
             simple_template=simple_template,
             streaming=streaming,
         )
+    elif mode == ResponseMode.ACCUMULATE:
+        return Accumulate(
+            service_context=service_context,
+            text_qa_template=text_qa_template,
+            streaming=streaming,
+            use_async=use_async,
+        )
     else:
         raise ValueError(f"Unknown mode: {mode}")
diff --git a/llama_index/indices/response/type.py b/llama_index/indices/response/type.py
index f0d283a584a50..ebc7f84c1e327 100644
--- a/llama_index/indices/response/type.py
+++ b/llama_index/indices/response/type.py
@@ -10,3 +10,4 @@ class ResponseMode(str, Enum):
     TREE_SUMMARIZE = "tree_summarize"
     GENERATION = "generation"
     NO_TEXT = "no_text"
+    ACCUMULATE = "accumulate"
diff --git a/tests/indices/response/test_response_builder.py b/tests/indices/response/test_response_builder.py
index cef8cb56057bd..f3b5bab88cc3e 100644
--- a/tests/indices/response/test_response_builder.py
+++ b/tests/indices/response/test_response_builder.py
@@ -1,5 +1,6 @@
 """Test response utils."""
 
+import asyncio
 from typing import List
 
 from llama_index.constants import MAX_CHUNK_OVERLAP, MAX_CHUNK_SIZE, NUM_OUTPUTS
@@ -134,3 +135,160 @@ def test_tree_summarize_response(mock_service_context: ServiceContext) -> None:
     )
     # TODO: fix this output, the \n join appends unnecessary results at the end
     assert str(response) == "What is?:This:is:a:bar:This:is:another:test"
+
+
+def test_accumulate_response(
+    mock_service_context: ServiceContext,
+    documents: List[Document],
+) -> None:
+    """Test accumulate response."""
+    # test response with ResponseMode.ACCUMULATE
+    # NOTE: here we want to guarante that prompts have 0 extra tokens
+    mock_qa_prompt_tmpl = "{context_str}{query_str}"
+    mock_qa_prompt = QuestionAnswerPrompt(mock_qa_prompt_tmpl)
+
+    # max input size is 11, prompt is two tokens (the query) --> 9 tokens
+    # --> padding is 1 --> 8 tokens
+    prompt_helper = PromptHelper(
+        11, 0, 0, tokenizer=mock_tokenizer, separator="\n\n", chunk_size_limit=4
+    )
+    service_context = mock_service_context
+    service_context.prompt_helper = prompt_helper
+    cur_chunk_size = prompt_helper.get_chunk_size_given_prompt("", 1, padding=1)
+    # outside of compact, assert that chunk size is 4
+    assert cur_chunk_size == 4
+
+    # within compact, make sure that chunk size is 8
+    query_str = "What is?"
+    texts = [
+        "This\nis\nbar",
+        "This\nis\nfoo",
+    ]
+    builder = get_response_builder(
+        service_context=service_context,
+        text_qa_template=mock_qa_prompt,
+        mode=ResponseMode.ACCUMULATE,
+    )
+
+    response = builder.get_response(text_chunks=texts, query_str=query_str)
+    expected = (
+        "Response 1: What is?:This\n"
+        "---------------------\n"
+        "Response 2: What is?:is\n"
+        "---------------------\n"
+        "Response 3: What is?:bar\n"
+        "---------------------\n"
+        "Response 4: What is?:This\n"
+        "---------------------\n"
+        "Response 5: What is?:is\n"
+        "---------------------\n"
+        "Response 6: What is?:foo"
+    )
+    assert str(response) == expected
+
+
+def test_accumulate_response_async(
+    mock_service_context: ServiceContext,
+    documents: List[Document],
+) -> None:
+    """Test accumulate response."""
+    # test response with ResponseMode.ACCUMULATE
+    # NOTE: here we want to guarante that prompts have 0 extra tokens
+    mock_qa_prompt_tmpl = "{context_str}{query_str}"
+    mock_qa_prompt = QuestionAnswerPrompt(mock_qa_prompt_tmpl)
+
+    # max input size is 11, prompt is two tokens (the query) --> 9 tokens
+    # --> padding is 1 --> 8 tokens
+    prompt_helper = PromptHelper(
+        11, 0, 0, tokenizer=mock_tokenizer, separator="\n\n", chunk_size_limit=4
+    )
+    service_context = mock_service_context
+    service_context.prompt_helper = prompt_helper
+    cur_chunk_size = prompt_helper.get_chunk_size_given_prompt("", 1, padding=1)
+    # outside of compact, assert that chunk size is 4
+    assert cur_chunk_size == 4
+
+    # within compact, make sure that chunk size is 8
+    query_str = "What is?"
+    texts = [
+        "This\nis\nbar",
+        "This\nis\nfoo",
+    ]
+    builder = get_response_builder(
+        service_context=service_context,
+        text_qa_template=mock_qa_prompt,
+        mode=ResponseMode.ACCUMULATE,
+        use_async=True,
+    )
+
+    response = builder.get_response(text_chunks=texts, query_str=query_str)
+    expected = (
+        "Response 1: What is?:This\n"
+        "---------------------\n"
+        "Response 2: What is?:is\n"
+        "---------------------\n"
+        "Response 3: What is?:bar\n"
+        "---------------------\n"
+        "Response 4: What is?:This\n"
+        "---------------------\n"
+        "Response 5: What is?:is\n"
+        "---------------------\n"
+        "Response 6: What is?:foo"
+    )
+    assert str(response) == expected
+
+
+def test_accumulate_response_aget(
+    mock_service_context: ServiceContext,
+    documents: List[Document],
+) -> None:
+    """Test accumulate response."""
+    # test response with ResponseMode.ACCUMULATE
+    # NOTE: here we want to guarante that prompts have 0 extra tokens
+    mock_qa_prompt_tmpl = "{context_str}{query_str}"
+    mock_qa_prompt = QuestionAnswerPrompt(mock_qa_prompt_tmpl)
+
+    # max input size is 11, prompt is two tokens (the query) --> 9 tokens
+    # --> padding is 1 --> 8 tokens
+    prompt_helper = PromptHelper(
+        11, 0, 0, tokenizer=mock_tokenizer, separator="\n\n", chunk_size_limit=4
+    )
+    service_context = mock_service_context
+    service_context.prompt_helper = prompt_helper
+    cur_chunk_size = prompt_helper.get_chunk_size_given_prompt("", 1, padding=1)
+    # outside of compact, assert that chunk size is 4
+    assert cur_chunk_size == 4
+
+    # within compact, make sure that chunk size is 8
+    query_str = "What is?"
+    texts = [
+        "This\nis\nbar",
+        "This\nis\nfoo",
+    ]
+    builder = get_response_builder(
+        service_context=service_context,
+        text_qa_template=mock_qa_prompt,
+        mode=ResponseMode.ACCUMULATE,
+    )
+
+    response = asyncio.run(
+        builder.aget_response(
+            text_chunks=texts,
+            query_str=query_str,
+            separator="\nWHATEVER~~~~~~\n",
+        )
+    )
+    expected = (
+        "Response 1: What is?:This\n"
+        "WHATEVER~~~~~~\n"
+        "Response 2: What is?:is\n"
+        "WHATEVER~~~~~~\n"
+        "Response 3: What is?:bar\n"
+        "WHATEVER~~~~~~\n"
+        "Response 4: What is?:This\n"
+        "WHATEVER~~~~~~\n"
+        "Response 5: What is?:is\n"
+        "WHATEVER~~~~~~\n"
+        "Response 6: What is?:foo"
+    )
+    assert str(response) == expected

The current response builders don't offer a way to accumulate new responses from each chunk of text. Refine, for example, is meant for iterating over chunks and modifying the answer with each iteration until you end up with 1 answer. Being able to accumulate responses allows people (like me) to use a prompt like "Summarize this text" and generate multiple summaries, 1 for each chunk.

Example:
```
query_engine = RetrieverQueryEngine.from_args(
    retriever=retriever,
    service_context=service_context,
    node_postprocessors=[
        SimilarityPostprocessor(similarity_cutoff=0.7)
    ],
    text_qa_template=QA_PROMPT,
    response_mode='accumulate'
)
```

**NOTE:** This response builder does not work for streaming (and will fail with an error saying so if you try). In order to get streaming working, I think we'd have to make it possible for `get_response` to return an array of generators which would probably break other consumers? Happy to discuss solutions though.