Merge pull request #1218 from stainless-sdks/pakrym/support-streaming-background-responses

Support streaming background responses

Author: kwhinnery-openai

examples/responses/stream_background.ts

@@ -0,0 +1,42 @@
+#!/usr/bin/env -S npm run tsn -T
+
+import OpenAI from 'openai';
+
+const openai = new OpenAI();
+
+async function main() {
+  const runner = openai.responses.stream({
+    model: 'gpt-4o-2024-08-06',
+    input: 'solve 8x + 31 = 2',
+    background: true,
+  });
+
+  let id: string | null = null;
+
+  for await (const event of runner) {
+    if (event.type == 'response.created') {
+      id = event.response.id;
+    }
+
+    console.log('event', event);
+    if (event.sequence_number == 10) {
+      break;
+    }
+  }
+
+  console.log('Interrupted. Continuing...');
+
+  const runner2 = openai.responses.stream({
+    response_id: id!,
+    starting_after: 10,
+  });
+
+  for await (const event of runner2) {
+    console.log('event', event);
+  }
+
+  const result = await runner2.finalResponse();
+  console.log(result);
+}
+
+main();

src/lib/ResponsesParser.ts

@@ -1,6 +1,7 @@
 import { OpenAIError } from '../error';
 import type { ChatCompletionTool } from '../resources/chat/completions';
 import {
+  ResponseTextConfig,
   type FunctionTool,
   type ParsedContent,
   type ParsedResponse,
@@ -20,7 +21,9 @@ export type ResponseCreateParamsWithTools = ResponseCreateParamsBase & {
   tools?: ParseableToolsParams;
 };
 
-export type ExtractParsedContentFromParams<Params extends ResponseCreateParamsWithTools> =
+type TextConfigParams = { text?: ResponseTextConfig };
+
+export type ExtractParsedContentFromParams<Params extends TextConfigParams> =
   NonNullable<Params['text']>['format'] extends AutoParseableTextFormat<infer P> ? P : null;
 
 export function maybeParseResponse<

src/lib/responses/ResponseStream.ts

@@ -1,4 +1,5 @@
 import {
+  ResponseTextConfig,
   type ParsedResponse,
   type Response,
   type ResponseCreateParamsBase,
@@ -10,12 +11,40 @@ import { APIUserAbortError, OpenAIError } from '../../error';
 import OpenAI from '../../index';
 import { type BaseEvents, EventStream } from '../EventStream';
 import { type ResponseFunctionCallArgumentsDeltaEvent, type ResponseTextDeltaEvent } from './EventTypes';
-import { maybeParseResponse } from '../ResponsesParser';
+import { maybeParseResponse, ParseableToolsParams } from '../ResponsesParser';
+import { Stream } from 'openai/streaming';
 
-export type ResponseStreamParams = Omit<ResponseCreateParamsBase, 'stream'> & {
+export type ResponseStreamParams = ResponseCreateAndStreamParams | ResponseStreamByIdParams;
+
+export type ResponseCreateAndStreamParams = Omit<ResponseCreateParamsBase, 'stream'> & {
   stream?: true;
 };
 
+export type ResponseStreamByIdParams = {
+  /**
+   * The ID of the response to stream.
+   */
+  response_id: string;
+  /**
+   * If provided, the stream will start after the event with the given sequence number.
+   */
+  starting_after?: number;
+  /**
+   * Configuration options for a text response from the model. Can be plain text or
+   * structured JSON data. Learn more:
+   *
+   * - [Text inputs and outputs](https://platform.openai.com/docs/guides/text)
+   * - [Structured Outputs](https://platform.openai.com/docs/guides/structured-outputs)
+   */
+  text?: ResponseTextConfig;
+
+  /**
+   * An array of tools the model may call while generating a response. When continuing a stream, provide
+   * the same tools as the original request.
+   */
+  tools?: ParseableToolsParams;
+};
+
 type ResponseEvents = BaseEvents &
   Omit<
     {
@@ -52,7 +81,7 @@ export class ResponseStream<ParsedT = null>
   ): ResponseStream<ParsedT> {
     const runner = new ResponseStream<ParsedT>(params as ResponseCreateParamsStreaming);
     runner._run(() =>
-      runner._createResponse(client, params, {
+      runner._createOrRetrieveResponse(client, params, {
         ...options,
         headers: { ...options?.headers, 'X-Stainless-Helper-Method': 'stream' },
       }),
@@ -65,11 +94,17 @@ export class ResponseStream<ParsedT = null>
     this.#currentResponseSnapshot = undefined;
   }
 
-  #addEvent(this: ResponseStream<ParsedT>, event: ResponseStreamEvent) {
+  #addEvent(this: ResponseStream<ParsedT>, event: ResponseStreamEvent, starting_after: number | null) {
     if (this.ended) return;
 
+    const maybeEmit = (name: string, event: ResponseStreamEvent & { snapshot?: string }) => {
+      if (starting_after == null || event.sequence_number > starting_after) {
+        this._emit(name as any, event);
+      }
+    };
+
     const response = this.#accumulateResponse(event);
-    this._emit('event', event);
+    maybeEmit('event', event);
 
     switch (event.type) {
       case 'response.output_text.delta': {
@@ -86,7 +121,7 @@ export class ResponseStream<ParsedT = null>
             throw new OpenAIError(`expected content to be 'output_text', got ${content.type}`);
           }
 
-          this._emit('response.output_text.delta', {
+          maybeEmit('response.output_text.delta', {
             ...event,
             snapshot: content.text,
           });
@@ -99,16 +134,15 @@ export class ResponseStream<ParsedT = null>
           throw new OpenAIError(`missing output at index ${event.output_index}`);
         }
         if (output.type === 'function_call') {
-          this._emit('response.function_call_arguments.delta', {
+          maybeEmit('response.function_call_arguments.delta', {
             ...event,
             snapshot: output.arguments,
           });
         }
         break;
       }
       default:
-        // @ts-ignore
-        this._emit(event.type, event);
+        maybeEmit(event.type, event);
         break;
     }
   }
@@ -128,9 +162,9 @@ export class ResponseStream<ParsedT = null>
     return parsedResponse;
   }
 
-  protected async _createResponse(
+  protected async _createOrRetrieveResponse(
     client: OpenAI,
-    params: ResponseStreamingParams,
+    params: ResponseStreamParams,
     options?: Core.RequestOptions,
   ): Promise<ParsedResponse<ParsedT>> {
     const signal = options?.signal;
@@ -140,13 +174,25 @@ export class ResponseStream<ParsedT = null>
     }
     this.#beginRequest();
 
-    const stream = await client.responses.create(
-      { ...params, stream: true },
-      { ...options, signal: this.controller.signal },
-    );
+    let stream: Stream<ResponseStreamEvent> | undefined;
+    let starting_after: number | null = null;
+    if ('response_id' in params) {
+      stream = await client.responses.retrieve(
+        params.response_id,
+        { stream: true },
+        { ...options, signal: this.controller.signal, stream: true },
+      );
+      starting_after = params.starting_after ?? null;
+    } else {
+      stream = await client.responses.create(
+        { ...params, stream: true },
+        { ...options, signal: this.controller.signal },
+      );
+    }
+
     this._connected();
     for await (const event of stream) {
-      this.#addEvent(event);
+      this.#addEvent(event, starting_after);
     }
     if (stream.controller.signal?.aborted) {
       throw new APIUserAbortError();

src/resources/responses/responses.ts

@@ -112,21 +112,44 @@ export class Responses extends APIResource {
    * );
    * ```
    */
+
   retrieve(
     responseId: string,
-    query?: ResponseRetrieveParams,
+    query?: ResponseRetrieveParamsNonStreaming,
     options?: Core.RequestOptions,
   ): Core.APIPromise<Response>;
+  retrieve(
+    responseId: string,
+    query?: ResponseRetrieveParamsStreaming,
+    options?: Core.RequestOptions,
+  ): Core.APIPromise<Stream<ResponseStreamEvent>>;
   retrieve(responseId: string, options?: Core.RequestOptions): Core.APIPromise<Response>;
+  retrieve(
+    responseId: string,
+    query: ResponseRetrieveParams | Core.RequestOptions,
+    options?: Core.RequestOptions,
+  ): Core.APIPromise<Response> | Core.APIPromise<Stream<ResponseStreamEvent>>;
   retrieve(
     responseId: string,
     query: ResponseRetrieveParams | Core.RequestOptions = {},
     options?: Core.RequestOptions,
-  ): Core.APIPromise<Response> {
-    if (isRequestOptions(query)) {
+  ): Core.APIPromise<Response> | Core.APIPromise<Stream<ResponseStreamEvent>> {
+    if (isRequestOptions(query) && options === undefined) {
       return this.retrieve(responseId, {}, query);
     }
-    return this._client.get(`/responses/${responseId}`, { query, ...options });
+    return (
+      this._client.get(`/responses/${responseId}`, {
+        query,
+        ...options,
+        stream: query.stream ?? false,
+      }) as APIPromise<Response> | APIPromise<Stream<ResponseStreamEvent>>
+    )._thenUnwrap((rsp) => {
+      if ('object' in rsp && rsp.object === 'response') {
+        addOutputText(rsp as Response);
+      }
+
+      return rsp;
+    }) as APIPromise<Response> | APIPromise<Stream<ResponseStreamEvent>>;
   }
 
   /**
@@ -4204,6 +4227,11 @@ export interface ResponseWebSearchCallCompletedEvent {
    */
   item_id: string;
 
+  /**
+   * A sequence number for this chunk of the stream response.
+   */
+  sequence_number: number;
+
   /**
    * The index of the output item that the web search call is associated with.
    */
@@ -4224,6 +4252,11 @@ export interface ResponseWebSearchCallInProgressEvent {
    */
   item_id: string;
 
+  /**
+   * A sequence number for this chunk of the stream response.
+   */
+  sequence_number: number;
+
   /**
    * The index of the output item that the web search call is associated with.
    */
@@ -4244,6 +4277,11 @@ export interface ResponseWebSearchCallSearchingEvent {
    */
   item_id: string;
 
+  /**
+   * A sequence number for this chunk of the stream response.
+   */
+  sequence_number: number;
+
   /**
    * The index of the output item that the web search call is associated with.
    */
@@ -4825,14 +4863,29 @@ export interface ResponseCreateParamsStreaming extends ResponseCreateParamsBase
   stream: true;
 }
 
-export interface ResponseRetrieveParams {
+export type ResponseRetrieveParams = ResponseRetrieveParamsStreaming | ResponseRetrieveParamsNonStreaming;
+export interface ResponseRetrieveParamsBase {
   /**
    * Additional fields to include in the response. See the `include` parameter for
    * Response creation above for more information.
    */
   include?: Array<ResponseIncludable>;
+
+  starting_after?: number | null;
+  stream?: boolean | null;
 }
 
+export interface ResponseRetrieveParamsStreaming extends ResponseRetrieveParamsBase {
+  stream: true;
+}
+export interface ResponseRetrieveParamsNonStreaming extends ResponseRetrieveParamsBase {
+  stream?: false | null;
+}
+
+export namespace ResponseRetrieveParams {
+  export type ResponseRetrieveParamsStreaming = ResponsesAPI.ResponseRetrieveParamsStreaming;
+  export type ResponseRetrieveParamsNonStreaming = ResponsesAPI.ResponseRetrieveParamsNonStreaming;
+}
 Responses.InputItems = InputItems;
 
 export declare namespace Responses {