AIEditor自定义大语言模型监听机制实战：从原理到实现

1次阅读

没有评论

共计 3156 个字符，预计需要花费 8 分钟才能阅读完成。

在 AIEditor 中集成自定义大语言模型时，监听机制主要用于以下几种场景：

实时进度反馈 ：在模型训练或推理过程中，向用户实时展示当前进度和状态
资源监控 ：监控模型运行的 CPU、GPU、内存等资源消耗情况
错误报警 ：在模型运行出现异常时及时通知用户
交互式调试 ：在模型开发阶段提供实时日志和调试信息

在实现监听功能时，WebSocket 和 Server-Sent Events(SSE) 是最常见的两种技术方案：

WebSocket
全双工通信，客户端和服务器可以同时发送消息
需要维护持久连接，适合高频交互场景
协议相对复杂，需要处理连接状态
SSE
单向通信，服务器向客户端推送消息
基于 HTTP 协议，实现简单
自动处理重连，适合低频更新场景

选型建议 ：对于大语言模型监听这种需要实时双向通信的场景，WebSocket 通常是更好的选择。

整个监听系统采用事件驱动架构，主要组件包括：

事件生产者（模型服务）
事件总线（WebSocket 服务）
事件消费者（前端界面）

使用 ws 库实现 WebSocket 服务：

import WebSocket, {WebSocketServer} from 'ws';
import {v4 as uuidv4} from 'uuid';

interface ModelEvent {
  type: 'progress' | 'status' | 'error';
  data: any;
  timestamp: number;
}

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

// 处理新连接
wss.on('connection', (ws) => {const clientId = uuidv4();
  clients.set(clientId, ws);

  // 心跳检测
  const heartbeatInterval = setInterval(() => {if (ws.readyState === WebSocket.OPEN) {ws.ping();
    }
  }, 30000);

  // 消息处理
  ws.on('message', (data) => {
    try {const message = JSON.parse(data.toString());
      // 处理客户端消息
    } catch (err) {console.error('消息解析错误:', err);
    }
  });

  // 连接关闭
  ws.on('close', () => {clearInterval(heartbeatInterval);
    clients.delete(clientId);
  });
});

// 广播模型事件
function broadcastEvent(event: ModelEvent) {const message = JSON.stringify(event);
  clients.forEach((client) => {if (client.readyState === WebSocket.OPEN) {client.send(message);
    }
  });
}

前端实现包含重连机制和心跳检测：

class ModelEventListener {
  private ws: WebSocket | null = null;
  private reconnectAttempts = 0;
  private maxReconnectAttempts = 5;
  private reconnectDelay = 1000;
  private heartbeatInterval: NodeJS.Timeout | null = null;

  constructor(private url: string, private callbacks: {onMessage: (event: ModelEvent) => void;
    onError?: (error: Error) => void;
    onReconnect?: (attempt: number) => void;
  }) {}

  connect() {this.ws = new WebSocket(this.url);

    this.ws.onopen = () => {
      this.reconnectAttempts = 0;
      // 启动心跳检测
      this.heartbeatInterval = setInterval(() => {if (this.ws?.readyState === WebSocket.OPEN) {this.ws.send('ping');
        }
      }, 25000);
    };

    this.ws.onmessage = (event) => {
      try {const data = JSON.parse(event.data);
        this.callbacks.onMessage(data);
      } catch (err) {this.callbacks.onError?.(new Error('消息解析失败'));
      }
    };

    this.ws.onclose = () => {if (this.heartbeatInterval) {clearInterval(this.heartbeatInterval);
      }
      this.handleReconnect();};

    this.ws.onerror = (error) => {this.callbacks.onError?.(new Error('连接错误'));
    };
  }

  private handleReconnect() {if (this.reconnectAttempts < this.maxReconnectAttempts) {
      this.reconnectAttempts++;
      this.callbacks.onReconnect?.(this.reconnectAttempts);
      setTimeout(() => this.connect(), this.reconnectDelay);
      this.reconnectDelay *= 2; // 指数退避
    } else {this.callbacks.onError?.(new Error('最大重试次数达到'));
    }
  }

  disconnect() {if (this.ws) {this.ws.close();
    }
    if (this.heartbeatInterval) {clearInterval(this.heartbeatInterval);
    }
  }
}

单机 WebSocket 连接数受限于内存和文件描述符
解决方案：
使用多实例部署，通过负载均衡分发连接
考虑使用专业的 WebSocket 服务如 SocketCluster

使用二进制协议如 Protocol Buffers 替代 JSON
压缩大消息体
批量发送小消息

在 WebSocket 握手阶段进行鉴权：

// 服务端鉴权
wss.on('connection', (ws, req) => {const token = req.headers['sec-websocket-protocol'];
  if (!verifyToken(token)) {ws.close(1008, '未授权');
    return;
  }
  // ... 其他逻辑
});

// 前端连接时带上 token
const ws = new WebSocket(url, [jwtToken]);