How to stream query engine responses in LlamaIndex

A LlamaIndex query engine normally returns an answer after response synthesis finishes. Streaming exposes generated text fragments as they arrive, allowing a chat interface or long-running RAG request to display output before the final fragment is ready.

The high-level query engine API enables this behavior with streaming=True on index.as_query_engine(). The resulting query() call returns a StreamingResponse object whose response_gen generator can feed a terminal, web response, or application callback one fragment at a time.

The selected LLM must implement streaming or LlamaIndex raises NotImplementedError. Query engines that make several LLM calls stream only the final call, so retrieval and earlier synthesis work may still delay the first visible fragment.

1. Create query_engine_streaming_check.py with an offline index and mock LLM.

   query_engine_streaming_check.py

   from llama_index.core import Document, Settings, VectorStoreIndex
   from llama_index.core.embeddings import MockEmbedding
   from llama_index.core.llms import MockLLM
    
    
   Settings.llm = MockLLM()
   Settings.embed_model = MockEmbedding(embed_dim=8)
    
   documents = [
       Document(
           text=(
               "Acme support answers refund questions from the policy index. "
               "Refund requests remain eligible for 30 days after purchase."
           )
       )
   ]
   index = VectorStoreIndex.from_documents(documents)

   MockLLM and MockEmbedding keep the smoke test offline while exercising the same query engine interface used with a production provider.

2. Append the streaming query engine configuration after the index assignment.

   query_engine = index.as_query_engine(
       streaming=True,
       similarity_top_k=1,
       response_mode="simple_summarize",
   )

   An application's existing response mode can remain when it supports the selected streaming LLM. Manually assembled RetrieverQueryEngine instances receive the same setting through get_response_synthesizer().

3. Append the streaming response check after the query engine configuration.

   streaming_response = query_engine.query("How long is the Acme refund window?")
   fragments = list(streaming_response.response_gen)
   full_response = "".join(fragments)
    
   assert type(streaming_response).__name__ == "StreamingResponse"
   assert len(fragments) > 1
   assert "Refund requests remain eligible" in full_response
    
   print(f"response_class={type(streaming_response).__name__}")
   print(f"fragment_count={len(fragments)}")
   print(f"contains_policy_context={'Refund requests remain eligible' in full_response}")

   Production handlers should forward each value directly while iterating over response_gen instead of converting the generator to a list. LLM providers may yield tokens, substrings, or individual characters.

4. Run the completed smoke test to confirm the response streams multiple fragments with retrieved policy context.

   $ python query_engine_streaming_check.py
   response_class=StreamingResponse
   fragment_count=323
   contains_policy_context=True