基于deepseek的智能语音客服【第二讲】后端异步接口调用封装
本篇内容主要讲前端请求（不包含）访问后端服务接口，接口通过检索知识库，封装提示词，调用deepseek的，并返回给前端的全过程，非完整代码，不可直接运行。
1.基于servlet封装异步请求
为什么要进行异步分装？因为前段需要流式输出，以减少用户长时间等待造成的不良体验
集成HttpServlet 实现POST方法，get方式多伦对话有数据了限制
@WebServlet(
urlPatterns = "/ds",
asyncSupported = true // 启用异步支持
)
public class DeepseekApi extends HttpServlet
2.设置跨域（如果没有前后端分离可以忽略此步骤）
response.setHeader("Access-Control-Allow-Origin", "*");
response.setHeader("Access-Control-Allow-Methods", "POST, GET, PUT, OPTIONS, DELETE");
response.setHeader("Access-Control-Max-Age", "3600");
response.setHeader("Access-Control-Allow-Headers", "x-requested-with, Content-Type");
response.setHeader("Access-Control-Allow-Credentials", "true");
// 设置SSE头
response.setContentType("text/event-stream");
response.setCharacterEncoding("UTF-8");
response.setHeader("Cache-Control", "no-cache");
response.setHeader("Connection", "keep-alive");
// 处理OPTIONS预检请求
if ("OPTIONS".equalsIgnoreCase(request.getMethod())) {
response.setStatus(HttpServletResponse.SC_OK);
asyncContext.complete();
return;
}
3.获取参数
我这里前段直接封装好多轮对话参数和当前问题，当前问题为什么要分开，应为问题需要在知识库做增强检索，这样好取参数
// 获取问题参数
String question = request.getParameter("question");
String his = request.getParameter("his");
4.封装异步任务
asyncContext
// 获取异步上下文
final AsyncContext asyncContext = request.startAsync();
asyncContext.setTimeout(3*60*1000); // 超时 60 秒
writer = response.getWriter();
processRequest(asyncContext, writer, question,his);
5.设置异步事件监听
asyncContext.addListener(new AsyncListener() {
@Override
public void onComplete(AsyncEvent event) {
// 确保资源释放
}
@Override
public void onTimeout(AsyncEvent event) {
writer.write("event:error\ndata:请求超时\n\n");
writer.flush();
asyncContext.complete();
}
@Override
public void onError(AsyncEvent event) {
asyncContext.complete();
}
@Override
public void onStartAsync(AsyncEvent event) {}
});
6.检索向量库
// 检索向量库
List<Map<String, Object>> kl = KnowledgeBaseService.searchKnowledge(question, 40);
7.构建提示词
private static String buildPrompt(String question, List<Map<String, Object>> knowledge) {
System.out.println("提示词封装！");
String txtString="";
for (Map<String, Object> map : knowledge) {
txtString+=map.get("title")+"\n"+map.get("text")+"\n";
}
return "问题：\n" + question + "\n\n参考知识：" +txtString+ "\n\n请以参考知识为主，给出简明扼要的回复，如果参考知识与问题没有相关性或不存在请拒绝答复：";
}
8.构建请求体
// 新的对话内容
JSONObject newMessage = new JSONObject();
newMessage.put("role", "user");
newMessage.put("content", prompt);
// 插入新的对话
messages.add(newMessage);
System.out.println(messages.toJSONString());
// 构建请求体
Map<String, String> headers = new HashMap<>();
headers.put("Authorization", "Bearer " + Consist.DEEPSEEK_API_KEY);
headers.put("Content-Type", "application/json");
JSONObject requestBody = new JSONObject();
requestBody.put("model", Consist.MODEL_NAME);
requestBody.put("messages", messages);
requestBody.put("stream", true);
requestBody.put("max_tokens", Consist.MAX_TOKENS);
9.进行异步调用
sendAsyncRequestWithCallback(
Consist.DEEPSEEK_API_URL,
headers,
requestBody.toJSONString(),
new StreamCallback() {
@Override
public void onDataReceived(String content) {
// System.out.print(content);
writer.write("data:" + content+"\n\n");
writer.flush();
}
@Override
public void onComplete() {
System.out.println("调用完成！");
writer.write("event:done\ndata:\n\n");
writer.flush();
asyncContext.complete();
}
@Override
public void onError(Exception ex) {
// ex.printStackTrace();
System.err.println("报错！");
writer.write("event:error\ndata:发生错误\n\n");
writer.flush();
asyncContext.complete();
}
}
);
10.监察调用状态，防止客户端掉线造成的异常
if (asyncClient == null || !asyncClient.isRunning()) {
synchronized (DeepseekR1WebApiPost.class) {
if (asyncClient == null || !asyncClient.isRunning()) {
try {
if (asyncClient != null) {
asyncClient.close();
}
asyncClient = HttpAsyncClients.createDefault();
asyncClient.start();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
11.封装客户端参数，调用deepseek官网接口
HttpPost request = new HttpPost(new URI(url));
request.setEntity(new StringEntity(requestBody, "UTF-8"));
headers.forEach(request::setHeader);
HttpHost target = new HttpHost(
new URI(url).getHost(),
new URI(url).getPort(),
new URI(url).getScheme()
);
12.执行异步请求，并处理返回数据
@Override
protected void onContentReceived(ContentDecoder decoder, IOControl ioctrl) {
try {
ByteBuffer bb = ByteBuffer.allocate(Consist.MAX_TOKENS);
int read;
while ((read = decoder.read(bb)) > 0) {
bb.flip();
byte[] bytes = new byte[bb.remaining()];
bb.get(bytes);
buffer.write(bytes);
processBuffer(callback);
bb.clear();
}
} catch (Exception e) {
e.printStackTrace();
System.out.println("报错: " + e.getMessage().toString());
callback.onError(e);
}
}
13.解析流式数据返回给前端
private void processBuffer(StreamCallback callback) throws Exception {
String chunk = buffer.toString("UTF-8");
buffer.reset();
// 按行分割并过滤空行
String[] lines = chunk.split("\\r?\\n");
for (String line : lines) {
if (line.isEmpty()) continue;
if (line.startsWith("data: ")) {
String jsonStr = line.substring(6).trim();
if ("[DONE]".equals(jsonStr)) {
callback.onComplete();
return; // 提前返回避免后续处理
}
try {
if(isJsonComplete(jsonStr)) {
JsonObject responseJson = JsonParser.parseString(jsonStr).getAsJsonObject();
JsonArray choices = responseJson.getAsJsonArray("choices");
callback.onDataReceived(choices.toString());
};
//
// callback.onDataReceived(jsonStr);
} catch (Exception e) {
callback.onError(new RuntimeException("解析 JSON 失败: " + jsonStr, e));
continue;
}
}
}
}