Streaming

[markwylde]

I see on the roadmap is streaming as 0% but I don't think it's been started?

Have you thought about the type of API you would like?

I was thinking it might be cool to implement something like this:

import { Chat } from "ragged"
const c = Chat.with('openai', { apiKey: process.env.OPENAI_API_KEY });

// call and wait for all messages
const messages = await c.chat('What is a rickroll?');

// stream messages as an iterator
for each (const message of c.chat('What is a rickroll?')) {
  console.log(message)
}

This could be done by making c.chat return a custom iterator/thenable.

To achieve the desired functionality of having c.chat act as both an iterator and a promise, we could implement a custom iterator that can handle both scenarios. This involves creating a class that implements the iterator protocol and wraps the promise-based method to provide streamed results. Here’s an example of what I'm thinking. I hadn't dug too much into this part of the code so this is kinda pseudo.

class Chat {
  constructor(provider, options) {
    this.provider = provider;
    this.options = options;
  }

  static with(provider, options) {
    return new Chat(provider, options);
  }

  async chat(prompt) {
    const response = await this._getChatResponse(prompt);
    const messages = this._parseMessages(response);
    return messages;
  }

  async *_chatGenerator(prompt) {
    const response = await this._getChatResponse(prompt);
    const messages = this._parseMessages(response);
    for (const message of messages) {
      yield message;
    }
  }

  _getChatResponse(prompt) {
    // Simulate the API call to get chat response
    return new Promise((resolve) => {
      setTimeout(() => {
        resolve(`Response for prompt: ${prompt}`);
      }, 1000);
    });
  }

  _parseMessages(response) {
    // Simulate parsing the response into individual messages
    return response.split(' ');
  }

  [Symbol.asyncIterator]() {
    return this._chatGenerator(this.prompt);
  }
}

class ChatWrapper {
  constructor(chatInstance, prompt) {
    this.chatInstance = chatInstance;
    this.prompt = prompt;
    this.generator = null;
  }

  async then(resolve, reject) {
    try {
      const messages = await this.chatInstance.chat(this.prompt);
      resolve(messages);
    } catch (error) {
      reject(error);
    }
  }

  [Symbol.asyncIterator]() {
    if (!this.generator) {
      this.generator = this.chatInstance._chatGenerator(this.prompt);
    }
    return this.generator;
  }
}

// Extend the Chat class to use the ChatWrapper
class ExtendedChat extends Chat {
  chat(prompt) {
    return new ChatWrapper(this, prompt);
  }
}

Or is this a bit too magical? Is it better to just provide a { stream: true } as an option, that then returns an iterator or promise?

So I guess the three proposals I'm thinking:

Option 1

import { Chat } from "ragged"
const c = Chat.with('openai', { apiKey: process.env.OPENAI_API_KEY });

// call and wait for all messages
const messages = await c.chat('What is a rickroll?');

// stream messages as an iterator
for each (const message of c.chat('What is a rickroll?')) {
  console.log(message)
}

Option 2

import { Chat } from "ragged"
const c = Chat.with('openai', { apiKey: process.env.OPENAI_API_KEY });

// call and wait for all messages
const messages = await c.chat('What is a rickroll?');

// stream messages as an iterator
for each (const message of c.chat('What is a rickroll?', { stream: true })) {
  console.log(message)
}

Option 3

import { Chat } from "ragged"
const c = Chat.with('openai', { apiKey: process.env.OPENAI_API_KEY });

// call and wait for all messages
const messages = await c.chat('What is a rickroll?');

// stream messages as an iterator
for each (const message of c.streamChat('What is a rickroll?')) {
  console.log(message)
}

Do you have a better idea? What are you thinking?

[monarchwadia]

We should definitely support streaming -- I had done some work on this before, if you want to check it out, you should see the 0.3.7 tag. I had based it on rxjs back then, and was using the OpenAI API SDK. I ripped it out because it generated a lot of API complexity for Ragged, I felt we should first focus on making the non-streaming use cases clean. Here is a streaming poetry example: https://github.com/monarchwadia/ragged/blob/v0.3.7/examples/comprehensive-api-usage/sub-examples/streaming-poetry.ts

Re: which option, there are a few considerations. I'm about to change the return API for the c.chat and e.embed methods in order to support things like getting the original response & request objects including API headers; count tokens and rate limiting information; and some additional utility functions like getLastMessage which would make life easier (instead of doing messages.at(-1)?.text which is a mouthful). So chat is going to turn into something like the following:

const {
  history,
  incomingMessages,
  rawResponse,
  rawRequest,
  getLastMessage
} = await c.chat("What is a rickroll?")

Given that this is the case, i think option 3 is the sensible option. This way we can provide dedicated fine-grained controls for streaming calls that are not relevant for non-streaming calls. Maybe something like...

import { Chat } from "ragged"
const c = Chat.with('openai', { apiKey: process.env.OPENAI_API_KEY });

// call and wait for all messages
const { stream, abortStream } = await c.chatStream('What is a rickroll?');

// stream messages as an iterator with a small chance of aborting the connection randomly
for each (const message of stream) {
  if (Math.random() > 0.95) abortStream(); // just illustrating a point
  console.log(message);
}

But I think we need to first do the non-streaming API changes first.

I'll start a separate issue for that.

[monarchwadia]

Here is the new issue #18