Mastra pour le streaming de données en temps réel : créer des agents pilotés par événements avec des intégrations WebSocket

Le paysage du développement d'applications IA évolue rapidement, passant du traitement par lots à la réactivité en temps réel. Dans les applications d'entreprise et grand public modernes, les agents IA ne peuvent pas se permettre d'attendre la fin d'une session complète avant de réagir. Ils doivent traiter les données en flux, maintenir l'état à travers des interactions continues et fournir des retours en moins d'une seconde. C'est là que Mastra, un framework d'agents IA axé sur TypeScript, excelle. En prenant nativement en charge les flux de travail asynchrones et les intégrations avec des couches de transport en temps réel comme les WebSockets, Mastra permet aux développeurs de créer des agents pilotés par événements qui semblent véritablement vivants.

Pourquoi l'architecture pilotée par événements est essentielle pour les agents IA

Les modèles traditionnels de requête-réponse sont mal adaptés à l'IA conversationnelle ou aux systèmes de surveillance intensifs en données. Une approche pilotée par événements permet à un agent de réagir à des occurrences discrètes — telles qu'un nouveau message utilisateur, une lecture de capteur ou une variation financière — sans bloquer le thread principal. Combinée aux WebSockets, qui fournissent des canaux de communication en duplex intégral, la latence entre l'occurrence de l'événement et l'action de l'agent est minimisée. Cela est crucial pour des applications comme les bots de trading en direct, l'assistance client en temps réel ou les assistants de codage collaboratif, où chaque milliseconde compte.

Mise en place de l'agent Mastra

Mastra simplifie la création d'agents IA en permettant de définir des outils, des invites et la logique de manière déclarative. Pour commencer, vous devez installer le package principal et mettre en place une structure d'agent de base. L'agent sera responsable du traitement des messages WebSocket entrants et de la génération de réponses contextuelles.

import { createAgent } from 'mastra';
import { openai } from '@mastra/core/models';

export const realtimeAgent = createAgent({
  name: 'Realtime Support Agent',
  description: 'An agent that processes real-time user queries',
  model: openai('gpt-4o'),
  instructions: `
    You are a helpful assistant. 
    Keep your responses concise as they are being streamed.
    Acknowledge immediate user intents before providing detailed answers.
  `,
  tools: {
    lookupOrder: async ({ orderId }) => {
      // Simulate a fast database lookup
      return { status: 'processing', estimatedTime: '2 days' };
    }
  }
});

Intégration des WebSockets avec Mastra

La puissance de cette configuration réside dans la manière dont nous connectons l'agent au client. Nous utilisons un serveur WebSocket pour maintenir une connexion persistante. Lorsqu'un message arrive, nous déclenchons l'agent Mastra pour diffuser sa réponse directement vers le socket. Cela évite la surcharge liée à la fermeture et à la réouverture des connexions HTTP pour chaque interaction.

Ci-dessous, un exemple pratique utilisant Node.js et la bibliothèque ws pour gérer la couche WebSocket. Notez comment nous exploitons les capacités de streaming de Mastra pour envoyer les jetons au client au fur et à mesure de leur génération.

import WebSocket, { WebSocketServer } from 'ws';
import { server } from 'http';
import { realtimeAgent } from './agent'; // Import from above

const server = http.createServer();
const wss = new WebSocketServer({ server });

wss.on('connection', (ws) => {
  console.log('Client connected');

  ws.on('message', async (message) => {
    try {
      const userPrompt = message.toString();
      
      // Use Mastra's stream method for real-time output
      const stream = await realtimeAgent.stream(userPrompt);

      // Forward chunks as they arrive
      for await (const chunk of stream) {
        if (chunk.type === 'text-delta') {
          ws.send(JSON.stringify({ 
            type: 'token', 
            content: chunk.text 
          }));
        }
      }
      
      ws.send(JSON.stringify({ type: 'done' }));
    } catch (error) {
      console.error('Agent error:', error);
      ws.send(JSON.stringify({ type: 'error', message: 'Something went wrong' }));
    }
  });

  ws.on('close', () => {
    console.log('Client disconnected');
  });
});

server.listen(8080, () => {
  console.log('WebSocket server running on ws://localhost:8080');
});

Gestion de l'état et de la concurrence

Dans les scénarios en temps réel, plusieurs messages peuvent arriver en succession rapide. Les agents Mastra sont sans état par défaut pour chaque appel, mais ils peuvent maintenir le contexte via des modules de mémoire ou des magasins d'état externes. Pour les serveurs WebSocket à forte concurrence, il est crucial de s'assurer que chaque connexion WebSocket est associée à un contexte d'agent ou à un ID de session spécifique. Cela empêche la fuite de données entre les utilisateurs et garantit que l'« historique de la conversation » est correctement limité à la connexion client spécifique. Vous pouvez étendre l'exemple ci-dessus en passant un sessionId depuis les métadonnées WebSocket vers les options de l'agent Mastra pour récupérer les interactions passées pertinentes.

Conclusion

La création d'agents IA en temps réel nécessite plus que simplement connecter un LLM à une base de données ; cela exige une architecture robuste pilotée par événements capable de gérer les connexions persistantes et les données en flux. Mastra fournit la couche d'abstraction idéale pour gérer la logique de l'agent, tandis que les WebSockets fournissent le mécanisme de transport pour une communication à faible latence. En combinant ces technologies, les développeurs peuvent créer des applications réactives et intelligentes qui réagissent au monde en temps réel, offrant une expérience utilisateur supérieure dans des environnements dynamiques.