Terminate the pipeline once the request connect through FastAPI is disconnected

[cloudissue]

I am totally new with Haystack and I am wondering if there are some way to terminate the running pipeline.
I have built the RAG based on Haystack as an open API.
The original codes should be something as below.

@app.post('/api')
async def api(request: Request, request_body: RequestBody):
    if request_body.searchType == "chat":
        response = await asyncio.wait_for( run_chat(),  timeout=60)
    ......

async def run_chat():
    pipeline = Pipeline()
    ........
    return pipeline.run()

Once the request is disconnected during the process, I'd like to stop the running process of the pipeline.
It seems that once the pipeline is started, it could not be stopped.
I have followed the discussion in FastAPI project to handle the event as "Connection Disconnected".
The test passed with sleep function, but it does work with the pipeline.
The pipeline will be run completely after I stop the connection script.

from fastapi import FastAPI, Request
import anyio
import httpx

app = FastAPI()


async def disconnected(request: Request, cancel_scope: anyio.CancelScope) -> None:
    while True:
        message = await request.receive()
        if message["type"] == "http.disconnect":
            cancel_scope.cancel()
            break


@app.get("/", status_code=204)
async def home(request: Request) -> None:
    async with anyio.create_task_group() as tg:
        tg.start_soon(disconnected, request, tg.cancel_scope)
        await anyio.sleep(10)

Is there any way or function to handle the lifecycle of the running pipeline or could be used to stop the process?
Could you please kindly share your suggestions and thanks a lot for your time.

[anakin87]

Hello! Could you please explain more about your use case?

I also recommend you take a look at Hayhooks: our project to deploy Haystack pipelines behind a REST API. It also makes it easy to deploy and undeploy Pipelines.

(FYI @mpangrazzi)

[cloudissue]

Thanks for your reply and sorry to make you confused on the description.
I'd like to share more information about the case.

Common process:
Originally the request access the FastAPI and trigger the Haystack pipeline running in the background.
This may take some time to wait for the pipeline running completely and return the response.

This case:
The request access the FastAPI and trigger the pipeline running in the background as above.
This time the request is disconnected caused by the network or user cancel operations.
The connection between the client and the FastAPI server is not available any more.
But the pipeline is still running in the background.
I'd like to terminate or interrupt or stop the running pipeline at this situation.
Because the response could not be returned as the request is canceled or disconnected.

So I'd like to have the lifecycle of the running pipeline or function which I could use to kill the pipeline.
Maybe I could get the running instance ID and stop the components or something like this.

Meanwhile, I have checked the Hayhooks you shared above.
It looks great and may I know if it could provide the solutions on the above case?
Thanks

[mpangrazzi]

Hi @cloudissue! To achieve what you want, you need to implement something which basically continuously polls the request and check the status of the client side.

You can take this answer as a starting point: this user has implement an async context manager which will cancel when a client request disconnects. It should be enough to handle your use case.

Out of curiosity, can you be more specific about your use case? Does the client requests coming from a browser? So e.g. if one closes the tab while awaiting the response, the request will be cancelled?

[cloudissue]

Thanks a lot for the quick response.

Actually, I have tried with the suggestion from FastAPI side such as the answer you shared above and this one.
It works fine with the testing sleep function and does not terminate the pipeline already started.
That is the reason I checked here for specific functions to work with the pipeline.

I will try with that again and maybe share some sample codes later.
So may I know once the pipeline.run() function is called, could it be stopped or not?

---

Out of curiosity, can you be more specific about your use case? Does the client requests coming from a browser? So e.g. if one closes the tab while awaiting the response, the request will be cancelled?

Sure, the request may from the browser and client both. We have a webUI and also provide it as an open API. The user could access the API endpoint through the web browser or the scripts. The example you shared above is the one we care about. Once the user close the tab while awaiting the response, the request will be disconnected and the pipeline running in the background should be canceled to save the cost of LLM.

[mpangrazzi]

Thank you for disclosing the use case! Actually, currently there's no method like e.g. pipeline.cancel() to actually stop the execution of pipeline's components. But since IMHO the use case is valid, we may want to internally discuss about a possible implementation.

So, ATM you may try to run the pipeline on a separate process and forcefully terminating it if client disconnects. But this would be not the best way if you have to handle a fair number of requests.

[cloudissue]

I see. Thanks a lot for your information and suggestions. I understand the current situations. I will discuss with the team for further solutions. Looking forward to the updates on this point with Haystack in the future if necessary after the discussion. Thank you very much for your time.