AI SDK Streaming

Test AI streaming applications with realistic chaos scenarios using Cruel’s AI SDK integration.

Basic AI SDK Integration

Wrap AI SDK models with chaos:

import { openai } from '@ai-sdk/openai'
import { generateText } from 'ai'
import { cruelModel } from 'cruel/ai-sdk'

const model = cruelModel(openai('gpt-4o'), {
  rateLimit: 0.05,
  slowTokens: [30, 100]
})

const result = await generateText({
  model,
  prompt: 'Write a haiku about code'
})

console.log(result.text)

Stream Chaos

Inject failures during streaming:

Stream Cut
Slow Tokens
Corrupt Chunks

Simulate stream interruptions:

import { openai } from '@ai-sdk/openai'
import { streamText } from 'ai'
import { cruelModel } from 'cruel/ai-sdk'

const model = cruelModel(openai('gpt-4o'), {
  streamCut: 0.3 // 30% chance stream cuts mid-response
})

try {
  const result = streamText({
    model,
    prompt: 'Write a long story'
  })

  for await (const chunk of result.textStream) {
    process.stdout.write(chunk)
  }
} catch (error) {
  console.error('\nStream cut:', error.message)
}

Simulate slow token generation:

const model = cruelModel(openai('gpt-4o'), {
  slowTokens: [50, 200] // 50-200ms delay per token
})

const start = Date.now()
const result = streamText({ model, prompt: 'Count to 10' })

for await (const chunk of result.textStream) {
  process.stdout.write(chunk)
}

console.log(`\nTook ${Date.now() - start}ms`)

Simulate corrupted stream data:

const model = cruelModel(openai('gpt-4o'), {
  corruptChunks: 0.1 // 10% of chunks may be corrupted
})

const result = streamText({
  model,
  prompt: 'Write a poem'
})

let text = ''
for await (const chunk of result.textStream) {
  text += chunk
}

// May contain \uFFFD replacement characters
console.log(text)

Handling Full Stream Events

Process all stream event types:

import { openai } from '@ai-sdk/openai'
import { streamText } from 'ai'
import { cruelModel } from 'cruel/ai-sdk'

const model = cruelModel(openai('gpt-4o'), {
  streamCut: 0.2,
  slowTokens: [30, 100],
  onChaos: (event) => {
    console.log('Chaos event:', event.type)
  }
})

const result = streamText({
  model,
  prompt: 'Write a haiku about testing'
})

for await (const part of result.fullStream) {
  switch (part.type) {
    case 'text-delta':
      process.stdout.write(part.textDelta)
      break
    
    case 'finish':
      console.log('\n\nFinish reason:', part.finishReason)
      console.log('Usage:', part.usage)
      break
    
    case 'error':
      console.error('\nError:', part.error)
      break
  }
}

Complete Example with Retries

Robust streaming with retry logic:

import { openai } from '@ai-sdk/openai'
import { streamText } from 'ai'
import { cruelModel } from 'cruel/ai-sdk'

async function streamWithRetry(prompt: string, maxRetries = 3) {
  const model = cruelModel(openai('gpt-4o'), {
    streamCut: 0.3,
    slowTokens: [20, 80],
    rateLimit: 0.1,
    overloaded: 0.05
  })

  for (let attempt = 0; attempt < maxRetries; attempt++) {
    try {
      const result = streamText({ model, prompt })
      let fullText = ''

      for await (const chunk of result.textStream) {
        fullText += chunk
        process.stdout.write(chunk)
      }

      console.log('\n\nSuccess!')
      return fullText
    } catch (error) {
      console.error(`\nAttempt ${attempt + 1} failed:`, error.message)
      
      if (attempt < maxRetries - 1) {
        const delay = Math.pow(2, attempt) * 1000
        console.log(`Retrying in ${delay}ms...`)
        await new Promise(r => setTimeout(r, delay))
      }
    }
  }

  throw new Error('All retry attempts failed')
}

await streamWithRetry('Write a story about resilience')

AI Provider Presets

Realistic chaos presets for AI testing:

import { cruelModel, presets } from 'cruel/ai-sdk'

const model = cruelModel(openai('gpt-4o'), presets.realistic)
// rateLimit: 0.02
// overloaded: 0.01
// delay: [50, 200]
// slowTokens: [20, 80]

Middleware Approach

Use middleware for application-wide chaos:

import { openai } from '@ai-sdk/openai'
import { generateText } from 'ai'
import { cruelMiddleware } from 'cruel/ai-sdk'

const result = await generateText({
  model: openai('gpt-4o'),
  prompt: 'Explain quantum computing',
  experimental_telemetry: {
    isEnabled: true,
    functionId: 'quantum-explain'
  },
  experimental_transform: cruelMiddleware({
    streamCut: 0.1,
    slowTokens: [30, 100],
    rateLimit: 0.05
  })
})

console.log(result.text)

Provider-Level Chaos

Apply chaos to all models from a provider:

import { openai } from '@ai-sdk/openai'
import { cruelProvider } from 'cruel/ai-sdk'
import { generateText } from 'ai'

const chaosOpenAI = cruelProvider(openai, {
  rateLimit: 0.1,
  slowTokens: [30, 100],
  models: {
    'gpt-4o': {
      streamCut: 0.15  // Extra chaos for GPT-4
    },
    'gpt-4o-mini': {
      overloaded: 0.2  // More overload for mini
    }
  }
})

// All models from this provider have chaos
const gpt4 = chaosOpenAI('gpt-4o')
const mini = chaosOpenAI('gpt-4o-mini')

const result1 = await generateText({ model: gpt4, prompt: 'Test 1' })
const result2 = await generateText({ model: mini, prompt: 'Test 2' })

Chaos Event Tracking

Monitor chaos events during AI operations:

import { cruelModel, diagnostics } from 'cruel/ai-sdk'
import { openai } from '@ai-sdk/openai'
import { streamText } from 'ai'

const ctx = diagnostics.context()
const track = diagnostics.tracker(ctx)

const model = cruelModel(openai('gpt-4o'), {
  rateLimit: 0.1,
  streamCut: 0.1,
  slowTokens: [50, 150],
  onChaos: track
})

const requestId = 1
diagnostics.before(ctx, requestId)

const start = Date.now()
try {
  const result = streamText({ model, prompt: 'Write a haiku' })
  let text = ''
  
  for await (const chunk of result.textStream) {
    text += chunk
  }
  
  diagnostics.success(ctx, requestId, Date.now() - start, text)
} catch (error) {
  diagnostics.failure(ctx, requestId, Date.now() - start, error)
}

const stats = diagnostics.stats(ctx)
console.log('Stats:', {
  total: stats.total,
  succeeded: stats.succeeded,
  successRate: (stats.successRate * 100).toFixed(1) + '%',
  events: stats.events,
  avgLatency: stats.latency.success.avg + 'ms'
})

Testing Streaming with Bun

Comprehensive streaming tests:

import { describe, test, expect } from 'bun:test'
import { openai } from '@ai-sdk/openai'
import { streamText } from 'ai'
import { cruelModel } from 'cruel/ai-sdk'

describe('AI streaming resilience', () => {
  test('handles stream cut gracefully', async () => {
    const model = cruelModel(openai('gpt-4o'), {
      streamCut: 1 // Always cut
    })

    try {
      const result = streamText({
        model,
        prompt: 'Count to 100'
      })

      for await (const _ of result.textStream) {
        // Should throw before completion
      }

      expect(true).toBe(false) // Should not reach
    } catch (error) {
      expect(error.message).toContain('stream')
    }
  })

  test('measures token generation speed', async () => {
    const model = cruelModel(openai('gpt-4o'), {
      slowTokens: 100 // Fixed 100ms per token
    })

    const start = Date.now()
    const result = streamText({
      model,
      prompt: 'Say "test" five times'
    })

    let chunks = 0
    for await (const _ of result.textStream) {
      chunks++
    }

    const elapsed = Date.now() - start
    // Should take at least chunks * 100ms
    expect(elapsed).toBeGreaterThanOrEqual((chunks - 1) * 90)
  })

  test('recovers from rate limits', async () => {
    let attempts = 0

    const attemptStream = async () => {
      attempts++
      const model = cruelModel(openai('gpt-4o'), {
        rateLimit: 0.5
      })

      const result = streamText({
        model,
        prompt: 'Test'
      })

      let text = ''
      for await (const chunk of result.textStream) {
        text += chunk
      }
      return text
    }

    for (let i = 0; i < 3; i++) {
      try {
        await attemptStream()
        break
      } catch (error) {
        if (i === 2) throw error
        await new Promise(r => setTimeout(r, 1000))
      }
    }

    expect(attempts).toBeLessThanOrEqual(3)
  })
})

Production Pattern

Production-ready streaming with monitoring:

import { openai } from '@ai-sdk/openai'
import { streamText } from 'ai'
import { cruelModel, diagnostics } from 'cruel/ai-sdk'

class ResilientAIStream {
  private ctx = diagnostics.context()
  private requestId = 0

  async stream(prompt: string, maxRetries = 3) {
    for (let attempt = 0; attempt < maxRetries; attempt++) {
      const id = ++this.requestId
      
      const model = cruelModel(openai('gpt-4o'), {
        ...( process.env.NODE_ENV === 'development' 
          ? { streamCut: 0.1, slowTokens: [30, 100] }
          : {}
        ),
        onChaos: diagnostics.tracker(this.ctx)
      })

      diagnostics.before(this.ctx, id)
      const start = Date.now()

      try {
        const result = streamText({ model, prompt })
        let fullText = ''

        for await (const chunk of result.textStream) {
          fullText += chunk
          process.stdout.write(chunk)
        }

        diagnostics.success(this.ctx, id, Date.now() - start, fullText)
        return fullText
      } catch (error) {
        diagnostics.failure(this.ctx, id, Date.now() - start, error)
        
        if (attempt < maxRetries - 1) {
          await new Promise(r => setTimeout(r, Math.pow(2, attempt) * 1000))
        } else {
          throw error
        }
      }
    }
  }

  getStats() {
    return diagnostics.stats(this.ctx)
  }
}

const ai = new ResilientAIStream()
await ai.stream('Explain chaos engineering')

console.log('\nStats:', ai.getStats())

Basic Usage

AI Applications

Patterns

AI SDK Streaming

AI SDK Streaming

Basic AI SDK Integration

Stream Chaos

Handling Full Stream Events

Complete Example with Retries

AI Provider Presets

Middleware Approach

Provider-Level Chaos

Chaos Event Tracking

Testing Streaming with Bun

Production Pattern

Next Steps

Build docs developers (and LLMs) love

Basic Usage

AI Applications

Patterns

Documentation Index

​AI SDK Streaming

​Basic AI SDK Integration

​Stream Chaos

​Handling Full Stream Events

​Complete Example with Retries

​AI Provider Presets

​Middleware Approach

​Provider-Level Chaos

​Chaos Event Tracking

​Testing Streaming with Bun

​Production Pattern

​Next Steps

Build docs developers (and LLMs) love

AI SDK Streaming

Basic AI SDK Integration

Stream Chaos

Handling Full Stream Events

Complete Example with Retries

AI Provider Presets

Middleware Approach

Provider-Level Chaos

Chaos Event Tracking

Testing Streaming with Bun

Production Pattern

Next Steps