How to stream API with session id??

[hawksung1]

I'm trying to use stream API with langflow

I found out that I can run my flow with API like below.

### not stream
POST {{base_url}}/api/v1/run/11
Content-Type: application/json

{
    "input_value": "what is my name?",
    "input_type": "chat",
    "output_type": "chat",
   "session_id": 11,
    "tweaks": {
        "TextInput-OGfkB": {
            "input_value": "11"
            }
    }
}


### get stream url
POST {{base_url}}/api/v1/run/11?stream=true
Content-Type: application/json

{
    "input_value": "what is my name?",
    "input_type": "chat",
    "output_type": "chat",
   "session_id": 11,
    "tweaks": {
        "TextInput-OGfkB": {
            "input_value": "11"
            }
    }
}

### stream
GET {{base_url}}/api/v1/build/11/ChatOutput-ZjJ3t/stream?session_id=11
Accept: text/event-stream

but what I found is session_id is not working for final stream api 'api/v1/build/{flow_id}/{id}/stream'
In code there is no code using session_id

How can I stream with session id??

[dosubot[bot]]

To stream the API with a session ID using Langflow, ensure that the session_id is correctly passed to the endpoint api/v1/build/{flow_id}/{vertex_id}/stream. Here is how you can do it:

1. Ensure the session_id is included in the query parameters:

   GET {{base_url}}/api/v1/build/11/ChatOutput-ZjJ3t/stream?session_id=11
   Accept: text/event-stream

2. Verify the backend code handles the session_id:
   The build_vertex_stream function in the backend should handle the session_id to load the session data. Here is the relevant part of the code:

   @router.get("/build/{flow_id}/{vertex_id}/stream", response_class=StreamingResponse)
   async def build_vertex_stream(
       flow_id: str,
       vertex_id: str,
       session_id: Optional[str] = None,
       chat_service: "ChatService" = Depends(get_chat_service),
       session_service: "SessionService" = Depends(get_session_service),
   ):
       try:
           async def stream_vertex():
               try:
                   if not session_id:
                       cache = await chat_service.get_cache(flow_id)
                       if not cache:
                           raise ValueError(f"No cache found for {flow_id}.")
                       else:
                           graph = cache.get("result")
                   else:
                       session_data = await session_service.load_session(session_id, flow_id=flow_id)
                       graph, artifacts = session_data if session_data else (None, None)
                       if not graph:
                           raise ValueError(f"No graph found for {flow_id}.")
   
                   vertex: "InterfaceVertex" = graph.get_vertex(vertex_id)
                   if not hasattr(vertex, "stream"):
                       raise ValueError(f"Vertex {vertex_id} does not support streaming")
                   if isinstance(vertex._built_result, str) and vertex._built_result:
                       stream_data = StreamData(
                           event="message",
                           data={"message": f"Streaming vertex {vertex_id}"},
                       )
                       yield str(stream_data)
                       stream_data = StreamData(
                           event="message",
                           data={"chunk": vertex._built_result},
                       )
                       yield str(stream_data)
                   elif not vertex.frozen or not vertex._built:
                       logger.debug(f"Streaming vertex {vertex_id}")
                       stream_data = StreamData(
                           event="message",
                           data={"message": f"Streaming vertex {vertex_id}"},
                       )
                       yield str(stream_data)
                       async for chunk in vertex.stream():
                           stream_data = StreamData(
                               event="message",
                               data={"chunk": chunk},
                           )
                           yield str(stream_data)
                   elif vertex.result is not None:
                       stream_data = StreamData(
                           event="message",
                           data={"chunk": vertex._built_result},
                       )
                       yield str(stream_data)
                   else:
                       raise ValueError(f"No result found for vertex {vertex_id}")
   
               except Exception as exc:
                   logger.exception(f"Error building vertex: {exc}")
                   exc_message = parse_exception(exc)
                   if exc_message == "The message must be an iterator or an async iterator.":
                       exc_message = "This stream has already been closed."
                   yield str(StreamData(event="error", data={"error": exc_message}))
               finally:
                   logger.debug("Closing stream")
                   yield str(StreamData(event="close", data={"message": "Stream closed"}))
   
           return StreamingResponse(stream_vertex(), media_type="text/event-stream")
       except Exception as exc:
           raise HTTPException(status_code=500, detail="Error building vertex") from exc

This function ensures that if a session_id is provided, it loads the session data using the session_service and proceeds with streaming the vertex data. If session_id is not provided, it retrieves the graph from the cache using the chat_service ^[1][2].

To continue talking to Dosu, mention @dosu.

[hawksung1]

Please show me example chat flow that uses session_id.
In my case, when I request, response 'No graph found for {flow_id}'

I request first

###
POST {{base_url}}/api/v1/run/11?stream=true
Content-Type: application/json

{
    "input_value": "what is my name?",
    "input_type": "chat",
    "output_type": "chat",
    "session_id": "1",
    "tweaks": {
    "TextInput-OGfkB": {
        "input_value": "1"
        }
    }
}

Second

###
GET {{base_url}}/api/v1/build/11/ChatOutput-ZjJ3t/stream?session_id=1
Accept: text/event-stream

[dosubot[bot]]

To resolve the "No graph found for {flow_id}" error, ensure that the flow_id you are using is valid and exists in the database. Here is an example chat flow that uses session_id:

1. POST request to run the flow:

POST {{base_url}}/api/v1/run/11?stream=true
Content-Type: application/json

{
    "input_value": "what is my name?",
    "input_type": "chat",
    "output_type": "chat",
    "session_id": "1",
    "tweaks": {
        "TextInput-OGfkB": {
            "input_value": "1"
        }
    }
}

2. GET request to stream the output:

GET {{base_url}}/api/v1/build/11/ChatOutput-ZjJ3t/stream?session_id=1
Accept: text/event-stream

Ensure that the flow_id (in this case, 11) is correct and exists in your system. If the flow ID is invalid or does not exist, you will encounter the "No graph found for {flow_id}" error.

Additionally, make sure that the session data is properly loaded when using session_id. If session_id is provided, the system will attempt to load the session data using the session_service. If the session data is not found, you will also encounter the "No graph found for {flow_id}" error.

To further validate your setup, you can refer to the test cases provided in the tests/test_endpoints.py file. For example, the test_successful_run_with_input_type_chat function demonstrates a successful run with a chat input type, ensuring that the session ID and outputs are correctly handled:

def test_successful_run_with_input_type_chat(client, starter_project, created_api_key):
    headers = {"x-api-key": created_api_key.api_key}
    flow_id = starter_project["id"]
    payload = {
        "input_type": "chat",
        "output_type": "debug",
        "input_value": "value1",
    }
    response = client.post(f"/api/v1/run/{flow_id}", headers=headers, json=payload)
    assert response.status_code == status.HTTP_200_OK, response.text
    json_response = response.json()
    assert "session_id" in json_response
    assert "outputs" in json_response
    outer_outputs = json_response["outputs"]
    assert len(outer_outputs) == 1
    outputs_dict = outer_outputs[0]
    assert len(outputs_dict) == 2
    assert "inputs" in outputs_dict
    assert "outputs" in outputs_dict
    assert outputs_dict.get("inputs") == {"input_value": "value1"}
    assert isinstance(outputs_dict.get("outputs"), list)
    assert len(outputs_dict.get("outputs")) == 4
    chat_input_outputs = [output for output in outputs_dict.get("outputs") if "ChatInput" in output.get("component_id")]
    assert len(chat_input_outputs) == 1
    assert all([output.get("results").get("text") == "value1" for output in chat_input_outputs]), chat_input_outputs

This test case ensures that the session ID and outputs are correctly handled, which can help you debug and validate your flow setup ^[1][2][3].

To continue talking to Dosu, mention @dosu.