Http流式响应接口封装

1. 背景

需要对接某个大模型的流式http对话接口，因为该项目中还需要对接另外的大模型服务但是另一个请求方式为websocket，为了实现模型服务切换以及对客户端做统一的流式响应，需要对大模型服务的调用做统一的流式响应封装。

环境：JDK11

2. 代码

2.1 回调函数

@FunctionalInterface
public interface LLMChatCallback {void callback(String message);}

2.2 统一对话接口

public interface LLMService {/*** 模型对话** @param payload 用于请求大模型的参数* @param callback 响应回调 调用者通过实现callback函数接收消息**/void dialogue(String payload, LLMChatCallback callback);}

2.3 StringSubscriber

import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.Flow;/*** http响应式流订阅**/
@Slf4j
public class StringSubscriber implements Flow.Subscriber<String> {Flow.Subscription subscription;/*** 消息回调接口*/private final LLMChatCallback callback;public StringSubscriber(LLMChatCallback callback) {this.callback = callback;}@Overridepublic void onSubscribe(Flow.Subscription subscription) {this.subscription = subscription;subscription.request(1);}@Overridepublic void onNext(String item) {log.debug("onNext: {}", item);if (callback != null) {// 可做业务处理callback.callback(item);}subscription.request(1);}@Overridepublic void onError(Throwable throwable) {log.error("StringSubscriber error", throwable);}@Overridepublic void onComplete() {log.debug("StringSubscriber complete");}}

2.4 大模型Service实现

import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.time.Duration;/*** B大模型服务**/
@Slf4j
@Service
public class BLMService implements LLMService {@Overridepublic void dialogue(String payload, LLMChatCallback callback) {HttpRequest httpRequest = HttpRequest.newBuilder(URI.create("this is url")).POST(HttpRequest.BodyPublishers.ofString("this is payload")).build();HttpClient httpClient = HttpClient.newBuilder().connectTimeout(Duration.ofMinutes(5)).version(HttpClient.Version.HTTP_2).build();HttpResponse.BodySubscriber<Void> bodySubscriber = HttpResponse.BodySubscribers.fromLineSubscriber(new StringSubscriber(callback));httpClient.sendAsync(httpRequest, responseInfo -> bodySubscriber).join();}}

2.5 使用

@Resource
private BLMService bLMService;public void void dialogue() {// 构建请求负载String payload = "this is payload";// 可以扩展使用工厂及策略模式获取对应的模型对话servicebLMService.dialogue(payload, message -> {// 处理业务接收返回的信息 比如通过grpc的流式响应 onNext(message) 同步返回System.out.println(message);});
}