nsarrazin HF Staff commited on
Commit
9dea7ad
·
unverified ·
1 Parent(s): fd14f95

fix(stream): add keep alive stream event

Browse files
src/lib/server/textGeneration/index.ts CHANGED
@@ -21,15 +21,31 @@ import type { TextGenerationContext } from "./types";
21
  import type { ToolResult } from "$lib/types/Tool";
22
  import { toolHasName } from "../tools/utils";
23
 
 
 
 
 
 
 
 
 
 
 
24
  export async function* textGeneration(ctx: TextGenerationContext) {
25
- yield* mergeAsyncGenerators([
26
- textGenerationWithoutTitle(ctx),
27
- generateTitleForConversation(ctx.conv),
28
- ]);
 
 
 
 
 
29
  }
30
 
31
  async function* textGenerationWithoutTitle(
32
- ctx: TextGenerationContext
 
33
  ): AsyncGenerator<MessageUpdate, undefined, undefined> {
34
  yield {
35
  type: MessageUpdateType.Status,
@@ -69,4 +85,5 @@ async function* textGenerationWithoutTitle(
69
 
70
  const processedMessages = await preprocessMessages(messages, webSearchResult, convId);
71
  yield* generate({ ...ctx, messages: processedMessages }, toolResults, preprompt);
 
72
  }
 
21
  import type { ToolResult } from "$lib/types/Tool";
22
  import { toolHasName } from "../tools/utils";
23
 
24
+ async function* keepAlive(done: AbortSignal): AsyncGenerator<MessageUpdate, undefined, undefined> {
25
+ while (!done.aborted) {
26
+ yield {
27
+ type: MessageUpdateType.Status,
28
+ status: MessageUpdateStatus.KeepAlive,
29
+ };
30
+ await new Promise((resolve) => setTimeout(resolve, 5000));
31
+ }
32
+ }
33
+
34
  export async function* textGeneration(ctx: TextGenerationContext) {
35
+ const done = new AbortController();
36
+
37
+ const titleGen = generateTitleForConversation(ctx.conv);
38
+ const textGen = textGenerationWithoutTitle(ctx, done);
39
+ const keepAliveGen = keepAlive(done.signal);
40
+
41
+ // keep alive until textGen is done
42
+
43
+ yield* mergeAsyncGenerators([titleGen, textGen, keepAliveGen]);
44
  }
45
 
46
  async function* textGenerationWithoutTitle(
47
+ ctx: TextGenerationContext,
48
+ done: AbortController
49
  ): AsyncGenerator<MessageUpdate, undefined, undefined> {
50
  yield {
51
  type: MessageUpdateType.Status,
 
85
 
86
  const processedMessages = await preprocessMessages(messages, webSearchResult, convId);
87
  yield* generate({ ...ctx, messages: processedMessages }, toolResults, preprompt);
88
+ done.abort();
89
  }
src/lib/types/MessageUpdate.ts CHANGED
@@ -25,6 +25,7 @@ export enum MessageUpdateStatus {
25
  Started = "started",
26
  Error = "error",
27
  Finished = "finished",
 
28
  }
29
  export interface MessageStatusUpdate {
30
  type: MessageUpdateType.Status;
 
25
  Started = "started",
26
  Error = "error",
27
  Finished = "finished",
28
+ KeepAlive = "keepAlive",
29
  }
30
  export interface MessageStatusUpdate {
31
  type: MessageUpdateType.Status;
src/routes/conversation/[id]/+page.svelte CHANGED
@@ -220,11 +220,6 @@
220
  messageUpdatesAbortController.abort();
221
  return;
222
  }
223
- if (update.type === "finalAnswer") {
224
- loading = false;
225
- pending = false;
226
- break;
227
- }
228
 
229
  // Remove null characters added due to remote keylogging prevention
230
  // See server code for more details
 
220
  messageUpdatesAbortController.abort();
221
  return;
222
  }
 
 
 
 
 
223
 
224
  // Remove null characters added due to remote keylogging prevention
225
  // See server code for more details
src/routes/conversation/[id]/+server.ts CHANGED
@@ -409,7 +409,7 @@ export async function POST({ request, locals, params, getClientAddress }) {
409
  controller.enqueue(JSON.stringify(event) + "\n");
410
 
411
  // Send 4096 of spaces to make sure the browser doesn't blocking buffer that holding the response
412
- if (event.type === "finalAnswer") {
413
  controller.enqueue(" ".repeat(4096));
414
  }
415
  }
 
409
  controller.enqueue(JSON.stringify(event) + "\n");
410
 
411
  // Send 4096 of spaces to make sure the browser doesn't blocking buffer that holding the response
412
+ if (event.type === MessageUpdateType.FinalAnswer) {
413
  controller.enqueue(" ".repeat(4096));
414
  }
415
  }