stream method

@override Future<Stream<StreamingMessage>> stream(List<LLMMessage> messages, { List<Tool>? tools, ToolChoice? toolChoice, required ModelConfig modelConfig, bool? jsonOutput, CancelToken? cancelToken, })

Implementation

@override
Future<Stream<StreamingMessage>> stream(
  List<LLMMessage> messages, {
  List<Tool>? tools,
  ToolChoice? toolChoice,
  required ModelConfig modelConfig,
  bool? jsonOutput,
  CancelToken? cancelToken,
}) async {
  final url = '$baseUrl/responses';
  final body = _createRequestBody(
    messages,
    tools: tools,
    toolChoice: toolChoice,
    modelConfig: modelConfig,
    stream: true,
    jsonOutput: jsonOutput,
    autoPreviousResponseId: autoPreviousResponseId,
    extraAllowedKeys: extraAllowedKeys,
  );

  StreamController<StreamingMessage> controller =
      StreamController<StreamingMessage>();
  int retryCount = 0;
  int currentDelayMs = initialRetryDelayMs;

  Future<void> waitForRetry(String reason) async {
    _logger.warning(
      'Responses API Stream: $reason. Retrying in ${currentDelayMs}ms... (Attempt ${retryCount + 1}/$maxRetries)',
    );
    await Future.delayed(Duration(milliseconds: currentDelayMs));
    retryCount++;
    currentDelayMs = (currentDelayMs * 2);
    if (currentDelayMs > maxRetryDelayMs) {
      currentDelayMs = maxRetryDelayMs;
    }
  }

  void pumpStream() async {
    while (true) {
      try {
        _logger.info(
          'Sending streaming request to OpenAI Responses API, baseUrl: $baseUrl, timeout: ${timeout.inSeconds} seconds, proxy:${proxyUrl ?? 'none'}, message length: ${messages.length}, tools: ${tools?.length}, model: ${modelConfig.model}',
        );
        final startTime = DateTime.now();

        final response = await _client.post(
          url,
          data: body,
          options: Options(
            responseType: ResponseType.stream,
            sendTimeout: timeout,
            receiveTimeout: timeout,
            headers: {
              'Content-Type': 'application/json',
              'Authorization': 'Bearer $_apiKey',
            },
            validateStatus: (code) => true,
          ),
          cancelToken: cancelToken,
        );
        final endTime = DateTime.now();

        _logger.info(
          'Received streaming response from OpenAI Responses API, status code: ${response.statusCode}, duration: ${endTime.difference(startTime).inMilliseconds} ms',
        );

        if (response.statusCode != 200) {
          if (response.statusCode != null &&
              (response.statusCode == 429 || response.statusCode! >= 500)) {
            if (retryCount < maxRetries) {
              await waitForRetry(
                'Returned status code ${response.statusCode}',
              );
              controller.add(
                StreamingMessage(
                  controlMessage: StreamingControlMessage(
                    controlFlag: StreamingControlFlag.retry,
                    data: {
                      'retryReason':
                          'Returned status code ${response.statusCode}',
                    },
                  ),
                ),
              );
              continue;
            }
          }

          final responseBody = await utf8.decodeStream(
            (response.data.stream as Stream).cast<List<int>>(),
          );
          throw Exception(
            'Failed to stream from OpenAI Responses API: ${response.statusCode} ${response.statusMessage} $responseBody',
          );
        }

        final stream = (response.data.stream as Stream).cast<List<int>>();

        final transformedStream = stream
            .transform(utf8.decoder)
            .transform(const LineSplitter())
            .transform(ResponsesChunkDecoder()) // Reusing SSE decoder
            .transform(ResponsesAPIResponseTransformer(modelConfig))
            .map((chunk) => StreamingMessage(modelMessage: chunk));

        await for (final message in transformedStream) {
          controller.add(message);
        }

        controller.close();
        break;
      } on DioException catch (e) {
        if (isLlmRequestCancelled(e)) {
          controller.addError(e);
          controller.close();
          break;
        }
        if (retryCount < maxRetries) {
          await waitForRetry('DioException: ${e.message}');
          controller.add(
            StreamingMessage(
              controlMessage: StreamingControlMessage(
                controlFlag: StreamingControlFlag.retry,
                data: {'retryReason': 'DioException: ${e.message}'},
              ),
            ),
          );
          continue;
        }
        controller.addError(e);
        controller.close();
        break;
      } catch (e) {
        controller.addError(e);
        controller.close();
        break;
      }
    }
  }

  pumpStream();
  return controller.stream;
}