Repository: run-llama/llama_index
Pull request: Adding streaming support for SQLAutoVectorQueryEngine
# Description

Streaming support is not present for SQLAutoVectorQueryEngine

Fixes #13713

## New Package? 
No

Did I fill in the `tool.llamahub` section in the `pyproject.toml` and provide a detailed README.md for my new integration or package?

- [ ] Yes
- [x] No

## Version Bump?

Did I bump the version in the `pyproject.toml` file of the package I am updating? (Except for the `llama-index-core` package)

- [ ] Yes
- [x] No

## Type of Change

Please delete options that are not relevant.

- [ ] Bug fix (non-breaking change which fixes an issue)
- [x] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to not work as expected)
- [ ] This change requires a documentation update

## How Has This Been Tested?

Please describe the tests that you ran to verify your changes. Provide instructions so we can reproduce. Please also list any relevant details for your test configuration

- [ ] Added new unit/integration tests
- [ ] Added new notebook (that tests end-to-end)
- [ ] I stared at the code and made sure it makes sense
- [x] I changed the code locally and made sure that it works

## Suggested Checklist:

- [x] I have performed a self-review of my own code
- [x] I have commented my code, particularly in hard-to-understand areas
- [ ] I have made corresponding changes to the documentation
- [ ] I have added Google Colab support for the newly added notebooks.
- [ ] My changes generate no new warnings
- [ ] I have added tests that prove my fix is effective or that my feature works
- [ ] New and existing unit tests pass locally with my changes
- [ ] I ran `make format; make lint` to appease the lint gods


Problem statement:
[Question]: while using SQLAutoVectorQueryEngine as query engine, how  to stream responses from openai llm?
### Question Validation

- [X] I have searched both the documentation and discord for an answer.

### Question

Im recently having issues in streaming responses. Im using sqlautovectorqueryengine where the vector store is in pinecone and structured data is stored in sqldb.how to stream the response?

Hints from issue discussion:
<!-- Greeting -->
Hey @nibinprasad! I'm here to assist you with any bugs, questions, or contributions you might have. Let's tackle this together!

<!-- Answer -->
To stream responses from OpenAI's LLM while using `SQLAutoVectorQueryEngine` with a Pinecone vector store, you need to use the `stream` parameter in the OpenAI API call and configure the query engine to support streaming. Here is how you can modify the `SQLAutoVectorQueryEngine` setup to enable streaming:

```python
from llama_index.llms.openai import OpenAI
from llama_index.core.query_engine import SQLAutoVectorQueryEngine
from llama_index.core import get_response_synthesizer

# Initialize the OpenAI LLM with streaming enabled
llm = OpenAI(model="gpt-4", stream=True)

# Create a response synthesizer with streaming enabled
synth = get_response_synthesizer(streaming=True)

# Define the SQLAutoVectorQueryEngine with the streaming LLM and response synthesizer
query_engine = SQLAutoVectorQueryEngine(
    sql_tool, vector_tool, llm=llm, response_synthesizer=synth
)

# Query the engine and handle the streaming response
streaming_response = query_engine.query(
    "Tell me about the arts and culture of the city with the highest population"
)

# Print the streaming response as it arrives
streaming_response.print_response_stream()
```

In this setup, the `OpenAI` instance is initialized with `stream=True`, and the `get_response_synthesizer` function is used to create a response synthesizer with streaming enabled. The `SQLAutoVectorQueryEngine` is then configured with both the streaming LLM and the response synthesizer. Finally, the `print_response_stream` method is used to print the streaming response as it arrives <sup>[[1]](https://github.com/run-llama/llama_index/blob/main/docs/docs/examples/query_engine/SQLAutoVectorQueryEngine.ipynb)</sup><sup>[[2]](https://github.com/run-llama/llama_index/blob/main/docs/docs/module_guides/deploying/query_engine/streaming.md)</sup>.


<!-- Dosu Comment Footer -->
<details open>
<summary>

#### Details
</summary>

**About Dosu**
This response is meant to be useful and save you time. It is **not meant to be a precise solution**, but rather a starting point for your own research.

Help me be more useful! Please leave a 👍 if this is helpful and 👎 if it is irrelevant.
</details>

*To continue the conversation, mention @dosu.*

@dosu TypeError: SQLAutoVectorQueryEngine.__init__() got an unexpected keyword argument 'response_synthesizer'

<!-- Answer -->
To resolve the `TypeError` and properly configure `SQLAutoVectorQueryEngine` to support streaming responses from OpenAI's LLM, you need to ensure that you are setting `streaming=True` when building the query engine. Here is how you can do it using both the high-level and low-level APIs:

### High-Level API

```python
query_engine = index.as_query_engine(streaming=True, similarity_top_k=1)
```

### Low-Level API

```python
from llama_index.core import get_response_synthesizer

synth = get_response_synthesizer(streaming=True, ...)
query_engine = RetrieverQueryEngine(response_synthesizer=synth, ...)
```

After configuring the query engine, you can call `query` to get a `StreamingResponse` object:

```python
streaming_response = query_engine.query(
    "What did the author do growing up?",
)
```

You can then iterate over the tokens as they arrive:

```python
for text in streaming_response.response_gen:
    # do something with text as they arrive.
    pass
```

Or print the text as they arrive:

```python
streaming_response.print_response_stream()
```

Make sure that the LLM you are using supports streaming, otherwise a `NotImplementedError` will be raised.

Additionally, when using `SQLAutoVectorQueryEngine`, ensure that the query engines are correctly instantiated and passed. Here is an example of how to create an instance of `SQLAutoVectorQueryEngine`:

```python
from llama_index.legacy.query_engine.sql_vector_query_engine import SQLAutoVectorQueryEngine
from llama_index.legacy.indices.struct_store.sql_query import BaseSQLTableQueryEngine
from llama_index.legacy.query_engine.retriever_query_engine import RetrieverQueryEngine
from llama_index.legacy.tools.query_engine import QueryEngineTool

sql_query_engine = BaseSQLTableQueryEngine(...)
vector_query_engine = RetrieverQueryEngine(...)

sql_query_tool = QueryEngineTool.from_defaults(
    sql_query_engine, name="SQL Tool", description="SQL Query Engine Tool"
)
vector_query_tool = QueryEngineTool.from_defaults(
    vector_query_engine, name="Vector Tool", description="Vector Query Engine Tool"
)

query_engine = SQLAutoVectorQueryEngine.from_sql_and_vector_query_engines(
    sql_query_engine=sql_query_engine,
    sql_tool_name="SQL Tool",
    sql_tool_description="SQL Query Engine Tool",
    vector_auto_retriever=vector_query_engine,
    vector_tool_name="Vector Tool",
    vector_tool_description="Vector Query Engine Tool",
    streaming=True
)
```

This ensures that the `SQLAutoVectorQueryEngine` is properly configured to handle both SQL and vector queries, and supports streaming responses <sup>[[1]](https://github.com/run-llama/llama_index/blob/main/docs/docs/module_guides/deploying/query_engine/streaming.md)</sup><sup>[[2]](https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/query_engine/sql_vector_query_engine.py)</sup><sup>[[3]](https://github.com/run-llama/llama_index/blob/main/llama-index-legacy/llama_index/legacy/query_engine/sql_vector_query_engine.py)</sup>.


<!-- Dosu Comment Footer -->
<details>
<summary>

#### Details
</summary>

**About Dosu**
This response is meant to be useful and save you time. It is **not meant to be a precise solution**, but rather a starting point for your own research.

Help me be more useful! Please leave a 👍 if this is helpful and 👎 if it is irrelevant.
</details>

*To continue the conversation, mention @dosu.*

@dosu SQLAutoVectorQueryEngine doesnt have a parameter called streaming.
<!-- Answer -->
To enable streaming in `SQLAutoVectorQueryEngine`, you need to configure the underlying components to support streaming. Here is how you can do it:

1. **Configure the LLM with streaming enabled:**

```python
from llama_index.llms.openai import OpenAI

# Initialize the OpenAI LLM with streaming enabled
llm = OpenAI(model="gpt-4", stream=True)
```

2. **Create the SQL and Vector Query Engines:**

```python
from llama_index.legacy.indices.struct_store.sql_query import BaseSQLTableQueryEngine
from llama_index.legacy.query_engine.retriever_query_engine import RetrieverQueryEngine
from llama_index.legacy.indices.vector_store.retrievers.auto_retriever import VectorIndexAutoRetriever

# Create SQL query engine
sql_query_engine = BaseSQLTableQueryEngine(...)

# Create vector auto retriever
vector_auto_retriever = VectorIndexAutoRetriever(...)

# Create vector query engine
vector_query_engine = RetrieverQueryEngine(retriever=vector_auto_retriever, llm=llm)
```

3. **Create Query Engine Tools:**

```python
from llama_index.legacy.tools.query_engine import QueryEngineTool

sql_query_tool = QueryEngineTool.from_defaults(
    sql_query_engine, name="SQL Tool", description="SQL Query Engine Tool"
)
vector_query_tool = QueryEngineTool.from_defaults(
    vector_query_engine, name="Vector Tool", description="Vector Query Engine Tool"
)
```

4. **Instantiate `SQLAutoVectorQueryEngine` with streaming enabled:**

```python
from llama_index.legacy.query_engine.sql_vector_query_engine import SQLAutoVectorQueryEngine

query_engine = SQLAutoVectorQueryEngine.from_sql_and_vector_query_engines(
    sql_query_engine=sql_query_engine,
    sql_tool_name="SQL Tool",
    sql_tool_description="SQL Query Engine Tool",
    vector_auto_retriever=vector_query_engine,
    vector_tool_name="Vector Tool",
    vector_tool_description="Vector Query Engine Tool",
    streaming=True  # Enable streaming
)
```

5. **Query the engine and handle the streaming response:**

```python
streaming_response = query_engine.query(
    "Tell me about the arts and culture of the city with the highest population"
)

# Print the streaming response as it arrives
for text in streaming_response.response_gen:
    print(text, end="")
```

This setup ensures that the `SQLAutoVectorQueryEngine` is correctly configured to handle both SQL and vector queries, with streaming responses enabled <sup>[[1]](https://github.com/run-llama/llama_index/blob/main/docs/docs/module_guides/deploying/query_engine/streaming.md)</sup><sup>[[2]](https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/query_engine/sql_vector_query_engine.py)</sup><sup>[[3]](https://github.com/run-llama/llama_index/blob/main/docs/docs/examples/query_engine/SQLAutoVectorQueryEngine.ipynb)</sup><sup>[[4]](https://github.com/run-llama/llama_index/blob/main/llama-index-legacy/llama_index/legacy/query_engine/sql_vector_query_engine.py)</sup>.


<!-- Dosu Comment Footer -->
<details>
<summary>

#### Details
</summary>

**About Dosu**
This response is meant to be useful and save you time. It is **not meant to be a precise solution**, but rather a starting point for your own research.

Help me be more useful! Please leave a 👍 if this is helpful and 👎 if it is irrelevant.
</details>

*To continue the conversation, mention @dosu.*

 SQLAutoVectorQueryEngine.from_sql_and_vector_query_engines(
    sql_query_engine=sql_query_engine,
    sql_tool_name="SQL Tool",
    sql_tool_description="SQL Query Engine Tool",
    vector_auto_retriever=vector_query_engine,
    vector_tool_name="Vector Tool",
    vector_tool_description="Vector Query Engine Tool",
    streaming=True  # Enable streaming
) there is no parameter streaming=True in sqlautovectorqueryengine class @dosu
<!-- Answer -->
To properly configure `SQLAutoVectorQueryEngine` to support streaming responses from OpenAI's LLM, you need to use the low-level API to compose the query engine and pass `streaming=True` when constructing the `Response Synthesizer`. Here is an example:

```python
from llama_index.core import get_response_synthesizer
from llama_index.query_engine import SQLAutoVectorQueryEngine
from llama_index.core.query_engine.retriever_query_engine import RetrieverQueryEngine
from llama_index.core.indices.struct_store.sql_query import BaseSQLTableQueryEngine
from llama_index.legacy.tools.query_engine import QueryEngineTool
from llama_index.llms.openai import OpenAI

# Set up the OpenAI LLM with streaming enabled
llm = OpenAI(model="gpt-3.5-turbo")

# Construct the response synthesizer with streaming enabled
synth = get_response_synthesizer(streaming=True, llm=llm, ...)

# Create the SQL and vector query engines
sql_query_engine = BaseSQLTableQueryEngine(...)
vector_query_engine = RetrieverQueryEngine(...)

# Create the SQLAutoVectorQueryEngine using the class method
query_engine = SQLAutoVectorQueryEngine.from_sql_and_vector_query_engines(
    sql_query_engine=sql_query_engine,
    sql_tool_name="SQL Tool",
    sql_tool_description="Description of SQL Tool",
    vector_auto_retriever=vector_query_engine,
    vector_tool_name="Vector Tool",
    vector_tool_description="Description of Vector Tool",
    response_synthesizer=synth,
    ...
)

# Perform a query and get a streaming response
streaming_response = query_engine.query("What did the author do growing up?")

# Iterate over the tokens as they arrive
for text in streaming_response.response_gen:
    # do something with text as they arrive.
    pass

# Alternatively, print the text as they arrive
streaming_response.print_response_stream()
```

This setup ensures that the `SQLAutoVectorQueryEngine` is configured to support streaming responses from OpenAI's LLM. The `from_sql_and_vector_query_engines` class method allows you to pass additional keyword arguments, such as the `response_synthesizer`, to configure the engine appropriately <sup>[[1]](https://github.com/run-llama/llama_index/blob/main/docs/docs/module_guides/deploying/query_engine/streaming.md)</sup><sup>[[2]](https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/query_engine/sql_vector_query_engine.py)</sup><sup>[[3]](https://github.com/run-llama/llama_index/blob/main/llama-index-legacy/llama_index/legacy/query_engine/sql_vector_query_engine.py)</sup>.


<!-- Dosu Comment Footer -->
<details>
<summary>

#### Details
</summary>

**About Dosu**
This response is meant to be useful and save you time. It is **not meant to be a precise solution**, but rather a starting point for your own research.

Help me be more useful! Please leave a 👍 if this is helpful and 👎 if it is irrelevant.
</details>

*To continue the conversation, mention @dosu.*

Commit message: fixing the error lint error

Patch to review:
diff --git a/llama-index-core/llama_index/core/query_engine/sql_join_query_engine.py b/llama-index-core/llama_index/core/query_engine/sql_join_query_engine.py
index 705282abe7..820dbee13f 100644
--- a/llama-index-core/llama_index/core/query_engine/sql_join_query_engine.py
+++ b/llama-index-core/llama_index/core/query_engine/sql_join_query_engine.py
@@ -4,7 +4,11 @@
 from typing import Callable, Dict, Optional, Union
 
 from llama_index.core.base.base_query_engine import BaseQueryEngine
-from llama_index.core.base.response.schema import RESPONSE_TYPE, Response
+from llama_index.core.base.response.schema import (
+    RESPONSE_TYPE,
+    Response,
+    StreamingResponse,
+)
 from llama_index.core.callbacks.base import CallbackManager
 from llama_index.core.indices.query.query_transform.base import BaseQueryTransform
 from llama_index.core.indices.struct_store.sql_query import (
@@ -198,6 +202,7 @@ def __init__(
         use_sql_join_synthesis: bool = True,
         callback_manager: Optional[CallbackManager] = None,
         verbose: bool = True,
+        streaming: bool = False,
         # deprecated
         service_context: Optional[ServiceContext] = None,
     ) -> None:
@@ -228,6 +233,7 @@ def __init__(
         )
         self._use_sql_join_synthesis = use_sql_join_synthesis
         self._verbose = verbose
+        self._streaming = streaming
 
     def _get_prompt_modules(self) -> PromptMixinType:
         """Get prompt sub-modules."""
@@ -282,26 +288,49 @@ def _query_sql_other(self, query_bundle: QueryBundle) -> RESPONSE_TYPE:
             print_text(f"query engine response: {other_response}\n", color="pink")
         logger.info(f"> query engine response: {other_response}")
 
-        response_str = self._llm.predict(
-            self._sql_join_synthesis_prompt,
-            query_str=query_bundle.query_str,
-            sql_query_str=sql_query,
-            sql_response_str=str(sql_response),
-            query_engine_query_str=new_query.query_str,
-            query_engine_response_str=str(other_response),
-        )
-        if self._verbose:
-            print_text(f"Final response: {response_str}\n", color="green")
-        response_metadata = {
-            **(sql_response.metadata or {}),
-            **(other_response.metadata or {}),
-        }
-        source_nodes = other_response.source_nodes
-        return Response(
-            response_str,
-            metadata=response_metadata,
-            source_nodes=source_nodes,
-        )
+        if self._streaming:
+            response_str = self._llm.stream(
+                self._sql_join_synthesis_prompt,
+                query_str=query_bundle.query_str,
+                sql_query_str=sql_query,
+                sql_response_str=str(sql_response),
+                query_engine_query_str=new_query.query_str,
+                query_engine_response_str=str(other_response),
+            )
+            if self._verbose:
+                print_text(f"Final response: {response_str}\n", color="green")
+            response_metadata = {
+                **(sql_response.metadata or {}),
+                **(other_response.metadata or {}),
+            }
+            source_nodes = other_response.source_nodes
+            return StreamingResponse(
+                response_str,
+                metadata=response_metadata,
+                source_nodes=source_nodes,
+            )
+        else:
+            response_str = self._llm.predict(
+                self._sql_join_synthesis_prompt,
+                query_str=query_bundle.query_str,
+                sql_query_str=sql_query,
+                sql_response_str=str(sql_response),
+                query_engine_query_str=new_query.query_str,
+                query_engine_response_str=str(other_response),
+            )
+
+            if self._verbose:
+                print_text(f"Final response: {response_str}\n", color="green")
+            response_metadata = {
+                **(sql_response.metadata or {}),
+                **(other_response.metadata or {}),
+            }
+            source_nodes = other_response.source_nodes
+            return Response(
+                response_str,
+                metadata=response_metadata,
+                source_nodes=source_nodes,
+            )
 
     def _query(self, query_bundle: QueryBundle) -> RESPONSE_TYPE:
         """Query and get response."""
@@ -320,10 +349,10 @@ def _query(self, query_bundle: QueryBundle) -> RESPONSE_TYPE:
                     f"Querying other query engine: {result.reason}\n", color="blue"
                 )
             logger.info(f"> Querying other query engine: {result.reason}")
-            response = self._other_query_tool.query_engine.query(query_bundle)
-            if self._verbose:
-                print_text(f"Query Engine response: {response}\n", color="pink")
-            return response
+            return self._other_query_tool.query_engine.query(query_bundle)
+            # if self._verbose:
+            #     print_text(f"Query Engine response: {response}\n", color="pink")
+            # return response
         else:
             raise ValueError(f"Invalid result.ind: {result.ind}")
 


Code review comments:
diff_hunk: @@ -282,26 +288,49 @@ def _query_sql_other(self, query_bundle: QueryBundle) -> RESPONSE_TYPE:
             print_text(f"query engine response: {other_response}\n", color="pink")
         logger.info(f"> query engine response: {other_response}")
 
-        response_str = self._llm.predict(
-            self._sql_join_synthesis_prompt,
-            query_str=query_bundle.query_str,
-            sql_query_str=sql_query,
-            sql_response_str=str(sql_response),
-            query_engine_query_str=new_query.query_str,
-            query_engine_response_str=str(other_response),
-        )
-        if self._verbose:
-            print_text(f"Final response: {response_str}\n", color="green")
-        response_metadata = {
-            **(sql_response.metadata or {}),
-            **(other_response.metadata or {}),
-        }
-        source_nodes = other_response.source_nodes
-        return Response(
-            response_str,
-            metadata=response_metadata,
-            source_nodes=source_nodes,
-        )
+        if self._streaming:
+            response_str = self._llm.stream(
+                self._sql_join_synthesis_prompt,
+                query_str=query_bundle.query_str,
+                sql_query_str=sql_query,
+                sql_response_str=str(sql_response),
+                query_engine_query_str=new_query.query_str,
+                query_engine_response_str=str(other_response),
+            )
+            if self._verbose:
+                print_text(f"Final response: {response_str}\n", color="green")
path: llama-index-core/llama_index/core/query_engine/sql_join_query_engine.py
text: @user1:
printing this will consume the stream I guess lol

@author:
surprisingly this doesn't do it, but if you try to print a `StreamingResponse` object, the stream gets consumed. I have raised a issue for the same at #13893 and that is the reason i have commented the line 353 to 355. 

Though I feel that commenting is not the correct action, because we need some kind of verbosity in place, without consuming the stream, I just dont know how to achieve that.

@user1:
If it's the final response, why do we need to print it? It will be returned to the user 

@author:
so do you want me to remove this?

@user1:
Yea

@author:
done

diff_hunk: @@ -320,10 +349,10 @@ def _query(self, query_bundle: QueryBundle) -> RESPONSE_TYPE:
                     f"Querying other query engine: {result.reason}\n", color="blue"
                 )
             logger.info(f"> Querying other query engine: {result.reason}")
-            response = self._other_query_tool.query_engine.query(query_bundle)
-            if self._verbose:
-                print_text(f"Query Engine response: {response}\n", color="pink")
-            return response
+            return self._other_query_tool.query_engine.query(query_bundle)
+            # if self._verbose:
+            #     print_text(f"Query Engine response: {response}\n", color="pink")
+            # return response
path: llama-index-core/llama_index/core/query_engine/sql_join_query_engine.py
text: @user1:
we should just remove this ?

@author:
depends can we fix the [__str__()](https://github.com/run-llama/llama_index/blob/4bfa6dfefa45a1dff940bf25d3a6c3c5ee4cb66a/llama-index-core/llama_index/core/base/response/schema.py#L120) function here? 
If not, then we should remove this. 

@user1:
I don't think there's anything to fix there 😅 We just shouldn't cast a stream as string unless you want to exhaust the generator 

@user1:
I don't think we need the verbosity here in the first place, since we are printing this and then returning directly to the user 

@author:
should i delete the commented code too?

@user1:
Yessir (commented code is dead code, almost always remove it imo)

@author:
done