CancelledError when exiting from subgraph

[madityaar]

Hello everyone, i'm currently experiencing unsolved error from my LangGraph implementation. i kinda modified the tutorial from Introduction to LangGraph on langchain academy specifically on module 4 to my own use case.
Heres the graph design:

i attach this subgraph as one of a tool for my main agent capabilities. the problem starts when i add cancel_analyst_node to exit the subgraph. when i implement it i receive this error:

  File "/usr/local/lib/python3.12/site-packages/langgraph/pregel/__init__.py", line 2655, in astream
    async for _ in runner.atick(


  File "/usr/local/lib/python3.12/site-packages/langgraph/pregel/runner.py", line 367, in atick
    done, inflight = await asyncio.wait(
                     ^^^^^^^^^^^^^^^^^^^


  File "/usr/local/lib/python3.12/asyncio/tasks.py", line 464, in wait
    return await _wait(fs, timeout, return_when, loop)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^


  File "/usr/local/lib/python3.12/asyncio/tasks.py", line 550, in _wait
    await waiter


asyncio.exceptions.CancelledError: Cancelled by cancel scope ffff8e347bc0



During handling of the above exception, another exception occurred:



Traceback (most recent call last):


  File "/usr/local/lib/python3.12/site-packages/langgraph/pregel/__init__.py", line 2596, in astream
    async with AsyncPregelLoop(
               ^^^^^^^^^^^^^^^^


  File "/usr/local/lib/python3.12/site-packages/langgraph/pregel/loop.py", line 1393, in __aexit__
    return await exit_task
           ^^^^^^^^^^^^^^^


  File "/usr/local/lib/python3.12/contextlib.py", line 707, in __aexit__
    async def __aexit__(self, *exc_details):


asyncio.exceptions.CancelledError: ('Cancelled by cancel scope ffff8e347bc0', <Task cancelled name='Task-124' coro=<AsyncExitStack.__aexit__() done, defined at /usr/local/lib/python3.12/contextlib.py:707>>)

and heres my node implementation:

...
async def human_feedback(state: GenerateAnalystsState):
    """ No-op node that should be interrupted on """
    determine_feedback_prompt=f"""Determine whether user having feedback or not about this LLM generated results of analyst:
    <result>
    {state["question_for_user"]}
    </result>
    you will determine feedback_type and feedback_value
    If user is having a clear feedback about the result and have something to say then fill type on 'Positive' and fill value with user feedback.
    if user is having an unclear feedback about the result then type 'Unclear' and fill nothing on value until its clear.
    if user is satified with the result, not having any feedback like saying 'no', saying continue, or something similar then type 'Negative' and fill nothing on value.
    if user want to cancel the operation by saying 'cancel' or something relevant to abort the operation then type 'Cancel' and fill "cancel" on value
    Here is the user response:
    """
    question=state["question_for_user"]
    while True:
        value = interrupt( 
            {
                "question_for_user": question
            })
        
        result=await model.with_structured_output(FeedbackResponse).ainvoke([SystemMessage(content=determine_feedback_prompt),HumanMessage(content=value)])
        if result.feedback_type == "Positive":
            break
        elif result.feedback_type == "Unclear":
            question=await model.ainvoke([SystemMessage(content=f"""identify user language from this user instruction:
                <instruction>
                {state["refined_question"]}
                </instruction>
                Ask user to be provide a more clear feedback. try to tell user to clearly state if they dont have any feedback. use the same language as user.
            """)]).content
            #question="Could you be more clear with your feedback? if you dont have any feedback then say 'No'"
            value=None
            continue
        else:
            break
        
    if result.feedback_type=="Positive":
        return {
                "human_analyst_feedback": value
                }
    elif result.feedback_type=="Cancel":
        return {
            "human_analyst_feedback": "cancel"
        }
    else:
        return {
                "human_analyst_feedback": None
                }
    pass
def cancel_analyst_node(state: GenerateAnalystsState, config: RunnableConfig, store: BaseStore):
    tool_calls = state['messages'][-1].tool_calls
    return {"messages":[{"role": "tool", "content": "the user have cancelled the operation.", "tool_call_id":tool_calls[0]['id']}]}
def initiate_all_interviews(state: ResearchGraphState,writer: StreamWriter):
    """ This is the "map" step where we run each interview sub-graph using Send API """    
    writer({"message_to_user": "Conducting research for the report"})
    # Check if human feedback
    human_analyst_feedback=state.get('human_analyst_feedback', None)
    if human_analyst_feedback:
        # Return to create_analysts
        if "cancel" in human_analyst_feedback:
            return "cancel_analyst_node"
        else:
            return "refine_user_query"

report_creation = StateGraph(ResearchGraphState)
report_creation.add_node("conduct_interview", interview_builder.compile().with_config(run_name="Conduct Interviews"))
report_creation.add_node("human_feedback",human_feedback)
report_creation.add_node("cancel_analyst_node",cancel_analyst_node)


report_creation.add_conditional_edges("human_feedback", initiate_all_interviews, ["refine_user_query", "conduct_interview","cancel_analyst_node"])
create_report = report_creation.compile(checkpointer=checkpointer)
...
builder = StateGraph(TaskMaestroExtendedState)
builder.add_node("create_report", create_report)
graph = builder.compile(checkpointer=checkpointer, store=store)

could anyone help me debug this? any feedback would be appreciated, thanks!

[madityaar]

It turns out that the problem is on my FastApi StreamingResponse SSE implementation. It is causing an unhandled error exception on the Chatbot UI side, which also triggered the error as mentioned above.

[satyaprakash1729]

@madityaar bunch of people are running into similar issue. Could you help how you resolved this?

[rishabh208gupta]

Hey @madityaar, I am facing a similar issue. How did you resolve this? Could you share your implementation of StreamingResponse SSE

[ehsanahmadkhan525]

any updates?

[rishabh208gupta]

I had issues with my custom SSE implementation as well as mentioned by @madityaar. My code for custom SSE implementation

import logging
from contextlib import asynccontextmanager
import httpx
from fastapi import FastAPI
import anyio
from starlette.types import Send
from sse_starlette import EventSourceResponse
from sse_starlette.event import ensure_bytes

class StreamingResponseWithStatusCode(EventSourceResponse):
    """
    Variation of StreamingResponse that can dynamically decide the HTTP status code,
    based on the return value of the content iterator (parameter `content`).
    Expects the content to yield either just str content as per the original `StreamingResponse`
    or else tuples of (`content`: `str`, `status_code`: `int`).
    """

    async def _stream_response(self, send: Send) -> None:
        """Send out SSE data to the client as it becomes available in the iterator."""
        await send(
            {
                "type": "http.response.start",
                "status": self.status_code,
                "headers": self.raw_headers,
            }
        )

        async for data in self.body_iterator:
            if isinstance(data, tuple):
                content, self.status_code = data
            else:
                content = data
            if self.status_code != 200:
                content = f'{{"error": "{content}", "status": "{self.status_code}"}}'
            chunk = ensure_bytes(content, self.sep)
            logger.debug("chunk: %s", chunk)
            with anyio.move_on_after(self.send_timeout) as cancel_scope:
                await send(
                    {"type": "http.response.body", "body": chunk, "more_body": True}
                )

            if cancel_scope and cancel_scope.cancel_called:
                if hasattr(self.body_iterator, "aclose"):
                    await self.body_iterator.aclose()
                raise SendTimeoutError()

        async with self._send_lock:
            self.active = False
            await send({"type": "http.response.body", "body": b"", "more_body": False})