fix: assemble streamed conversation message chunks into conversation.answer - #371
OsamaAnsar wants to merge 1 commit into
Conversation
…answer combineMessageChunks() logged how many streamed message chunks it found but never used them to build anything -- it just returned the last complete-search-response chunk (or the first one found), whose `conversation.answer` is deliberately left empty by the server since the answer was already streamed token-by-token as separate message chunks. Callers using conversational/RAG streaming search got back an empty conversation.answer even though every token streamed correctly. Assemble the message chunks into the answer and fill it in only when the metadata chunk's conversation.answer is still empty, so a server-provided answer (if one is ever sent) is never overwritten. Adds regression tests covering: the streamed-answer assembly, that an already-populated answer is left untouched, and that plain non-conversation streams are unaffected.
tharropoulos
left a comment
tharropoulos
left a comment
There was a problem hiding this comment.
Thank you for contributing! Couple of questions before this goes further. Could you attach a raw stream captured from a real Typesense server on sse? The fix depends on the final chunk having conversation.answer set to an empty string rather than no conversation field at all, and if it's the latter this change wouldn't do anything.
| } | ||
|
|
||
| if (metadataChunk.conversation && !metadataChunk.conversation.answer) { | ||
| metadataChunk.conversation.answer = assembledAnswer; |
There was a problem hiding this comment.
this mutates the chunk in place, and that same object was already handed to the user in onChunk. anyone holding onto chunks will see the metadata one change after the fact. could we return a copy instead? something like spreading metadataChunk and its conversation into a new object with the assembled answer
| // `conversation.answer` is left empty (the answer was already streamed | ||
| // incrementally). Assemble it here so callers get the complete answer | ||
| // instead of silently getting back an empty string. | ||
| const assembledAnswer = messagesChunks |
There was a problem hiding this comment.
processStreamingLine never really fails. anything that isn't a json object comes back as a message chunk with conversation_id unknown, including sse comments, event or id lines, and data: [DONE] if the line ends in \r. before this pr those were ignored but now they end up inside the answer. can we at least only join chunks whose conversation_id matches the one on the metadata chunk?
| throw new Error("No metadata chunk found"); | ||
| } | ||
|
|
||
| if (metadataChunk.conversation && !metadataChunk.conversation.answer) { |
There was a problem hiding this comment.
answer === "" says what you mean more clearly than !answer. also if the final chunk has no conversation field at all this skips and the answer is lost. that's a case that the server shouldn't ever lie in, but always good to have guardrails around this stuff.
| // combineStreamingChunks/combineMessageChunks are private: they hold the | ||
| // actual chunk-assembly logic, so they're exercised directly here rather | ||
| // than through the full Node/browser stream-parsing machinery. | ||
| const combineStreamingChunks = (chunks: unknown[]) => |
There was a problem hiding this comment.
I would avoid casting everything into unknown, this throws type checking off.
| expect(result.conversation.answer).toBe("already set"); | ||
| }); | ||
|
|
||
| it("still returns the metadata chunk unchanged for non-conversation streams", () => { |
There was a problem hiding this comment.
this one has a single chunk, which hits the chunks.length === 1 early return and never reaches combineMessageChunks. it'd need a few non conversation chunks to actually test what the name says
|
You were right to ask for a real capture — I traced the actual server-side implementation (src/conversation_model.cpp, src/core_api.cpp) instead of assuming, and the premise behind this fix doesn't hold up. Every provider (OpenAI, Cloudflare, vLLM, Gemini, Azure) accumulates the full answer text into My original reproduction was a hand-constructed mock stream with |
2 participants
Summary
For conversational/RAG search streaming, the server sends the LLM answer token-by-token as
{conversation_id, message}chunks, then a final metadata chunk (found/hits/search_time_ms/...) whoseconversation.answeris left empty, since the answer was already streamed incrementally.combineMessageChunks()insrc/Typesense/ApiCall.tslogs how many message chunks it found (Found N message chunks to combine), but themessagesChunksparameter is never actually used to build a result — the function just returns the last chunk if it's already a complete search response, or otherwise the first complete-search-response chunk it finds. The streamed tokens are silently dropped, so callers get backconversation.answer === ""even though every token streamed correctly over the wire.The function's own doc comment says combining chunks "is critical for ensuring we return the complete data rather than just the last chunk" — this fix makes that true for conversation streams too, not just regular search streams.
Fix
combineMessageChunks()now assemblesmessagesChunks.map(c => c.message).join("")and fills it into the metadata chunk'sconversation.answer, but only when that field is still empty — so a non-empty answer the server might already provide is never overwritten.Test plan
Added three cases to
test/Typesense/ApiCall.spec.ts(exercising the private chunk-combination logic directly, since it's pure and has no I/O):conversation.answercorrectly.conversation.answeris left untouched.Ran the full
ApiCall.spec.tssuite locally (31/31 passing) andeslinton the changed files (clean). Also confirmed the new tests actually catch the regression: reverted only theApiCall.tschange and reran — the streamed-answer test fails withconversation.answer === ""as described above; restored the fix and it passes again.