High-Throughput NestJS & Node.js Backends for LLM Agent Streams

// NestJS High-Throughput SSE Streaming Controller with Redis Semantic Cache & Abort Cleanup
import { Controller, Sse, Query, Req, MessageEvent, Logger } from '@nestjs/common';
import { Observable } from 'rxjs';
import { Request } from 'express';
import { RedisVectorCacheService } from './redis-vector-cache.service';
import { PythonAgentBridgeService } from './python-agent-bridge.service';

@Controller('v1/agents')
export class AgentStreamController {
  private readonly logger = new Logger(AgentStreamController.name);

  constructor(
    private readonly vectorCache: RedisVectorCacheService,
    private readonly agentBridge: PythonAgentBridgeService,
  ) {}

  @Sse('stream')
  streamAgentReasoning(
    @Query('prompt') prompt: string,
    @Query('tenantId') tenantId: string,
    @Req() req: Request,
  ): Observable<MessageEvent> {
    return new Observable<MessageEvent>((subscriber) => {
      const abortController = new AbortController();

      // Immediately release upstream Python worker & sockets if client disconnects
      req.on('close', () => {
        this.logger.debug(`[SSE] Client disconnected for tenant ${tenantId}, aborting stream`);
        abortController.abort();
        subscriber.complete();
      });

      (async () => {
        try {
          // 1. Emit instant 200 OK heartbeat frame (<15ms) to prevent AWS ALB 504 timeouts
          subscriber.next({ type: 'status', data: { state: 'connected', ts: Date.now() } });

          // 2. Check sub-5ms Redis HNSW semantic vector cache before invoking LLM
          const cached = await this.vectorCache.lookupSimilar(tenantId, prompt, 0.94);
          if (cached) {
            subscriber.next({ type: 'token', data: { text: cached.response, cached: true } });
            subscriber.next({ type: 'done', data: { source: 'redis-semantic-cache' } });
            subscriber.complete();
            return;
          }

          // 3. Stream incremental tokens over gRPC/HTTP2 from Python LangGraph/PydanticAI worker
          for await (const chunk of this.agentBridge.streamInference(prompt, tenantId, abortController.signal)) {
            if (abortController.signal.aborted) break;
            subscriber.next({ type: 'token', data: { text: chunk.delta, step: chunk.agentNode } });
          }

          subscriber.next({ type: 'done', data: { source: 'live-agent-pipeline' } });
          subscriber.complete();
        } catch (err) {
          this.logger.error(`[SSE Error] ${(err as Error).message}`);
          subscriber.next({ type: 'error', data: { message: 'Stream interrupted, retrying...' } });
          subscriber.complete();
        }
      })();
    });
  }
}