Stream never closes with LangChainStream using postman #97

Otoris · 2023-06-16T05:27:04Z

Have a few issues with LangChainStream:

Appears stream never closes when API is called
Tokens are not getting set or sent down stream, but are coming through when adding callback functions to LangChainStream

Example code:

import { ChatAnthropic } from "langchain/chat_models/anthropic";
import { ChatOpenAI } from 'langchain/chat_models/openai';

import { BaseChatMessage } from "langchain/dist/schema";
import {
  ChatPromptTemplate,
  HumanMessagePromptTemplate,
  PromptTemplate,
  SystemMessagePromptTemplate,
} from "langchain/prompts";
import { StreamingTextResponse, LangChainStream } from "ai";
import { CallbackManager } from "langchain/callbacks";

export const runtime = "edge";

if (!process.env.ANTHROPIC_API_KEY) {
  throw new Error("Missing env var from Anthropic");
}

const createSummaryPrompt = async (
  persona: string,
  preprocessed_data: string
): Promise<BaseChatMessage[]> => {
  const chatPrompt = PromptTemplate.fromTemplate(
      `This is your title: {title}. blah blah blah. \User Data: {userData}.`
);

  const formattedPrompt = await chatPrompt.formatPromptValue({
    persona,
    transcriptData: JSON.stringify(preprocessed_data),
  });

  return formattedPrompt.toChatMessages();
};

export const POST = async (req: Request) => {
  const { title, user_data } =
    (await req.json()) as { title?: string; user_data: string };

  if (!preprocessed_data) {
    return new Response("No transcript in the request", { status: 400 });
  }

  const { stream, handlers } = LangChainStream({
    onStart: async () => console.log("start"),
    onToken: async (token) => console.log(token),
    onCompletion: async () => console.log("end"),
  });

  const llm = new ChatAnthropic({
    apiKey: process.env.ANTHROPIC_API_KEY,
    modelName: "claude-v1.3-100k",
    temperature: 0.1,
    maxTokensToSample: 100000,
    streaming: true,
    verbose: true,
    callbacks: [handlers],
  });

  const prompt = await createSummaryPrompt(persona, preprocessed_data);

  llm
    .call(prompt)
    .then((res) => console.log(stream))
    .catch(console.error).finally(() => console.log("done"));
    
  console.log(stream)
  return new StreamingTextResponse(stream);
};

Example cURL:

curl --location 'http://localhost:3000/api/generate-summary' \
--header 'Content-Type: text/plain' \
--header 'Cookie: <secret>' \
--data '{"user_data":"blah blah blah","title":"sales chief"}'

Postman screenshot - Connection stays open and never ends. No data returns or streams.

Console output:

start
<ref *1> ReadableStream {
  _state: 'readable',
  _reader: undefined,
  _storedError: undefined,
  _disturbed: false,
  _readableStreamController: ReadableStreamDefaultController {
  _controlledReadableStream: [Circular *1],
  _queue: S {
  _cursor: 0,
  _size: 0,
  _front: { _elements: [], _next: undefined },
  _back: { _elements: [], _next: undefined }
},
  _queueTotalSize: 0,
  _started: true,
  _closeRequested: false,
  _pullAgain: false,
  _pulling: false,
  _strategySizeAlgorithm: [Function],
  _strategyHWM: 0,
  _pullAlgorithm: [Function],
  _cancelAlgorithm: [Function]
}
}
[llm/start] [1:llm:ChatAnthropic] Entering LLM run with input: {
  "messages": [
    [
      {
        "type": "human",
        "data": {
          "content": "redacted",
          "additional_kwargs": {}
        }
      }
    ]
  ]
}
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
token!
[llm/end] [1:llm:ChatAnthropic] [7.56s] Exiting LLM run with output: {
  "generations": [
    [
      {
        "text": "redacted",
        "message": {
          "type": "ai",
          "data": {
            "content": " redacted",
            "additional_kwargs": {}
          }
        }
      }
    ]
  ]
}
<ref *1> ReadableStream {
  _state: 'readable',
  _reader: ReadableStreamDefaultReader {
  _ownerReadableStream: [Circular *1],
  _closedPromise_resolve: [Function],
  _closedPromise_reject: [Function],
  _closedPromise: Promise {
  [Symbol(async_id_symbol)]: 1974,
  [Symbol(trigger_async_id_symbol)]: 0,
  [Symbol(kResourceStore)]: undefined,
  [Symbol(kResourceStore)]: undefined,
  [Symbol(kResourceStore)]: undefined
},
  _readRequests: S {
  _cursor: 77,
  _size: 1,
  _front: {
  _elements: [
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  {
  _chunkSteps: [Function: _chunkSteps],
  _closeSteps: [Function: _closeSteps],
  _errorSteps: [Function: _errorSteps]
}
],
  _next: undefined
},
  _back: {
  _elements: [
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  {
  _chunkSteps: [Function: _chunkSteps],
  _closeSteps: [Function: _closeSteps],
  _errorSteps: [Function: _errorSteps]
}
],
  _next: undefined
}
}
},
  _storedError: undefined,
  _disturbed: true,
  _readableStreamController: ReadableStreamDefaultController {
  _controlledReadableStream: [Circular *1],
  _queue: S {
  _cursor: 0,
  _size: 0,
  _front: { _elements: [], _next: undefined },
  _back: { _elements: [], _next: undefined }
},
  _queueTotalSize: 0,
  _started: true,
  _closeRequested: false,
  _pullAgain: false,
  _pulling: true,
  _strategySizeAlgorithm: [Function],
  _strategyHWM: 0,
  _pullAlgorithm: [Function],
  _cancelAlgorithm: [Function]
}
}
done
token!
token!
token!
token!
token!
token!
done!!!
token!
token!
token!
token!
token!
token!
token!
[llm/end] [1:llm:ChatAnthropic] [8.54s] Exiting LLM run with output: {
  "generations": [
    [
      {
        "text": " redacted",
        "message": {
          "type": "ai",
          "data": {
            "content": " redactedl",
            "additional_kwargs": {}
          }
        }
      }
    ]
  ]
}
<ref *1> ReadableStream {
  _state: 'readable',
  _reader: ReadableStreamDefaultReader {
  _ownerReadableStream: [Circular *1],
  _closedPromise_resolve: [Function],
  _closedPromise_reject: [Function],
  _closedPromise: Promise {
  [Symbol(async_id_symbol)]: 2452,
  [Symbol(trigger_async_id_symbol)]: 2239,
  [Symbol(kResourceStore)]: undefined,
  [Symbol(kResourceStore)]: undefined,
  [Symbol(kResourceStore)]: undefined,
  [Symbol(kResourceStore)]: undefined,
  [Symbol(kResourceStore)]: undefined,
  [Symbol(kResourceStore)]: undefined
},
  _readRequests: S {
  _cursor: 100,
  _size: 1,
  _front: {
  _elements: [
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  {
  _chunkSteps: [Function: _chunkSteps],
  _closeSteps: [Function: _closeSteps],
  _errorSteps: [Function: _errorSteps]
}
],
  _next: undefined
},
  _back: {
  _elements: [
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  undefined,
  {
  _chunkSteps: [Function: _chunkSteps],
  _closeSteps: [Function: _closeSteps],
  _errorSteps: [Function: _errorSteps]
}
],
  _next: undefined
}
}
},
  _storedError: undefined,
  _disturbed: true,
  _readableStreamController: ReadableStreamDefaultController {
  _controlledReadableStream: [Circular *1],
  _queue: S {
  _cursor: 0,
  _size: 0,
  _front: { _elements: [], _next: undefined },
  _back: { _elements: [], _next: undefined }
},
  _queueTotalSize: 0,
  _started: true,
  _closeRequested: false,
  _pullAgain: false,
  _pulling: true,
  _strategySizeAlgorithm: [Function],
  _strategyHWM: 0,
  _pullAlgorithm: [Function],
  _cancelAlgorithm: [Function]
}
}
done

Something isn't connecting here I guess if we're getting an array of undefined. Any ideas where I'm going wrong here? Any help much appreciated!

( additional side note: Would be so cool to have a useStream() for features that are not chat and are not completions! e.g. one off API calls formed from app data like mine. I am passing things to useChat({body: {title:, user_data}}) )

The text was updated successfully, but these errors were encountered:

Itsnotaka · 2023-06-16T19:40:06Z

Had the same issue and the workaround I used was to just overwrite the callback after llm.call

Otoris · 2023-06-17T00:19:28Z

@Itsnotaka Could I trouble you to share an example?

e-roy · 2023-06-17T12:43:51Z

I noticed the same issue with the example.

One workaround to get to work is add an wrappedCall

  const { messages } = await req.json()

  const { stream, handlers } = LangChainStream({
    onStart: async () => console.log('start'),
    onToken: async token => console.log(token),
    onCompletion: async () => console.log('end')
  })

  const llm = new ChatOpenAI({
    streaming: true,
    callbackManager: CallbackManager.fromHandlers(handlers)
  })

  async function wrappedCall(messages: Message[], onCompletion: () => void) {
    try {
      await llm.call(
        messages.map(m =>
          m.role == 'user'
            ? new HumanChatMessage(m.content)
            : new AIChatMessage(m.content)
        )
      )
    } catch (error) {
      console.error(error)
    } finally {
      onCompletion()
    }
  }

  wrappedCall(messages as Message[], () => {
    console.log('end')
  })

  return new StreamingTextResponse(stream)

The issue might be within the models in the langchain library to expose a completion event or callback.

nfcampos · 2023-06-17T16:02:40Z

I can confirm Langchain does call handleLLMEnd, see the screenshot attached, so it must be an issue with something in this vercel library

e-roy · 2023-06-17T18:03:49Z

Here is another approach to trigger onCompletion with no wrapper

import { StreamingTextResponse, LangChainStream, Message } from 'ai'
import { CallbackManager } from 'langchain/callbacks'
import { ChatOpenAI } from 'langchain/chat_models/openai'
import { AIChatMessage, HumanChatMessage } from 'langchain/schema'

export const runtime = 'edge'

export async function POST(req: Request) {
  const { messages } = await req.json()

  const { stream, handlers } = LangChainStream({
    onStart: async () => console.log('start'),
    // onToken: async token => console.log(token),
    onCompletion: async () => console.log('end')
  })

  const llm = new ChatOpenAI({
    streaming: true,
    callbackManager: CallbackManager.fromHandlers(handlers)
  })

  llm
    .call(
      (messages as Message[]).map(m =>
        m.role == 'user'
          ? new HumanChatMessage(m.content)
          : new AIChatMessage(m.content)
      )
    )
    .catch(console.error)
    .finally(() => {
      // Call handleStreamEnd when the chat or stream ends
      handlers.handleChainEnd()
    })

  return new StreamingTextResponse(stream)
}

MichaelReaNZ · 2023-06-18T04:28:45Z

Have had the same issue which makes onCompletion not work but also the onLoading in the UI.

MichaelReaNZ · 2023-06-18T04:51:42Z

Thanks e-roy.

.finally(() => {
      // Call handleStreamEnd when the chat or stream ends
      handlers.handleChainEnd()
    })

Is a good workaround 🙂

feliche93 · 2023-06-20T11:38:12Z

@e-roy thanks for the snippet, also solved my problem 👏

jasan-s · 2023-06-20T22:12:05Z

another solution:

new ChatOpenAI({
	temperature: 0,
	modelName: 'gpt-3.5-turbo',
	maxTokens: 512, // Choose the max allowed tokens in completion
	streaming: true,
	callbackManager: CallbackManager.fromHandlers({
		async handleLLMNewToken(token: string) {
		// console.log({ token })
		handlers.handleLLMNewToken(token)
	        },
		async handleLLMEnd(output) {
		console.log('End of stream...', output)
		handlers.handleChainEnd(output)
		},
		async handleLLMError(e) {
		console.log('Error in stream...', e)
		handlers.handleLLMError(e)
		},
	}),
})

aranlucas · 2023-06-21T04:35:30Z

This also causes isLoading from useChat() to never go back to false when using LangChainStream.

However, e-roy's fix seems to work.

aranlucas · 2023-06-22T16:37:35Z

I think the issue is caused by be90740

When you do what @nfcampos suggested overwriting the returned handlers

  const { stream, handlers } = LangChainStream();

  const model = new ChatOpenAI({
    temperature: 0,
    streaming: true,
    callbacks: CallbackManager.fromHandlers({
      handleLLMNewToken: handlers.handleLLMNewToken,
      handleChainEnd: async () => {
        console.log("handleChainEnd");
        await handlers.handleChainEnd();
      },
      handleLLMEnd: async () => {
        console.log("handleLLMEnd");
        await handlers.handleChainEnd();
      },
      handleLLMError: handlers.handleLLMError,
    }),
  });

handleChainEnd never ends up in showing in the logs. (For my use case, using ConversationalRetrievalQAChain)

It seems that depending on which chain/model you use, handleChainEnd is not guaranteed to be called.

It looks like adding handleLLMEnd (keeping handleChainEnd) should fix this issue, but I'm not sure what issue @jaredpalmer might have been fixing when switching handleLLMEnd -> handleChainEnd

nfcampos · 2023-06-27T09:26:22Z

@aranlucas the issue is actually that you're passing handlers for chain events to the constructor of a chat model, therefore they will never be called. You should pass the handlers either to the constructor of the chain you're using or to the .call() method

aranlucas · 2023-06-27T15:08:24Z

Yeah took me a while to get to that conclusion in #205 (comment).

Itsnotaka mentioned this issue Jun 17, 2023

Streaming is not working on M1 mac browsers & Postman #113

Closed

This was referenced Jun 22, 2023

LangChain stream not updating the 'isLoading' boolean #181

Closed

Getting function invocation timeout in serverless function streaming #170

Closed

This was referenced Jun 22, 2023

fix: langchain stream not closing #201

Merged

Question: Vercel AI library streaming not closing due to callbacks langchain-ai/langchainjs#1740

Closed

Langchain handlers being use case specific #205

Closed

jaredpalmer closed this as completed in #201 Jun 29, 2023

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Stream never closes with LangChainStream using postman #97

Stream never closes with LangChainStream using postman #97

Otoris commented Jun 16, 2023

Itsnotaka commented Jun 16, 2023

Otoris commented Jun 17, 2023

e-roy commented Jun 17, 2023

nfcampos commented Jun 17, 2023

e-roy commented Jun 17, 2023

MichaelReaNZ commented Jun 18, 2023

MichaelReaNZ commented Jun 18, 2023

feliche93 commented Jun 20, 2023

jasan-s commented Jun 20, 2023 •

edited

aranlucas commented Jun 21, 2023

aranlucas commented Jun 22, 2023 •

edited

nfcampos commented Jun 27, 2023

aranlucas commented Jun 27, 2023

Stream never closes with LangChainStream using postman #97

Stream never closes with LangChainStream using postman #97

Comments

Otoris commented Jun 16, 2023

Itsnotaka commented Jun 16, 2023

Otoris commented Jun 17, 2023

e-roy commented Jun 17, 2023

nfcampos commented Jun 17, 2023

e-roy commented Jun 17, 2023

MichaelReaNZ commented Jun 18, 2023

MichaelReaNZ commented Jun 18, 2023

feliche93 commented Jun 20, 2023

jasan-s commented Jun 20, 2023 • edited

aranlucas commented Jun 21, 2023

aranlucas commented Jun 22, 2023 • edited

nfcampos commented Jun 27, 2023

aranlucas commented Jun 27, 2023

jasan-s commented Jun 20, 2023 •

edited

aranlucas commented Jun 22, 2023 •

edited