对话日志

import nexus.io.db.activerecord.Db;
import nexus.io.db.activerecord.Row;
import nexus.io.mosskb.model.MossKbApplicationChat;
import nexus.io.mosskb.model.MossKbApplicationChatRecord;
import nexus.io.mosskb.vo.MossKbChatRequestVo;
import nexus.io.model.result.ResultVo;
import nexus.io.tio.boot.http.TioRequestContext;
import nexus.io.tio.utils.snowflake.SnowflakeIdUtils;

public ResultVo ask(ChannelContext channelContext, Long chatId, MossKbChatRequestVo vo) {
  String quesiton = vo.getMessage();
  Row record = Db.findFirst(MossKbApplicationChat.tableName, "application_id,chat_type,client_id",
      Row.by("id", chatId).set("is_deleted", false));
  if (record == null) {
    return ResultVo.fail("会话不存在");
  }
  Long caller = TioRequestContext.getUserIdLong();
  if (caller == null || (caller != 1L && !caller.equals(record.getLong("client_id")))) {
    return ResultVo.fail("无权访问此会话");
  }
  Long application_id = record.getLong("application_id");
  if (!ApplicationAccess.canChat(caller, application_id)) {
    return ResultVo.fail("应用已停用或无权访问");
  }
  if (!ChatExecution.begin(chatId)) {
    return ResultVo.fail("当前会话正在处理上一条问题，请等待完成");
  }
  long messageId = SnowflakeIdUtils.id();
  // 先保存问题行，回答生成结束后回填答案与明细
  int countTokens = TokenCounter.countTokens(quesiton);
  Row chatRecord = Row.by("id", messageId).set("problem_text", quesiton).set("message_tokens", countTokens).set("chat_id", chatId);
  Db.save(MossKbApplicationChatRecord.tableName, chatRecord);
  Db.update("update moss_kb_application_chat set abstract=coalesce(abstract,?),update_time=now() where id=?",
      quesiton.substring(0, Math.min(quesiton.length(), 100)), chatId);
  // 检索与生成省略，见迭代检索章节
}

import org.postgresql.util.PGobject;

import nexus.io.db.activerecord.Db;
import nexus.io.db.activerecord.Row;
import nexus.io.kit.PgObjectUtils;
import nexus.io.mosskb.constant.MossKbTableNames;
import nexus.io.mosskb.vo.MossKbChatRecordDetail;
import nexus.io.mosskb.vo.MossKbStreamChatVo;
import nexus.io.mosskb.utils.TokenCounter;
import nexus.io.tio.http.server.util.SseEmitter;
import nexus.io.tio.utils.json.JsonUtils;

StringBuffer completionContent = processResponseBody(chatId, messageId, channelContext, responseBody);
double runTime = ((double) (System.currentTimeMillis() - start)) / 1000;
int answer_tokens = TokenCounter.countTokens(completionContent.toString());
chatStep.setRun_time(runTime).setAnswer_tokens(answer_tokens);

MossKbChatRecordDetail detail = new MossKbChatRecordDetail(chatStep, searchStep);
PGobject pgobject = PgObjectUtils.json(JsonUtils.toJson(detail));
Row record = Row.by("id", messageId).set("run_time", runTime).set("answer_text", completionContent.toString())
    .set("answer_tokens", answer_tokens).set("message_tokens", chatStep.getMessage_tokens())
    .set("details", pgobject);
Db.update(MossKbTableNames.moss_kb_application_chat_record, record);
// 先落库，再发送结束事件：紧接着的追问能读到这一轮
MossKbStreamChatVo end = new MossKbStreamChatVo();
end.setContent("").setChat_id(chatId).setId(messageId).setChat_record_id(messageId.toString());
end.setOperate(true).setIs_end(true).setNode_is_end(true);
SseEmitter.pushSSEChunk(channelContext, JsonUtils.toJson(end));

package nexus.io.mosskb.service.kb;

import nexus.io.db.activerecord.Db;
import nexus.io.db.activerecord.Row;

public final class ApplicationAccess {
  private ApplicationAccess() {}

  public static boolean owns(Long userId, Long applicationId) {
    if (userId == null || applicationId == null) {
      return false;
    }
    Row app = Db.findById("moss_kb_application", applicationId);
    return app != null && (userId == 1L || userId.equals(app.getLong("user_id")));
  }

  public static boolean canChat(Long clientId, Long applicationId) {
    if (owns(clientId, applicationId)) {
      return true;
    }
    return clientId != null && Db.queryLong("select count(*) from moss_kb_application_public_access_client c join moss_kb_application_access_token t on t.application_id=c.application_id where c.client_id=? and c.application_id=? and t.is_active=true and t.deleted=0", clientId, applicationId) > 0;
  }

  public static boolean canReadChat(Long clientId, Long applicationId, Long chatId) {
    Row chat = Db.findById("moss_kb_application_chat", chatId);
    return chat != null && !Boolean.TRUE.equals(chat.getBoolean("is_deleted")) && applicationId.equals(chat.getLong("application_id"))
        && (owns(clientId, applicationId) || (canChat(clientId, applicationId) && clientId.equals(chat.getLong("client_id"))));
  }
}

if (!member && !(path.equals("/api/application/profile")
    || path.matches("/api/application/\\d+/chat/open")
    || path.matches("/api/application/chat_message/\\d+")
    || path.matches("/api/application/\\d+/chat/\\d+/context(/compact)?")
    || path.matches("/api/application/\\d+/chat/client/\\d+(/\\d+)?")
    || path.matches("/api/application/\\d+/chat/\\d+/chat_record/\\d+/\\d+")
    || path.matches("/api/application/\\d+/chat/\\d+/chat_record/\\d+(/vote)?"))) {
  HttpResponse denied = TioRequestContext.getResponse();
  denied.setStatus(HttpResponseStatus.C403);
  return denied;
}

import java.time.LocalDate;
import java.time.OffsetDateTime;
import java.time.ZoneId;
import java.util.ArrayList;
import java.util.List;

import com.jfinal.kit.Kv;

import nexus.io.db.activerecord.Db;
import nexus.io.db.activerecord.Row;
import nexus.io.jfinal.aop.Aop;
import nexus.io.model.result.ResultVo;

public class MossKbChatHistoryService {

  public ResultVo chats(Long user, Long application, int page, int size, boolean clientOnly, Query query) {
    if (clientOnly ? !ApplicationAccess.canChat(user, application) : !ApplicationAccess.owns(user, application)) {
      return ResultVo.fail("无权访问对话日志");
    }
    page = Math.max(1, page);
    size = Math.max(1, Math.min(100, size));
    Sql sql = listSql(application, clientOnly ? user : null, query, false);
    long total = Db.queryLong("select count(*) from (" + sql.select + ") chat_log", sql.args.toArray());
    List<Object> pageArgs = new ArrayList<>(sql.args);
    pageArgs.add(size);
    pageArgs.add((page - 1) * size);
    List<Row> rows = Db.find(sql.select + " order by c.update_time desc,c.id desc limit ? offset ?", pageArgs.toArray());
    List<Kv> records = new ArrayList<>();
    for (Row row : rows) {
      records.add(toChat(row));
    }
    return ResultVo.ok(Kv.by("current", page).set("size", size).set("total", total).set("records", records));
  }

  /** 统计子查询 + 用户反查 + 条件拼装，导出与会话分页共用。 */
  private Sql listSql(Long application, Long clientId, Query query, boolean export) {
    StringBuilder select = new StringBuilder();
    select.append("select c.*, coalesce(s.chat_record_count,0) chat_record_count, coalesce(s.star_num,0) star_num,")
        .append(" coalesce(s.trample_num,0) trample_num, coalesce(s.mark_sum,0) mark_sum, coalesce(s.tokens_num,0) tokens_num,")
        .append(" u.username asker_name, u.nick_name asker_nick_name")
        .append(" from moss_kb_application_chat c")
        .append(" left join (select r.chat_id, count(*) chat_record_count,")
        .append(" sum(case when r.vote_status='0' then 1 else 0 end) star_num,")
        .append(" sum(case when r.vote_status='1' then 1 else 0 end) trample_num,")
        .append(" sum(case when r.improve_paragraph_id_list is null then 0 else array_length(r.improve_paragraph_id_list,1) end) mark_sum,")
        .append(" sum(coalesce(r.answer_tokens,0)+coalesce(r.message_tokens,0)) tokens_num")
        .append(" from moss_kb_application_chat_record r group by r.chat_id) s on s.chat_id=c.id")
        .append(" left join moss_kb_user u on u.id=c.client_id");
    List<Object> args = new ArrayList<>();
    select.append(" where c.application_id=? and c.is_deleted=false");
    args.add(application);
    if (clientId != null) {
      select.append(" and c.client_id=? and c.chat_type=0");
      args.add(clientId);
    }
    if (query.hasAbstract()) {
      select.append(" and c.abstract ilike ?");
      args.add("%" + query.abstractText + "%");
    }
    OffsetDateTime from = startOfDay(query.startTime);
    if (from != null) {
      select.append(" and c.update_time >= ?");
      args.add(from);
    }
    OffsetDateTime to = endOfDay(query.endTime);
    if (to != null) {
      select.append(" and c.update_time <= ?");
      args.add(to);
    }
    if (query.hasVoteFilter()) {
      String star = "coalesce(s.star_num,0) >= ?";
      String trample = "coalesce(s.trample_num,0) >= ?";
      if (query.minStar != null && query.minTrample != null) {
        select.append(" and (").append(star).append(query.isOrComparer() ? " or " : " and ").append(trample).append(")");
        args.add(query.minStar);
        args.add(query.minTrample);
      } else if (query.minStar != null) {
        select.append(" and ").append(star);
        args.add(query.minStar);
      } else {
        select.append(" and ").append(trample);
        args.add(query.minTrample);
      }
    }
    Sql sql = new Sql();
    sql.select = select.toString();
    sql.args = args;
    return sql;
  }

  public ResultVo records(Long user, Long application, Long chat, int page, int size, boolean ascending) {
    if (!ApplicationAccess.canReadChat(user, application, chat)) {
      return ResultVo.fail("无权访问会话");
    }
    page = Math.max(1, page);
    size = Math.max(1, Math.min(100, size));
    long total = Db.queryLong("select count(*) from moss_kb_application_chat_record where chat_id=?", chat);
    List<Long> ids = Db.queryListLong("select id from moss_kb_application_chat_record where chat_id=? order by id "
        + (ascending ? "asc" : "desc") + " limit ? offset ?", chat, size, (page - 1) * size);
    List<Object> records = new ArrayList<>();
    for (Long id : ids) {
      records.add(Aop.get(MossKbApplicationCharRecordService.class).get(user, application, chat, id).getData());
    }
    return ResultVo.ok(Kv.by("current", page).set("size", size).set("total", total).set("records", records));
  }

  public ResultVo vote(Long user, Long application, Long chat, Long record, String vote) {
    if (!ApplicationAccess.canReadChat(user, application, chat) || vote == null || !List.of("-1", "0", "1").contains(vote)) {
      return ResultVo.fail("无效的评价请求");
    }
    return ResultVo.ok(Db.update("update moss_kb_application_chat_record set vote_status=? where id=? and chat_id=?", vote, record, chat) > 0);
  }
}

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.Objects;

import org.postgresql.util.PGobject;

import com.jfinal.kit.Kv;

import nexus.io.db.activerecord.Db;
import nexus.io.db.activerecord.Row;
import nexus.io.mosskb.constant.MossKbTableNames;
import nexus.io.mosskb.vo.MossKbChatRecordDetail;
import nexus.io.mosskb.vo.ParagraphSearchResultVo;
import nexus.io.model.result.ResultVo;
import nexus.io.tio.utils.hutool.StrUtil;
import nexus.io.tio.utils.json.JsonUtils;

public class MossKbApplicationCharRecordService {

  public ResultVo get(Long userId, Long applicationId, Long chatId, Long recordId) {
    if (!ApplicationAccess.canReadChat(userId, applicationId, chatId)) {
      return ResultVo.fail("会话不存在或无权访问");
    }
    Row queryRecord = Row.by("id", recordId).set("chat_id", chatId);
    Row record = Db.findFirst(MossKbTableNames.moss_kb_application_chat_record, queryRecord);
    if (record == null) {
      return ResultVo.ok();
    }
    Object object = record.get("details");
    record.remove("details");
    MossKbChatRecordDetail detail = null;
    if (object instanceof PGobject) {
      String value = ((PGobject) object).getValue();
      if (StrUtil.isNotBlank(value)) {
        detail = JsonUtils.parse(value, MossKbChatRecordDetail.class);
      }
    } else if (object instanceof String) {
      String value = (String) object;
      if (StrUtil.isNotBlank(value)) {
        detail = JsonUtils.parse(value, MossKbChatRecordDetail.class);
      }
    }

    Kv kv = record.toKv();
    String answer = record.getStr("answer_text");
    kv.set("answer_text_list", List.of(answer == null ? "" : answer));
    kv.set("create_time", Objects.toString(record.getObject("create_time")));
    kv.set("update_time", Objects.toString(record.getObject("update_time")));
    if (detail != null) {
      kv.set("agent_trace", detail.getSearch_step().getIterations());
      kv.set("agent_stop_reason", detail.getSearch_step().getStop_reason());
      kv.set("context_info", detail.getSearch_step().getContext());
      List<ParagraphSearchResultVo> paragraph_list = detail.getSearch_step().getParagraph_list();
      List<Kv> dataset_list = new ArrayList<>();
      Set<Long> seenIds = new HashSet<>();
      for (ParagraphSearchResultVo paragraph : paragraph_list) {
        if (seenIds.add(paragraph.getDataset_id())) {
          dataset_list.add(Kv.by("id", paragraph.getDataset_id()).set("name", paragraph.getDataset_name()));
        }
      }
      kv.set("paragraph_list", paragraph_list);
      kv.set("dataset_list", dataset_list);
    }
    return ResultVo.ok(kv);
  }
}

  Object improveParagraphIds = record.get("improve_paragraph_id_list");
  record.remove("improve_paragraph_id_list");

  Kv kv = record.toKv();
  kv.set("improve_paragraph_id_list", ChatLogArrays.toLongList(improveParagraphIds));

import java.util.ArrayList;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;

import com.alibaba.fastjson2.JSONObject;
import com.jfinal.kit.Kv;

import nexus.io.db.activerecord.Db;
import nexus.io.db.activerecord.Row;
import nexus.io.jfinal.aop.Aop;
import nexus.io.mosskb.constant.MossKbTableNames;
import nexus.io.model.result.ResultVo;
import nexus.io.tio.utils.snowflake.SnowflakeIdUtils;

public class MossKbChatLogService {

  public ResultVo improve(Long userId, Long applicationId, Long chatId, Long recordId, Long datasetId, Long documentId, JSONObject body) {
    ResultVo denied = check(userId, applicationId, chatId, recordId, datasetId, documentId);
    if (denied != null) {
      return denied;
    }
    String content = body == null ? null : body.getString("content");
    if (content == null || content.isBlank()) {
      return ResultVo.fail("段落内容不能为空");
    }
    String title = body.getString("title");
    String problemText = body.getString("problem_text");

    Long paragraphId = Aop.get(MossKbParagraphServcie.class).insertParagraph(datasetId, documentId, title, content);
    saveProblem(datasetId, documentId, paragraphId, problemText, recordId);
    appendParagraph(recordId, chatId, paragraphId);
    return Aop.get(MossKbApplicationCharRecordService.class).get(userId, applicationId, chatId, recordId);
  }

  private void appendParagraph(Long recordId, Long chatId, Long paragraphId) {
    Db.update("update moss_kb_application_chat_record set improve_paragraph_id_list=array_append(coalesce(improve_paragraph_id_list,'{}'),?::bigint) where id=? and chat_id=?",
        paragraphId, recordId, chatId);
  }
}

  /** 插入段落并计算向量，返回段落 ID；调用方需自行完成归属校验。 */
  public Long insertParagraph(Long datasetId, Long documentId, String title, String content) {
    Row dataset = Db.findById(MossKbTableNames.moss_kb_dataset, datasetId);
    if (dataset == null) {
      throw new IllegalArgumentException("知识库不存在");
    }
    Long embedding_mode_id = dataset.getLong("embedding_mode_id");
    KbEmbeddingService mossKbEmbeddingService = Aop.get(KbEmbeddingService.class);

    PGobject contentVector = mossKbEmbeddingService.getVectorForModel(content, embedding_mode_id);
    PGobject titleVector = title == null || title.isBlank() ? null : mossKbEmbeddingService.getVectorForModel(title, embedding_mode_id);
    long id = SnowflakeIdUtils.id();
    Row record = Row.by("id", id)
        .set("source_type", "md")
        .set("title", title)
        .set("content", content)
        .set("md5", Md5Utils.md5Hex(content))
        .set("status", "2")
        .set("hit_num", 0)
        .set("is_active", true).set("dataset_id", datasetId).set("document_id", documentId)
        .set("embedding", contentVector).set(MossKbParagraph.titleEmbedding, titleVector);

    Db.save(MossKbTableNames.moss_kb_paragraph, record);
    refreshStatistics(documentId);
    return id;
  }

  public ResultVo improveBatch(Long userId, Long applicationId, Long datasetId, JSONObject body) {
    if (!ApplicationAccess.owns(userId, applicationId)) {
      return ResultVo.fail("应用不存在或无权访问");
    }
    Long documentId = body == null ? null : body.getLong("document_id");
    if (!DatasetAccess.ownsDocument(userId, datasetId, documentId)) {
      return ResultVo.fail("文档不存在或无权访问");
    }
    ...
    List<Long> paragraphIds = new ArrayList<>();
    for (Long chatId : chatIds) {
      if (!ApplicationAccess.canReadChat(userId, applicationId, chatId)) {
        return ResultVo.fail("会话不存在或无权访问");
      }
      List<Row> records = Db.find("select id,problem_text,answer_text from moss_kb_application_chat_record where chat_id=? order by id", chatId);
      for (Row record : records) {
        String answer = record.getStr("answer_text");
        if (answer == null || answer.isBlank()) {
          continue;
        }
        Long paragraphId = paragraphService.insertParagraph(datasetId, documentId, record.getStr("problem_text"), answer);
        saveProblem(datasetId, documentId, paragraphId, record.getStr("problem_text"), record.getLong("id"));
        appendParagraph(record.getLong("id"), chatId, paragraphId);
        paragraphIds.add(paragraphId);
      }
    }
    return ResultVo.ok(Kv.by("paragraph_ids", paragraphIds).set("dataset_id", datasetId));
  }

  public static ExcelWriterBuilder getExcelWriteBuilder(OutputStream outputStream) {
    ExcelWriterBuilder excelWriterBuilder = EasyExcel.write(outputStream)
        .autoCloseStream(false)
        .registerWriteHandler(new MaxColumnWidthStyleStrategy(15, 30))
        // 表头与内容统一左对齐
        .registerWriteHandler(new HorizontalCellStyleStrategy(leftAlignStyle(), leftAlignStyle()))
        .registerConverter(new LocalDateTimeConverter()).registerConverter(new TimestampStringConverter())
        .registerConverter(new StringArrayConverter());
    return excelWriterBuilder;
  }

  /** 左对齐样式，其余属性沿用 EasyExcel 默认值 */
  public static WriteCellStyle leftAlignStyle() {
    WriteCellStyle writeCellStyle = new WriteCellStyle();
    writeCellStyle.setHorizontalAlignment(HorizontalAlignment.LEFT);
    return writeCellStyle;
  }

  public static HttpResponse excel(HttpResponse response, byte[] bodyBytes, String filename) {
    return download(response, bodyBytes, CONTENT_TYPE_EXCEL, filename);
  }

  public static HttpResponse download(HttpResponse response, byte[] bodyBytes, String contentType, String filename) {
    bytesWithContentType(response, bodyBytes, contentType);
    return contentDisposition(response, filename);
  }

import java.io.ByteArrayOutputStream;
import java.time.LocalDate;
import java.util.ArrayList;
import java.util.List;

import nexus.io.db.activerecord.Db;
import nexus.io.db.activerecord.Row;
import nexus.io.table.utils.EasyExcelUtils;

public class MossKbChatLogExportService {

  private static final String[] HEADERS = { "会话 ID", "标题", "用户问题", "改写后的问题", "回答", "用户反馈", "引用分段数", "分段标题 + 内容", "标注", "用户", "消耗 tokens", "耗时(秒)", "提问时间" };

  /** 导出文件名带日期，避免多次下载互相覆盖。 */
  public String fileName(Long applicationId) {
    Row application = Db.findById("moss_kb_application", applicationId);
    String name = application == null ? null : application.getStr("name");
    if (name == null || name.isBlank()) {
      name = "chat_log";
    }
    return name + "_" + LocalDate.now() + ".xlsx";
  }

  public byte[] export(Long user, Long applicationId, MossKbChatHistoryService.Query query) {
    MossKbChatHistoryService history = new MossKbChatHistoryService();
    List<Row> chats = history.chatsForExport(user, applicationId, query);
    List<List<Object>> rows = new ArrayList<>();
    for (Row chat : chats) {
      List<Row> records = Db.find("select id,problem_text,answer_text,vote_status,message_tokens,answer_tokens,run_time,details,"
          + "improve_paragraph_id_list,create_time from moss_kb_application_chat_record where chat_id=? order by id", chat.getLong("id"));
      for (Row record : records) {
        rows.add(toRow(chat.getLong("id"), chat.getStr("abstract"), askerName(chat), record));
      }
    }

    ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
    EasyExcelUtils.getExcelWriteBuilder(outputStream)
        .sheet("对话日志")
        .head(EasyExcelUtils.head(HEADERS))
        .doWrite(rows);
    return outputStream.toByteArray();
  }
}

  @Post("/{app}/chat/export")
  public HttpResponse export(Long app, HttpRequest request) {
    Long userId = TioRequestContext.getUserIdLong();
    MossKbChatHistoryService.Query query = query(request);
    JSONObject body = body(request);
    ...
    MossKbChatLogExportService exportService = Aop.get(MossKbChatLogExportService.class);
    byte[] data = exportService.export(userId, app, query);
    return Resps.excel(TioRequestContext.getResponse(), data, exportService.fileName(app));
  }

public class ChatLogCleanupService {

  public static final int DEFAULT_RETAIN_DAYS = 180;

  public record Application(Long id, Integer cleanTime) {
    public int retainDays() {
      return cleanTime == null || cleanTime <= 0 ? DEFAULT_RETAIN_DAYS : cleanTime;
    }
  }

  /** 返回删除的问答记录条数。 */
  public int cleanExpired() {
    int deleted = 0;
    for (Application application : applications()) {
      OffsetDateTime cutoff = cutoff(application);
      List<Long> chatIds = expiredChatIds(application.id(), cutoff);
      if (chatIds.isEmpty()) {
        continue;
      }
      deleted += deleteExpiredRecords(application.id(), cutoff);
      List<Long> emptyChats = chatsWithoutRecords(chatIds);
      if (!emptyChats.isEmpty()) {
        deleteChats(emptyChats);
      }
    }
    return deleted;
  }

  protected int deleteExpiredRecords(Long applicationId, OffsetDateTime cutoff) {
    return Db.update("delete from moss_kb_application_chat_record r using moss_kb_application_chat c"
        + " where r.chat_id=c.id and c.application_id=? and r.create_time < ?", applicationId, cutoff);
  }
}