智能问答

本文档详细介绍了校园领域搜索问答系统中的智能问答功能模块的设计与实现。该模块旨在通过接收用户提问、校验会话合法性、问题重写、生成提示词并交由大模型推理，管理历史记录以及控制推理过程，提供高效、精准的智能问答服务。本文将涵盖功能概述、接口设计、控制器实现、服务层实现以及代码优化。

功能概述

智能问答模块主要负责以下功能：

1. 接收用户提问：通过 API 接口接收用户的提问内容。
2. 校验会话是否合法：确保用户的会话 ID 有效且属于当前用户。
3. 问题重写：对用户的提问进行优化和重写，提高大模型理解的准确性。
4. 生成提示词并交给大模型推理：根据用户提问生成相应的提示词，调用大模型（如 GPT-4）进行推理。
5. 管理历史记录：记录用户的提问和系统的回答，支持查询和管理历史记录。
6. 停止推理：提供接口以便用户在需要时停止正在进行的推理过程。

接口设计

1. 发送用户提问

• 请求方法：POST

• 接口路径：/api/v1/chat/ask

• 请求头：

  Content-Type: application/json

• 请求体：

  {
    "type": "general",
    "session_id": "452057544733368320",
    "school_id": 1,
    "app_id": "",
    "messages": [
      {
        "role": "user",
        "content": "When is the last day of Kapiolani Community College in spring 2025"
      }
    ],
    "stream": true,
    "tools":[],
  }

• SSE 数据示例：

  event:delta
  data:{"content":"- Think about your question: When is the last day of Kapiolani Community College in spring 2025\r\n"}
  
  event:progress
  data:The number of history records to be queried:0
  
  event:message_id
  data:{"question_id":"452093442603847680"}
  
  event:progress
  data:Serach it is processed using ppl
  
  event:question
  data:When is the last day of Kapiolani Community College in spring 2025?
  
  event:delta
  data:{"content":"- Understand your intention: When is the last day of Kapiolani Community College in spring 2025?\r\n"}
  
  event:delta
  data:{"content":"- Searching... \r\n"}
  
  event:progress
  data:default
  
  event:input
  data:[]
  
  event:citations
  data:["https://www.bhcc.edu/academic-calendar/academiccalendar-spring2025/","https://kellogg.edu/about/academic-calendar/","https://hawaii.hawaii.edu/sites/default/files/assets/catalog/docs/02-academic_calendar.pdf","https://www.kapiolani.hawaii.edu/classes/academic-calendar/","https://www.kapiolani.hawaii.edu/classes/"]
  
  event:reasoning
  data:{"model":"deepseek-r1-250528","content":"Alright"}
  
  event:delta
  data:{"model":"deepseek-r1-250528","content":"\n"}
  
  event:message_id
  data:{"answer_id":"452093453534203904"}

2. 停止推理

• 请求方法：POST

• 接口路径：/api/v1/chat/stop

• 请求参数：

  参数	类型	描述	是否必填
  session_id	Long	会话 ID	是

• 请求示例：

  POST /api/v1/chat/stop?session_id=452057544733368320

• 响应示例：

  {
    "data": null,
    "code": 1,
    "msg": null,
    "ok": true
  }

控制器实现

ApiChatHandler类负责处理与智能问答相关的 HTTP 请求，包括发送用户提问和停止推理。以下是该类的实现代码及说明：

package nexus.io.llm.handler;

import com.jfinal.kit.Kv;
import nexus.io.db.TableResult;
import nexus.io.jfinal.aop.Aop;
import nexus.io.llm.service.LlmAiChatService;
import nexus.io.llm.service.LlmChatHistoryService;
import nexus.io.llm.service.LlmChatMessageService;
import nexus.io.llm.service.LlmRewriteQuestionService;
import nexus.io.llm.service.UserAskQuesitonService;
import nexus.io.model.body.RespBodyVo;
import nexus.io.model.page.Page;
import nexus.io.tio.boot.http.TioRequestContext;
import nexus.io.tio.http.common.HttpRequest;
import nexus.io.tio.http.common.HttpResponse;
import nexus.io.tio.http.server.util.CORSUtils;
import nexus.io.tio.utils.json.FastJson2Utils;
import nexus.io.tio.utils.json.JsonUtils;
import nexus.io.tio.utils.hutool.StrUtil;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;

import lombok.extern.slf4j.Slf4j;

@Slf4j
public class ApiChatHandler {

  private final LlmAiChatService aiChatService = Aop.get(LlmAiChatService.class);

  /**
   * 发送用户提问
   *
   * @param request HTTP请求对象
   * @return HTTP响应对象
   */
  public HttpResponse send(HttpRequest request) {
    HttpResponse response = TioRequestContext.getResponse();
    CORSUtils.enableCORS(response);

    JSONObject payload = FastJson2Utils.parseObject(request.getBodyString());
    Long sessionId = payload.getLong("session_id");
    Boolean validateChatId = payload.getBoolean("validateChatId");
    Long schoolId = payload.getLong("school_id");
    String type = payload.getString("type");
    Long appId = StrUtil.isNotBlank(payload.getString("app_id")) ? payload.getLong("app_id") : null;
    JSONArray messages = payload.getJSONArray("messages");
    Boolean stream = payload.getBoolean("stream");

    String userId = TioRequestContext.getUserIdString();

    // 校验会话是否合法
    if (validateChatId != null && validateChatId) {
      boolean exists = Aop.get(LlmChatHistoryService.class).exists(sessionId, userId);
      if (!exists) {
        return response.fail(RespBodyVo.fail("invalid session_id"));
      }
    }

    // 调用智能问答服务
    RespBodyVo respBody = aiChatService.index(TioRequestContext.getChannelContext(), stream, userId, sessionId, type, appId, 1, schoolId, messages);
    return response.setJson(respBody);
  }

  /**
   * 停止推理
   *
   * @param request HTTP请求对象
   * @return HTTP响应对象
   */
  public HttpResponse stop(HttpRequest request) {
    HttpResponse response = TioRequestContext.getResponse();
    CORSUtils.enableCORS(response);

    Long sessionId = request.getLong("session_id");
    if (sessionId == null) {
      return response.fail(RespBodyVo.fail("session_id can not be empty"));
    }

    okhttp3.Call call = ChatStreamCallCan.stop(sessionId);
    if (call != null) {
      return response.setJson(RespBodyVo.ok("Inference stopped successfully"));
    } else {
      return response.fail(RespBodyVo.fail("No active inference found for the given session_id"));
    }
  }
}

控制器代码说明

• send：

  • 处理/api/v1/chat/send的POST请求。
  • 从请求体中解析session_id、validateChatId、school_id、type、app_id、messages和stream参数。
  • 如果validateChatId为true，则调用LlmChatHistoryService.exists方法校验会话是否合法。
  • 调用服务层LlmAiChatService.index方法处理用户提问，并返回响应结果。

• stop：

  • 处理/api/v1/chat/stop的POST请求。
  • 获取并校验session_id参数。
  • 调用ChatStreamCallCan.stop方法尝试停止正在进行的推理过程。
  • 根据停止结果返回相应的响应。

服务层实现

1.ChatMessagesVo

package nexus.io.llm.vo;

import java.util.List;

import nexus.io.openai.chat.ChatMessage;

import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;

@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
public class ChatMessagesVo {
  private boolean hasImage;
  private List<ChatMessage> messages;
  private String textQuestion;
  private String inputQuesiton;
}

2. AiChatEventName（聊天事件名称常量接口）

定义聊天过程中使用的事件名称，确保事件的一致性和可维护性。

package nexus.io.llm.consts;

public interface AiChatEventName {
  String question = "question";
  String progress = "progress";
  String markdown = "markdown";
  String delta = "delta";
  String citations = "citations";
  String message_id = "message_id";

  String summary_question = "summary_question";
  String table = "table";
  String input = "input";
  String rerank = "rerank";
  String need_login = "need_login";
  String error = "error";
  String paragraph = "paragraph";
  String documents = "documents";
}

3. ChatStreamCallCan（聊天流调用管理类）

负责管理聊天流的 HTTP 调用，支持停止和移除特定会话的推理过程。

package nexus.io.llm.can;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import okhttp3.Call;

public class ChatStreamCallCan {
  public static Map<Long, Call> callMap = new ConcurrentHashMap<>();

  /**
   * 停止特定会话的推理过程
   *
   * @param id 会话ID
   * @return 被停止的Call对象
   */
  public static Call stop(Long id) {
    Call call = callMap.get(id);
    if (call != null && !call.isCanceled()) {
      call.cancel();
      return callMap.remove(id);
    }
    return null;
  }

  /**
   * 移除特定会话的Call对象
   *
   * @param id 会话ID
   * @return 被移除的Call对象
   */
  public static Call remove(Long id) {
    return callMap.remove(id);
  }

  /**
   * 添加会话的Call对象
   *
   * @param chatId 会话ID
   * @param call   Call对象
   */
  public static void put(Long chatId, Call call) {
    callMap.put(chatId, call);
  }
}

4. LlmChatMessageService(消息解析服务)

package nexus.io.llm.service;

import java.util.ArrayList;
import java.util.List;

import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import nexus.io.llm.vo.ChatMessagesVo;
import nexus.io.openai.chat.ChatMessage;
import nexus.io.openai.chat.ChatRequestImage;
import nexus.io.openai.chat.ChatRequestMultiContent;

public class LlmChatMessageService {

  public ChatMessagesVo parse(JSONArray reqMessages) {
    List<ChatMessage> messages = new ArrayList<>();
    boolean hasImage = false;
    String textQuestion = null;


    for (int i = 0; i < reqMessages.size(); i++) {
      JSONObject message = reqMessages.getJSONObject(i);
      String role = message.getString("role");
      Object content = message.get("content");

      if ("system".equals(role)) {
        textQuestion = content.toString();
        messages.add(new ChatMessage().role(role).content(content.toString()));

      } else if ("user".equals(role)) {
        if (content instanceof String) {
          textQuestion = content.toString();
          // 文本消息单独返回,不添加到最终的消息体中
          // messages.add(new ChatMessage().role(role).content(content.toString()));

        } else if (content instanceof JSONArray) {
          JSONArray contentsArray = (JSONArray) content;
          for (int j = 0; j < contentsArray.size(); j++) {
            JSONObject contentObj = contentsArray.getJSONObject(j);
            String type = contentObj.getString("type");
            if ("image_url".equals(type)) {
              hasImage = true;
              JSONObject imageUrl = contentObj.getJSONObject("image_url");
              String url = imageUrl.getString("url");

              if (url.startsWith("data:image/")) {
                ChatRequestImage image = new ChatRequestImage();
                image.setUrl(url);
                image.setDetail(imageUrl.getString("detail"));

                ChatRequestMultiContent multiContent = new ChatRequestMultiContent();
                multiContent.setType("image_url");
                multiContent.setImage_url(image);

                List<ChatRequestMultiContent> multiContents = new ArrayList<>();
                multiContents.add(multiContent);

                messages.add(new ChatMessage().role(role).multiContents(multiContents));
              } else {
                throw new RuntimeException("image is not encoded with base64");
              }
            } else if ("text".equals(type)) {
              messages.add(new ChatMessage().role(role).content(contentObj.getString("text")));
            }
          }
        }
      }
    }

    return ChatMessagesVo.builder().hasImage(hasImage).messages(messages).textQuestion(textQuestion).build();
  }
}

5. LlmAiChatService（智能问答服务）

负责处理用户的提问，包括校验、重写、生成提示词、调用大模型推理、管理历史记录以及控制推理过程。

package nexus.io.llm.service;

import java.util.ArrayList;
import java.util.List;

import com.alibaba.fastjson2.JSONArray;
import com.jfinal.kit.Kv;
import nexus.io.db.TableResult;
import nexus.io.db.activerecord.Row;
import nexus.io.jfinal.aop.Aop;
import nexus.io.llm.can.ChatStreamCallCan;
import nexus.io.llm.consts.AiChatEventName;
import nexus.io.llm.dao.SchoolDictDao;
import nexus.io.llm.utils.LarkBotQuestionUtils;
import nexus.io.llm.vo.AiChatResponseVo;
import nexus.io.llm.vo.ChatMessagesVo;
import nexus.io.llm.vo.SchoolDict;
import nexus.io.model.body.RespBodyVo;
import nexus.io.openai.chat.ChatMessage;
import nexus.io.openai.chat.ChatResponseVo;
import nexus.io.openai.chat.OpenAiChatRequestVo;
import nexus.io.openai.client.OpenAiClient;
import nexus.io.openai.consts.OpenAiModels;
import nexus.io.tio.core.ChannelContext;
import nexus.io.tio.core.Tio;
import nexus.io.http.common.sse.SsePacket;
import nexus.io.tio.http.server.util.SseEmitter;
import nexus.io.tio.utils.environment.EnvUtils;
import nexus.io.tio.utils.hutool.StrUtil;
import nexus.io.tio.utils.json.JsonUtils;
import nexus.io.tio.utils.snowflake.SnowflakeIdUtils;
import nexus.io.tio.utils.thread.TioThreadUtils;

import lombok.extern.slf4j.Slf4j;
import okhttp3.Call;
import okhttp3.Callback;

@Slf4j
public class LlmAiChatService {
  ChatStreamCommonService chatStreamCommonService = Aop.get(ChatStreamCommonService.class);
  LLmAiChatSearchService aiChatSearchService = Aop.get(LLmAiChatSearchService.class);

  public RespBodyVo index(ChannelContext channelContext, Boolean stream,
      //
      String userId, Long sessionId,
      //
      String type, Long appId, Integer chatType, Long schoolId, JSONArray messages) {

    ChatMessagesVo chatMessages = Aop.get(LlmChatMessageService.class).parse(messages);
    return predict(channelContext, stream, userId, sessionId, type, appId, schoolId, chatMessages);
  }

  public RespBodyVo predict(ChannelContext channelContext, Boolean stream,
      //
      String userId, Long sessionId, String type, Long appId, Long schoolId, ChatMessagesVo chatMessages) {
    String textQuestion = chatMessages.getTextQuestion();

    chatMessages.setInputQuesiton(textQuestion);
    if (textQuestion != null) {
      if (stream) {
        Kv kv = Kv.by("content", "- Think about your question: " + textQuestion + "\r\n");
        SsePacket packet = new SsePacket(AiChatEventName.progress, JsonUtils.toJson(kv));
        Tio.bSend(channelContext, packet);
      }

    }
    if (textQuestion.startsWith("__echo:")) {
      String[] split = textQuestion.split(":");
      if (stream) {
        SsePacket packet = new SsePacket(AiChatEventName.delta, JsonUtils.toJson(Kv.by("content", "\r\n\r\n")));
        Tio.bSend(channelContext, packet);

        packet = new SsePacket(AiChatEventName.delta, JsonUtils.toJson(Kv.by("content", split[1])));
        Tio.bSend(channelContext, packet);

        packet = new SsePacket(AiChatEventName.delta, JsonUtils.toJson(Kv.by("content", "end")));
        Tio.bSend(channelContext, packet);

        SseEmitter.closeSeeConnection(channelContext);
      }
      return RespBodyVo.ok(new AiChatResponseVo(split[1]));
    }

    if (schoolId == null) {
      schoolId = 1L;
    }
    SchoolDict schoolDict = null;

    try {
      schoolDict = Aop.get(SchoolDictDao.class).getNameById(schoolId.longValue());
    } catch (Exception e) {
      e.printStackTrace();

      String error = e.getMessage();
      if (stream) {
        SsePacket ssePacket = new SsePacket(AiChatEventName.error, error.getBytes());
        Tio.bSend(channelContext, ssePacket);
        SseEmitter.closeSeeConnection(channelContext);
      }
      return RespBodyVo.fail(error);
    }
    if (schoolDict == null) {
      String error = "schoolId not not found";
      if (stream) {
        SsePacket ssePacket = new SsePacket(AiChatEventName.error, error.getBytes());
        Tio.bSend(channelContext, ssePacket);
        SseEmitter.closeSeeConnection(channelContext);
      }
      return RespBodyVo.fail(error);
    }

    List<Row> histories = null;
    try {
      histories = Aop.get(LlmChatHistoryService.class).getHistory(sessionId);
    } catch (Exception e) {
      e.printStackTrace();
      String error = e.getMessage();
      if (stream) {
        SsePacket ssePacket = new SsePacket(AiChatEventName.error, error);
        Tio.bSend(channelContext, ssePacket);
        SseEmitter.closeSeeConnection(channelContext);
      }
      return RespBodyVo.fail(error);
    }

    int size = 0;
    if (histories != null) {
      size = histories.size();
    }

    if (stream) {
      SsePacket ssePacket = new SsePacket(AiChatEventName.progress, ("The number of history records to be queried:" + size).getBytes());
      Tio.bSend(channelContext, ssePacket);
    }

    boolean isFirstQuestion = false;
    if (histories == null || size < 1) {
      isFirstQuestion = true;
    }

    List<ChatMessage> historyMessage = new ArrayList<>();
    for (Row record : histories) {
      String role = record.getStr("role");
      String content = record.getStr("content");
      historyMessage.add(new ChatMessage(role, content));
    }

    AiChatResponseVo aiChatResponseVo = new AiChatResponseVo();
    // save to the user question to db
    if (StrUtil.isNotEmpty(textQuestion)) {
      long questionId = SnowflakeIdUtils.id();
      TableResult<Kv> ts = Aop.get(LlmChatHistoryService.class).saveUser(questionId, sessionId, textQuestion);
      if (ts.getCode() != 1) {
        log.error("Failed to save message:{}", ts.toString());
      } else {
        if (stream) {
          Kv kv = Kv.by("question_id", questionId);
          SsePacket packet = new SsePacket(AiChatEventName.message_id, JsonUtils.toJson(kv));
          Tio.bSend(channelContext, packet);
        }
        aiChatResponseVo.setQuesitonId(questionId);
      }
    }

    if (StrUtil.isNotEmpty(textQuestion)) {
      StringBuffer stringBuffer = new StringBuffer();

      stringBuffer.append("app env:").append(EnvUtils.getStr("app.env")).append("\n")
          //
          .append("userId:").append(userId).append("\n")//
          .append("schooL id:").append(schoolId).append("\n")
          //
          .append("schooL name:").append(schoolDict.getFullName()).append("\n")
          //
          .append("user question:").append(textQuestion).append("\n")
          //
          .append("type:").append(type);
      if (appId != null) {
        stringBuffer.append("app id:").append(appId);
      }

      log.info("question:{}", stringBuffer.toString());

      if (!EnvUtils.isDev()) {
        String thatTextQuestion = textQuestion;
        TioThreadUtils.submit(() -> {
          LarkBotQuestionUtils.send(stringBuffer.toString());
          if (stream) {
            SsePacket packet = new SsePacket(AiChatEventName.progress, "send message to lark");
            Tio.send(channelContext, packet);
          }
          // save to db
          Aop.get(UserAskQuesitonService.class).save(thatTextQuestion);
        });
      }
    }
    if (size > 20) {
      String message = "Dear user, your conversation count has exceeded the maximum length for multiple rounds of conversation. "
          //
          + "Please start a new session. Your new question might be:" + textQuestion;

      long answerId = SnowflakeIdUtils.id();
      aiChatResponseVo.setAnswerId(answerId);

      Aop.get(LlmChatHistoryService.class).saveAssistant(answerId, sessionId, message);
      Kv kv = Kv.by("answer_id", answerId);
      if (stream) {
        SsePacket ssePacket = new SsePacket(AiChatEventName.progress, JsonUtils.toJson(Kv.by("content", message)));
        Tio.bSend(channelContext, ssePacket);
        SsePacket packet = new SsePacket(AiChatEventName.message_id, JsonUtils.toJson(kv));
        Tio.send(channelContext, packet);
        SseEmitter.closeSeeConnection(channelContext);
      }
      aiChatResponseVo.setContent(message);
      return RespBodyVo.ok(message);
    }
    if (isFirstQuestion && textQuestion != null) {
      textQuestion += " at " + schoolDict.getFullName() + " in Fall 2024";
    }

    if (chatMessages.isHasImage()) {
      if (stream) {
        SsePacket packet = new SsePacket(AiChatEventName.progress, "Because there are pictures, it is processed using gpt4o");
        Tio.bSend(channelContext, packet);
      }
      String answer = processMessageByChatModel(sessionId, chatMessages, isFirstQuestion, stream, channelContext);
      aiChatResponseVo.setContent(answer);
      return RespBodyVo.ok(aiChatResponseVo);

    } else if (textQuestion != null && textQuestion.startsWith("4o:")) {
      if (stream) {
        SsePacket packet = new SsePacket(AiChatEventName.progress, "The user specifies that the gpt4o model is used for message processing");
        Tio.bSend(channelContext, packet);
      }
      String answer = processMessageByChatModel(sessionId, chatMessages, isFirstQuestion, stream, channelContext);
      aiChatResponseVo.setContent(answer);
      return RespBodyVo.ok(aiChatResponseVo);

    } else {
      if (stream) {
        SsePacket packet = new SsePacket(AiChatEventName.progress, "Serach it is processed using ppl");
        Tio.bSend(channelContext, packet);
      }

      if (textQuestion != null) {
        // rewrite question
        textQuestion = Aop.get(LlmRewriteQuestionService.class).rewrite(textQuestion, historyMessage);
        log.info("rewrite question:{}", textQuestion);

        if (stream != null) {
          SsePacket packet = new SsePacket(AiChatEventName.question, textQuestion);
          Tio.bSend(channelContext, packet);
          Kv kv = Kv.by("content", "- Understand your intention: " + textQuestion + "\r\n");
          packet = new SsePacket(AiChatEventName.progress, JsonUtils.toJson(kv));
          Tio.bSend(channelContext, packet);
        }
        aiChatResponseVo.setRewrite(textQuestion);
        chatMessages.setTextQuestion(textQuestion);
      }

      aiChatSearchService.processMessageBySearchModel(schoolId, sessionId, chatMessages, isFirstQuestion,
          //
          historyMessage, stream, channelContext, aiChatResponseVo);
      return RespBodyVo.ok(aiChatResponseVo);
    }
  }

  public String processMessageByChatModel(Long sessionId, ChatMessagesVo chatMessages, boolean isFirstQuestion, Boolean stream, ChannelContext channelContext) {
    long start = System.currentTimeMillis();
    // 添加文本
    List<ChatMessage> messages = chatMessages.getMessages();
    messages.add(new ChatMessage("user", chatMessages.getTextQuestion()));

    OpenAiChatRequestVo chatRequestVo = new OpenAiChatRequestVo().setModel(OpenAiModels.gpt_4o_mini).setMessages(messages);

    if (stream) {
      Kv kv = Kv.by("content", "- Reply to your question.\r\n\r\n");
      SsePacket packet = new SsePacket(AiChatEventName.delta, JsonUtils.toJson(kv));
      Tio.bSend(channelContext, packet);

      chatRequestVo.setStream(true);
      Callback callback = chatStreamCommonService.getCallback(channelContext, sessionId, start);
      Call call = OpenAiClient.chatCompletions(chatRequestVo, callback);
      log.info("add call:{}", sessionId);
      ChatStreamCallCan.put(sessionId, call);
      return null;

    } else {
      ChatResponseVo chatCompletions = OpenAiClient.chatCompletions(chatRequestVo);
      String content = chatCompletions.getChoices().get(0).getMessage().getContent();
      return content;
    }

  }
}

6. LLmAiChatSearchService（搜索模型处理服务）

负责使用搜索模型（如 Perplexity）处理用户的提问，并返回系统回答。

package nexus.io.llm.service;

import java.util.ArrayList;
import java.util.List;

import com.jfinal.kit.Kv;
import nexus.io.jfinal.aop.Aop;
import nexus.io.llm.can.ChatStreamCallCan;
import nexus.io.llm.consts.AiChatEventName;
import nexus.io.llm.vo.AiChatResponseVo;
import nexus.io.llm.vo.ChatMessagesVo;
import nexus.io.openai.chat.ChatMessage;
import nexus.io.openai.chat.ChatResponseVo;
import nexus.io.openai.chat.OpenAiChatRequestVo;
import nexus.io.openai.client.OpenAiClient;
import nexus.io.openai.consts.PerplexityConstants;
import nexus.io.openai.consts.PerplexityModels;
import nexus.io.tio.core.ChannelContext;
import nexus.io.tio.core.Tio;
import nexus.io.http.common.sse.SsePacket;
import nexus.io.tio.utils.environment.EnvUtils;
import nexus.io.tio.utils.json.JsonUtils;
import nexus.io.tio.utils.snowflake.SnowflakeIdUtils;

import lombok.extern.slf4j.Slf4j;
import okhttp3.Call;
import okhttp3.Callback;

@Slf4j
public class LLmAiChatSearchService {

  /**
   * 使用搜索模型处理消息
   *
   * @param schoolId        学校ID
   * @param sessionId       会话ID
   * @param chatMessages    消息对象
   * @param isFirstQuestion 是否为首次提问
   * @param history  关联消息
   * @param stream          是否使用流式响应
   * @param channelContext  通道上下文
   * @param aiChatResponseVo
   * @return 响应对象
   */
  public AiChatResponseVo processMessageBySearchModel(Long schoolId, Long sesionId, ChatMessagesVo chatMessages, boolean isFirstQuestion,
      //
      List<ChatMessage> history, Boolean stream, ChannelContext channelContext, AiChatResponseVo aiChatResponseVo) {

    // 发送搜索进度
    if (stream && channelContext != null) {
      Kv kv = Kv.by("content", "- Searching... \r\n");
      SsePacket packet = new SsePacket(AiChatEventName.progress, JsonUtils.toJson(kv));
      Tio.bSend(channelContext, packet);
    }

    List<ChatMessage> messages = chatMessages.getMessages();
    if (messages == null) {
      messages = new ArrayList<>();
    }

    // 添加初始提示词
    String textQuestion = chatMessages.getTextQuestion();
    String initPrompt = Aop.get(SearchPromptService.class).index(schoolId, textQuestion, stream, channelContext);
    messages.add(0, new ChatMessage("system", initPrompt));
    //添加历史
    messages.addAll(history);
    // 添加用户问题
    messages.add(new ChatMessage("user", textQuestion));
    if (stream) {
      SsePacket packet = new SsePacket(AiChatEventName.input, JsonUtils.toJson(history));
      Tio.bSend(channelContext, packet);
    }

    OpenAiChatRequestVo chatRequestVo = new OpenAiChatRequestVo().setModel(PerplexityModels.llama_3_1_sonar_large_128k_online)
        //
        .setMessages(messages).setMax_tokens(3000);

    log.info("chatRequestVo:{}", JsonUtils.toJson(chatRequestVo));
    String pplApiKey = EnvUtils.get("PERPLEXITY_API_KEY");
    if (stream) {
      // 发送回复提示
      Kv kv = Kv.by("content", "- Reply to your question.\r\n\r\n");
      SsePacket replyPacket = new SsePacket(AiChatEventName.delta, JsonUtils.toJson(kv));
      Tio.bSend(channelContext, replyPacket);

      chatRequestVo.setStream(true);
      long start = System.currentTimeMillis();

      Callback callback = Aop.get(ChatStreamCommonService.class).getCallback(channelContext, sesionId, start);
      Call call = OpenAiClient.chatCompletions(PerplexityConstants.server_url, pplApiKey, chatRequestVo, callback);
      ChatStreamCallCan.put(sesionId, call);
      return null;
    } else {
      ChatResponseVo chatCompletions = OpenAiClient.chatCompletions(PerplexityConstants.server_url, pplApiKey, chatRequestVo);
      List<String> citations = chatCompletions.getCitations();
      String answerContent = chatCompletions.getChoices().get(0).getMessage().getContent();
      long answerId = SnowflakeIdUtils.id();
      Aop.get(LlmChatHistoryService.class).saveAssistant(answerId, sesionId, answerContent);
      aiChatResponseVo.setContent(answerContent);
      aiChatResponseVo.setAnswerId(answerId);
      aiChatResponseVo.setCition(citations);
      return aiChatResponseVo;
    }
  }
}

7. ChatStreamCommonService（聊天流服务）

负责管理聊天流的回调和响应处理，确保流式数据的正确发送和连接的管理。

package nexus.io.llm.service;

import java.io.IOException;
import java.util.List;

import com.jfinal.kit.Kv;
import nexus.io.db.TableResult;
import nexus.io.jfinal.aop.Aop;
import nexus.io.llm.can.ChatStreamCallCan;
import nexus.io.llm.consts.AiChatEventName;
import nexus.io.openai.chat.ChatResponseVo;
import nexus.io.openai.chat.Choice;
import nexus.io.openai.chat.Delta;
import nexus.io.openai.chat.OpenAiChatRequestVo;
import nexus.io.openai.client.OpenAiClient;
import nexus.io.tio.core.ChannelContext;
import nexus.io.tio.core.Tio;
import nexus.io.http.common.sse.SsePacket;
import nexus.io.tio.http.server.util.SseEmitter;
import nexus.io.tio.utils.json.FastJson2Utils;
import nexus.io.tio.utils.json.JsonUtils;
import nexus.io.tio.utils.snowflake.SnowflakeIdUtils;

import lombok.extern.slf4j.Slf4j;
import okhttp3.Call;
import okhttp3.Callback;
import okhttp3.Response;
import okhttp3.ResponseBody;
import okio.BufferedSource;

@Slf4j
public class ChatStreamCommonService {

  /**
   * 启动聊天流
   *
   * @param chatRequestVo  聊天请求对象
   * @param chatId        会话ID
   * @param channelContext 通道上下文
   * @param start         开始时间
   */
  public void stream(OpenAiChatRequestVo chatRequestVo, Long chatId, ChannelContext channelContext, long start) {
    Call call = OpenAiClient.chatCompletions(chatRequestVo, getCallback(channelContext, chatId, start));
    ChatStreamCallCan.put(chatId, call);
  }

  /**
   * 获取回调对象
   *
   * @param channelContext 通道上下文
   * @param chatId         会话ID
   * @param start          开始时间
   * @return 回调对象
   */
  public Callback getCallback(ChannelContext channelContext, Long chatId, long start) {
    return new Callback() {
      @Override
      public void onResponse(Call call, Response response) throws IOException {
        if (!response.isSuccessful()) {
          SsePacket packet = new SsePacket(AiChatEventName.progress, "Chat model response an unsuccessful message:" + response.body().string());
          Tio.bSend(channelContext, packet);
        }

        try (ResponseBody responseBody = response.body()) {
          if (responseBody == null) {
            String message = "response body is null";
            log.error(message);
            SsePacket ssePacket = new SsePacket(AiChatEventName.progress, message);
            Tio.bSend(channelContext, ssePacket);
            return;
          }
          StringBuffer completionContent = onChatGptResponseSuccess(channelContext, responseBody, start);

          if (completionContent != null && !completionContent.toString().isEmpty()) {
            long answerId = SnowflakeIdUtils.id();
            TableResult<Kv> tr = Aop.get(LlmChatHistoryService.class).saveAssistant(answerId, chatId, completionContent.toString());
            if (tr.getCode() != 1) {
              log.error("Failed to save assistant answer: {}", tr);
            } else {
              Kv kv = Kv.by("answer_id", answerId);
              SsePacket packet = new SsePacket(AiChatEventName.message_id, JsonUtils.toJson(kv));
              Tio.bSend(channelContext, packet);
            }
          }
        }
        ChatStreamCallCan.remove(chatId);
        SseEmitter.closeSeeConnection(channelContext);
      }

      @Override
      public void onFailure(Call call, IOException e) {
        SsePacket packet = new SsePacket(AiChatEventName.progress, "error: " + e.getMessage());
        Tio.bSend(channelContext, packet);
        ChatStreamCallCan.remove(chatId);
        SseEmitter.closeSeeConnection(channelContext);
      }
    };
  }

  /**
   * 处理ChatGPT成功响应
   *
   * @param channelContext 通道上下文
   * @param responseBody    响应体
   * @return 完整内容
   * @throws IOException
   */
  public StringBuffer onChatGptResponseSuccess(ChannelContext channelContext, ResponseBody responseBody, Long start) throws IOException {
    StringBuffer completionContent = new StringBuffer();
    BufferedSource source = responseBody.source();
    String line;
    boolean sentCitations = false;
    while ((line = source.readUtf8Line()) != null) {
      if (line.length() < 1) {
        continue;
      }
      // 处理数据行
      if (line.length() > 6) {
        String data = line.substring(6);
        if (data.endsWith("}")) {
          ChatResponseVo chatResponse = FastJson2Utils.parse(data, ChatResponseVo.class);
          List<String> citations = chatResponse.getCitations();
          if (citations != null && !sentCitations) {
            SsePacket ssePacket = new SsePacket(AiChatEventName.citations, JsonUtils.toJson(citations));
            Tio.bSend(channelContext, ssePacket);
            sentCitations = true;
          }
          List<Choice> choices = chatResponse.getChoices();
          if (!choices.isEmpty()) {
            Delta delta = choices.get(0).getDelta();
            String part = delta.getContent();
            if (part != null && !part.isEmpty()) {
              completionContent.append(part);
              SsePacket ssePacket = new SsePacket(AiChatEventName.delta, JsonUtils.toJson(delta));
              Tio.bSend(channelContext, ssePacket);
            }
          }
        } else {
          log.info("Data does not end with }:{}", line);
        }
      }
    }

    // 关闭连接
    long end = System.currentTimeMillis();
    log.info("finish llm in {} (ms)", (end - start));
    return completionContent;
  }
}

总结

本文档详细介绍了校园领域搜索问答系统中智能问答功能模块的设计与实现。通过定义清晰的 API 接口、设计合理的控制器和服务层代码，并对关键服务类进行了优化，该模块能够有效地接收用户提问、校验会话合法性、重写问题、生成提示词并调用大模型进行推理，管理历史记录以及控制推理过程。未来，可以进一步扩展智能问答功能，例如集成更多大模型、优化意图识别算法、增强多轮对话能力，以及与其他系统模块的深度集成，以提升系统的整体用户体验和功能丰富性。