Issue streaming from ConversationalRetrievalQAChain

[apps2net]

I'm trying to stream the response from ConversationalRetrievalQAChain. It works the first time but when you ask another question with a chat history it streams out the modified question and then my code closes the controller. When that actual answer is returned the streaming causes an error because the controller is already closed. Is there any way around this or have I fault in my code?

My code is like this:

`const encoder = new TextEncoder();
const decoder = new TextDecoder();
const stream: ReadableStream = new ReadableStream({
async start(controller) {

      const llm = new ChatOpenAI({
        openAIApiKey: process.env.OPENAI_API_KEY,
        modelName: 'gpt-3.5-turbo',
        temperature: 0,
        streaming: true,	
        callbacks: [
          {
            async handleLLMNewToken(token) {
            const queue = encoder.encode(token);
            controller.enqueue(queue);
            },
            async handleLLMEnd() {
              controller.close();
            },
          },
        ],	  
      });

      const chain = ConversationalRetrievalQAChain.fromLLM(
        llm,
        vectorStore.asRetriever(), 
        {
          returnSourceDocuments: false,
        }
      );
      const res = await chain.call({ question: payload.question, chat_history: payload.messages_string });
  }});

  return stream;
`

[dosubot[bot]]

🤖

Hello,

The issue you're experiencing might be due to the fact that the ConversationalRetrievalQAChain object is not designed to maintain an open connection for multiple calls. Each call to the chain.call() method is treated as a separate conversation, and the connection is closed after the response is returned.

To prevent this issue, you can modify your code to keep the connection open until all your questions have been answered. You can achieve this by creating a new method that accepts an array of questions and iterates over them, calling the chain.call() method for each one. Here's an example:

async function askQuestions(chain, questions, chatHistory = "") {
  for (const question of questions) {
    const res = await chain.call({ question, chat_history: chatHistory });
    console.log(res);
    chatHistory += `${question}\n${res.text}`;
  }
}

/* Ask it a series of questions */
const questions = ["What did the president say about Justice Breyer?", "Was that nice?"];
askQuestions(chain, questions);

In this example, the askQuestions function accepts the chain, an array of questions, and an optional chat history. It then iterates over the questions, calling chain.call() for each one and appending the question and response to the chat history. This way, the connection remains open until all questions have been answered.

In the context shared, the chain.call() method is called twice in a row without maintaining the chat history. This is likely causing the connection to close after the first call, resulting in an error when the second call is made. By using the askQuestions function as shown above, you can ensure that the connection remains open until all your questions have been answered.

You can find more information about how the ConversationalRetrievalQAChain handles streaming in the LangChain codebase in the langchain/src/chains/conversational_retrieval_chain.ts file.

I hope this helps! If you have any other questions, feel free to ask.

Best,
Dosu

Sources

• Question: Vercel AI library streaming not closing due to callbacks
• examples/src/chains/conversational_qa_external_memory.ts
• examples/src/chains/conversational_qa.ts
• langchain/src/chains/conversational_retrieval_chain.ts

---

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.

If you want to continue the conversation, start your reply with @dosu-bot.