Can't seem to stream from a chain

[latifs]

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 answerGenerationChainPrompt = ChatPromptTemplate.fromMessages([
    ['system', ANSWER_CHAIN_SYSTEM_TEMPLATE],
    new MessagesPlaceholder('history'),
    [
      'human',
      'Now, answer this question using the previous context and chat history:\n{standalone_question}',
    ],
  ]);

const llm = new ChatOpenAI({
    modelName: LLM,
    temperature: TEMPERATURE,
    streaming: true,
  });

const rephraseQuestionChain = RunnableSequence.from([
    rephraseQuestionChainPrompt,
    llm,
    new StringOutputParser(),
  ]);

  const vectorStore = await PineconeStore.fromExistingIndex(embeddings, {
    pineconeIndex,
    maxConcurrency: 5,
    namespace: botId,
  });

 const documentRetrievalChain = RunnableSequence.from([
    (input: { question: string }) => input.question,
    async (standaloneQ: string) => {
      // const { documents } = await retriever(standaloneQ, botId, model);
      const documents = await vectorStore.similaritySearchWithScore(
        standaloneQ,
        topK || TOP_K
      );
      // console.log('documents retrieved:', documents);
      return documents;
    },
    convertDocsToString1,
  ]);

  const conversationalRetrievalChain = RunnableSequence.from([
    RunnablePassthrough.assign({ standalone_question: rephraseQuestionChain }),
    RunnablePassthrough.assign({ context: documentRetrievalChain }),
    answerGenerationChainPrompt,
    llm,
    (input: any) => {
      console.log('input0:', input);
      console.log('input.content:', input.content);
      return new StringOutputParser().stream(input.content);
    },
  ]);

  const messageHistory = new InMemoryChatMessageHistory();
  formatMessagesToInstances(formattedMessages, messageHistory);

  const finalRetrievalChain = new RunnableWithMessageHistory({
    runnable: conversationalRetrievalChain,
    getMessageHistory: (_sessionId: string) => messageHistory,
    historyMessagesKey: 'history',
    inputMessagesKey: 'question',
  });

  const finalAnswer = await finalRetrievalChain.stream(
    { question: content },
    { configurable: { sessionId: botId } }
  );

 if (finalAnswer && typeof finalAnswer.pipeThrough === 'function') {
    console.log('finalAnswer is a valid stream:', finalAnswer);

    return new Response(finalAnswer, {
      headers: { 'Content-Type': 'text/event-stream' },
    });
  } else {
    console.error('finalAnswer is not a valid stream:', finalAnswer);
    return NextResponse.json(
      { error: 'Failed to create a valid stream' },
      { status: 500 }
    );
  }

Description

I can't seem to be able to stream in this chain, even the console.log sends back the complete answer instead of chunks

System Info

none

[dosubot[bot]]

I found some similar discussions and issues that might help you with streaming in your chain:

1. Stream answer from MultiPromptChain ^[1]
2. Streaming OpenAI Function Call output in LangChainStreaming ^[2]
3. Langchain Agent Streaming LLM & Function Call & Tools ^[3]
4. Unable to obtain the streaming transmission method, it always returns only after receiving the entire text ^[4]
5. Is it possible to use streaming for Custom LLM in chains? ^[5]

To enable streaming in your chain so that console.log outputs chunks instead of the complete answer, you can use the following code:

const query = "What is 3 * 12? Also, what is 11 + 49?";

const stream = await modelWithTools.stream(query);

for await (const chunk of stream) {
  console.log(chunk.tool_call_chunks);
}

This code sets up a query and streams the output, logging each chunk of the tool call as it is received.

Additionally, you can add some logic to compile your stream as it's being returned:

const output = {};
let currentKey: string | null = null;

for await (const chunk of await modelWithTools.stream(query)) {
  for (const key of Object.keys(chunk)) {
    if (output[key] === undefined) {
      output[key] = chunk[key];
    } else {
      output[key] += chunk[key];
    }

    if (key !== currentKey) {
      console.log(`\n\n${key}: ${JSON.stringify(chunk[key])}`);
    } else {
      console.log(chunk[key]);
    }
    currentKey = key;
  }
}

This additional logic helps to compile the stream as it is being returned, providing a more structured output.

For a more detailed example, you can refer to the LangChain documentation on streaming final outputs:

for await (const chunk of await ragChainWithSource.stream("What is task decomposition?")) {
  console.log(chunk)
}

This example demonstrates how to stream outputs as they are generated using LangChain.

Additionally, you can refer to the langchain-core tests for more examples on how to handle streaming:

test("Stream model output", async () => {
  const remote = new RemoteRunnable({ url: `${BASE_URL}/b` });
  const stream = await remote.stream({ text: "What are the 5 best apples?" });
  let chunkCount = 0;
  let accumulator: AIMessageChunk | null = null;
  for await (const chunk of stream) {
    const innerChunk = chunk as AIMessageChunk;
    accumulator = accumulator ? accumulator.concat(innerChunk) : innerChunk;
    chunkCount += 1;
  }
  expect(chunkCount).toBe(18);
  expect(accumulator?.content).toEqual(
    '"object1, object2, object3, object4, object5"'
  );
});

This test case shows how to stream model output and accumulate chunks to form the final response.

For a more comprehensive understanding of streaming in LangChain, you can refer to the LangChain documentation on streaming:

import { ChatOpenAI } from "@langchain/openai";

const model = new ChatOpenAI({
  model: "gpt-4o",
  temperature: 0,
});

const stream = await model.stream("Hello! Tell me about yourself.");
const chunks = [];
for await (const chunk of stream) {
  chunks.push(chunk);
  console.log(`${chunk.content}|`)
}

This example shows how to stream the output from a model token by token, making the application feel more responsive by showing intermediate progress ^{[6][7][8][9][10]}.

To continue talking to Dosu, mention @dosu.