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

java模拟GPT流式问答_realeventsource

作者：人工智能uu | 2024-08-11 07:11:08

赞

踩

realeventsource

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

1）java流式获取gpt答案

1、读取文件流的方式

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

  1. /**
  2. org.apache.http.client.methods
  3. **/
  4. @SneakyThrows
  5.     private void chatStream(List<ChatParamMessagesBO> 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.okhttp3</groupId>
  3.             <artifactId>okhttp</artifactId>
  4.         </dependency>
  5.         <dependency>
  6.             <groupId>com.squareup.okhttp3</groupId>
  7.             <artifactId>okhttp-sse</artifactId>
  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<String, List<WebSocketClient>> 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<WebSocketClient> 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<WebSocketClient> 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<ChatParamMessagesBO> messagesBOList) {
  92.        // TODO 和GPT的访问请求
  93.     }
  94. }
测试，postman建立链接

2、SSE

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

前端：
  1. <!DOCTYPE html>
  2. <html lang="en">
  3.  
  4. <head>
  5.     <meta charset="UTF-8">
  6.     <title>SseEmitter</title>
  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 + '<br/>';
  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<String> 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<String> ids) {
  108.         ids.forEach(userId -> sendMessage(userId, message));
  109.     }
  110.  
  111.  
  112.     /**
  113.      * 获取当前连接信息
  114.      */
  115. //    public List<String> 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<Throwable> 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<String, SseEmitter> sseCache = new ConcurrentHashMap<>();
  29.  
  30.  
  31.     /**
  32.      * 添加用户
  33.      * @author pengbin <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 <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 <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<ChatParamMessagesBO> 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.         }

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

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

  

闽ICP备14008679号