Java使用WebFlux调用大模型实现智能对话

     <dependency>
            <groupid>org.springframework.boot</groupid>
            <artifactid>spring-boot-starter-webflux</artifactid>
            <exclusions>
                <exclusion>
                    <groupid>org.springframework.boot</groupid>
                    <artifactid>spring-boot-starter-tomcat</artifactid>
                </exclusion>
            </exclusions>

        </dependency>
         <dependency>
            <groupid>org.projectlombok</groupid>
            <artifactid>lombok</artifactid>
        </dependency>

@data
public class aipromptdto {
    /**
     * 大模型id
     */
    private string serviceid;
    /**
     * 用户输入
     */
    private string userinput;
    /**
     * sessionid
     */
    private string sessionid;
    /**
     * 请求id
     */
    private string requestid;
    /**
     * 获取token
     */
    private string token;
    }

@data
@allargsconstructor
public class answerchunk {
    /**
     * 返回的内容
     */
    private string content;
    private string sessionid;
}

reactor:
  netty:
    response:
      buffer-size: 0

public interface aiservice {
    /**
     * 根据请求获取流式返回的答案
     * @param request
     * @return
     */
    flux<answerchunk>  processstream(aipromptdto request);
    }

import org.springframework.web.reactive.function.client.webclient;
@service
public class aiserviceimpl implements aiservice {
    private final webclient webclient;
    //初始化webclient，并ssl校验，生产环境不要跳过
    public aiserviceimpl(webclient.builder webclientbuilder) {
        // 使用insecuretrustmanagerfactory来信任所有证书
        sslcontextbuilder sslcontextbuilder = sslcontextbuilder.forclient()
                .trustmanager(insecuretrustmanagerfactory.instance);
        httpclient httpclient = httpclient.create()
                .secure(sslcontextspec -> sslcontextspec.sslcontext(sslcontextbuilder))
                .responsetimeout(duration.ofminutes(timeout));
        this.webclient = webclientbuilder.clientconnector(
                new reactorclienthttpconnector(httpclient)
        ).baseurl(aiforceurl).build();
    }
  @override
    public flux<answerchunk> processstream(aipromptdto request) {
   string body = jsonutil.tojsonstr(request);//参数都转化为json字符串
   return webclient.post()
                    .uri(aiforceurl + "/aiforceplatformapi/openapi/llm/debugsse")//大模型地址
                    .bodyvalue(body)//body参数
                    .header("token", request.gettoken())//设置请求头
                    .header("content-type", "application/json")
                    .retrieve()//retrieve 方法会从服务器响应中提取数据
                    .bodytoflux(string.class)//响应体解析为一个流式的 string 类型序列
                    .map(chunk -> {//解析数据以供存储
                        //system.out.println("chunk = " + chunk);
                        string content = "";
                        // 解析大模型返回数据
                        if (!chunk.contains("[done]")) {//结束标志
                            if (chunk.contains("inquirylist")) {//处理返回的关联查询列表
                                content = parsechunk(chunk);
                                finalanswer[0].setquerylist(content);
                            }else if (chunk.contains("messageid")&&chunk.contains("messagecontent")) {//处理返回提示message
                                parsemessage(chunk, messagemap);
                            }  else if (chunk.contains("data")) {//处理返回的问题答案
                                content = parsechunk(chunk);
                                redistemplate.opsforvalue().append(request.getrequestid() + "_result", content);
                            } else if (chunk.contains("question")) {//处理返回question
                                //先删除
                                questionservice.deletequestionsbypreviousidandrequest(questionid, requestid);
                                //保存ai返回的question
                            } else if (chunk.contains("image")) {//处理图片
                                parseimages(chunk, imagesurl);
                            } else if (chunk.contains("referenceinfo")) {//处理参照信息
                                parsereference(chunk, aianswerreferencelist);
                            }
                        } else {
                            // 处理结束
                            end.set("[done]")；
                            finalanswer[0].setstate("done");
                        }
                        if (stringutils.isempty(chunk)) {
                            chunk = "";
                        }
                        return new answerchunk(chunk, request.getrequestid());
                    })
                    .dooncomplete(() -> {//答案都完成后存储对应数据到数据库中
                        string finalcontent = redistemplate.opsforvalue().get(request.getrequestid() + "_result");
                        redistemplate.delete(request.getrequestid());
                        //保存答案
                        string returnanswer = "";
                        jsonobject answer1 = new jsonobject().putonce("data", finalcontent);
                        //具体实现
                    })
                    .onerrorresume(e -> {//错误情况处理
                        finalanswer[0].setstate("failed");
                        answerservice.saveorupdate(finalanswer[0]);
                        return flux.error(e);
                    });
}
}

@restcontroller
@requestmapping("/aiagent")
public class aiforcecontroller {
   /**
     * 获取内容
     *
     * @param request mediatype.text_event_stream_value 流式输出，否则会一次返回
     *                charset=utf-8 字符集，不设置会乱码
     *                注意：使用get会中文乱码
     * @return
     */
    @postmapping(value = "/stream", produces = mediatype.text_event_stream_value + ";charset=utf-8")
    public flux<serversentevent<string>> streamresponse(@requestbody aiforcepromptdto request) {
        return aiservice.processstream(request)
                .limitrate(100) // 限制每秒最大请求数
                .onbackpressurebuffer(100,//背压策略：缓冲区大小为 100
                buffer -> {
                    logger.warn("backpressure buffer overflow, dropping {} items", buffer);
                }).publishon(schedulers.boundedelastic(),1) // 单线程调度确保顺序
                .flatmap(chunk -> { // 使用 flatmap 将一个异步流中的每个元素映射为另一个流，并将这些流合并为一个单一的流
                    string content = chunk.getcontent();
                    if (stringutils.isnotblank(content)) {
                        string processedcontent = content.replaceall("`{3}", "\n```"); // 规范代码块格式
                        return flux.just(serversentevent.<string>builder()
                                .id(request.getrequestid())
                                .data(processedcontent)
                                .build());
                    }
                    return flux.empty();//如果内容为空，就返回空的flux
                }, 1) // 设置并发度为 1，确保逐条发送
                .doonnext(event -> logger.info("streaming chunk: {}", event.data())); // 日志记录每次发送的数据
    }
}
// flux<serversentevent<string>> 实现 sse（server-sent events），以便客户端可以实时接收服务器推送的消息