ai.matey.utils

Shared utility functions for streaming, validation, and conversions.

Part of the ai.matey monorepo.

Installation

npm install ai.matey.utils

Stream Utilities

Comprehensive utilities for working with IR chat streams.

Basic Stream Operations

import {
  collectStream,        // Collect chunks into array
  collectStreamFull,    // Collect with rich metadata
  streamToText,         // Get accumulated text
  streamToResponse,     // Convert to IR response
} from 'ai.matey.utils';

// Collect all chunks
const chunks = await collectStream(stream);

// Collect with full metadata
const result = await collectStreamFull(stream);
console.log(result.content);      // Full text
console.log(result.message);      // IR message
console.log(result.finishReason); // 'stop', 'length', etc.
console.log(result.usage);        // Token counts

// Get just the text
const text = await streamToText(stream);

Processing Streams

import { processStream, streamToLines, streamToTextIterator } from 'ai.matey.utils';

// Process with callbacks
const result = await processStream(stream, {
  onStart: (requestId) => console.log('Started:', requestId),
  onContent: (delta, accumulated) => process.stdout.write(delta),
  onDone: (result) => console.log('\nTokens:', result.usage?.totalTokens),
  onError: (error) => console.error('Error:', error),
});

// Iterate over text chunks
for await (const text of streamToTextIterator(stream)) {
  process.stdout.write(text);
}

// Buffer and yield complete lines
for await (const line of streamToLines(stream)) {
  console.log('Line:', line);
}

Transforming Streams

import {
  transformStream,
  filterStream,
  mapStream,
  tapStream,
} from 'ai.matey.utils';

// Transform chunks
const transformed = transformStream(stream, (chunk) => ({
  ...chunk,
  delta: chunk.type === 'content' ? chunk.delta.toUpperCase() : chunk.delta,
}));

// Filter chunks
const contentOnly = filterStream(stream, (chunk) => chunk.type === 'content');

// Map chunks
const textOnly = mapStream(stream, (chunk) =>
  chunk.type === 'content' ? chunk.delta : ''
);

// Tap for side effects (logging, etc.)
const logged = tapStream(stream, (chunk) => console.log('Chunk:', chunk.type));

Stream Control

import {
  throttleStream,
  rateLimitStream,
  teeStream,
  splitStream,
} from 'ai.matey.utils';

// Throttle updates (batches content chunks)
// Great for limiting UI update frequency
for await (const chunk of throttleStream(stream, 50)) {
  updateUI(chunk); // Max every 50ms
}

// Rate limit (delays chunks to max N/second)
const limited = rateLimitStream(stream, 10); // Max 10 chunks/second

// Split stream for parallel processing
const [stream1, stream2] = teeStream(stream, 2);

await Promise.all([
  processStream(stream1, { onContent: (d) => logger.log(d) }),
  processStream(stream2, { onContent: (d) => ui.append(d) }),
]);

Stream Validation

import { validateStream, validateChunkSequence } from 'ai.matey.utils';

// Validate stream structure
const validated = validateStream(stream, {
  requireStart: true,
  requireDone: true,
  requireContent: true,
});

// Check if chunk sequence is valid
const chunks = await collectStream(stream);
const isValid = validateChunkSequence(chunks);

Error Handling

import { catchStreamErrors, streamWithTimeout, createStreamError } from 'ai.matey.utils';

// Catch and handle errors
const safe = catchStreamErrors(stream, (error) => {
  console.error('Stream error:', error);
});

// Add timeout
const withTimeout = streamWithTimeout(stream, 30000); // 30s timeout

// Create error chunks
const errorChunk = createStreamError(new Error('Something went wrong'));

Stream Accumulator

Build responses incrementally from chunks:

import {
  createStreamAccumulator,
  accumulateChunk,
  accumulatorToMessage,
  accumulatorToResponse,
} from 'ai.matey.utils';

const accumulator = createStreamAccumulator();

for await (const chunk of stream) {
  accumulateChunk(accumulator, chunk);
  console.log('Current text:', accumulator.content);
}

const message = accumulatorToMessage(accumulator);
const response = accumulatorToResponse(accumulator);

Content Utilities

import { getContentDeltas, isContentChunk, isDoneChunk, isErrorChunk } from 'ai.matey.utils';

// Get just the text deltas
for await (const delta of getContentDeltas(stream)) {
  process.stdout.write(delta);
}

// Type guards
for await (const chunk of stream) {
  if (isContentChunk(chunk)) {
    console.log('Content:', chunk.delta);
  } else if (isDoneChunk(chunk)) {
    console.log('Done:', chunk.finishReason);
  } else if (isErrorChunk(chunk)) {
    console.error('Error:', chunk.error);
  }
}

Web Stream Conversion

import {
  asyncGeneratorToReadableStream,
  readableStreamToAsyncGenerator,
} from 'ai.matey.utils';

// Convert async generator to Web ReadableStream
const readable = asyncGeneratorToReadableStream(stream);

// Convert back to async generator
const generator = readableStreamToAsyncGenerator(readable);

Types

import type {
  CollectedStream,      // Rich result from collectStreamFull
  ProcessStreamOptions, // Options for processStream
  StreamValidationOptions,
} from 'ai.matey.utils';

API Reference

Collection Functions

| Function | Description | |----------|-------------| | collectStream(stream) | Collect chunks into array | | collectStreamFull(stream) | Collect with rich metadata | | streamToText(stream) | Get accumulated text | | streamToResponse(stream) | Convert to IR response |

Processing Functions

| Function | Description | |----------|-------------| | processStream(stream, options) | Process with callbacks | | streamToLines(stream) | Yield complete lines | | streamToTextIterator(stream) | Yield text deltas | | getContentDeltas(stream) | Yield only content |

Transform Functions

| Function | Description | |----------|-------------| | transformStream(stream, fn) | Transform each chunk | | filterStream(stream, predicate) | Filter chunks | | mapStream(stream, fn) | Map chunks to new values | | tapStream(stream, fn) | Side effects without modification |

Control Functions

| Function | Description | |----------|-------------| | throttleStream(stream, ms) | Batch chunks by time interval | | rateLimitStream(stream, rate) | Limit chunks per second | | teeStream(stream, count) | Split into multiple streams | | splitStream(stream, count) | Split into multiple streams |

Validation Functions

| Function | Description | |----------|-------------| | validateStream(stream, options) | Validate stream structure | | validateChunkSequence(chunks) | Check sequence validity |

Error Functions

| Function | Description | |----------|-------------| | catchStreamErrors(stream, handler) | Handle errors gracefully | | streamWithTimeout(stream, ms) | Add timeout | | createStreamError(error) | Create error chunk |

License

MIT - see LICENSE for details.