devbox
知新_RL
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

开源模型应用落地-业务优化篇（一）_开源大模型调优

作者：知新_RL | 2024-04-30 02:38:01

赞

踩

开源大模型调优

一、前言

    通过参与“开源模型应用落地-业务整合系列篇”的学习，我们已经成功建立了基本的业务流程。然而，这只是迈出了万里长征的第一步。现在我们要对整个项目进行优化，以提高效率。我们计划利用线程池来加快处理速度，使用redis来实现排队需求，以及通过多级环境来减轻负载压力。这些优化措施将有助于我们进一步改进项目的性能和效果。

二、术语

2.1. 线程池

    是一种用于线程管理的技术，它包含一组预先创建的线程，用于执行任务。线程池维护着一个任务队列，当有任务到达时，线程池中的线程会自动分配任务并执行。

    线程池的主要目的是重用线程，避免频繁地创建和销毁线程带来的开销。通过使用线程池，可以在程序初始化时创建一组线程，并将任务提交给线程池进行处理，而不需要为每个任务都创建一个新的线程。这样可以有效地管理系统中的线程数量，控制并发度，提高系统的性能和资源利用率。

线程池通常包含以下几个关键组件：

  1. 任务队列（Task Queue）：用于存储待执行的任务，通常是一个队列结构。当有新的任务到达时，会被添加到任务队列中。
  2. 线程池管理器（Thread Pool Manager）：负责管理线程池的创建、销毁和线程的调度。它会监视任务队列的状态，并根据需要动态地创建或回收线程。
  3. 工作线程（Worker Threads）：线程池中的线程，用于执行任务。它们会从任务队列中获取任务，并执行任务的处理逻辑。

三、前置条件

3.1. 已搭建WebSocket与AI服务调用链路

四、技术实现

4.1. 调整业务逻辑处理类

     对于每次交互的chat对话，都需要经过以下步骤，包括但不限于：

  1.   对用户输入的内容进行自定义违规词检测
  2.   对用户输入的内容进行第三方在线违规词检测
  3.   对用户输入的内容进行组装成Prompt
  4.   对Prompt根据业务进行增强（完善prompt的内容）
  5.   对history进行裁剪或总结（检测history是否操作模型支持的上下文长度，例如qwen-7b支持的上下文长度为8192）

特别是调用第三方在线违规词检测，例如：某某云的内容安全审核服务，是非常耗时，会阻塞正常线程的执行，导致吞吐量的下降。

所以，我们就要对下面这块的处理逻辑进行调整，通过自定义线程池的方式，去处理核心的Chat交互流程

调整后：

4.2. 新增线程处理类

  1. import io.netty.channel.ChannelHandlerContext;
  2. import lombok.extern.slf4j.Slf4j;
  3. import org.springframework.beans.factory.annotation.Autowired;
  4. import org.springframework.stereotype.Component;
  5.  
  6. import java.util.List;
  7. import java.util.concurrent.ExecutorService;
  8. import java.util.concurrent.Executors;
  9.  
  10. @Component
  11. @Slf4j
  12. public class TaskUtils{
  13.     private static ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() * 2);
  14.     @Autowired
  15.     private AIChatUtils aiChatUtils;
  16.  
  17.      public void execute(AITaskReqMessage aiTaskReqMessage) {
  18.  
  19.         executorService.execute(() -> {
  20.             Long userId = aiTaskReqMessage.getUserId();
  21.  
  22.             if (null == userId || (long) userId < 10000) {
  23.                 log.warn("用户身份标识有误!");
  24.                 return;
  25.             }
  26.  
  27.             ChannelHandlerContext channelHandlerContext = AbstractBusinessLogicHandler.getContextByUserId(userId);
  28.  
  29.             if (channelHandlerContext != null) {
  30.                 try {
  31.                     aiChatUtils.chatStream(aiTaskReqMessage);
  32.  
  33.                 } catch (Throwable exception) {
  34.                     exception.printStackTrace();
  35.                 }
  36.             }
  37.         });
  38.     }
  39.  
  40.     public static void destory(){
  41.         executorService.shutdownNow();
  42.         executorService = null;
  43.     }
  44.  
  45. }

4.3. 新增线程处理实体类

  1. import lombok.Builder;
  2. import lombok.Getter;
  3. import lombok.Setter;
  4.  
  5. import java.util.List;
  6.  
  7. @Builder
  8. @Setter
  9. @Getter
  10. public class AITaskReqMessage {
  11.  
  12.     private String messageId;
  13.     private Long userId;
  14.     private String contents;
  15.     private List<ChatContext> history;
  16. }

五、测试

在线测试方式：WebSocket在线测试工具

5.1.  建立连接

5.2.  业务初始化

服务端输出：

5.3.  业务对话

服务端输出

5.4.  关闭连接

六、附带说明

6.1. 可以使用jmeter进行websocket压测，以评估各项性能指标是否符合预期（下一篇）

6.2. BusinessHandler完整代码

  1. import com.alibaba.fastjson.JSON;
  2. import io.netty.channel.ChannelHandler;
  3. import lombok.extern.slf4j.Slf4j;
  4. import io.netty.channel.ChannelHandlerContext;
  5. import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
  6. import org.apache.commons.lang3.StringUtils;
  7. import org.springframework.beans.factory.annotation.Autowired;
  8. import org.springframework.stereotype.Component;
  9.  
  10. import java.util.List;
  11.  
  12.  
  13. /**
  14.  * @Description: 处理消息的handler
  15.  */
  16. @Slf4j
  17. @ChannelHandler.Sharable
  18. @Component
  19. public class BusinessHandler extends AbstractBusinessLogicHandler<TextWebSocketFrame> {
  20.     @Autowired
  21.     private TaskUtils taskExecuteUtils;
  22.  
  23.     @Override
  24.     public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
  25.         String channelId = ctx.channel().id().asShortText();
  26.         log.info("add client,channelId：{}", channelId);
  27.     }
  28.  
  29.     @Override
  30.     public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
  31.         String channelId = ctx.channel().id().asShortText();
  32.         log.info("remove client,channelId：{}", channelId);
  33.     }
  34.  
  35.  
  36.     @Override
  37.     protected void channelRead0(ChannelHandlerContext channelHandlerContext, TextWebSocketFrame textWebSocketFrame)
  38.             throws Exception {
  39.         // 获取客户端传输过来的消息
  40.         String content = textWebSocketFrame.text();
  41.         // 兼容在线测试
  42.         if (StringUtils.equals(content, "PING")) {
  43.             buildResponse(channelHandlerContext, ApiRespMessage.builder().code(String.valueOf(StatusCode.SUCCESS.getCode()))
  44.                     .respTime(String.valueOf(System.currentTimeMillis()))
  45.                     .msgType(String.valueOf(MsgType.HEARTBEAT.getCode()))
  46.                     .contents("心跳测试，很高兴收到你的心跳包")
  47.                     .build());
  48.  
  49.             return;
  50.         }
  51.         log.info("接收到客户端发送的信息: {}", content);
  52.  
  53.         Long userIdForReq;
  54.         String msgType = "";
  55.         String contents = "";
  56.  
  57.         try {
  58.             ApiReqMessage apiReqMessage = JSON.parseObject(content, ApiReqMessage.class);
  59.             msgType = apiReqMessage.getMsgType();
  60.             contents = apiReqMessage.getContents();
  61.  
  62.  
  63.             userIdForReq = apiReqMessage.getUserId();
  64.             // 用户身份标识校验
  65.             if (null == userIdForReq || (long) userIdForReq <= 10000) {
  66.                 ApiRespMessage apiRespMessage = ApiRespMessage.builder().code(String.valueOf(StatusCode.SYSTEM_ERROR.getCode()))
  67.                         .respTime(String.valueOf(System.currentTimeMillis()))
  68.                         .contents("用户身份标识有误!")
  69.                         .msgType(String.valueOf(MsgType.SYSTEM.getCode()))
  70.                         .build();
  71.                 buildResponseAndClose(channelHandlerContext, apiRespMessage);
  72.                 return;
  73.             }
  74.  
  75.  
  76.             if (StringUtils.equals(msgType, String.valueOf(MsgType.CHAT.getCode()))) {
  77.                 // 对用户输入的内容进行自定义违规词检测
  78.                 // 对用户输入的内容进行第三方在线违规词检测
  79.                 // 对用户输入的内容进行组装成Prompt
  80.                 // 对Prompt根据业务进行增强（完善prompt的内容）
  81.                 // 对history进行裁剪或总结（检测history是否操作模型支持的上下文长度，例如qwen-7b支持的上下文长度为8192）
  82.                 // ...
  83.                 String messageId = apiReqMessage.getMessageId();
  84.                 List<ChatContext> history = apiReqMessage.getHistory();
  85.                 AITaskReqMessage aiTaskReqMessage = AITaskReqMessage.builder().messageId(messageId).userId(userIdForReq).contents(contents).history(history).build();
  86.                 taskExecuteUtils.execute(aiTaskReqMessage);
  87.  
  88.  
  89.             } else if (StringUtils.equals(msgType, String.valueOf(MsgType.INIT.getCode()))) {
  90.                 //一、业务黑名单检测（多次违规，永久锁定）
  91.  
  92.                 //二、账户锁定检测（临时锁定）
  93.  
  94.                 //三、多设备登录检测
  95.  
  96.                 //四、剩余对话次数检测
  97.  
  98.                 //检测通过，绑定用户与channel之间关系
  99.                 addChannel(channelHandlerContext, userIdForReq);
  100.                 String respMessage = "用户标识: " + userIdForReq + " 登录成功";
  101.  
  102.                 buildResponse(channelHandlerContext, ApiRespMessage.builder().code(String.valueOf(StatusCode.SUCCESS.getCode()))
  103.                         .respTime(String.valueOf(System.currentTimeMillis()))
  104.                         .msgType(String.valueOf(MsgType.INIT.getCode()))
  105.                         .contents(respMessage)
  106.                         .build());
  107.  
  108.             } else if (StringUtils.equals(msgType, String.valueOf(MsgType.HEARTBEAT.getCode()))) {
  109.  
  110.                 buildResponse(channelHandlerContext, ApiRespMessage.builder().code(String.valueOf(StatusCode.SUCCESS.getCode()))
  111.                         .respTime(String.valueOf(System.currentTimeMillis()))
  112.                         .msgType(String.valueOf(MsgType.HEARTBEAT.getCode()))
  113.                         .contents("心跳测试，很高兴收到你的心跳包")
  114.                         .build());
  115.             }
  116.             else {
  117.                 log.info("用户标识: {}, 消息类型有误，不支持类型: {}", userIdForReq, msgType);
  118.             }
  119.  
  120.  
  121.         } catch (Exception e) {
  122.             log.warn("【BusinessHandler】接收到请求内容：{}，异常信息：{}", content, e.getMessage(), e);
  123.             // 异常返回
  124.             return;
  125.         }
  126.  
  127.     }
  128.  
  129. }

6.3.  AIChatUtils完整代码

  1. import com.alibaba.fastjson.JSON;
  2. import lombok.extern.slf4j.Slf4j;
  3. import okhttp3.MediaType;
  4. import okhttp3.Request;
  5. import okhttp3.RequestBody;
  6. import okhttp3.Response;
  7. import org.apache.commons.lang3.StringUtils;
  8. import org.springframework.beans.factory.annotation.Autowired;
  9. import org.springframework.stereotype.Component;
  10.  
  11. import java.io.ByteArrayOutputStream;
  12. import java.io.InputStream;
  13. import java.nio.charset.StandardCharsets;
  14. import java.security.MessageDigest;
  15. import java.util.List;
  16. import java.util.Objects;
  17.  
  18. @Slf4j
  19. @Component
  20. public class AIChatUtils {
  21.     @Autowired
  22.     private AIConfig aiConfig;
  23.  
  24.     private Request buildRequest(Long userId, String prompt) throws Exception {
  25.         //创建一个请求体对象(body)
  26.         MediaType mediaType = MediaType.parse("application/json");
  27.         RequestBody requestBody = RequestBody.create(mediaType, prompt);
  28.  
  29.         return buildHeader(userId, new Request.Builder().post(requestBody))
  30.                 .url(aiConfig.getUrl()).build();
  31.     }
  32.  
  33.     private Request.Builder buildHeader(Long userId, Request.Builder builder) throws Exception {
  34.         return builder
  35.                 .addHeader("Content-Type", "application/json")
  36.                 .addHeader("userId", String.valueOf(userId))
  37.                 .addHeader("secret",generateSecret(userId))
  38.     }
  39.  
  40.  
  41.  
  42.     /**
  43.      * 生成请求密钥
  44.      *
  45.      * @param userId 用户ID
  46.      * @return
  47.      */
  48.     private String generateSecret(Long userId) throws Exception {
  49.         String key = aiConfig.getServerKey();
  50.         String content = key + userId + key;
  51.  
  52.         MessageDigest digest = MessageDigest.getInstance("SHA-256");
  53.         byte[] hash = digest.digest(content.getBytes(StandardCharsets.UTF_8));
  54.  
  55.         StringBuilder hexString = new StringBuilder();
  56.         for (byte b : hash) {
  57.             String hex = Integer.toHexString(0xff & b);
  58.             if (hex.length() == 1) {
  59.                 hexString.append('0');
  60.             }
  61.             hexString.append(hex);
  62.         }
  63.         return hexString.toString();
  64.     }
  65.  
  66.     public String chatStream(AITaskReqMessage aiTaskReqMessage) throws Exception {
  67.  
  68.         String messageId = aiTaskReqMessage.getMessageId();
  69.         Long userId = aiTaskReqMessage.getUserId();
  70.         String contents = aiTaskReqMessage.getContents();
  71.         List<ChatContext> history = aiTaskReqMessage.getHistory();
  72.  
  73.         if(StringUtils.isEmpty(contents) || StringUtils.isBlank(contents)){
  74.             log.warn("用户输入内容不能为空！");
  75.             return null;
  76.         }
  77.  
  78.         //定义请求的参数
  79.         String prompt = JSON.toJSONString(AIChatReqVO.init(contents, history));
  80.         log.info("【AIChatUtils】调用AI聊天，用户（{}），prompt：{}",  userId, prompt);
  81.  
  82.         //创建一个请求对象
  83.         Request request = buildRequest(userId, prompt);
  84.  
  85.         InputStream is = null;
  86.         try {
  87.  
  88.             // 从线程池获取http请求并执行
  89.             Response response =OkHttpUtils.getInstance(aiConfig).getOkHttpClient().newCall(request).execute();
  90.  
  91.             // 响应结果
  92.             StringBuffer resultBuff = new StringBuffer();
  93.             //正常返回
  94.             if (response.code() == 200) {
  95.                 //打印返回的字符数据
  96.                 is = response.body().byteStream();
  97.                 byte[] bytes = new byte[1024];
  98.  
  99.                 int len = is.read(bytes);
  100.                 while (len != -1) {
  101.                     ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
  102.                     outputStream.write(bytes, 0, len);
  103.                     outputStream.flush();
  104.                     // 本轮读取到的数据
  105.                     String result = new String(outputStream.toByteArray(), StandardCharsets.UTF_8);
  106.                     resultBuff.append(result);
  107.  
  108.                     len = is.read(bytes);
  109.  
  110.                     // 将数据逐个传输给用户
  111.                     AbstractBusinessLogicHandler.pushChatMessageForUser(userId, result);
  112.                 }
  113.  
  114.                 // 正常响应
  115.                 return resultBuff.toString();
  116.             }
  117.             else {
  118.                 String result = response.body().string();
  119.                 log.warn("处理异常，异常描述：{}",result);
  120.             }
  121.         } catch (Throwable e) {
  122.             log.error("【AIChatUtils】消息（{}）调用AI聊天 chatStream 异常，异常消息：{}", messageId, e.getMessage(), e);
  123.  
  124.         } finally {
  125.             if (!Objects.isNull(is)) {
  126.                 try {
  127.                     is.close();
  128.                 } catch (Exception e) {
  129.                     e.printStackTrace();
  130.                 }
  131.             }
  132.         }
  133.         return null;
  134.     }
  135.  
  136.  
  137. }

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/知新_RL/article/detail/511059
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号