437 lines
13 KiB
Go
437 lines
13 KiB
Go
package service
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"os"
|
||
"path/filepath"
|
||
"regexp"
|
||
"sort"
|
||
"strings"
|
||
"time"
|
||
|
||
"rag-local/common"
|
||
"rag-local/kb/consts"
|
||
"rag-local/kb/dao"
|
||
"rag-local/kb/model/domain"
|
||
"rag-local/kb/model/entity"
|
||
|
||
"github.com/cloudwego/eino/schema"
|
||
"github.com/gogf/gf/v2/errors/gerror"
|
||
"github.com/gogf/gf/v2/frame/g"
|
||
)
|
||
|
||
var AnnotationService = &annotationService{}
|
||
|
||
type annotationService struct{}
|
||
|
||
// Clause 合同条款（切分中间态）
|
||
type Clause struct {
|
||
Seq int
|
||
Title string
|
||
Content string
|
||
}
|
||
|
||
// 条款切分正则，按优先级探测（首个命中 >=2 条的采用）
|
||
var clausePatterns = []*regexp.Regexp{
|
||
regexp.MustCompile(`(?m)^\s*(第[一二三四五六七八九十百千\d]+条[、\s:：]?)`),
|
||
regexp.MustCompile(`(?m)^\s*(\d{1,2}(\.\d{1,2})*[、.．]\s*)`),
|
||
regexp.MustCompile(`(?m)^\s*([一二三四五六七八九十]+[、.．]\s*)`),
|
||
}
|
||
|
||
var lawItemRe = regexp.MustCompile(`第[一二三四五六七八九十百千\d]+条`)
|
||
|
||
// annoCandidate 多数据集融合后的候选法条
|
||
type annoCandidate struct {
|
||
ChunkId int64
|
||
DatasetId int64
|
||
LawTitle string // 法律名 = dataset 名
|
||
Content string // chunk 内容（截断）
|
||
RrfScore float64
|
||
}
|
||
|
||
// StartAnnotationPoller 启动标注任务轮询：单 goroutine 串行消费
|
||
func (s *annotationService) StartAnnotationPoller(ctx context.Context) {
|
||
go func() {
|
||
g.Log().Info(ctx, "annotation task poller started")
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case <-time.After(consts.ParsePollIntervalSeconds * time.Second):
|
||
s.processOne(ctx)
|
||
}
|
||
}
|
||
}()
|
||
}
|
||
|
||
// SplitClauses 按行首标记切分条款；无结构时整篇作为单条
|
||
func (s *annotationService) SplitClauses(text string) []Clause {
|
||
var chosen *regexp.Regexp
|
||
for _, p := range clausePatterns {
|
||
if len(p.FindAllStringIndex(text, -1)) >= 2 {
|
||
chosen = p
|
||
break
|
||
}
|
||
}
|
||
if chosen == nil {
|
||
t := strings.TrimSpace(text)
|
||
if t == "" {
|
||
return nil
|
||
}
|
||
return []Clause{{Seq: 1, Title: "全文", Content: t}}
|
||
}
|
||
idxs := chosen.FindAllStringIndex(text, -1)
|
||
var out []Clause
|
||
for i, m := range idxs {
|
||
end := len(text)
|
||
if i+1 < len(idxs) {
|
||
end = idxs[i+1][0]
|
||
}
|
||
seg := strings.TrimSpace(text[m[0]:end])
|
||
if seg == "" {
|
||
continue
|
||
}
|
||
out = append(out, Clause{
|
||
Seq: len(out) + 1,
|
||
Title: strings.TrimSpace(text[m[0]:m[1]]),
|
||
Content: seg,
|
||
})
|
||
}
|
||
return out
|
||
}
|
||
|
||
// processOne 处理一个待处理标注任务：解析 → 切分 → 逐条款召回+判定 → 落库
|
||
func (s *annotationService) processOne(ctx context.Context) {
|
||
task, err := dao.ContractTask.NextPending(ctx)
|
||
if err != nil {
|
||
g.Log().Errorf(ctx, "next annotation task failed: %v", err)
|
||
return
|
||
}
|
||
if task == nil {
|
||
return
|
||
}
|
||
if err := dao.ContractTask.UpdateStatus(ctx, task.Id, consts.TaskStatusRunning, ""); err != nil {
|
||
g.Log().Errorf(ctx, "mark annotation task running failed: %v", err)
|
||
return
|
||
}
|
||
|
||
dsIds := parseDatasetIds(task.DatasetIds)
|
||
if len(dsIds) == 0 {
|
||
s.fail(ctx, task, "未选择法律语料数据集")
|
||
return
|
||
}
|
||
|
||
text, err := common.ParseFile(filepath.Join("workspace", task.FilePath))
|
||
if err != nil {
|
||
s.fail(ctx, task, "解析合同文件失败: "+err.Error())
|
||
return
|
||
}
|
||
|
||
// 断点续跑：已有条款则跳过切分（保留已完成条款状态），仅首次切分落库
|
||
clauses, err := dao.ContractClause.ListByTask(ctx, task.Id)
|
||
if err != nil {
|
||
s.fail(ctx, task, "读取条款失败: "+err.Error())
|
||
return
|
||
}
|
||
if len(clauses) == 0 {
|
||
split := s.SplitClauses(text)
|
||
if len(split) == 0 {
|
||
s.fail(ctx, task, "合同文本为空")
|
||
return
|
||
}
|
||
es := make([]entity.ContractClause, 0, len(split))
|
||
for _, c := range split {
|
||
es = append(es, entity.ContractClause{
|
||
Seq: c.Seq,
|
||
Title: c.Title,
|
||
Content: truncateRunes(c.Content, consts.AnnoMaxClauseChars),
|
||
})
|
||
}
|
||
if err := dao.ContractClause.InsertAll(ctx, task.Id, es); err != nil {
|
||
s.fail(ctx, task, "写入条款失败: "+err.Error())
|
||
return
|
||
}
|
||
if err := dao.ContractTask.UpdateProgress(ctx, task.Id, len(es), 0); err != nil {
|
||
g.Log().Warningf(ctx, "update annotation progress failed: %v", err)
|
||
}
|
||
clauses, err = dao.ContractClause.ListByTask(ctx, task.Id)
|
||
if err != nil {
|
||
s.fail(ctx, task, "读取条款失败: "+err.Error())
|
||
return
|
||
}
|
||
}
|
||
|
||
chatCfgId, err := dao.ModelConfig.GetDefault(ctx, consts.ModelTypeChat)
|
||
if err != nil || chatCfgId <= 0 {
|
||
s.fail(ctx, task, "未配置对话模型")
|
||
return
|
||
}
|
||
chatModel, err := BuildChatModel(ctx, chatCfgId)
|
||
if err != nil {
|
||
s.fail(ctx, task, "构建对话模型失败: "+err.Error())
|
||
return
|
||
}
|
||
|
||
embedders := make(map[int64]*OpenAIEmbedder)
|
||
dsNames := make(map[int64]string)
|
||
for _, dsId := range dsIds {
|
||
ds, err := dao.Dataset.GetOne(ctx, dsId)
|
||
if err != nil || ds == nil {
|
||
continue
|
||
}
|
||
dsNames[dsId] = ds.Name
|
||
if cfgId, err := dao.Dataset.GetEmbeddingCfgId(ctx, dsId); err == nil && cfgId > 0 {
|
||
if emb, err := BuildEmbedder(ctx, cfgId); err == nil {
|
||
embedders[dsId] = emb
|
||
} else {
|
||
g.Log().Warningf(ctx, "dataset %d embedder build failed: %v", dsId, err)
|
||
}
|
||
}
|
||
}
|
||
|
||
failed := 0
|
||
for _, cl := range clauses {
|
||
if cl.Status == consts.TaskStatusDone {
|
||
continue
|
||
}
|
||
if err := dao.ContractClause.UpdateStatus(ctx, cl.Id, consts.TaskStatusRunning, ""); err != nil {
|
||
g.Log().Errorf(ctx, "mark clause running failed: %v", err)
|
||
continue
|
||
}
|
||
cands, err := s.recallCandidates(ctx, cl, dsIds, dsNames, embedders)
|
||
if err != nil {
|
||
failed++
|
||
_ = dao.ContractClause.UpdateStatus(ctx, cl.Id, consts.TaskStatusFailed, err.Error())
|
||
continue
|
||
}
|
||
if len(cands) == 0 {
|
||
// 无候选视为完成（无标注），避免卡住进度
|
||
_ = dao.ContractClause.UpdateStatus(ctx, cl.Id, consts.TaskStatusDone, "")
|
||
s.updateProgress(ctx, task.Id)
|
||
continue
|
||
}
|
||
marks, err := s.judgeClause(ctx, chatModel, cl, cands)
|
||
if err != nil {
|
||
failed++
|
||
_ = dao.ContractClause.UpdateStatus(ctx, cl.Id, consts.TaskStatusFailed, err.Error())
|
||
continue
|
||
}
|
||
// 幂等：重跑前清旧标注，避免断点续跑产生重复 mark
|
||
if err := dao.ContractMark.DeleteByClause(ctx, cl.Id); err != nil {
|
||
g.Log().Warningf(ctx, "clear old marks failed: %v", err)
|
||
}
|
||
for _, m := range marks {
|
||
m.ClauseId = cl.Id
|
||
}
|
||
if err := dao.ContractMark.InsertAll(ctx, marks); err != nil {
|
||
failed++
|
||
_ = dao.ContractClause.UpdateStatus(ctx, cl.Id, consts.TaskStatusFailed, err.Error())
|
||
continue
|
||
}
|
||
_ = dao.ContractClause.UpdateStatus(ctx, cl.Id, consts.TaskStatusDone, "")
|
||
s.updateProgress(ctx, task.Id)
|
||
}
|
||
|
||
msg := ""
|
||
if failed > 0 {
|
||
msg = fmt.Sprintf("%d 条条款标注失败", failed)
|
||
}
|
||
if err := dao.ContractTask.UpdateStatus(ctx, task.Id, consts.TaskStatusDone, msg); err != nil {
|
||
g.Log().Errorf(ctx, "mark annotation task done failed: %v", err)
|
||
}
|
||
}
|
||
|
||
// updateProgress 以库内实际完成数更新任务进度（断点续跑时跳过已 done 条款也能算对）
|
||
func (s *annotationService) updateProgress(ctx context.Context, taskId int64) {
|
||
doneList, err := dao.ContractClause.ListByTask(ctx, taskId)
|
||
if err != nil {
|
||
return
|
||
}
|
||
done := 0
|
||
for _, c := range doneList {
|
||
if c.Status == consts.TaskStatusDone {
|
||
done++
|
||
}
|
||
}
|
||
if err := dao.ContractTask.UpdateProgress(ctx, taskId, len(doneList), done); err != nil {
|
||
g.Log().Warningf(ctx, "update annotation progress failed: %v", err)
|
||
}
|
||
}
|
||
|
||
// recallCandidates 多数据集召回：每数据集向量+FTS 各取 AnnoRecallTopK，全局 RRF 融合截断
|
||
func (s *annotationService) recallCandidates(ctx context.Context, clause *entity.ContractClause, dsIds []int64,
|
||
dsNames map[int64]string, embedders map[int64]*OpenAIEmbedder) ([]annoCandidate, error) {
|
||
merged := make(map[int64]*annoCandidate)
|
||
clauseText := clause.Title + " " + clause.Content
|
||
ftsText := clause.Title + " " + clause.Content
|
||
if rs := []rune(ftsText); len(rs) > 200 {
|
||
ftsText = string(rs[:200])
|
||
}
|
||
ftsQuery := common.TokenizeQuery(ftsText)
|
||
for _, dsId := range dsIds {
|
||
if emb := embedders[dsId]; emb != nil {
|
||
vecs, err := emb.EmbedStrings(ctx, []string{clauseText})
|
||
if err != nil {
|
||
g.Log().Warningf(ctx, "clause embed failed (dataset %d): %v", dsId, err)
|
||
} else if len(vecs) > 0 {
|
||
hits, err := dao.Chunk.VecSearch(ctx, dsId, domain.VecJsonF64(vecs[0]), consts.AnnoRecallTopK)
|
||
if err != nil {
|
||
g.Log().Warningf(ctx, "vec search failed (dataset %d): %v", dsId, err)
|
||
} else {
|
||
for i, h := range hits {
|
||
s.mergeHit(merged, h.ChunkId, dsId, dsNames[dsId], float64(i))
|
||
}
|
||
}
|
||
}
|
||
}
|
||
hits, err := dao.Chunk.FtsSearch(ctx, dsId, ftsQuery, consts.AnnoRecallTopK)
|
||
if err != nil {
|
||
g.Log().Warningf(ctx, "fts search failed (dataset %d): %v", dsId, err)
|
||
continue
|
||
}
|
||
for i, h := range hits {
|
||
s.mergeHit(merged, h.ChunkId, dsId, dsNames[dsId], float64(i))
|
||
}
|
||
}
|
||
cands := make([]annoCandidate, 0, len(merged))
|
||
for _, c := range merged {
|
||
cands = append(cands, *c)
|
||
}
|
||
sort.Slice(cands, func(i, j int) bool { return cands[i].RrfScore > cands[j].RrfScore })
|
||
if len(cands) > consts.AnnoMaxCandidates {
|
||
cands = cands[:consts.AnnoMaxCandidates]
|
||
}
|
||
for i := range cands {
|
||
if chunk, err := dao.Chunk.GetOne(ctx, cands[i].ChunkId); err == nil && chunk != nil {
|
||
cands[i].Content = truncateRunes(chunk.Content, consts.RerankMaxChars)
|
||
}
|
||
}
|
||
return cands, nil
|
||
}
|
||
|
||
func (s *annotationService) mergeHit(merged map[int64]*annoCandidate, chunkId, dsId int64, lawTitle string, rank float64) {
|
||
c := merged[chunkId]
|
||
if c == nil {
|
||
c = &annoCandidate{ChunkId: chunkId, DatasetId: dsId, LawTitle: lawTitle}
|
||
merged[chunkId] = c
|
||
}
|
||
c.RrfScore += 1 / (float64(consts.RrfK) + rank + 1)
|
||
}
|
||
|
||
// judgeClause LLM 判定：一次非流式调用对全部候选打分并给出理由，全部保留按分降序
|
||
func (s *annotationService) judgeClause(ctx context.Context, model *OpenAIChatModel, clause *entity.ContractClause, cands []annoCandidate) ([]*entity.ContractMark, error) {
|
||
var sb strings.Builder
|
||
sb.WriteString("你是资深法律顾问，负责对合同条款进行法律条文标注。请逐条判定每个候选法律条文与合同条款的相关性。\n\n【合同条款】\n")
|
||
sb.WriteString(clause.Title + " " + clause.Content)
|
||
sb.WriteString("\n\n【候选法律条文】\n")
|
||
for i, c := range cands {
|
||
sb.WriteString(fmt.Sprintf("[%d]《%s》%s\n", i+1, c.LawTitle, c.Content))
|
||
}
|
||
sb.WriteString("\n请为每条候选输出 score（0-10 整数，10=直接适用，0=完全无关）与 reason（一句话说明该条文与合同条款的关联及适用性）。\n注意：候选内容可能合并了多条法条，law_item 必须是候选内容中与合同条款最相关的那一条法条的编号（如「第四十四条」），不要输出候选内容里没有的编号。\n只输出 JSON，不要其他内容：")
|
||
sb.WriteString(`{"marks":[{"cand_id":1,"law_item":"第四十四条","score":9,"reason":"..."}]}`)
|
||
|
||
msg, err := model.Generate(ctx, []*schema.Message{{Role: schema.User, Content: sb.String()}})
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
content := msg.Content
|
||
if i, j := strings.Index(content, "{"), strings.LastIndex(content, "}"); i >= 0 && j > i {
|
||
content = content[i : j+1]
|
||
}
|
||
var resp struct {
|
||
Marks []struct {
|
||
CandId int `json:"cand_id"`
|
||
LawItem string `json:"law_item"`
|
||
Score float64 `json:"score"`
|
||
Reason string `json:"reason"`
|
||
} `json:"marks"`
|
||
}
|
||
if err := json.Unmarshal([]byte(content), &resp); err != nil {
|
||
return nil, gerror.Wrap(err, "解析标注结果失败: "+msg.Content)
|
||
}
|
||
if len(resp.Marks) == 0 {
|
||
return nil, gerror.New("标注结果为空")
|
||
}
|
||
marks := make([]*entity.ContractMark, 0, len(resp.Marks))
|
||
for _, m := range resp.Marks {
|
||
if m.CandId < 1 || m.CandId > len(cands) {
|
||
continue
|
||
}
|
||
c := cands[m.CandId-1]
|
||
lawItem := strings.TrimSpace(m.LawItem)
|
||
if lawItem == "" {
|
||
lawItem = lawItemRe.FindString(c.Content)
|
||
}
|
||
marks = append(marks, &entity.ContractMark{
|
||
ChunkId: c.ChunkId,
|
||
DatasetId: c.DatasetId,
|
||
LawTitle: c.LawTitle,
|
||
LawItem: lawItem,
|
||
Content: truncateRunes(c.Content, 800),
|
||
Reason: m.Reason,
|
||
Score: m.Score,
|
||
})
|
||
}
|
||
sort.Slice(marks, func(i, j int) bool { return marks[i].Score > marks[j].Score })
|
||
return marks, nil
|
||
}
|
||
|
||
func (s *annotationService) fail(ctx context.Context, task *entity.ContractTask, msg string) {
|
||
_ = dao.ContractTask.UpdateStatus(ctx, task.Id, consts.TaskStatusFailed, msg)
|
||
g.Log().Errorf(ctx, "annotation task %d failed: %s", task.Id, msg)
|
||
}
|
||
|
||
// List 任务列表
|
||
func (s *annotationService) List(ctx context.Context, page, pageSize int) ([]*entity.ContractTask, int, error) {
|
||
return dao.ContractTask.List(ctx, page, pageSize)
|
||
}
|
||
|
||
// Delete 删除任务及关联数据与文件
|
||
func (s *annotationService) Delete(ctx context.Context, id int64) error {
|
||
task, err := dao.ContractTask.GetOne(ctx, id)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if task == nil {
|
||
return gerror.New("任务不存在")
|
||
}
|
||
if err := dao.ContractMark.DeleteByTask(ctx, id); err != nil {
|
||
return err
|
||
}
|
||
if err := dao.ContractClause.DeleteByTask(ctx, id); err != nil {
|
||
return err
|
||
}
|
||
if err := dao.ContractTask.Delete(ctx, id); err != nil {
|
||
return err
|
||
}
|
||
if task.FilePath != "" {
|
||
_ = os.Remove(filepath.Join("workspace", task.FilePath))
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func parseDatasetIds(s string) []int64 {
|
||
var out []int64
|
||
for _, part := range strings.Split(s, ",") {
|
||
part = strings.TrimSpace(part)
|
||
if part == "" {
|
||
continue
|
||
}
|
||
var id int64
|
||
if _, err := fmt.Sscanf(part, "%d", &id); err == nil && id > 0 {
|
||
out = append(out, id)
|
||
}
|
||
}
|
||
return out
|
||
}
|
||
|
||
func truncateRunes(s string, n int) string {
|
||
rs := []rune(s)
|
||
if len(rs) > n {
|
||
return string(rs[:n])
|
||
}
|
||
return s
|
||
}
|