can not stream workflow when chaining llm

[sancelot]

Bug Description

Hi,

using api , it is not possible to launch and stream output of flow , when we chain llms

There is no streaming reply, it replies with done.

Reproduction

const response = await axios.post(
  `${this.baseUrl}/v1/run/${flow}?stream=true`,
  requestBody,
  {
    headers: {
      //                "Authorization": "Bearer <TOKEN>",
      "Content-Type": "application/json",
      //              "x-api-key": "sk-wcChzPLjRj2LKIPriY7hB4l-PpB4fEky_bvEfAPLw8Q",
    },
    responseType: "stream",
    adapter: "fetch",
  },
);

// Get the response body as a ReadableStream
const stream = response.data;

// Create a reader from the stream
const reader = stream.getReader();

// Process the stream chunks
try {
  while (true) {
    const { done, value } = await reader.read();
    
    // Exit the loop when the stream is complete
    if (done) {
      console.log("Stream complete");
      break;
    }
    
    // Convert the chunk (Uint8Array) to a string
    const chunk = new TextDecoder().decode(value);
    
    try {
      // Each chunk might be one or more JSON objects separated by newlines
      const lines = chunk.split('\n').filter(line => line.trim() !== '');
      
      for (const line of lines) {
        // Parse the JSON data from each line
        const data = JSON.parse(line);
        
        // Process the streaming data based on your needs
        if (data.type === 'message') {
          console.log('Received message:', data.content);
          // Update UI or process message content
        } else if (data.type === 'error') {
          console.error('Error:', data.error);
          // Handle error
        } else if (data.type === 'end') {
          console.log('Stream ended with status:', data.status);
          // Handle completion
        }
      }
    } catch (parseError) {
      console.error('Error parsing chunk:', parseError);
      console.log('Problematic chunk:', chunk);
    }
  }
} catch (streamError) {
  console.error('Stream reading error:', streamError);
} finally {
  // Always release the reader when done
  reader.releaseLock();
}

Expected behavior

should wait end of my flow and stream output

Who can help?

@italojohnny @anovazzi1 @nicoloboschi

Operating System

win10

Langflow Version

1.2.0

Python Version

3.12

Screenshot

No response

Flow File

test.json

[dosubot]

A similar issue was previously reported and closed. The issue titled Support stream output was resolved by a pull request that implemented streaming functionality ^[1].

In Langflow version 1.2.0, the streaming functionality does not work with the /api/v1/run/<flow_id>?stream=true endpoint because it is deprecated. Instead, you should use the /build endpoint for streaming responses. A better streaming option is expected to be provided in API v2 ^[2].

To enable streaming output when chaining LLMs, you can use the AsyncStreamingLLMCallbackHandler class and append the ?stream=true query parameter to your API request. This will allow LLM chat responses to be streamed back as token events until the end event closes the connection ^[3].

To continue talking to Dosu, mention @dosu.

---

^{Help Dosu learn! Give it feedback: Great Response | Irrelevant Answer | Incorrect Sources | Too Verbose | Hallucination | Bug Report | Other}

[sancelot]

@dosu , the /build endpoint does not permit to stream , after trying it, it replies only with {"job_id":"dcf94393-41c9-4493-a5f0-8d283f09c003"}

[dosubot]

To stream the output using the /build endpoint, you need to follow a two-step process:

1. Obtain a Job ID: First, send a POST request to the /build/{flow_id}/flow endpoint to initiate the build and get a job ID.

2. Stream the Execution Results: Use the obtained job ID to stream the execution results by sending a GET request to the /build/{job_id}/events endpoint. This will allow you to receive the events in real-time.

Here's how you can do it:

1. Send a POST request to get the job ID:

   curl -X POST \
     "$LANGFLOW_URL/api/v1/build/$FLOW_ID/flow" \
     -H "accept: application/json" \
     -H "Content-Type: application/json" \
     -d '{
       "inputs": {
         "input_value": "Tell me a story"
       }
     }'

2. Use the job ID to stream the execution results:

   curl -X GET \
     "$LANGFLOW_URL/api/v1/build/{job_id}/events" \
     -H "accept: application/json"

This approach will stream the execution events as they occur. The events endpoint accepts an optional stream query parameter, which defaults to true, allowing you to receive events in real-time ^[1][2].

To continue talking to Dosu, mention @dosu.

---

^{Help Dosu learn! Give it feedback: Great Response | Irrelevant Answer | Incorrect Sources | Too Verbose | Hallucination | Bug Report | Other}

[sancelot]

@dosu, ok. I have now some data streamed, but it is not very clear what do I have to parse to fill in my chatbot output.

I received successive end_vertex events.

[dosubot]

To fill in your chatbot output from the streamed data, you should focus on parsing the end_vertex events. For each end_vertex event, extract the text part from the nested JSON structure and append it to your result string. Here's a code snippet to guide you:

result = ""
async for line in response.body_iterator:
    if not line:
        continue
    event_data = json.loads(line)
    if event_data.get("event") == "end_vertex":
        text_part = (
            event_data.get("data", {})
            .get("build_data", "")
            .get("data", {})
            .get("results", {})
            .get("message", {})
            .get("text", "")
        )
        result += text_part

This code will help you collect and concatenate the text from each end_vertex event into the result string, which you can then use for your chatbot output ^[1].

To continue talking to Dosu, mention @dosu.

---

^{Help Dosu learn! Give it feedback: Great Response | Irrelevant Answer | Incorrect Sources | Too Verbose | Hallucination | Bug Report | Other}

[sancelot]

@dosu. Sorry but I am using typescript code. Thanks I managed to do the trick.

[dosubot]

Great to hear you managed to get it working with TypeScript! If you have any more questions or need further assistance, feel free to ask. Happy coding!

To continue talking to Dosu, mention @dosu.

---

^{Help Dosu learn! Give it feedback: Great Response | Irrelevant Answer | Incorrect Sources | Too Verbose | Hallucination | Bug Report | Other}

[sancelot]

@dosu is it possible to use a session id in this mode ?

// Step 1: Initiate the build and get the job ID
        const jobResponse = await axios.post(
          `${this.baseUrl}/v1/build/${flowId}/flow?stream=true`,
          requestBody,
          {
            headers: {
              "Content-Type": "application/json",
            },
          },
        );

        const jobId = jobResponse.data?.job_id;
        if (!jobId) {
          throw new Error("Failed to obtain job ID from LangFlow.");
        }

        console.log("Job ID obtained:", jobId);

        // Step 2: Stream the execution results
        const response = await axios.get(
          `${this.baseUrl}/v1/build/${jobId}/events`,
          {
            headers: {
              "Content-Type": "application/json",
              accept: "application/json",
            },
            responseType: "stream",
            adapter: "fetch", // Specify the fetch adapter
          },
        );

[dosubot]

Yes, you can use a session ID in this mode. To do so, include the session_id in the request body when initiating the build. This will ensure that the session context is maintained throughout the process. Here's how you can modify your TypeScript code to include a session ID:

// Step 1: Initiate the build and get the job ID
const jobResponse = await axios.post(
  `${this.baseUrl}/v1/build/${flowId}/flow?stream=true`,
  {
    ...requestBody,
    session_id: "your-session-id", // Include the session ID here
  },
  {
    headers: {
      "Content-Type": "application/json",
    },
  },
);

const jobId = jobResponse.data?.job_id;
if (!jobId) {
  throw new Error("Failed to obtain job ID from LangFlow.");
}

console.log("Job ID obtained:", jobId);

// Step 2: Stream the execution results
const response = await axios.get(
  `${this.baseUrl}/v1/build/${jobId}/events`,
  {
    headers: {
      "Content-Type": "application/json",
      accept: "application/json",
    },
    responseType: "stream",
    adapter: "fetch", // Specify the fetch adapter
  },
);

By including the session_id in the request body, you ensure that the session is correctly tracked and managed during the execution ^[1].

To continue talking to Dosu, mention @dosu.

---

^{Help Dosu learn! Give it feedback: Great Response | Irrelevant Answer | Incorrect Sources | Too Verbose | Hallucination | Bug Report | Other}

[philnash]

I don't think @dosubot has the right response here. The /build endpoint is intended to be used by the Langflow frontend for building flows. The documentation says:

This endpoint is meant to be used by the frontend and is not optimized for external use. To run your flow, use the /run endpoint instead.

It is an issue that the /run endpoint doesn't stream results when there is a chain of LLMs.

[lucaseduoli]

@italojohnny would you mind taking a peek at this?