Building Real-Time AI Applications

// Server-side streaming endpoint
export async function POST(req: Request) {
  const { messages } = await req.json();
  
  const stream = await openai.chat.completions.create({
    model: 'gpt-4',
    messages,
    stream: true,
  });

  const encoder = new TextEncoder();
  
  const readable = new ReadableStream({
    async start(controller) {
      for await (const chunk of stream) {
        const text = chunk.choices[0]?.delta?.content || '';
        if (text) {
          controller.enqueue(encoder.encode(`data: ${JSON.stringify({ text })}

`));
        }
      }
      controller.enqueue(encoder.encode('data: [DONE]

'));
      controller.close();
    },
  });

  return new Response(readable, {
    headers: {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache',
      'Connection': 'keep-alive',
    },
  });
}

// Client-side streaming consumer
async function streamChat(message: string, onToken: (text: string) => void) {
  const response = await fetch('/api/chat', {
    method: 'POST',
    body: JSON.stringify({ messages: [{ role: 'user', content: message }] }),
  });

  const reader = response.body?.getReader();
  const decoder = new TextDecoder();

  while (true) {
    const { done, value } = await reader!.read();
    if (done) break;

    const chunk = decoder.decode(value);
    const lines = chunk.split('
').filter(line => line.startsWith('data: '));

    for (const line of lines) {
      const data = line.slice(6);
      if (data === '[DONE]') return;
      
      const { text } = JSON.parse(data);
      onToken(text);
    }
  }
}

// WebSocket server with AI streaming
import { WebSocketServer } from 'ws';

const wss = new WebSocketServer({ port: 8080 });
const connections = new Map<string, WebSocket>();

wss.on('connection', (ws, req) => {
  const userId = req.headers['x-user-id'] as string;
  connections.set(userId, ws);
  
  ws.on('message', async (data) => {
    const message = JSON.parse(data.toString());
    
    switch (message.type) {
      case 'chat':
        await handleChatMessage(ws, message);
        break;
      case 'ping':
        ws.send(JSON.stringify({ type: 'pong' }));
        break;
    }
  });
  
  ws.on('close', () => {
    connections.delete(userId);
  });
});

async function handleChatMessage(ws: WebSocket, message: any) {
  // Send typing indicator
  ws.send(JSON.stringify({ type: 'typing', status: true }));
  
  try {
    const stream = await openai.chat.completions.create({
      model: 'gpt-4',
      messages: message.messages,
      stream: true,
    });
    
    for await (const chunk of stream) {
      const text = chunk.choices[0]?.delta?.content || '';
      if (text) {
        ws.send(JSON.stringify({ type: 'token', text }));
      }
    }
    
    ws.send(JSON.stringify({ type: 'complete' }));
  } catch (error) {
    ws.send(JSON.stringify({ type: 'error', message: error.message }));
  } finally {
    ws.send(JSON.stringify({ type: 'typing', status: false }));
  }
}