What is Proper Way to Cancel A Stream? Does Cancelling the Reader Work?

[mgreystone]

Checked other resources

• I added a very descriptive title to this question.
• I searched the LangChain documentation with the integrated search.
• I used the GitHub search to find a similar question and didn't find it.

Commit to Help

• I commit to help with one of those options 👆

Example Code

const stream = await chain.stream({ topic: "Bonobos" })
const reader = stream.getReader()

reader.read().then(function process({ done, value }) {
  if (done) return
  // do something with `value`
  return reader.read().then(process)
})

// attempt to cancel the stream after 3 seconds
setTimeout(() => reader.cancel(), 3000)

Description

I'm working with streams & i am interested in the proper way to cancel a stream.

I have found the documentation here describing how to use an AbortController to cancel a request. The provided example shows how to cancel a stream. But I'm unclear on the details. I can see that the iterator now throws, but is the underlying stream terminated? Is merely the request that is canceled? Or is the underlying response still being received in the background?

ReadableStream's can be cancelled by cancelling a reader. When piping streams to other targets, such as a ServerResponse, cancelling the downstream is generally expected to also cancel upstream.

Would the following code also work to cancel a streamed response? Or with this result in a leak?

const stream = await chain.stream({ topic: "Bonobos" })
const reader = stream.getReader()

reader.read().then(function process({ done, value }) {
  if (done) return
  // do something with `value`
  return reader.read().then(process)
})

// attempt to cancel the stream after 3 seconds
setTimeout(() => reader.cancel(), 3000)

System Info

[email protected]
linux (docker)
node 18.20.2
yarn 1.22.15

[dosubot[bot]]

Using reader.cancel() to terminate a stream, as shown in your code snippet, is indeed a correct approach for managing streams in JavaScript and will effectively cancel the stream without causing resource leaks. This method directly cancels the reading operation and should clean up resources associated with the stream.

For LangChain specifically, when dealing with streams that involve external requests or operations that might be long-running, it's recommended to use an AbortController to manage cancellation more comprehensively. This method allows you to cancel not just the stream reading but also any ongoing external requests initiated by LangChain, ensuring a more thorough cancellation process. Here's a concise example of how to implement this with LangChain:

import { ChatOpenAI } from "@langchain/openai";
import { PromptTemplate } from "@langchain/core/prompts";

const controller = new AbortController();
const llm = new ChatOpenAI({ temperature: 0.9 });
const model = llm.bind({ signal: controller.signal });
const prompt = PromptTemplate.fromTemplate("Please write a 500 word essay about {topic}.");
const chain = prompt.pipe(model);

setTimeout(() => controller.abort(), 3000);

try {
  const stream = await chain.stream({ topic: "Bonobos" });
  for await (const chunk of stream) {
    console.log(chunk);
  }
} catch (e) {
  console.log(e); // Handle or log the error
}

This approach ensures that the cancellation signal is propagated through all layers of the request, effectively managing both local and external operations. It's particularly useful for ensuring that your application remains efficient and responsive, avoiding potential resource leaks when dealing with complex or long-running operations.

Details

Sources

• langchain-core/src/utils/stream.ts
• docs/core_docs/docs/expression_language/how_to/cancellation.mdx
• examples/src/chains/llm_chain_cancellation.ts
• examples/src/guides/expression_language/how_to_cancellation.ts

About Dosu
This response is meant to be useful and save you time. It is not meant to be a precise solution, but rather a starting point for your own research.

Help me be more useful! Please leave a 👍 if this is helpful and 👎 if it is irrelevant.

To continue the conversation, mention @dosu.