2023java使用WebClient实现chatGPT调用建立web socket连接

??目录

一、介绍
- 1、使用技术
- 2、效果
二、代码
- 1、前端代码
- 2、后端代码
- - 2.1、maven依赖
  - 2.2、model
  - - 2.2.1、请求接口的格式
    - 2.2.2、响应数据对象
  - 2.3、工具类
  - - 2.3.1、??使用WebClient调用`chatgpt`方法
    - 2.3.2、?? webSocket连接对话方法
  - 2.4、Controller

一、介绍

通过java实现对chatGPT的API接口实现websocket流式输出以及接口调用两种方式代码

1、使用技术

使用到的技术包括WebClient、webSocket加thymeleaf

WebClient：客户端的使用可以开 ??java http客户端
webSocket：可以看??webSokcet 使用
thymeleaf：可以网上直接找教程，我使用的比较简单会掉用就行，或者使用html静态页面也可以

2、效果

websocket的具体实现效果如下，非专业前端，页面比较简陋，可以自行优化页面
接口调用

二、代码

1、前端代码

<!DOCTYPE html>
<html>

<head>
    <meta charset="utf-8">
    <title>Java后端WebSocket的Tomcat实现</title>
    <script type="text/javascript" src="js/jquery.min.js"></script>
</head>

<body>

<strong>ChatGPT使用</strong>
    <br />
    <input id="text" type="text" onkeydown="checkEnter(event)"/>
<button onclick="send()">发送消息</button>
<hr />
<button onclick="closeWebSocket()">关闭WebSocket连接</button>
<hr />
<div id="message"></div>
</body>
<script type="text/javascript">
    var websocket = null;
    //判断当前浏览器是否支持WebSocket
    if ('WebSocket' in window) {
        //改成你的地址
        websocket = new WebSocket("ws://localhost:8088/websocket/100");
    } else {
        alert('当前浏览器 Not support websocket')
    }

    //连接发生错误的回调方法
    websocket.onerror = function () {
        setMessageInnerHTML("WebSocket连接发生错误");
    };

    //连接成功建立的回调方法
    websocket.onopen = function () {
        setMessageInnerHTML("WebSocket连接成功");
    }
    var U01data, Uidata, Usdata
    //接收到消息的回调方法
    websocket.onmessage = function (event) {
        console.log(event);
        if (event.data !== "conn_success") {
            setMessageInnerHTML(event.data);
            // setMessageInnerHTML(event);
            setechart()

        }
    }

    //连接关闭的回调方法
    websocket.onclose = function () {
        setMessageInnerHTML("WebSocket连接关闭");
    }

    // //监听窗口关闭事件，当窗口关闭时，主动去关闭websocket连接，防止连接还没断开就关闭窗口，server端会抛异常。
    window.onbeforeunload = function () {
        closeWebSocket();
    }

    //将消息显示在网页上
    function setMessageInnerHTML(innerHTML) {
        document.getElementById('message').innerHTML += innerHTML ;
    }

    //关闭WebSocket连接
    function closeWebSocket() {
        websocket.close();
    }

    //发送消息
    function send() {
        var message = document.getElementById('text').value;
        websocket.send('{"msg":"' + message + '"}');
        setMessageInnerHTML("<br>--------------发送消息：" + message + "<br>");
        document.getElementById('text').value = null;
    }

    function checkEnter(event) {
        if (event.keyCode === 13) {
            send();
        }
    }
</script>
</html>

2、后端代码

2.1、maven依赖

       <!--thymeleaf模版的--> 
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-thymeleaf</artifactId>
            <!--websocket的--> 
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-websocket</artifactId>
        </dependency>
         <!--webClient-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-webflux</artifactId>
        </dependency>       

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
         <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
         <dependency>
            <groupId>cn.hutool</groupId>
            <artifactId>hutool-all</artifactId>
            <version>5.8.22</version>
        </dependency>

2.2、model

2.2.1、请求接口的格式

如果想简单的话也可以不用封装对象，直接使用json
在这里插入图片描述

ChatGptRequestParameter

@Data
@NoArgsConstructor
@AllArgsConstructor
public class ChatGptRequestParameter {
    private String model = "gpt-3.5-turbo-16k-0613";
//     是否支持流式输出
    private boolean stream = true;

//    请求对象数组
    List<ChatGptMessage> messages=new ArrayList();

    public void addMessages(ChatGptMessage message) {
        this.messages.add(message);
    }
}

ChatGptMessage

@Data
@NoArgsConstructor
@AllArgsConstructor
public class ChatGptMessage implements Serializable {

    String role;
    String content;
}

2.2.2、响应数据对象

在这里插入图片描述

ChatGptResponseParameter

@Data
@NoArgsConstructor
@AllArgsConstructor
public class ChatGptResponseParameter implements Serializable {
    String id;
    String object;
    String created;
    String model;
    Usage usage;
    List<Choices> choices;
    String system_fingerprint;
}

Choices

@Data
@NoArgsConstructor
@AllArgsConstructor
public class Choices implements Serializable {
    ChatGptMessage delta;
    String finish_reason;
    Integer index;
    String logprobs;
}

ChatGptMessage

@Data
@NoArgsConstructor
@AllArgsConstructor
public class ChatGptMessage implements Serializable {
    String role;
    String content;
}

2.3、工具类

2.3.1、??使用WebClient调用chatgpt方法

@Component
@Slf4j
public class WebChatGPT {

    /**
     * 自己chatGpt的ApiKey
     */
    private String apiKey = "sk-***";

    private ChatGptRequestParameter chatGptRequestParameter = new ChatGptRequestParameter();


    /**
     * 推送
     * @param session webSocket会话对象
     * @param str 请求数据
     * @return Flux<String>
     */
    public Flux<String> getAnswer(Session session, String str) {


        ChatGptMessage chatGptMessage = new ChatGptMessage("user", str);
        //        保存消息，用于记录前面的消息，方便上下文记忆，因为3.5不能自动实现上下文记忆
        chatGptRequestParameter.addMessages(chatGptMessage);

        return webClient().post()
                .accept(MediaType.TEXT_EVENT_STREAM) //接收text/event-stream流的数据
                .body(BodyInserters.fromValue(chatGptRequestParameter)) //参数
                .retrieve() //执行请求并获取响应结果
                .bodyToFlux(String.class) // 将响应体转换为 Flux 类型，这里是将 SSE 流转换为字符串类型。
                .map(s -> {  //对每个响应元素进行处理
                    log.info("Gpt输出：{}", s);
                    if (!Objects.equals(s, "[DONE]")) {

                        ChatGptMessage message = JSONUtil.toBean(s, ChatGptResponseParameter.class).getChoices().get(0).getDelta();
                        String content = message.getContent();
                        if (content != null) {
                            try {
                                if (session != null) {
//                                通过websocket回写到页面
                                    session.getBasicRemote().sendText(content);
                                }

                                log.info("Gpt输出：{}", s);

                            } catch (IOException e) {
                                e.printStackTrace();
                            }
                            return content;
                        }
                    }
                    return "";
                })
                .onErrorResume(WebClientResponseException.class, ex -> Flux.just(ex.getResponseBodyAsString())) //请求失败
                .doFinally(signalType -> log.info("完成{}", signalType));  //在 Flux 完成时执行，无论是成功还是错误，都会打印日志表示请求完成。
              
    }

    private WebClient webClient(){
        return  WebClient.builder()
//                .clientConnector(new ReactorClientHttpConnector(
//                        HttpClient.create().proxy(proxy -> proxy.type(ProxyProvider.Proxy.HTTP).host("127.0.0.1").port(1080)) //代理
//                ))
                .defaultHeader(HttpHeaders.ACCEPT, MediaType.APPLICATION_JSON_VALUE)
                .defaultHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE + ";charset=UTF-8")
                .defaultHeader(HttpHeaders.AUTHORIZATION, "Bearer " + apiKey)
                .baseUrl("https://api.openai.com/v1/chat/completions") //官方api请求地址
                .build();
    }
}

2.3.2、?? webSocket连接对话方法

通过该方法和html页面建立websocket连接

@Slf4j
@Service
@ServerEndpoint("/websocket/{uid}")
@Component
public class WebSocketServer2 {

    //连接建立时长
    private static final long sessionTimeout = 600000;

    // 用来存放每个客户端对应的WebSocketServer对象
    private static Map<String, WebSocketServer2> webSocketMap = new ConcurrentHashMap<>();

    // 与某个客户端的连接会话，需要通过它来给客户端发送数据
    private Session session;

    // 接收id
    private String uid;

    private static WebChatGPT webChatGPT;


    static {
         webChatGPT = SpringUtil.getBean(WebChatGPT.class);
    }

    /**
     * 连接建立成功调用的方法
     * @author zhengfuping
     * @date 2023/8/22
     * @param session
     * @param uid
     */
    @OnOpen
    public void onOpen(Session session , @PathParam("uid") String uid){
        session.setMaxIdleTimeout(sessionTimeout);
        this.session = session;
        this.uid = uid;
        if (webSocketMap.containsKey(uid)){
            webSocketMap.remove(uid);
        }
        webSocketMap.put(uid,this);
        log.info("websocket连接成功编号uid: " + uid + "，当前在线数: " + getOnlineClients());

        try{
            // 响应客户端实际业务数据！
            sendMessage("conn_success");
        }catch (Exception e){
            log.error("websocket发送连接成功错误编号uid: " + uid + "，网络异常!!!");
        }
    }

    /**
     * 连接关闭调用的方法
     * @author zhengfuping
     * @date 2023/8/22
     */
    @OnClose
    public void onClose(){
        try {
            if (webSocketMap.containsKey(uid)){
                webSocketMap.remove(uid);
            }
            log.info("websocket退出编号uid: " + uid + "，当前在线数为: " + getOnlineClients());
        } catch (Exception e) {
            log.error("websocket编号uid连接关闭错误: " + uid + "，原因: " + e.getMessage());
        }
    }


    /**
     * 收到客户端消息后调用chatGpt的方法
     * @param message 客户端发送过来的消息
     * @param session
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        try {

            JSON parse = JSONUtil.parse(message);
            String msg = parse.getByPath("msg").toString();

            if (StrUtil.isNotEmpty(msg)){
//                调用
                Flux<String> answer = webChatGPT.getAnswer(session, msg);

//                 触发实际的请求操作
                answer.subscribe();

            }

        } catch (Exception e) {
            log.error("websocket发送消息失败编号uid为: " + uid + ",报文: " + message);
        }

    }

    /**
     * 发生错误时调用
     * @param session
     * @param error
     */
    @OnError
    public void onError(Session session, Throwable error) {
        log.error("websocket编号uid错误: " + this.uid + "原因: " + error.getMessage());
        error.printStackTrace();
    }

    /**
     * 实现服务器主动推送
     * @author yingfeng
     * @date 2023/8/22 10:11
     * @Param * @param null
     * @return
     */

    public void sendMessage(String message) throws IOException {
        this.session.getBasicRemote().sendText(message);
    }

    /**
     * 获取客户端在线数
     * @author zhengfuping
     * @date 2023/8/22 10:11
     * @param
     */
    public static synchronized int getOnlineClients() {
        if (Objects.isNull(webSocketMap)) {
            return 0;
        } else {
            return webSocketMap.size();
        }
    }
 }

2.4、Controller

跳转thymeleaf页的接口

@Controller
public class ChatWeb {


    @RequestMapping("/webSocket")
    public String webSocket(){
        return "webSocket";
    }
}

ChatGptController：页面请求访问

@RestController
@RequestMapping("/chat")
public class ChatGptController {
    @Resource
    private WebChatGPT webChatGPT;


    @GetMapping(value ="/stringFlux", produces = "application/json; charset=utf-8")
    public Flux<String> stringFlux(String c) {
        Flux<String> flux = webChatGPT.getAnswer(null,c);
        return flux;
    }
}