stream method

Stream<Map<String, Object?>> stream({ required List<Map<String, Object?>> messages, required ChatCancellation cancellation, List<Map<String, Object?>> tools = const [], })

Implementation

Stream<Map<String, Object?>> stream({
  required List<Map<String, Object?>> messages,
  required ChatCancellation cancellation,
  List<Map<String, Object?>> tools = const [],
}) async* {
  cancellation.check();
  final client = clientFactory();
  var closed = false;
  void close() {
    if (!closed) {
      closed = true;
      client.close();
    }
  }

  unawaited(cancellation.whenCancelled.then((_) => close()));
  final timeout = Timer(requestTimeout, close);
  try {
    final request = http.Request('POST', endpoint)
      ..followRedirects = false
      ..headers.addAll({
        ...headers,
        'Content-Type': 'application/json',
        'Accept': 'text/event-stream',
        if (apiKey?.isNotEmpty == true) 'Authorization': 'Bearer $apiKey',
      })
      ..body = jsonEncode({
        ...extraBody,
        'model': model,
        'messages': messages,
        'stream': true,
        if (tools.isNotEmpty) 'tools': tools,
      });
    final response = await cancellation.bind(
      client.send(request).timeout(requestTimeout),
    );
    if (response.statusCode < 200 || response.statusCode >= 300) {
      // Never put credentials or provider echo bodies into UI diagnostics.
      throw ChatApiException(
        'The model request failed',
        statusCode: response.statusCode,
      );
    }
    final events = decodeServerSentEvents(
      response.stream,
      maxEventBytes: maxEventBytes,
      maxResponseBytes: maxResponseBytes,
    );
    final iterator = StreamIterator(events);
    var done = false;
    try {
      while (await cancellation.bind(
        iterator.moveNext().timeout(requestTimeout),
      )) {
        cancellation.check();
        final data = iterator.current;
        if (data.trim() == '[DONE]') {
          done = true;
          break;
        }
        final dynamic value;
        try {
          value = jsonDecode(data);
        } on FormatException {
          throw const ChatApiException(
            'The model returned malformed stream data.',
          );
        }
        if (value is! Map<String, dynamic>) {
          throw const ChatApiException(
            'The model returned an invalid stream event.',
          );
        }
        if (value.containsKey('error')) {
          throw const ChatApiException(
            'The model reported a streaming error.',
          );
        }
        yield Map<String, Object?>.from(value);
      }
    } finally {
      await iterator.cancel();
    }
    cancellation.check();
    if (!done) {
      throw const ChatApiException(
        'The response ended before the completion marker.',
      );
    }
  } on ChatCancelled {
    rethrow;
  } on ChatApiException {
    rethrow;
  } catch (_) {
    cancellation.check();
    throw const ChatApiException(
      'Could not complete the model connection. Check the endpoint and try again.',
    );
  } finally {
    timeout.cancel();
    close();
  }
}