Cancelled error with langgraph runs

[kedarsp-informa]

I have an inconsistent issue as CancelledError().
I have a langgraph and the error seems to be within langgraph library.
I am also using asyncio in my code as:

workflow.add_node(RAG, lambda state: asyncio.run(generate_with_rag(state=state, generation_rag_chain=generation_rag_chain)))
workflow.add_node(STORE_CHAT, lambda state: asyncio.run(store_chat(state=state, memory=memory)))

and

if chat_history:
        task1 = asyncio.create_task(router_chain.ainvoke({"question": question, "chat_history": state['chat_history']}))
        task2 = asyncio.create_task(rephrase_chain.ainvoke({"question": question, "chat_history": state['chat_history']}))
        source, follow_up_question = await asyncio.gather(task1, task2)
    else:
        source = await asyncio.shield(router_chain.ainvoke({"question": question, "chat_history": state['chat_history']}))
        follow_up_question = question

The below is the error trace:
_CancelledError()Traceback (most recent call last):
File "/usr/local/lib/python3.11/site-packages/langgraph/pregel/init.py", line 1189, in astream
done, inflight = await asyncio.wait(
^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/asyncio/tasks.py", line 418, in wait
return await _wait(fs, timeout, return_when, loop)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/asyncio/tasks.py", line 525, in wait
await waiter
asyncio.exceptions.CancelledError
During handling of the above exception, another exception occurred:
Traeback (most recent call last):
File "/usr/local/lib/python3.11/site-packages/langgraph/pregel/init.py", line 1132, in astream
async with AsyncPregelLoop(
asyncio.exceptions.CancelledError
If anyone has encountered this error before or any idea on this would be helpful. Thanks

[delorenzotommaso]

I get the same error when execution time reaches about 60s

[nfcampos]

Hi, is this in langgraph cloud?

[tommasodelorenzo]

Hi, nope. It is a graph called from a fastAPI endpoint deployed as an azure container app. Logging everything to LangSmith

[nfcampos]

Hi @kedarsp-informa you shouldn't use asyncio.run in add_node we have native support for async functions in nodes. Using asyncio.run would create a new event loop for each node, which is very inefficient and can cause issues such as cancellation not propagating from graph to node etc