feat: streamText() with streaming fallbacks (0.2.0)

README.md

@@ -35,7 +35,19 @@ const { object } = await generateObject({
 });
 ```
 
-`generateText` and `generateObject` share one router per process (`getAi()`), so rotation and parking see every call. Build your own with `createAiProviders({ env, c0mpute, openai, anthropic, router })`.
+### Streaming
+
+```js
+import { streamText } from '@profullstack/ai-providers';
+
+const s = streamText({ system: 'Translate to French.', user: post.body, maxTokens: 2000 });
+for await (const piece of s.textStream) res.write(piece);
+const { provider, model, text } = await s.result;
+```
+
+c0mpute streams its job's `token` events; OpenAI and Anthropic stream with `stream: true`. Fallback works as for `generateText` until the first piece of text has gone out. After that a failure is final (the error has `partial: true`), because the caller has already shown that text. `onToken` in the request is an alternative to iterating, and `getAi().streamText(req)` returns just the final result.
+
+`generateText`, `generateObject` and `streamText` share one router per process (`getAi()`), so rotation and parking see every call. Build your own with `createAiProviders({ env, c0mpute, openai, anthropic, router })`.
 
 ### Keep your own provider code
 
@@ -75,6 +87,7 @@ A thrown error with `status` 401/402/403, or a 429 or 400 that mentions credits
 npx @profullstack/ai-providers status
 npx @profullstack/ai-providers ask "one line about c0mpute"
 npx @profullstack/ai-providers ask --json 'Return {"ok": true}'
+npx @profullstack/ai-providers ask --stream "a haiku about peer nodes"
 ```
 
 ## What c0mpute does not do

ROLLOUT.md

@@ -41,7 +41,7 @@ Not conversions:
 
 | Need | Who | Plan |
 |---|---|---|
-| Streaming text | rssamplifier translate, parklookup trip generator | `streamText()` over the same SSE reader (0.2.0) |
+| Streaming text | rssamplifier translate, parklookup trip generator | **Done in 0.2.0:** `streamText()` over the same SSE reader, paid fallbacks stream too |
 | Tool calling | c0upons agent loop | JSON-schema tool emulation on c0mpute, real tools on paid fallbacks |
 | Vision | aiornot, imageroster, mediaanalyzer, linkedin-networking-bot | stays on vendor until a c0mpute vision model is a house model |
 | OpenAI-compatible endpoint | vu1nz (Python) and anything non-JS | Infernet `/v1/chat/completions` shim over `/api/chat` |

index.d.ts

@@ -39,6 +39,22 @@ export interface ProviderOutput {
 export interface TextProvider {
   readonly name: ProviderName;
   generate(req: TextRequest): Promise<ProviderOutput>;
+  /** Optional. Without it, streamText sends the whole generate() answer as one token. */
+  stream?(req: TextRequest, onToken: (text: string) => void): Promise<ProviderOutput>;
+}
+
+export interface StreamRequest extends TextRequest {
+  /** Called with each piece of text as it arrives. */
+  onToken?: (text: string) => void;
+  /** Only these providers, by name. */
+  only?: string[];
+}
+
+export interface TextStream extends AsyncIterable<string> {
+  /** Each piece of text; throws if the call fails. */
+  textStream: AsyncGenerator<string, void, unknown>;
+  /** The same result generateText returns, once the stream is complete. */
+  result: Promise<TextResult>;
 }
 
 export interface Attempt {
@@ -116,6 +132,8 @@ export declare class AiRouter {
     opts?: { feature?: string; only?: string[] },
   ): Promise<{ provider: ProviderName; value: T; attempts: Attempt[] }>;
   generateText(req: TextRequest): Promise<TextResult>;
+  /** After the first token a failure is final and carries `partial: true`. */
+  streamText(req: StreamRequest): Promise<TextResult>;
   generateObject<T = unknown>(req: ObjectRequest<T>): Promise<ObjectResult<T>>;
 }
 
@@ -132,6 +150,8 @@ export declare class ProviderError extends Error {
 }
 export declare class AllProvidersFailedError extends Error {
   readonly attempts: Attempt[];
+  /** true when streamText had already sent text before the failure. */
+  readonly partial?: boolean;
   readonly skipped: { name: string; waitMs: number; reason: string }[];
 }
 export declare function classify(status: number, message?: string): FailureKind;
@@ -155,6 +175,7 @@ export declare class C0mputeProvider implements TextProvider {
   constructor(opts?: C0mputeOptions);
   readonly name: 'c0mpute';
   generate(req: TextRequest): Promise<ProviderOutput>;
+  stream(req: TextRequest, onToken: (text: string) => void): Promise<ProviderOutput>;
 }
 
 export interface PaidOptions {
@@ -170,11 +191,13 @@ export declare class OpenAIProvider implements TextProvider {
   constructor(opts?: PaidOptions);
   readonly name: 'openai';
   generate(req: TextRequest): Promise<ProviderOutput>;
+  stream(req: TextRequest, onToken: (text: string) => void): Promise<ProviderOutput>;
 }
 export declare class AnthropicProvider implements TextProvider {
   constructor(opts?: PaidOptions);
   readonly name: 'anthropic';
   generate(req: TextRequest): Promise<ProviderOutput>;
+  stream(req: TextRequest, onToken: (text: string) => void): Promise<ProviderOutput>;
 }
 
 export declare function extractJson(text: string): unknown;
@@ -194,3 +217,8 @@ export declare function createAiProviders(opts?: CreateOptions): AiRouter;
 export declare function getAi(): AiRouter;
 export declare function generateText(req: TextRequest): Promise<TextResult>;
 export declare function generateObject<T = unknown>(req: ObjectRequest<T>): Promise<ObjectResult<T>>;
+export declare function streamText(req: StreamRequest, router?: AiRouter): TextStream;
+export declare function sseEvents(
+  body: ReadableStream<Uint8Array>,
+  onChunk?: () => void,
+): AsyncGenerator<{ event?: string; data: string }, void, unknown>;

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@profullstack/ai-providers",
-  "version": "0.1.2",
+  "version": "0.2.0",
   "type": "module",
   "description": "One text-model call for every Profullstack app: c0mpute first, OpenAI and Anthropic as alternating fallbacks, dead keys parked. Zero dependencies.",
   "keywords": [

src/c0mpute.js

@@ -16,6 +16,7 @@
  * so "any node" is off unless `anyModel` is set.
  */
 import { ProviderError, toProviderError } from './errors.js';
+import { sseEvents } from './sse.js';
 
 export const DEFAULT_C0MPUTE_URL = 'https://infernetprotocol.com';
 export const DEFAULT_C0MPUTE_MODELS = [
@@ -56,7 +57,23 @@ export class C0mputeProvider {
   }
 
   /** @param {import('../index.js').TextRequest} req */
-  async generate(req) {
+  generate(req) {
+    return this.#call(req);
+  }
+
+  /**
+   * Same as generate, calling `onToken` with each piece of text as the node
+   * writes it. Once a token has gone out, a stall or a dropped stream is an
+   * error rather than a quiet retry on the next model.
+   *
+   * @param {import('../index.js').TextRequest} req
+   * @param {(text: string) => void} onToken
+   */
+  stream(req, onToken) {
+    return this.#call(req, onToken);
+  }
+
+  async #call(req, onToken) {
     const release = await this.slots.acquire();
     const controller = new AbortController();
     const timer = setTimeout(() => controller.abort(), req.timeoutMs ?? this.timeoutMs);
@@ -119,6 +136,13 @@ export class C0mputeProvider {
           }, this.stallMs);
         };
         let out;
+        let emitted = false;
+        const emit = onToken
+          ? (t) => {
+              emitted = true;
+              onToken(t);
+            }
+          : undefined;
         try {
           arm();
           const stream = await this.fetch(`${this.baseUrl}/api/chat/stream/${job.jobId}`, {
@@ -134,9 +158,16 @@ export class C0mputeProvider {
               kind: 'capacity',
             });
           }
-          out = await readStream(stream.body, arm);
+          out = await readStream(stream.body, arm, emit);
         } catch (err) {
           if (!stalled) throw err;
+          if (emitted) {
+            throw new ProviderError(
+              'c0mpute',
+              `${model ?? 'any'}: stopped streaming for ${Math.round(this.stallMs / 1000)}s mid-answer (job ${job.jobId})`,
+              { status: 504, kind: 'capacity' },
+            );
+          }
           last = new ProviderError(
             'c0mpute',
             `${model ?? 'any'}: no answer for ${Math.round(this.stallMs / 1000)}s (job ${job.jobId})`,
@@ -180,49 +211,36 @@ export function toMessages(req) {
   ];
 }
 
-/** Reads `meta`, `token` and `done` events until `done` or `error`. */
-export async function readStream(body, onEvent) {
-  const reader = body.getReader();
-  const decoder = new TextDecoder();
-  let buffer = '';
+/**
+ * Reads `meta`, `token` and `done` events until `done` or `error`.
+ * `onToken` gets each token as it arrives; if `done` carries a fuller text than
+ * the tokens added up to, the missing tail is emitted too.
+ */
+export async function readStream(body, onEvent, onToken) {
   let text = '';
   let model;
-  for (;;) {
-    const { value, done } = await reader.read();
-    if (value?.length) onEvent?.();
-    if (value) buffer += decoder.decode(value, { stream: true });
-    buffer = buffer.replace(/\r\n/g, '\n');
-    let cut;
-    while ((cut = buffer.indexOf('\n\n')) >= 0) {
-      const block = buffer.slice(0, cut);
-      buffer = buffer.slice(cut + 2);
-      const event = /^event:\s*(.+)$/m.exec(block)?.[1]?.trim();
-      const raw = block
-        .split('\n')
-        .filter((l) => l.startsWith('data:'))
-        .map((l) => l.slice(5).trimStart())
-        .join('\n');
-      if (!event || !raw) continue;
-      let data;
-      try {
-        data = JSON.parse(raw);
-      } catch {
-        continue;
-      }
-      if (event === 'meta' && typeof data.model === 'string') model = data.model;
-      else if (event === 'token' && typeof data.text === 'string') text += data.text;
-      else if (event === 'done') {
-        await reader.cancel().catch(() => {});
-        return { text: typeof data.text === 'string' && data.text ? data.text : text, model };
-      } else if (event === 'error') {
-        await reader.cancel().catch(() => {});
-        throw new ProviderError('c0mpute', String(data.message ?? data.error ?? 'job failed'), {
-          status: 502,
-          kind: 'capacity',
-        });
-      }
+  for await (const { event, data: raw } of sseEvents(body, onEvent)) {
+    if (!event || !raw) continue;
+    let data;
+    try {
+      data = JSON.parse(raw);
+    } catch {
+      continue;
+    }
+    if (event === 'meta' && typeof data.model === 'string') model = data.model;
+    else if (event === 'token' && typeof data.text === 'string') {
+      text += data.text;
+      if (data.text) onToken?.(data.text);
+    } else if (event === 'done') {
+      const final = typeof data.text === 'string' && data.text ? data.text : text;
+      if (onToken && final.length > text.length && final.startsWith(text)) onToken(final.slice(text.length));
+      return { text: final, model };
+    } else if (event === 'error') {
+      throw new ProviderError('c0mpute', String(data.message ?? data.error ?? 'job failed'), {
+        status: 502,
+        kind: 'capacity',
+      });
     }
-    if (done) break;
   }
   // No done event: the node or the connection died mid-answer.
   throw new ProviderError(

src/cli.js

@@ -3,6 +3,7 @@
  * ai-providers status            which providers are configured, in call order
  * ai-providers ask "prompt"      one call through the router; prints who answered
  * ai-providers ask --json "..."  same, asking for a JSON object
+ * ai-providers ask --stream "..." same, printing the text as it arrives
  */
 import { createAiProviders } from './index.js';
 
@@ -14,23 +15,27 @@ if (cmd === 'status' || !cmd) {
   console.log(JSON.stringify({ order: order.map((p) => p.name), providers: ai.status() }, null, 2));
 } else if (cmd === 'ask') {
   const json = rest[0] === '--json';
-  const prompt = (json ? rest.slice(1) : rest).join(' ');
+  const stream = rest[0] === '--stream';
+  const prompt = (json || stream ? rest.slice(1) : rest).join(' ');
   if (!prompt) {
-    console.error('usage: ai-providers ask [--json] "prompt"');
+    console.error('usage: ai-providers ask [--json|--stream] "prompt"');
     process.exit(2);
   }
   try {
     const r = json
       ? await ai.generateObject({ user: prompt, maxTokens: 800 })
-      : await ai.generateText({ user: prompt, maxTokens: 800 });
+      : stream
+        ? await ai.streamText({ user: prompt, maxTokens: 800, onToken: (t) => process.stdout.write(t) })
+        : await ai.generateText({ user: prompt, maxTokens: 800 });
     console.error(`# ${r.provider} ${r.model} in=${r.usage.inputTokens} out=${r.usage.outputTokens}`);
     for (const a of r.attempts) console.error(`# skipped: ${a.error.message}`);
-    console.log(json ? JSON.stringify(r.object, null, 2) : r.text);
+    if (stream) process.stdout.write('\n');
+    else console.log(json ? JSON.stringify(r.object, null, 2) : r.text);
   } catch (err) {
     console.error(err.message);
     process.exit(1);
   }
 } else {
-  console.error('usage: ai-providers status | ask [--json] "prompt"');
+  console.error('usage: ai-providers status | ask [--json|--stream] "prompt"');
   process.exit(2);
 }

src/index.js

@@ -14,6 +14,7 @@ export * from './c0mpute.js';
 export * from './paid.js';
 export * from './router.js';
 export * from './json.js';
+export * from './sse.js';
 
 const OFF = /^(0|off|false|no|none|disabled?)$/i;
 const list = (v) =>
@@ -99,3 +100,75 @@ export function getAi() {
 export const generateText = (req) => getAi().generateText(req);
 /** Shorthand for getAi().generateObject(req). */
 export const generateObject = (req) => getAi().generateObject(req);
+
+/**
+ * Streams text through the shared router (or `router`).
+ *
+ *   const s = streamText({ user: 'Translate ...' });
+ *   for await (const piece of s.textStream) res.write(piece);
+ *   const { provider, text } = await s.result;
+ *
+ * `textStream` (also the object's own async iterator) yields each piece of
+ * text and throws if the call fails. `result` resolves to the generateText
+ * result. `req.onToken`, if given, is called too.
+ *
+ * @param {import('../index.js').StreamRequest} req
+ * @param {AiRouter} [router]
+ */
+export function streamText(req, router = getAi()) {
+  const queue = [];
+  let wake;
+  let finished = false;
+  let failure;
+  const nudge = () => {
+    const w = wake;
+    wake = undefined;
+    w?.();
+  };
+  const result = router
+    .streamText({
+      ...req,
+      onToken: (t) => {
+        req.onToken?.(t);
+        queue.push(t);
+        nudge();
+      },
+    })
+    .then(
+      (r) => {
+        finished = true;
+        nudge();
+        return r;
+      },
+      (err) => {
+        failure = err;
+        finished = true;
+        nudge();
+        throw err;
+      },
+    );
+  // The failure also surfaces through textStream; a caller that only iterates
+  // must not get an unhandled rejection.
+  result.catch(() => {});
+  async function* pieces() {
+    for (;;) {
+      if (queue.length) {
+        yield queue.shift();
+        continue;
+      }
+      if (finished) {
+        if (failure) throw failure;
+        return;
+      }
+      await new Promise((resolve) => (wake = resolve));
+    }
+  }
+  const textStream = pieces();
+  return {
+    textStream,
+    result,
+    [Symbol.asyncIterator]() {
+      return textStream;
+    },
+  };
+}