Quick Start

Follow along to go from zero to a working streaming chat in just a few steps.

1. Install rxjs-ai

npm install rxjs-ai rxjs

2. Create a Model Adapter

A model adapter is any object with a complete method that returns an Observable of text chunks. Here is the simplest possible adapter — it echoes back whatever the user sent:

import { of } from "rxjs";
import type { ChatModelAdapter } from "rxjs-ai";

const echoModel: ChatModelAdapter = {
  complete: ({ messages }) => {
    const last = messages[messages.length - 1];
    return of(`Echo: ${last?.content ?? ""}`);
  },
};

That is all you need. In production you would swap this for a real LLM — see Model Adapters for streaming and fetch-based examples.

Tip

Want to see streaming in action? Replace of(...) with an Observable that emits chunks over time. The Model Adapters page has a streaming adapter example with interval-based token dripping and automatic cancellation teardown.

3. Create a Chat Controller

Pass your model adapter to createChatController:

import { createChatController } from "rxjs-ai";

const chat = createChatController(echoModel);

4. Subscribe to Status and Messages

The chat controller exposes Observables for every piece of state:

chat.status$.subscribe((status) => {
  console.log("Status:", status);
});

chat.messages$.subscribe((messages) => {
  const last = messages[messages.length - 1];
  if (last?.role === "assistant") {
    console.log("Assistant:", last.content);
  }
});

5. Send a Message

Call send() to add a user message and kick off streaming:

chat.send("Hello, rxjs-ai!");

You will see output like this:

Status: loading
Status: streaming
Assistant: Echo:
Assistant: Echo: Hello,
Assistant: Echo: Hello, rxjs-ai!
Status: idle

Each messages$ emission contains the full message array with the assistant’s content accumulated so far — perfect for rendering in a UI.

6. Cancel and Retry

Cancel a stream mid-flight with cancel(), then retry the last message with retryLast():

// Start streaming
chat.send("Tell me a long story");

// Cancel after 200ms
setTimeout(() => {
  chat.cancel();
  console.log("Cancelled!");

  // Retry the same message
  chat.retryLast();
}, 200);

Cancellation tears down the underlying Observable subscription immediately. No tokens are wasted, no cleanup callbacks to remember.

7. Clean Up

When you are done — for example, when the user navigates away — call destroy() to complete all internal Observables and release resources:

chat.destroy();

Full Example

Here is the complete script in one block:

import { of } from "rxjs";
import { createChatController } from "rxjs-ai";
import type { ChatModelAdapter } from "rxjs-ai";

// 1. Model adapter
const echoModel: ChatModelAdapter = {
  complete: ({ messages }) => {
    const last = messages[messages.length - 1];
    return of(`Echo: ${last?.content ?? ""}`);
  },
};

// 2. Chat controller
const chat = createChatController(echoModel);

// 3. Subscribe
chat.status$.subscribe((s) => console.log("Status:", s));
chat.messages$.subscribe((msgs) => {
  const last = msgs[msgs.length - 1];
  if (last?.role === "assistant") {
    console.log("Assistant:", last.content);
  }
});

// 4. Send
chat.send("Hello, rxjs-ai!");

// 5. Cleanup after 2 seconds
setTimeout(() => chat.destroy(), 2000);

Next Steps

• Chat Controller — full API reference
• Model Adapters — connect to real LLMs
• createStore — reactive state primitive
• Why rxjs-ai? — understand the RxJS advantage