fix(core): request/response events' stream discrimination. (#525)

Author: samchon

packages/core/src/events/AgenticaRequestEvent.ts

@@ -3,8 +3,25 @@ import type OpenAI from "openai";
 import type { AgenticaEventBase } from "./AgenticaEventBase";
 import type { AgenticaEventSource } from "./AgenticaEventSource";
 
-export interface AgenticaRequestEvent extends AgenticaEventBase<"request"> {
-  source: AgenticaEventSource;
-  body: OpenAI.ChatCompletionCreateParamsStreaming | OpenAI.ChatCompletionCreateParamsNonStreaming;
-  options?: OpenAI.RequestOptions | undefined;
+export type AgenticaRequestEvent =
+  | AgenticaRequestEvent.Streaming
+  | AgenticaRequestEvent.NonStreaming;
+export namespace AgenticaRequestEvent {
+  export type Streaming = Base<
+    true,
+    OpenAI.ChatCompletionCreateParamsStreaming
+  >;
+
+  export type NonStreaming = Base<
+    false,
+    OpenAI.ChatCompletionCreateParamsNonStreaming
+  >;
+
+  interface Base<Stream extends boolean, Body extends object>
+    extends AgenticaEventBase<"request"> {
+    source: AgenticaEventSource;
+    stream: Stream;
+    body: Body;
+    options?: OpenAI.RequestOptions | undefined;
+  }
 }

packages/core/src/events/AgenticaResponseEvent.ts

@@ -3,42 +3,30 @@ import type OpenAI from "openai";
 import type { AgenticaEventBase } from "./AgenticaEventBase";
 import type { AgenticaEventSource } from "./AgenticaEventSource";
 
-export interface AgenticaResponseEvent extends AgenticaEventBase<"response"> {
-  request_id: string;
-
-  /**
-   * The source agent of the response.
-   */
-  source: AgenticaEventSource;
-
-  /**
-   * Request body.
-   */
-  body: OpenAI.ChatCompletionCreateParamsStreaming;
-
-  /**
-   * The response data.
-   */
-  response: AgenticaResponseEvent.Response;
+export type AgenticaResponseEvent =
+  | AgenticaResponseEvent.Streaming
+  | AgenticaResponseEvent.NonStreaming;
+export namespace AgenticaResponseEvent {
+  export type Streaming = Base<
+    true,
+    OpenAI.ChatCompletionCreateParamsStreaming,
+    AsyncGenerator<OpenAI.ChatCompletionChunk, undefined, undefined>
+  >;
 
-  /**
-   * Options for the request.
-   */
-  options?: OpenAI.RequestOptions | undefined;
+  export type NonStreaming = Base<
+    false,
+    OpenAI.ChatCompletionCreateParamsNonStreaming,
+    OpenAI.ChatCompletion
+  >;
 
-  /**
-   * Wait the completion.
-   */
-  join: () => Promise<OpenAI.ChatCompletion>;
-}
-export namespace AgenticaResponseEvent {
-  export type Response = StreamResponse | NonStreamResponse;
-  export interface StreamResponse {
-    stream: true;
-    data: AsyncGenerator<OpenAI.ChatCompletionChunk, undefined, undefined>;
-  }
-  export interface NonStreamResponse {
-    stream: false;
-    data: OpenAI.ChatCompletion;
+  interface Base<Stream extends boolean, Body extends object, Completion extends object>
+    extends AgenticaEventBase<"response"> {
+    source: AgenticaEventSource;
+    request_id: string;
+    stream: Stream;
+    body: Body;
+    completion: Completion;
+    options?: OpenAI.RequestOptions | undefined;
+    join: () => Promise<OpenAI.ChatCompletion>;
   }
 }

packages/core/src/factory/events.ts

@@ -308,9 +308,12 @@ export function createDescribeEvent(props: {
 /* -----------------------------------------------------------
   API REQUESTS
 ----------------------------------------------------------- */
-export function createRequestEvent(props: {
+export function createRequestEvent<Stream extends boolean>(props: {
   source: AgenticaEventSource;
-  body: OpenAI.ChatCompletionCreateParamsStreaming | OpenAI.ChatCompletionCreateParamsNonStreaming;
+  stream: Stream;
+  body: Stream extends true
+    ? OpenAI.ChatCompletionCreateParamsStreaming
+    : OpenAI.ChatCompletionCreateParamsNonStreaming;
   options?: OpenAI.RequestOptions | undefined;
 }): AgenticaRequestEvent {
   const id: string = v4();
@@ -320,17 +323,23 @@ export function createRequestEvent(props: {
     id,
     created_at,
     source: props.source,
-    body: props.body,
+    stream: props.stream as false,
+    body: props.body as OpenAI.ChatCompletionCreateParamsNonStreaming,
     options: props.options,
   };
 }
 
-export function createResponseEvent(props: {
+export function createResponseEvent<Stream extends boolean>(props: {
   request_id: string;
   source: AgenticaEventSource;
-  body: OpenAI.ChatCompletionCreateParamsStreaming;
+  stream: Stream;
+  body: Stream extends true
+    ? OpenAI.ChatCompletionCreateParamsStreaming
+    : OpenAI.ChatCompletionCreateParamsNonStreaming;
   options?: OpenAI.RequestOptions | undefined;
-  response: AgenticaResponseEvent.Response;
+  completion: Stream extends true
+    ? AsyncGenerator<OpenAI.ChatCompletionChunk, undefined, undefined>
+    : OpenAI.ChatCompletion;
   join: () => Promise<OpenAI.ChatCompletion>;
 }): AgenticaResponseEvent {
   const id: string = v4();
@@ -341,9 +350,10 @@ export function createResponseEvent(props: {
     request_id: props.request_id,
     created_at,
     source: props.source,
-    body: props.body,
+    stream: props.stream as false,
+    body: props.body as OpenAI.ChatCompletionCreateParamsNonStreaming,
+    completion: props.completion as OpenAI.ChatCompletion,
     options: props.options,
-    response: props.response,
     join: props.join,
   };
 }

packages/core/src/utils/request.ts

@@ -29,6 +29,7 @@ export function getChatCompletionFunction(props: {
       : { stream: false };
     const event: AgenticaRequestEvent = createRequestEvent({
       source,
+      stream: streamOptions.stream,
       body: {
         ...body,
         model: props.vendor.model,
@@ -75,11 +76,9 @@ export function getChatCompletionFunction(props: {
         type: "response",
         request_id: event.id,
         source,
-        response: {
-          stream: false,
-          data: completion,
-        },
-        body: event.body as OpenAI.ChatCompletionCreateParamsStreaming,
+        stream: false,
+        body: event.body as OpenAI.ChatCompletionCreateParamsNonStreaming,
+        completion,
         options: event.options,
         join: async () => completion,
         created_at: new Date().toISOString(),
@@ -122,11 +121,9 @@ export function getChatCompletionFunction(props: {
       type: "response",
       request_id: event.id,
       source,
-      response: {
-        stream: true,
-        data: streamDefaultReaderToAsyncGenerator(streamForStream.getReader(), props.abortSignal),
-      },
+      stream: true,
       body: event.body as OpenAI.ChatCompletionCreateParamsStreaming,
+      completion: streamDefaultReaderToAsyncGenerator(streamForStream.getReader(), props.abortSignal),
       options: event.options,
       join: async () => {
         const chunks = await StreamUtil.readAll(streamForJoin, props.abortSignal);

test/src/features/test_base_streaming.ts

@@ -44,12 +44,12 @@ export async function test_base_streaming(): Promise<void | false> {
   });
 
   agent.on("response", async (event: AgenticaResponseEvent) => {
-    if (event.response.stream === false) {
+    if (event.stream === false) {
       throw new Error("Response is not a stream");
     }
     responseEventFired = true;
     // Test the stream
-    for await (const value of event.response.data) {
+    for await (const value of event.completion) {
       if (value.choices !== undefined && value.choices[0]?.delta?.content !== undefined) {
         streamContentPieces.push(value.choices[0].delta.content as string);
       }