Inserting a Final Payload onCompletion

[oalexdoda]

Hi,

Is there any way to stream a final JSON object after onCompletion() is triggered? From what I gather, once onCompletion is triggered and handlers.handleChainEnd(); is called, the stream closes and it's considered complete.

What I need to do is run a callback async function onCompletion, and send its final result to the client.

For example:

Chunk 1: Hello

Chunk 2: World

onCompletion( async (message) => {
    const finalChunk = await getFinalChunk( message );

    return JSON.stringify(finalChunk); // i.e. { final_message: "something", chat_id: 1 }
} );

Chunk 3: finalChunk

Working with the LangChain Stream. I tried creating a custom AI Stream, transformer / writer streams, etc. Just couldn't get to insert that final chunk before the stream closes.

[shuding]

I think this feature request makes sense. IMO a beforeCompletion callback would be interesting (more feedback needed though!)

[oalexdoda]

@shuding I tried a lot of different solutions and couldn’t get to hack around it. Got any ideas on how to bandage it until an official callback is available?

[shuding]

You can do stream.pipeThrough(transformStream) where transformStream appends an extra chunk on flush: https://developer.mozilla.org/en-US/docs/Web/API/TransformStream/TransformStream

[oalexdoda]

Hey @shuding , managed to get this done with a transformer, but is there any way to ensure that the JSON being sent doesn't get split in bad chunks? For example, instead of sending:

Here's a message being streamed

...as

Here 's a message being streamed

...and then follow up with the JSON inside of flush, such as { "done": true }

...it often sends the last chunk as streamed{ "done": true }, making it impossible to parse the JSON stringified on the server reliably on the client. Which causes the app the crash as a result.

I tried being a bit hacky about it by prepending a symbol, and checking If that symbol exists on the client (and then removing /parsing), but when pushing to Vercel:

1. It will behave differently than on local host, often sending the chunks mixed up vs. the flushed chunk separately.
2. It just renders the entire final message to the client as if it wasn't streaming at all.

Also, what exactly does StreamingTextResponse help with vs. just returning a regular Response? Can that cause extra unreliable chunks? I'd really appreciate any insights. Thanks!

[amar-sharma]

@altechzilla can you please help me out how you were able to append the final payload?

[amar-sharma]

You can do stream.pipeThrough(transformStream) where transformStream appends an extra chunk on flush: https://developer.mozilla.org/en-US/docs/Web/API/TransformStream/TransformStream

I want to append the sources returned from ConversationalRetrievalQAChain and append that to the stream. How can I go about it?

[oalexdoda]

@altechzilla can you please help me out how you were able to append the final payload?

I ended up writing it inside of the handleChainEnd callback and write a custom stream handler. Due to the stream helpers in the ai package sending the entire message at the end, I just wrote a custom one that doesn't do that and then at the very end sends the JSON payload I need instead.

And on the client side, you'd have to check for the payload to be a valid JSON, because chunks often get sent in batches and you won't always get the entire JSON the way you enqueue it.

[Jordan-Gilliam]

You can do stream.pipeThrough(transformStream) where transformStream appends an extra chunk on flush: https://developer.mozilla.org/en-US/docs/Web/API/TransformStream/TransformStream

I want to append the sources returned from ConversationalRetrievalQAChain and append that to the stream. How can I go about it?

Also trying to figure this out. I've got a working solution without using vercel/ai:

Stream

but it's a bit hacky and adds a ton of complexity.