• java模拟GPT流式问答

    流式请求gpt并且流式推送相关前端页面

    1）java流式获取gpt答案

    1、读取文件流的方式

    使用post请求数据，由于gpt是eventsource的方式返回数据，所以格式是data：，需要手动替换一下值

    1. /**
    2. org.apache.http.client.methods
    3. **/
    4. @SneakyThrows
    5.     private void chatStream(List messagesBOList) {
    6.         CloseableHttpClient httpclient = HttpClients.createDefault();
    7.         HttpPost httpPost = new HttpPost("https://api.openai.com/v1/chat/completions");
    8.         httpPost.setHeader("Authorization","xxxxxxxxxxxx");
    9.         httpPost.setHeader("Content-Type","application/json; charset=UTF-8");
    10.  
    11.         ChatParamBO build = ChatParamBO.builder()
    12.                 .temperature(0.7)
    13.                 .model("gpt-3.5-turbo")
    14.                 .messages(messagesBOList)
    15.                 .stream(true)
    16.                 .build();
    17.         System.out.println(JsonUtils.toJson(build));
    18.         httpPost.setEntity(new StringEntity(JsonUtils.toJson(build),"utf-8"));
    19.         CloseableHttpResponse response = httpclient.execute(httpPost);
    20.         try {
    21.             HttpEntity entity = response.getEntity();
    22.             if (entity != null) {
    23.                 InputStream inputStream = entity.getContent();
    24.                 BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream));
    25.  
    26.                 String line;
    27.                 while ((line = reader.readLine()) != null) {
    28.                     // 处理 event stream 数据
    29.  
    30.                     try {
    31. //                        System.out.println(line);
    32.                         ChatResultBO chatResultBO = JsonUtils.toObject(line.replace("data:", ""), ChatResultBO.class);
    33.                         String content = chatResultBO.getChoices().get(0).getDelta().getContent();
    34.                         log.info(content);
    35.                         
    36. //                        System.out.println(chatResultBO.getChoices().get(0).getMessage().getContent());
    37.                     } catch (Exception e) {
    38. //                        e.printStackTrace();
    39.                     }
    40.                 }
    41.             }
    42.         } finally {
    43.             response.close();
    44.         }
    45.     }

    2、sse链接的方式获取数据

    用到了okhttp

    需要先引用相关maven：

    1.         <dependency>
    2.             <groupId>com.squareup.okhttp3groupId>
    3.             <artifactId>okhttpartifactId>
    4.         dependency>
    5.         <dependency>
    6.             <groupId>com.squareup.okhttp3groupId>
    7.             <artifactId>okhttp-sseartifactId>
    8.         dependency>
    1.        
    2.         // 定义see接口
    3.         Request request = new Request.Builder().url("https://api.openai.com/v1/chat/completions")
    4.                 .header("Authorization","xxx")
    5.                 .post(okhttp3.RequestBody.create(okhttp3.MediaType.parse("application/json; charset=utf-8"),param.toJSONString()))
    6.                 .build();
    7.         OkHttpClient okHttpClient = new OkHttpClient.Builder()
    8.                 .connectTimeout(10, TimeUnit.MINUTES)
    9.                 .readTimeout(10, TimeUnit.MINUTES)//这边需要将超时显示设置长一点，不然刚连上就断开，之前以为调用方式错误被坑了半天
    10.                 .build();
    11.  
    12.         // 实例化EventSource，注册EventSource监听器
    13.         RealEventSource realEventSource = new RealEventSource(request, new EventSourceListener() {
    14.  
    15.             @Override
    16.             public void onOpen(EventSource eventSource, Response response) {
    17.                 log.info("onOpen");
    18.             }
    19.  
    20.             @SneakyThrows
    21.             @Override
    22.             public void onEvent(EventSource eventSource, String id, String type, String data) {
    23. //                log.info("onEvent");
    24.                 log.info(data);//请求到的数据
    25.             
    26.             }
    27.  
    28.             @Override
    29.             public void onClosed(EventSource eventSource) {
    30.                 log.info("onClosed");
    31. //                emitter.complete();
    32.             }
    33.  
    34.             @Override
    35.             public void onFailure(EventSource eventSource, Throwable t, Response response) {
    36.                 log.info("onFailure,t={},response={}",t,response);//这边可以监听并重新打开
    37. //                emitter.complete();
    38.             }
    39.         });
    40.         realEventSource.connect(okHttpClient);//真正开始请求的一步

    2）流式推送答案

    方法一：通过订阅式SSE/WebSocket

    原理是先建立链接，然后不断发消息就可以

    1、websocket

    创建相关配置：

    1.  
    2. import javax.websocket.Session;
    3.  
    4. import lombok.Data;
    5.  
    6. /**
    7.  * @description WebSocket客户端连接
    8.  */
    9. @Data
    10. public class WebSocketClient {
    11.  
    12.     // 与某个客户端的连接会话，需要通过它来给客户端发送数据
    13.     private Session session;
    14.  
    15.     //连接的uri
    16.     private String uri;
    17.  
    18. }
    1.  
    2.  
    3. import org.springframework.context.annotation.Bean;
    4. import org.springframework.context.annotation.Configuration;
    5. import org.springframework.web.socket.server.standard.ServerEndpointExporter;
    6.  
    7. @Configuration
    8. public class WebSocketConfig {
    9.     @Bean
    10.     public ServerEndpointExporter serverEndpointExporter() {
    11.         return new ServerEndpointExporter();
    12.     }
    13. }
    配置相关service
    1.  
    2.  
    3. @Slf4j
    4. @Component
    5. @ServerEndpoint("/websocket/chat/{chatId}")
    6. public class ChatWebsocketService {
    7.  
    8.     static final ConcurrentHashMap> webSocketClientMap= new ConcurrentHashMap<>();
    9.  
    10.     private String chatId;
    11.  
    12.     /**
    13.      * 连接建立成功时触发，绑定参数
    14.      * @param session 与某个客户端的连接会话，需要通过它来给客户端发送数据
    15.      * @param chatId 商户ID
    16.      */
    17.     @OnOpen
    18.     public void onOpen(Session session, @PathParam("chatId") String chatId){
    19.  
    20.         WebSocketClient client = new WebSocketClient();
    21.         client.setSession(session);
    22.         client.setUri(session.getRequestURI().toString());
    23.  
    24.         List webSocketClientList = webSocketClientMap.get(chatId);
    25.         if(webSocketClientList == null){
    26.             webSocketClientList = new ArrayList<>();
    27.         }
    28.         webSocketClientList.add(client);
    29.         webSocketClientMap.put(chatId, webSocketClientList);
    30.         this.chatId = chatId;
    31.     }
    32.  
    33.     /**
    34.      * 收到客户端消息后调用的方法
    35.      *
    36.      * @param message 客户端发送过来的消息
    37.      */
    38.     @OnMessage
    39.     public void onMessage(String message) {
    40.         log.info("chatId = {},message = {}",chatId,message);
    41.         // 回复消息
    42.         this.chatStream(BaseUtil.newList(ChatParamMessagesBO.builder().content(message).role("user").build()));
    43. //        this.sendMessage(chatId,message+"233");
    44.     }
    45.  
    46.     /**
    47.      * 连接关闭时触发，注意不能向客户端发送消息了
    48.      * @param chatId
    49.      */
    50.     @OnClose
    51.     public void onClose(@PathParam("chatId") String chatId){
    52.         webSocketClientMap.remove(chatId);
    53.     }
    54.  
    55.     /**
    56.      * 通信发生错误时触发
    57.      * @param session
    58.      * @param error
    59.      */
    60.     @OnError
    61.     public void onError(Session session, Throwable error) {
    62.         System.out.println("发生错误");
    63.         error.printStackTrace();
    64.     }
    65.  
    66.     /**
    67.      * 向客户端发送消息
    68.      * @param chatId
    69.      * @param message
    70.      */
    71.     public void sendMessage(String chatId,String message){
    72.         try {
    73.             List webSocketClientList = webSocketClientMap.get(chatId);
    74.             if(webSocketClientList!=null){
    75.                 for(WebSocketClient webSocketServer:webSocketClientList){
    76.                     webSocketServer.getSession().getBasicRemote().sendText(message);
    77.                 }
    78.             }
    79.         } catch (IOException e) {
    80.             e.printStackTrace();
    81.             throw new RuntimeException(e.getMessage());
    82.         }
    83.     }
    84.  
    85.     /**
    86.      * 流式调用查询gpt
    87.      * @param messagesBOList
    88.      * @throws IOException
    89.      */
    90.     @SneakyThrows
    91.     private void chatStream(List messagesBOList) {
    92.        // TODO 和GPT的访问请求
    93.     }
    94. }
    测试，postman建立链接

    2、SSE

    本质也是基于订阅推送方式

    前端：
    1. html>
    2. <html lang="en">
    3.  
    4. <head>
    5.     <meta charset="UTF-8">
    6.     <title>SseEmittertitle>
    7. head>
    8.  
    9. <body>
    10. <button onclick="closeSse()">关闭连接button>
    11. <div id="message">div>
    12. body>
    13. <script>
    14.     let source = null;
    15.  
    16.     // 用时间戳模拟登录用户
    17.     //const id = new Date().getTime();
    18. 		const id = '7829083B42464C5B9C445A087E873C7D';
    19.    
    20.     if (window.EventSource) {
    21.  
    22.         // 建立连接
    23. 	
    24.          source = new EventSource('http://172.28.54.27:8902/api/sse/connect?conversationId=' + id);
    25. 		setMessageInnerHTML("连接用户=" + id);
    26.         /**
    27.          * 连接一旦建立，就会触发open事件
    28.          * 另一种写法：source.onopen = function (event) {}
    29.          */
    30.         source.addEventListener('open', function(e) {
    31.             setMessageInnerHTML("建立连接。。。");
    32.         }, false);
    33.  
    34.         /**
    35.          * 客户端收到服务器发来的数据
    36.          * 另一种写法：source.onmessage = function (event) {}
    37.          */
    38.         source.addEventListener('message', function(e) {
    39. 			//console.log(e);
    40.             setMessageInnerHTML(e.data);
    41. 			
    42.         });
    43.  
    44. 		source.addEventListener("close", function (event) {
    45. 			// 在这里处理关闭事件
    46. 			console.log("Server closed the connection");
    47. 			
    48. 			// 可以选择关闭EventSource连接
    49. 			source.close();
    50. 		});
    51.  
    52.         /**
    53.          * 如果发生通信错误（比如连接中断），就会触发error事件
    54.          * 或者：
    55.          * 另一种写法：source.onerror = function (event) {}
    56.          */
    57.         source.addEventListener('error', function(e) {
    58. 			console.log(e);
    59.             if (e.readyState === EventSource.CLOSED) {
    60.                 setMessageInnerHTML("连接关闭");
    61.             } else {
    62.                 console.log(e);
    63.             }
    64.         }, false);
    65.  
    66.     } else {
    67.         setMessageInnerHTML("你的浏览器不支持SSE");
    68.     }
    69.  
    70.     // 监听窗口关闭事件，主动去关闭sse连接，如果服务端设置永不过期，浏览器关闭后手动清理服务端数据
    71.     window.onbeforeunload = function() {
    72.         //closeSse();
    73.     };
    74.  
    75.     // 关闭Sse连接
    76.     function closeSse() {
    77.         source.close();
    78.         const httpRequest = new XMLHttpRequest();
    79.         httpRequest.open('GET', 'http://172.28.54.27:8902/api/sse/disconnection?conversationId=' + id, true);
    80.         httpRequest.send();
    81.         console.log("close");
    82.     }
    83.  
    84.     // 将消息显示在网页上
    85.     function setMessageInnerHTML(innerHTML) {
    86.         document.getElementById('message').innerHTML += innerHTML + '
      '      ;
    87.     }
    88. script>
    89.  
    90. html>
    后端：
    controller
    1.  
    2.  
    3. import org.springframework.cloud.context.config.annotation.RefreshScope;
    4. import org.springframework.validation.annotation.Validated;
    5. import org.springframework.web.bind.annotation.GetMapping;
    6. import org.springframework.web.bind.annotation.RequestMapping;
    7. import org.springframework.web.bind.annotation.RestController;
    8. import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
    9.  
    10. import java.util.Set;
    11. import java.util.function.Consumer;
    12.  
    13. import javax.annotation.Resource;
    14.  
    15. import lombok.SneakyThrows;
    16. import lombok.extern.slf4j.Slf4j;
    17.  
    18. @Validated
    19. @RestController
    20. @RequestMapping("/api/sse")
    21. @Slf4j
    22. @RefreshScope  // 会监听变化实时变化值
    23. public class SseController {
    24.  
    25.     @Resource
    26.     private SseBizService sseBizService;
    27.  
    28.  
    29.     /**
    30.      * 创建用户连接并返回 SseEmitter
    31.      *
    32.      * @param conversationId 用户ID
    33.      * @return SseEmitter
    34.      */
    35.     @SneakyThrows
    36.     @GetMapping(value = "/connect", produces = "text/event-stream; charset=utf-8")
    37.     public SseEmitter connect(String conversationId) {
    38.         // 设置超时时间，0表示不过期。默认30秒，超过时间未完成会抛出异常：AsyncRequestTimeoutException
    39.         SseEmitter sseEmitter = new SseEmitter(0L);
    40.         // 注册回调
    41.         sseEmitter.onCompletion(completionCallBack(conversationId));
    42.         sseEmitter.onError(errorCallBack(conversationId));
    43.         sseEmitter.onTimeout(timeoutCallBack(conversationId));
    44.         log.info("创建新的sse连接，当前用户：{}", conversationId);
    45.         sseBizService.addConnect(conversationId,sseEmitter);
    46.         sseBizService.sendMsg(conversationId,"链接成功");
    47. //        sseCache.get(conversationId).send(SseEmitter.event().reconnectTime(10000).data("链接成功"),MediaType.TEXT_EVENT_STREAM);
    48.         return sseEmitter;
    49.     }
    50.  
    51.     /**
    52.      * 给指定用户发送信息  -- 单播
    53.      */
    54.     @GetMapping(value = "/send", produces = "text/event-stream; charset=utf-8")
    55.     public void sendMessage(String conversationId, String msg) {
    56.         sseBizService.sendMsg(conversationId,msg);
    57.     }
    58.  
    59.     /**
    60.      * 移除用户连接
    61.      */
    62.     @GetMapping(value = "/disconnection", produces = "text/event-stream; charset=utf-8")
    63.     public void removeUser(String conversationId) {
    64.         log.info("移除用户：{}", conversationId);
    65.         sseBizService.deleteConnect(conversationId);
    66.     }
    67.  
    68.     /**
    69.      * 向多人发布消息   -- 组播
    70.      * @param groupId 开头标识
    71.      * @param message 消息内容
    72.      */
    73.     public void groupSendMessage(String groupId, String message) {
    74.        /* if (!BaseUtil.isNullOrEmpty(sseCache)) {
    75.             *//*Set ids = sseEmitterMap.keySet().stream().filter(m -> m.startsWith(groupId)).collect(Collectors.toSet());
    76.             batchSendMessage(message, ids);*//*
    77.             sseCache.forEach((k, v) -> {
    78.                 try {
    79.                     if (k.startsWith(groupId)) {
    80.                         v.send(message, MediaType.APPLICATION_JSON);
    81.                     }
    82.                 } catch (IOException e) {
    83.                     log.error("用户[{}]推送异常:{}", k, e.getMessage());
    84.                     removeUser(k);
    85.                 }
    86.             });
    87.         }*/
    88.     }
    89.  
    90.     /**
    91.      * 群发所有人   -- 广播
    92.      */
    93.     public void batchSendMessage(String message) {
    94.         /*sseCache.forEach((k, v) -> {
    95.             try {
    96.                 v.send(message, MediaType.APPLICATION_JSON);
    97.             } catch (IOException e) {
    98.                 log.error("用户[{}]推送异常:{}", k, e.getMessage());
    99.                 removeUser(k);
    100.             }
    101.         });*/
    102.     }
    103.  
    104.     /**
    105.      * 群发消息
    106.      */
    107.     public void batchSendMessage(String message, Set ids) {
    108.         ids.forEach(userId -> sendMessage(userId, message));
    109.     }
    110.  
    111.  
    112.     /**
    113.      * 获取当前连接信息
    114.      */
    115. //    public List getIds() {
    116. //        return new ArrayList<>(sseCache.keySet());
    117. //    }
    118.  
    119.     /**
    120.      * 获取当前连接数量
    121.      */
    122. //    public int getUserCount() {
    123. //        return count.intValue();
    124. //    }
    125.  
    126.     private Runnable completionCallBack(String userId) {
    127.         return () -> {
    128.             log.info("结束连接：{}", userId);
    129.             removeUser(userId);
    130.         };
    131.     }
    132.  
    133.     private Runnable timeoutCallBack(String userId) {
    134.         return () -> {
    135.             log.info("连接超时：{}", userId);
    136.             removeUser(userId);
    137.         };
    138.     }
    139.  
    140.     private Consumer errorCallBack(String userId) {
    141.         return throwable -> {
    142.             log.info("连接异常：{}", userId);
    143.             removeUser(userId);
    144.         };
    145.     }
    146. }
    service
    1.  
    2.  
    3. import org.springframework.cloud.context.config.annotation.RefreshScope;
    4. import org.springframework.http.MediaType;
    5. import org.springframework.stereotype.Component;
    6. import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
    7.  
    8. import java.util.Map;
    9. import java.util.concurrent.ConcurrentHashMap;
    10. import java.util.concurrent.atomic.AtomicInteger;
    11.  
    12. import lombok.SneakyThrows;
    13. import lombok.extern.slf4j.Slf4j;
    14.  
    15. @Component
    16. @Slf4j
    17. @RefreshScope  // 会监听变化实时变化值
    18. public class SseBizService {
    19.     /**
    20.      * 
    21.      * 当前连接数
    22.      */
    23.     private AtomicInteger count = new AtomicInteger(0);
    24.  
    25.     /**
    26.      * 使用map对象，便于根据userId来获取对应的SseEmitter，或者放redis里面
    27.      */
    28.     private Map sseCache = new ConcurrentHashMap<>();
    29.  
    30.  
    31.     /**
    32.      * 添加用户
    33.      * @author pengbin 
    34.      * @date 2023/9/11 11:37
    35.      * @param
    36.      * @return
    37.      */
    38.     public void addConnect(String id,SseEmitter sseEmitter){
    39.         sseCache.put(id, sseEmitter);
    40.         // 数量+1
    41.         count.getAndIncrement();
    42.     }
    43.     /**
    44.      * 删除用户
    45.      * @author pengbin 
    46.      * @date 2023/9/11 11:37
    47.      * @param
    48.      * @return
    49.      */
    50.     public void deleteConnect(String id){
    51.         sseCache.remove(id);
    52.         // 数量+1
    53.         count.getAndDecrement();
    54.     }
    55.  
    56.     /**
    57.      * 发送消息
    58.      * @author pengbin 
    59.      * @date 2023/9/11 11:38
    60.      * @param
    61.      * @return
    62.      */
    63.     @SneakyThrows
    64.     public void sendMsg(String id, String msg){
    65.         if(sseCache.containsKey(id)){
    66.             sseCache.get(id).send(msg, MediaType.TEXT_EVENT_STREAM);
    67.         }
    68.     }
    69.  
    70. }

    方法二：SSE建立eventSource，使用完成后即刻销毁

    前端：在接收到结束标识后立即销毁

    1. /**
    2.          * 客户端收到服务器发来的数据
    3.          * 另一种写法：source.onmessage = function (event) {}
    4.          */
    5.         source.addEventListener('message', function(e) {
    6. 			//console.log(e);
    7.             setMessageInnerHTML(e.data);
    8. 			if(e.data == '[DONE]'){
    9. 				source.close();
    10. 			}
    11.         });

    后端：
     

    1. @SneakyThrows
    2.     @GetMapping(value = "/stream/sse", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    3.     public SseEmitter completionsStream(@RequestParam String conversationId){
    4.         //
    5.         List messagesBOList =new ArrayList();
    6.  
    7.         // 获取内容信息
    8.         ChatParamBO build = ChatParamBO.builder()
    9.                 .temperature(0.7)
    10.                 .stream(true)
    11.                 .model("xxxx")
    12.                 .messages(messagesBOList)
    13.                 .build();
    14.  
    15.         SseEmitter emitter = new SseEmitter();
    16.             
    17.         // 定义see接口
    18.         Request request = new Request.Builder().url("xxx")
    19.                 .header("Authorization","xxxx")
    20.                 .post(okhttp3.RequestBody.create(okhttp3.MediaType.parse("application/json; charset=utf-8"),JsonUtils.toJson(build)))
    21.                 .build();
    22.         OkHttpClient okHttpClient = new OkHttpClient.Builder()
    23.                 .connectTimeout(10, TimeUnit.MINUTES)
    24.                 .readTimeout(10, TimeUnit.MINUTES)//这边需要将超时显示设置长一点，不然刚连上就断开，之前以为调用方式错误被坑了半天
    25.                 .build();
    26.  
    27.         StringBuffer sb = new StringBuffer("");
    28.  
    29.         // 实例化EventSource，注册EventSource监听器
    30.         RealEventSource realEventSource = null;
    31.         realEventSource = new RealEventSource(request, new EventSourceListener() {
    32.  
    33.             @Override
    34.             public void onOpen(EventSource eventSource, Response response) {
    35.                 log.info("onOpen");
    36.             }
    37.  
    38.             @SneakyThrows
    39.             @Override
    40.             public void onEvent(EventSource eventSource, String id, String type, String data) {
    41.  
    42.                 log.info(data);//请求到的数据
    43.                 try {
    44.  
    45.                     ChatResultBO chatResultBO = JsonUtils.toObject(data.replace("data:", ""), ChatResultBO.class);
    46.                     String content = chatResultBO.getChoices().get(0).getDelta().getContent();
    47.                     sb.append(content);
    48.                     emitter.send(SseEmitter.event().data(JsonUtils.toJson(ChatContentBO.builder().content(content).build())));
    49.  
    50.                 } catch (Exception e) {
    51. //                        e.printStackTrace();
    52.                 }
    53.                 if("[DONE]".equals(data)){
    54.                     emitter.send(SseEmitter.event().data(data));
    55.                     emitter.complete();
    56.                     log.info("result={}",sb);
    57.                 }
    58.             }
    59.  
    60.             @Override
    61.             public void onClosed(EventSource eventSource) {
    62.                 log.info("onClosed,eventSource={}",eventSource);//这边可以监听并重新打开
    63. //                emitter.complete();
    64.             }
    65.  
    66.             @Override
    67.             public void onFailure(EventSource eventSource, Throwable t, Response response) {
    68.                 log.info("onFailure,t={},response={}",t,response);//这边可以监听并重新打开
    69. //                emitter.complete();
    70.             }
    71.         });
    72.         realEventSource.connect(okHttpClient);//真正开始请求的一步
    73.         return emitter;
    74.     }

    3）踩坑

    ngnix配置：

    后端配置需要添加：

     #gpt支持流式处理
      proxy_buffering off;

    1. location / {
    2.             proxy_pass http://backend;
    3.             proxy_redirect  default;
    4. 	    proxy_connect_timeout 90;
    5. 	    proxy_read_timeout 90;
    6. 	    proxy_send_timeout 90;
    7.     	    #gpt支持流式处理
    8.     	    proxy_buffering off;
    9.             #root   html;
    10.             #root /opt/project/;
    11.             index  index.html index.htm;
    12. 	    client_max_body_size 1024m;
    13. 			#设置正确的外网ip
    14.             proxy_set_header Host $host;
    15.             proxy_set_header X-Real-IP $remote_addr;
    16.             proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    17.         }

  • 相关阅读:
    动漫主题dreamweaver作业静态HTML网页设计——仿京东(海贼王)版本
    NPDP怎么报名？考试难度大吗？
    网安学习-应急响应3
    千年TGS服务器日志报错如何解决
    【100个 Unity实用技能】☀️ | Unity 将秒数转化为00:00:00时间格式
    4种方法教你如何查看java对象所占内存大小
    【机器学习】面试题：LSTM长短期记忆网络的理解？LSTM是怎么解决梯度消失的问题的？还有哪些其它的解决梯度消失或梯度爆炸的方法？
    redis的持久化
    【如何】求 矩阵 的 若尔当标准形J，求 初等因子
    桌面运维命令
  • 原文地址：https://blog.csdn.net/pengbin790000/article/details/133684395