streaming-parser.ts

/**
 * Streaming Parser Example
 *
 * Demonstrates real-time parsing of Harmony format tokens as they arrive
 * from an OpenAI model, similar to how you'd handle streaming responses.
 */

import {
  StreamableParser,
  loadHarmonyEncoding,
  Role,
  Channel,
  HARMONY_TOKENS,
  SPECIAL_TOKENS
} from '../src/index.js';

async function main(): Promise<void> {
  console.log('🌊 Harmony Protocol - Streaming Parser Example');
  console.log('===============================================');

  try {
    // Step 1: Initialize the encoding and streaming parser
    console.log('\n🔧 Step 1: Initializing streaming parser');
    const encoding = loadHarmonyEncoding('o200k_base');
    const parser = new StreamableParser(encoding, Role.ASSISTANT);
    console.log('✓ Created streaming parser for assistant role');

    // Step 2: Simulate streaming tokens from a model
    console.log('\n🎬 Step 2: Simulating streaming response');

    // This simulates what you'd receive from OpenAI's streaming API
    const simulatedResponse = [
      `${HARMONY_TOKENS.START}assistant${HARMONY_TOKENS.CHANNEL}analysis${HARMONY_TOKENS.MESSAGE}`,
      'I need to think about this step by step. ',
      'The user is asking about mathematical calculation, ',
      'so I should break down the problem carefully.',
      `${HARMONY_TOKENS.END}`,

      `${HARMONY_TOKENS.START}assistant${HARMONY_TOKENS.CHANNEL}commentary${HARMONY_TOKENS.MESSAGE}`,
      'This is a straightforward arithmetic problem. ',
      'I can solve this directly without complex reasoning.',
      `${HARMONY_TOKENS.END}`,

      `${HARMONY_TOKENS.START}assistant${HARMONY_TOKENS.CHANNEL}final${HARMONY_TOKENS.MESSAGE}`,
      'Let me solve this for you: ',
      '2 + 2 = 4. ',
      'This is a basic addition problem.',
      `${HARMONY_TOKENS.END}`
    ];

    console.log('✓ Created simulated streaming response with multiple channels');

    // Step 3: Process tokens as they arrive
    console.log('\n⚡ Step 3: Processing streaming tokens');
    let chunkCount = 0;

    for (const chunk of simulatedResponse) {
      chunkCount++;
      console.log(`\n--- Processing chunk ${chunkCount} ---`);
      console.log(`Raw chunk: "${chunk}"`);

      // Process the text chunk
      parser.processText(chunk);

      // Get real-time delta for UI updates
      const delta = parser.getLastContentDelta();
      if (delta) {
        console.log(`📝 Content delta: "${delta}"`);
      }

      // Show parsing statistics
      const stats = parser.getStats();
      console.log(`📊 Parser state: ${stats.totalMessages} messages, buffer: ${stats.bufferLength} chars`);

      // Small delay to simulate network latency
      await new Promise(resolve => setTimeout(resolve, 100));
    }

    // Step 4: Get final parsed messages
    console.log('\n✅ Step 4: Final results');
    const messages = parser.intoMessages();
    console.log(`✓ Parsed ${messages.length} complete messages`);

    for (let i = 0; i < messages.length; i++) {
      const message = messages[i];
      console.log(`\nMessage ${i + 1}:`);
      console.log(`  Role: ${message.role}`);
      console.log(`  Channel: ${message.channel || 'none'}`);
      console.log(`  Content: "${message.getTextContent()}"`);
    }

    // Step 5: Demonstrate token-by-token processing
    console.log('\n🔬 Step 5: Token-by-token processing demonstration');

    const parser2 = new StreamableParser(encoding, Role.ASSISTANT);

    // Encode the response to tokens and process them individually
    const responseText = simulatedResponse.join('');
    const responseTokens = encoding.encode(responseText);

    console.log(`✓ Encoded response to ${responseTokens.length} tokens`);

    let processedTokens = 0;
    console.log('\nProcessing tokens individually:');

    for (const token of responseTokens) {
      parser2.process(token);
      processedTokens++;

      // Show progress every 10 tokens
      if (processedTokens % 10 === 0 || processedTokens === responseTokens.length) {
        const stats = parser2.getStats();
        const delta = parser2.getLastContentDelta();

        console.log(`Token ${processedTokens}/${responseTokens.length}: `
          + `${stats.totalMessages} msgs, delta: "${delta || '(none)'}"`);
      }
    }

    const finalMessages = parser2.intoMessages();
    console.log(`✓ Token-by-token parsing resulted in ${finalMessages.length} messages`);

    // Step 6: Error handling demonstration
    console.log('\n🔥 Step 6: Error handling demonstration');

    const parser3 = new StreamableParser(encoding, Role.ASSISTANT);

    // Simulate malformed input
    const malformedInput = [
      `${HARMONY_TOKENS.START}unknown_role${HARMONY_TOKENS.MESSAGE}This has an invalid role`,
      `${HARMONY_TOKENS.START}assistant${HARMONY_TOKENS.CHANNEL}invalid_channel${HARMONY_TOKENS.MESSAGE}Invalid channel`,
      `${HARMONY_TOKENS.START}assistant${HARMONY_TOKENS.MESSAGE}This is valid content${HARMONY_TOKENS.END}`
    ];

    for (const input of malformedInput) {
      parser3.processText(input);
    }

    const robustMessages = parser3.intoMessages();
    console.log(`✓ Robust parsing handled errors gracefully: ${robustMessages.length} valid messages`);

    // Show final statistics
    const finalStats = parser.getStats();
    console.log('\n📈 Final Statistics:');
    console.log(`  Total messages parsed: ${finalStats.totalMessages}`);
    console.log(`  Buffer length: ${finalStats.bufferLength}`);
    console.log(`  Last role: ${finalStats.lastRole}`);
    console.log(`  Last channel: ${finalStats.lastChannel}`);

    console.log('\n🎉 Streaming parser example completed successfully!');
    console.log('\n💡 Integration tips:');
    console.log('  1. Use processText() for string chunks from HTTP streams');
    console.log('  2. Use process() for individual tokens from tokenized streams');
    console.log('  3. Use getLastContentDelta() for real-time UI updates');
    console.log('  4. Handle parsing errors gracefully - parser continues working');
    console.log('  5. Call intoMessages() when streaming is complete');

  } catch (error) {
    console.error('❌ Streaming parser example failed:', error);
  }
}

// Helper function to simulate typing effect
async function simulateTyping(text: string, delay: number = 50): Promise<void> {
  for (const char of text) {
    process.stdout.write(char);
    await new Promise(resolve => setTimeout(resolve, delay));
  }
  console.log(); // New line
}

// Run the example
main().catch(console.error);