issues with streaming with fastapi

[GSequist]

Checked other resources

• I added a very descriptive title to this question.
• I searched the LangChain documentation with the integrated search.
• I used the GitHub search to find a similar question and didn't find it.

Commit to Help

• I commit to help with one of those options 👆

Example Code

app = FastAPI()

async def handle_message(user_id, data, websocket):
    """main wrapper"""
    print(f"\n[handle_message]: received data: {data}")

    print(f"\n[handle_message] userid received: {user_id}")
    query = data.get("message")

    if query:
        print(f"\n[handle_message]: inserting chat text...{query}")

        # emit the user's new message
        message = json.dumps({"type": "new_user_message", "data": query})
        await websocket.send_text(message)

        inputs = {"messages": [HumanMessage(content=query)]}
        response_generator = await to_thread(invoke, inputs)

calling on invoke in another module:

app = workflow.compile()

def invoke(inputs):
    app.invoke(inputs)
    for output in app.stream(inputs):
        for key, value in output.items():
            # print(f"[invoke]: key: {key}, value: {value}")
            print(f"output: {output.items()}")
            yield key, value

this module however despite having graph compiled using streaming = True model does not yield chunk by chunk but instead the entire answer

Description

I am having issue as the response is not streamed chunk by chunk

System Info

pip freeze
aiohttp==3.9.1
aiosignal==1.3.1
annotated-types==0.6.0
anyio==4.2.0
attrs==23.2.0
backoff==2.2.1
beautifulsoup4==4.12.2
certifi==2023.11.17
cffi==1.16.0
chardet==5.2.0
charset-normalizer==3.3.2
click==8.1.7
curl-cffi==0.5.10
dataclasses-json==0.6.3
distro==1.9.0
duckduckgo_search==4.2
emoji==2.9.0
et-xmlfile==1.1.0
faiss-cpu==1.7.4
fastapi==0.109.0
filetype==1.2.0
frozenlist==1.4.1
gunicorn==21.2.0
h11==0.14.0
httpcore==1.0.2
httpx==0.26.0
idna==3.6
Jinja2==3.1.3
joblib==1.3.2
jsonpatch==1.33
jsonpath-python==1.0.6
jsonpointer==2.4
langchain==0.1.0
langchain-community==0.0.12
langchain-core==0.1.16
langchain-openai==0.0.5
langdetect==1.0.9
langgraph==0.0.24
langsmith==0.0.83
lxml==5.1.0
MarkupSafe==2.1.3
marshmallow==3.20.2
multidict==6.0.4
mypy-extensions==1.0.0
networkx==3.2.1
nltk==3.8.1
numpy==1.26.3
openai==1.10.0
openpyxl==3.1.2
packaging==23.2
pandas==2.1.4
pillow==10.2.0
pycparser==2.21
pydantic==2.5.3
pydantic_core==2.14.6
pypdf==3.17.4
python-dateutil==2.8.2
python-docx==1.1.0
python-dotenv==1.0.0
python-iso639==2024.1.2
python-magic==0.4.27
python-multipart==0.0.6
python-pptx==0.6.23
pytz==2023.3.post1
PyYAML==6.0.1
rapidfuzz==3.6.1
regex==2023.12.25
requests==2.31.0
six==1.16.0
sniffio==1.3.0
soupsieve==2.5
SQLAlchemy==2.0.25
starlette==0.35.1
tabulate==0.9.0
tenacity==8.2.3
termcolor==2.4.0
tiktoken==0.5.2
tqdm==4.66.1
typing-inspect==0.9.0
typing_extensions==4.9.0
tzdata==2023.4
unstructured==0.12.0
unstructured-client==0.15.2
urllib3==2.1.0
uvicorn==0.25.0
websockets==12.0
wrapt==1.16.0
XlsxWriter==3.1.9
yarl==1.9.4

[dosubot[bot]]

Hey @GSequist, great to see you diving into another challenge! Hope this one's a fun puzzle too. 🛠️

Based on the code you've provided, it seems like you're trying to stream the response from the invoke function in your FastAPI application. However, it's important to note that the invoke function in LangChain is not designed to be a generator function, and therefore it won't yield results chunk by chunk.

The invoke function is designed to execute a chain of language models and return the final result. If you're trying to stream the response, you might need to modify your approach.

One possible solution could be to refactor your code to use an asynchronous generator with FastAPI's StreamingResponse. Here's a simplified example:

from fastapi import FastAPI, StreamingResponse
import asyncio

app = FastAPI()

async def stream_results():
    for i in range(10):
        yield f"data: {i}\n\n"
        await asyncio.sleep(1)

@app.get("/stream")
async def get_stream():
    return StreamingResponse(stream_results())

In this example, stream_results is an asynchronous generator that yields results over time. The StreamingResponse takes this generator and sends the results to the client as they become available.

However, this approach might not be directly applicable to your use case, as it depends on how the invoke function and the underlying language models are implemented. If the language models are not designed to yield intermediate results, you might need to modify them or use a different approach.

Regarding the similar issues you found in the LangChain repository, they seem to be related but not identical to your problem. The discussions in these issues are about streaming responses from FastAPI endpoints, but they don't seem to involve the invoke function or the specific problem you're facing.

I hope this helps! If you have further questions or if something is not clear, please let me know.

Sources

• can't sql_agent stream with fastapi endpoint
• Stream a chain response word by word to React Native frontend.
• How to combine FastAPI Streaming Response with Langchain OpenAI Stream?
• libs/community/langchain_community/llms/openai.py
• libs/langchain/langchain/chains/base.py

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.

If you want to continue the conversation, start your reply with @dosu-bot.

[GSequist]

@dosu-bot

thanks but the issue is not with the fastapi implementation of streaming the issue is that when i call on the langgraph app.invoke and stream back here:

app = workflow.compile()

def invoke(inputs, metadata):
app.invoke(inputs, metadata)
for output in app.stream(inputs, metadata):
for key, value in output.items():
print(f"[invoke]: key: {key}, value: {value}")
yield key, value

i can see from prints this is not streaming it returns the answer in full once complete

(therefore also then as it returns back to fastapi handler i cannot stream back to front end naturally because its returned in one bulk

[dosubot[bot]]

The behavior you're experiencing is due to the design of the invoke function in LangChain. This function is designed to execute the chain operation synchronously and return the final output after the operation is complete. This is why you're seeing the entire output returned at once, rather than in chunks.

To achieve the chunk-by-chunk output streaming you're looking for, you would need to modify the invoke method to yield outputs as they become available, rather than waiting for the entire operation to complete. This could involve using Python's generator functions.

Here's a conceptual example of how you might start to modify the invoke method to support streaming outputs using a generator function:

def invoke_streaming(
    self,
    input: Dict[str, Any],
    config: Optional[RunnableConfig] = None,
    **kwargs: Any,
) -> Generator[Dict[str, Any], None, None]:
    # Similar setup code as in the original invoke method
    # ...

    try:
        # Assuming _call is modified to be a generator yielding outputs
        for output_chunk in self._call(inputs, run_manager=run_manager):
            yield self.prep_outputs(inputs, output_chunk, return_only_outputs)
    except BaseException as e:
        run_manager.on_chain_error(e)
        raise e
    finally:
        run_manager.on_chain_end(outputs)

This example assumes that _call is modified to yield intermediate results, which invoke_streaming then processes and yields. Implementing this functionality would require a thorough understanding of the chain's operations and might involve significant changes to the underlying logic of how chains and their components handle data processing and callbacks.

Remember, streaming outputs in this manner changes how clients interact with your Chain API, as they'll need to handle data incrementally rather than all at once. This approach can be beneficial for large datasets or real-time processing but requires careful design to ensure reliability and efficiency.

Please note that this is a significant change to the LangChain codebase and might not be compatible with the current design of the framework. It's recommended to thoroughly test this change to ensure it doesn't introduce any unexpected behavior or performance issues.

For more information on how the invoke function is implemented in the LangChain codebase, you can refer to the source code in libs/langchain/langchain/chains/base.py.

Sources

• libs/langchain/langchain/chains/base.py

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.

If you want to continue the conversation, start your reply with @dosu-bot.