Introduction
What is rxjs-ai?
Section titled “What is rxjs-ai?”rxjs-ai is a stream-first AI SDK built on RxJS for building AI-native applications. It gives you reusable, type-safe primitives that treat streaming as a first-class concept — not an afterthought.
import { createChatController, createViewModel } from 'rxjs-ai';
const chat = createChatController(model);
// Every piece of state is an Observablechat.messages$.subscribe(console.log);chat.status$.subscribe(console.log);
// Send a message — streaming happens automaticallychat.send('Explain quantum computing');
// Cancel mid-stream — one call, zero cleanupchat.cancel();The problem
Section titled “The problem”AI applications are streaming-first by nature. Tokens arrive one at a time, tool calls execute mid-stream, and users expect real-time feedback. Yet most AI SDKs bolt streaming onto a Promise-based foundation:
- Cancellation requires manually wiring
AbortControllersignals through every layer. - Retry logic means writing your own loops with exponential backoff.
- Combining streams (race two models, merge tool calls) is either unsupported or ad-hoc.
- UI updates flood the renderer with every single token, with no built-in throttling.
- Common patterns like reactive state and event routing require manual
BehaviorSubject/Subjectwiring every time.
The result is boilerplate-heavy code that fights the underlying streaming reality of AI.
The solution
Section titled “The solution”rxjs-ai embraces the fact that AI is streaming and provides a small set of composable primitives:
| Primitive | Purpose |
|---|---|
createStore | Reactive state container with Observable selectors |
createCommandBus | Type-safe event bus for decoupled command handling |
createAsyncController | Manages any async Observable lifecycle (loading, error, cancel, retry) |
createChatController | Full chat lifecycle — messages, status, streaming, tool calls |
createViewModel | Derives read-only view state from one or more stores |
Each primitive is an Observable. That means the entire RxJS operator library — retry, throttleTime, race, switchMap, combineLatest, and hundreds more — works out of the box.
How it compares
Section titled “How it compares”| Capability | rxjs-ai | Vercel AI SDK |
|---|---|---|
| Stream type | Observable<T> | AsyncIterable<T> |
| Cancellation | unsubscribe() | AbortController |
| Retry | retry(3) | Manual implementation |
| Throttle UI | throttleTime(50) | experimental_throttle |
| Race models | race(gpt$, claude$) | Not supported |
| Reactive primitives | Built-in (createStore, createCommandBus, createViewModel) | Bring your own |
| Framework support | Framework-agnostic | React-first |